Claude's Corner: Crosslayer Labs — The Princeton Team That Patched the Internet's Certificate Infrastructure

# Build Guide: Internet Impersonation Detection Platform (Crosslayer Labs Clone)

A step-by-step guide to building an outside-in internet infrastructure monitoring platform that detects BGP hijacks, certificate fraud, DNS tampering, and JavaScript supply chain attacks. Use Claude Code to scaffold each step.

---

## Step 1: Database Schema & Core Data Models

Design a PostgreSQL schema to track monitored domains and events.

```sql
-- Monitored assets per customer
CREATE TABLE monitored_domains (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID NOT NULL,
  fqdn TEXT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  UNIQUE(org_id, fqdn)
);

-- DNS snapshots
CREATE TABLE dns_snapshots (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  domain_id UUID REFERENCES monitored_domains(id),
  snapshot_at TIMESTAMPTZ DEFAULT NOW(),
  record_type TEXT NOT NULL,        -- A, CNAME, MX, NS, TXT
  record_value TEXT NOT NULL,
  ttl INTEGER,
  probe_location TEXT NOT NULL      -- which vantage point
);

-- BGP route observations
CREATE TABLE bgp_observations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  prefix CIDR NOT NULL,
  origin_asn INTEGER NOT NULL,
  as_path INTEGER[] NOT NULL,
  observed_at TIMESTAMPTZ DEFAULT NOW(),
  collector_id TEXT NOT NULL,       -- RIPE RIS, RouteViews, etc.
  is_anomalous BOOLEAN DEFAULT FALSE
);

-- Certificate Transparency log entries
CREATE TABLE ct_log_entries (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  domain TEXT NOT NULL,
  san_domains TEXT[],
  issuer_cn TEXT NOT NULL,
  not_before TIMESTAMPTZ NOT NULL,
  not_after TIMESTAMPTZ NOT NULL,
  serial_number TEXT UNIQUE NOT NULL,
  ct_log_id TEXT NOT NULL,
  logged_at TIMESTAMPTZ DEFAULT NOW(),
  is_authorized BOOLEAN           -- NULL = unknown, TRUE = customer confirmed
);

-- Cross-layer incidents
CREATE TABLE incidents (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID NOT NULL,
  domain_id UUID REFERENCES monitored_domains(id),
  severity TEXT CHECK (severity IN ('critical','high','medium','low')),
  attack_type TEXT NOT NULL,       -- bgp_hijack, dns_tamper, cert_fraud, js_injection
  description TEXT NOT NULL,
  evidence JSONB NOT NULL,         -- raw signals that triggered the incident
  remediation TEXT,
  status TEXT DEFAULT 'open',
  detected_at TIMESTAMPTZ DEFAULT NOW(),
  resolved_at TIMESTAMPTZ
);

-- JS supply chain snapshots
CREATE TABLE js_snapshots (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  domain_id UUID REFERENCES monitored_domains(id),
  page_url TEXT NOT NULL,
  script_src TEXT NOT NULL,
  sha256_hash TEXT NOT NULL,
  snapshot_at TIMESTAMPTZ DEFAULT NOW(),
  is_new BOOLEAN DEFAULT TRUE
);

CREATE INDEX ON ct_log_entries USING GIN (san_domains);
CREATE INDEX ON bgp_observations (prefix, observed_at DESC);
CREATE INDEX ON incidents (org_id, status, detected_at DESC);
```

**Claude Code prompt:** "Generate SQLAlchemy ORM models for this schema with async support and Alembic migrations."

---

## Step 2: BGP Monitoring Pipeline

Ingest BGP update feeds from public route collectors.

**Key sources:**
- RIPE RIS Live: `wss://ris-live.ripe.net/v1/ws/` — real-time BGP updates via WebSocket
- RouteViews: MRT dump files via HTTP, updated every 15 minutes
- RPKI ROA database: Cloudflare's `https://rpki.cloudflare.com/rpki.json`

```python
import asyncio
import websockets
import json
from datetime import datetime

async def ingest_ris_live(collector: str = "rrc00"):
    uri = f"wss://ris-live.ripe.net/v1/ws/?client=crosslayer-clone"
    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps({
            "type": "ris_subscribe",
            "data": {"type": "UPDATE", "host": f"{collector}.ripe.net"}
        }))
        async for message in ws:
            data = json.loads(message)
            if data.get("type") == "ris_message":
                await process_bgp_update(data["data"])

async def process_bgp_update(update: dict):
    # Extract prefix, origin ASN, AS path
    announcements = update.get("announcements", [])
    for ann in announcements:
        for prefix in ann.get("prefixes", []):
            origin_asn = ann["next_hop"]  # simplification; parse AS_PATH properly
            await check_prefix_anomaly(prefix, origin_asn, update["path"])

async def check_prefix_anomaly(prefix: str, origin_asn: int, as_path: list):
    # Compare against known-good baseline from RPKI
    # Flag if origin ASN doesn't match ROA or if prefix is more specific than expected
    pass
```

**Anomaly detection logic:**
- Origin ASN changed from known baseline → high severity
- More-specific prefix announced (prefix hijack) → critical
- AS path includes unexpected transit ASNs → medium

**Claude Code prompt:** "Build an async Python service that ingests RIPE RIS Live BGP updates, stores prefix-to-ASN mappings in Redis for fast lookup, and emits anomaly events to a Postgres queue table when the origin ASN changes."

---

## Step 3: Certificate Transparency Monitor

Stream all newly issued certificates from public CT logs.

```python
import httpx
import asyncio
from dataclasses import dataclass

CT_LOGS = [
    "https://ct.googleapis.com/logs/us1/argon2024/",
    "https://ct.cloudflare.com/logs/nimbus2024/",
    "https://oak.ct.letsencrypt.org/2024h1/",
]

@dataclass
class CTEntry:
    domain: str
    san_domains: list[str]
    issuer: str
    not_before: str
    not_after: str
    log_id: str

async def stream_ct_log(log_url: str, start_index: int = 0):
    async with httpx.AsyncClient(timeout=30) as client:
        while True:
            resp = await client.get(
                f"{log_url}ct/v1/get-entries",
                params={"start": start_index, "end": start_index + 255}
            )
            entries = resp.json().get("entries", [])
            if not entries:
                await asyncio.sleep(60)
                continue
            for i, entry in enumerate(entries):
                cert = parse_ct_entry(entry)
                if cert:
                    await match_against_monitored_domains(cert)
            start_index += len(entries)

async def match_against_monitored_domains(cert: CTEntry):
    # Check if any SAN matches a monitored domain
    # If match found, check if issuing CA is on the org's allowlist
    # If not authorized, create incident
    pass
```

**Claude Code prompt:** "Build a Python worker that continuously polls Certificate Transparency log APIs, parses X.509 certificate fields using the `cryptography` library, and matches new certs against a Postgres table of monitored domains. Flag any cert whose issuing CA org isn't in the domain owner's approved CA list."

---

## Step 4: Multi-Vantage-Point DNS Probing

DNS results can vary by location — that's the whole point of detecting hijacks.

```typescript
// Deploy as edge functions on Cloudflare Workers (free tier: 100k req/day)
// Each worker region = one vantage point

interface DnsProbeResult {
  domain: string;
  recordType: string;
  values: string[];
  resolvedFrom: string; // Cloudflare colo ID
  resolvedAt: string;
}

export default {
  async fetch(request: Request): Promise<Response> {
    const { domain, recordType } = await request.json() as any;
    
    // Cloudflare Workers have access to DNS-over-HTTPS natively
    const resp = await fetch(
      `https://cloudflare-dns.com/dns-query?name=${domain}&type=${recordType}`,
      { headers: { Accept: "application/dns-json" } }
    );
    const data = await resp.json() as any;
    
    const result: DnsProbeResult = {
      domain,
      recordType,
      values: (data.Answer || []).map((a: any) => a.data),
      resolvedFrom: request.cf?.colo as string,
      resolvedAt: new Date().toISOString(),
    };
    
    return Response.json(result);
  }
};
```

Deploy this worker to 10+ Cloudflare regions. Your orchestration layer sends probe requests to each region simultaneously and diffs the results. Inconsistency across regions = potential BGP/DNS hijack in progress.

**Claude Code prompt:** "Build a Node.js orchestration service that dispatches DNS probe requests to Cloudflare Worker endpoints in 15 regions simultaneously, diffs the responses, and writes a `dns_divergence` event to Postgres when two or more regions return different A records for the same domain."

---

## Step 5: JavaScript Supply Chain Monitor

Headless browser crawling to detect unexpected script changes.

```python
from playwright.async_api import async_playwright
import hashlib
import httpx

async def snapshot_js_for_domain(domain: str, page_url: str) -> list[dict]:
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        
        scripts_found = []
        
        # Intercept all script requests
        async def handle_route(route):
            request = route.request
            if request.resource_type == "script":
                response = await route.fetch()
                body = await response.body()
                sha256 = hashlib.sha256(body).hexdigest()
                scripts_found.append({
                    "src": request.url,
                    "sha256": sha256,
                    "size_bytes": len(body),
                })
            await route.continue_()
        
        await page.route("**/*", handle_route)
        await page.goto(page_url, wait_until="networkidle")
        await browser.close()
        
    return scripts_found

async def diff_against_baseline(domain_id: str, current_scripts: list[dict]):
    # Load last known snapshot from DB
    # Compare SHA256 hashes
    # New scripts or changed hashes → create incident
    pass
```

**Claude Code prompt:** "Build a Python microservice using Playwright that crawls monitored pages on a 1-hour schedule, extracts all loaded JavaScript URLs and their SHA256 hashes, diffs against the stored baseline, and creates a `js_change` incident in Postgres when a script hash changes or a new third-party origin appears."

---

## Step 6: Correlation Engine & Alert Generation

This is where the real value is created — correlating signals across layers.

```python
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import asyncpg

@dataclass
class CorrelationWindow:
    domain: str
    window_start: datetime
    window_end: datetime
    bgp_anomalies: list
    dns_changes: list
    new_certs: list
    js_changes: list

ATTACK_PATTERNS = [
    {
        "name": "bgp_certificate_attack",
        "description": "BGP route change followed by new certificate issuance — classic MPIC-circumvention attempt",
        "severity": "critical",
        "requires": ["bgp_anomaly", "new_cert"],
        "time_window_minutes": 60,
    },
    {
        "name": "dns_hijack_with_cert",
        "description": "DNS record changed AND new certificate issued to same domain",
        "severity": "critical",
        "requires": ["dns_change", "new_cert"],
        "time_window_minutes": 30,
    },
    {
        "name": "supply_chain_injection",
        "description": "New or modified JavaScript on monitored page",
        "severity": "high",
        "requires": ["js_change"],
        "time_window_minutes": 1440,
    },
]

async def run_correlation(domain_id: str, pool: asyncpg.Pool):
    window = await gather_signals(domain_id, pool, minutes=60)
    
    for pattern in ATTACK_PATTERNS:
        if matches_pattern(window, pattern):
            await create_incident(
                domain_id=domain_id,
                attack_type=pattern["name"],
                severity=pattern["severity"],
                evidence=build_evidence(window, pattern),
                pool=pool
            )
```

**Claude Code prompt:** "Implement a Python correlation engine that runs every 5 minutes, queries the `bgp_observations`, `dns_snapshots`, `ct_log_entries`, and `js_snapshots` tables for each monitored domain, applies the pattern definitions, and creates enriched incidents with structured evidence JSONB including links to the raw signals and an auto-generated remediation recommendation using Claude API."

---

## Step 7: API, Dashboard & Alerting

```typescript
// Hono.js REST API (fast, edge-deployable)
import { Hono } from 'hono'
import { zValidator } from '@hono/zod-validator'
import { z } from 'zod'

const app = new Hono()

// List incidents for org
app.get('/v1/incidents', async (c) => {
  const orgId = c.get('orgId') // from JWT middleware
  const { status, severity, limit = 20 } = c.req.query()
  
  const incidents = await db.query(
    `SELECT i.*, m.fqdn 
     FROM incidents i 
     JOIN monitored_domains m ON i.domain_id = m.id
     WHERE i.org_id = $1 
       AND ($2::text IS NULL OR i.status = $2)
       AND ($3::text IS NULL OR i.severity = $3)
     ORDER BY i.detected_at DESC 
     LIMIT $4`,
    [orgId, status || null, severity || null, limit]
  )
  return c.json({ incidents: incidents.rows })
})

// Add domain to monitoring
app.post('/v1/domains', zValidator('json', z.object({
  fqdn: z.string().regex(/^([a-z0-9-]+\.)+[a-z]{2,}$/)
})), async (c) => {
  const { fqdn } = c.req.valid('json')
  // Validate ownership via DNS TXT record challenge
  const challenge = await createOwnershipChallenge(fqdn)
  return c.json({ challenge_token: challenge.token, dns_record: `_crosslayer-verify.${fqdn}` })
})
```

**Deployment stack:**
- API: Hono on Cloudflare Workers
- Background workers: Python on Railway or Fly.io (needs persistent processes for BGP streaming)
- Database: Supabase (Postgres + real-time subscriptions for live dashboard)
- Alert delivery: Resend for email, Slack webhooks, PagerDuty API

**Dashboard key pages:**
1. Attack surface map (all domains, subdomains, IPs, ASNs)
2. Live incident feed with evidence timeline
3. BGP route visualization (prefix → ASN path graph)
4. Certificate inventory with issuance history

**Claude Code prompt:** "Build a Next.js dashboard with a live incident feed using Supabase real-time subscriptions. Each incident card should show the attack type, affected domain, severity badge, and an expandable evidence panel with a timeline of the correlated signals. Add a one-click 'Acknowledge' and 'Resolve' action."

---

## Estimated Build Time & Cost

| Phase | Time | Monthly Infra Cost |
|-------|------|--------------------|
| Schema + local dev env | 2 days | $0 |
| BGP + CT ingestion workers | 1 week | ~$50 (Railway) |
| DNS probing (Cloudflare Workers) | 2 days | ~$5 |
| JS crawler (Playwright) | 3 days | ~$30 (Fly.io) |
| Correlation engine | 1 week | included above |
| API + dashboard | 1 week | ~$25 (Supabase Pro) |
| **Total MVP** | **~5 weeks** | **~$110/month** |

The infrastructure cost is deceptively low for a prototype. What's not reflected: the 6-12 months needed to tune false-positive rates to enterprise-acceptable levels, and the years needed to build the distributed vantage point network that makes BGP anomaly detection meaningful.