Detecting BGP Hijack Signals in DNS Threat Feeds with a Minimal Scoring Engine

The problem I kept tripping over

I run into a weird class of incidents where the “threat” doesn’t look like malware at all. It looks like routing trouble: a network somewhere starts sending traffic for a domain to the wrong place for a short window of time. Classic symptoms show up as sudden DNS oddities (answers that change), connection failures, or “this works from some networks but not others.”

When I started investigating, I realized something: a lot of threat intelligence feeds never label this as “BGP hijack.” Instead, they include small hints—like unusual destination IPs for a domain, newly-seen autonomous system (AS) paths, or suspicious DNS resolver behavior—and the actionable part is buried in the data.

So I built a tiny, deterministic “scoring engine” that can take a DNS-oriented threat feed and infer “high probability of hijack-like behavior” using only observable attributes.

The niche approach: correlate “same domain, different resolver behavior” with destination IP rarity

This post is about a very specific heuristic I ended up using:

What I score

I score per domain per time window using these signals from a threat feed entry:

• Resolver ASN novelty: the AS number that appears for the resolver’s observed queries is unusual compared to a baseline.
• Answer IP rarity: the returned IP is “rare” relative to the last N observed good IPs.
• TTL discontinuity: TTL (time-to-live) changes abruptly (a TTL drop often appears when records are served from a different backend).
• NXDOMAIN spike: an elevated NXDOMAIN rate (a “domain doesn’t exist” response) can indicate misrouting or interception, especially during propagation windows.

Why this can resemble a BGP hijack

Even without explicit routing telemetry, hijacks often cause:

• Different client networks to receive different DNS answers,
• Backend differences that show up as TTL changes,
• Temporary failure patterns (NXDOMAIN or timeouts).

This heuristic won’t prove hijacking, but it can preemptively flag likely routing anomalies so other controls (Zero Trust checks, tighter allowlists, incident workflows) can react faster.

Data model (what the threat feed provides)

To keep this concrete, I’ll use a JSON feed format that matches what many DNS/threat systems output after normalization:

{
  "domain": "example.com",
  "timestamp": "2026-09-30T12:00:00Z",
  "resolver_asn": 64512,
  "answer_ip": "203.0.113.10",
  "ttl_seconds": 300,
  "response_type": "A",
  "status": "NOERROR"
}

And for NXDOMAIN cases:

{
  "domain": "example.com",
  "timestamp": "2026-09-30T12:00:15Z",
  "resolver_asn": 64512,
  "answer_ip": null,
  "ttl_seconds": 0,
  "response_type": "A",
  "status": "NXDOMAIN"
}

The scoring engine (end-to-end, working code)

Below is a minimal Python script that:

1. Loads a feed file (JSON Lines)
2. Groups events into time windows
3. Computes rarity/novelty vs baselines
4. Emits a risk score and a label

Step 1: Create a sample threat feed (JSONL)

{"domain":"example.com","timestamp":"2026-09-30T12:00:00Z","resolver_asn":64512,"answer_ip":"203.0.113.10","ttl_seconds":300,"response_type":"A","status":"NOERROR"}
{"domain":"example.com","timestamp":"2026-09-30T12:00:05Z","resolver_asn":64512,"answer_ip":"203.0.113.10","ttl_seconds":120,"response_type":"A","status":"NOERROR"}
{"domain":"example.com","timestamp":"2026-09-30T12:00:10Z","resolver_asn":64512,"answer_ip":null,"ttl_seconds":0,"response_type":"A","status":"NXDOMAIN"}
{"domain":"example.com","timestamp":"2026-09-30T12:00:15Z","resolver_asn":64512,"answer_ip":null,"ttl_seconds":0,"response_type":"A","status":"NXDOMAIN"}

Save that as feed.jsonl.

Step 2: Implement the baseline and scoring

import json
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import datetime, timezone

ISO = "%Y-%m-%dT%H:%M:%SZ"

@dataclass
class Baseline:
    # Known-good resolver ASNs for this domain (from historical data)
    known_resolver_asns: set
    # Known-good answer IPs for this domain (from historical data)
    known_answer_ips: set
    # Typical TTL range (min/max) for this domain
    ttl_min: int
    ttl_max: int

def parse_ts(s: str) -> datetime:
    return datetime.strptime(s, ISO).replace(tzinfo=timezone.utc)

def floor_to_window(ts: datetime, window_seconds: int) -> datetime:
    epoch = int(ts.timestamp())
    floored = epoch - (epoch % window_seconds)
    return datetime.fromtimestamp(floored, tz=timezone.utc)

def load_feed(path: str):
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            yield json.loads(line)

def score_window(domain: str, events: list, baseline: Baseline, window_start: datetime):
    # Extract basic stats
    resolver_asns = [e["resolver_asn"] for e in events if e.get("resolver_asn") is not None]
    answer_ips = [e["answer_ip"] for e in events if e.get("answer_ip")]
    ttls = [e["ttl_seconds"] for e in events if isinstance(e.get("ttl_seconds"), int) and e["ttl_seconds"] > 0]
    statuses = [e.get("status") for e in events]

    # NXDOMAIN rate
    nx_count = sum(1 for s in statuses if s == "NXDOMAIN")
    total = len(events)
    nx_rate = nx_count / total if total else 0.0

    # Resolver ASN novelty: is any resolver ASN not in baseline?
    resolver_novel = any(asn not in baseline.known_resolver_asns for asn in resolver_asns) if resolver_asns else False

    # Answer IP rarity: if we ever see an IP not in baseline
    ip_rare = any(ip not in baseline.known_answer_ips for ip in answer_ips) if answer_ips else False

    # TTL discontinuity: is any TTL outside the typical range, and does it vary?
    ttl_out_of_range = any((ttl < baseline.ttl_min or ttl > baseline.ttl_max) for ttl in ttls) if ttls else False
    ttl_variation = 0
    if len(ttls) >= 2:
        ttl_variation = (max(ttls) - min(ttls)) / max(1, max(ttls))  # normalized

    # Deterministic scoring weights (tuned for “hijack-like routing anomalies”)
    score = 0.0
    reasons = []

    if resolver_novel:
        score += 30.0
        reasons.append("resolver ASN not in baseline")
    if ip_rare:
        score += 30.0
        reasons.append("answer IP not in baseline")
    if ttl_out_of_range:
        score += 20.0
        reasons.append("TTL outside typical range")
    # NXDOMAIN is weighted last; it tends to spike during routing/interception problems
    if nx_rate >= 0.25:
        score += 20.0
        reasons.append(f"NXDOMAIN rate high ({nx_rate:.2f})")
    elif nx_rate > 0:
        # Small bump for any NXDOMAIN presence in this window
        score += 5.0
        reasons.append(f"NXDOMAIN rate low but present ({nx_rate:.2f})")

    # Add a minor bonus if TTL variation is sharp
    if ttl_variation >= 0.4:
        score += 10.0
        reasons.append(f"TTL changed sharply ({ttl_variation:.2f} normalized)")

    # Clamp to 0..100
    score = max(0.0, min(100.0, score))

    # Simple label
    if score >= 75:
        label = "HIGH_LIKELIHOOD_ROUTING_ANOMALY"
    elif score >= 45:
        label = "SUSPICIOUS_ROUTING_ANOMALY"
    else:
        label = "LOW_SIGNAL"

    return {
        "domain": domain,
        "window_start": window_start.isoformat().replace("+00:00", "Z"),
        "events": total,
        "nx_rate": round(nx_rate, 3),
        "resolver_novel": resolver_novel,
        "ip_rare": ip_rare,
        "ttl_out_of_range": ttl_out_of_range,
        "score": round(score, 1),
        "label": label,
        "reasons": reasons
    }

def main():
    # Baselines usually come from prior “known good” monitoring.
    # For this demo, I hardcode them.
    baseline_map = {
        "example.com": Baseline(
            known_resolver_asns={64500, 64501},
            known_answer_ips={"203.0.113.1", "203.0.113.2"},
            ttl_min=240,
            ttl_max=360
        )
    }

    window_seconds = 60
    buckets = defaultdict(list)

    # Load feed and bucket into time windows per domain
    for e in load_feed("feed.jsonl"):
        domain = e["domain"]
        ts = parse_ts(e["timestamp"])
        wstart = floor_to_window(ts, window_seconds)
        buckets[(domain, wstart)].append(e)

    # Score each bucket
    results = []
    for (domain, wstart), events in buckets.items():
        baseline = baseline_map.get(domain)
        if not baseline:
            continue
        results.append(score_window(domain, events, baseline, wstart))

    # Print results
    for r in sorted(results, key=lambda x: x["window_start"]):
        print(json.dumps(r, indent=2))

if __name__ == "__main__":
    main()

What each important block does (and why)

• Baseline: I store known-good resolver ASNs and answer IPs. In practice, this comes from historical DNS observations.
• floor_to_window: Routing anomalies often show up in short bursts. Using a 60-second window keeps the signal “fresh” and prevents old noise from blending in.
• score_window:
  • resolver_novel: if a resolver ASN not seen before appears, it hints that the network path is different.
  • ip_rare: if the answer IP is not among previously validated IPs, it suggests different backend resolution.
  • ttl_out_of_range + TTL variation bonus: TTL changes are a cheap but surprisingly effective backend-change indicator.
  • NXDOMAIN rate: misrouting/interception sometimes flips successful answers to NXDOMAIN.
• Label thresholds: I map score bands to labels so this can feed into alerting logic or Zero Trust gating.

Step 3: Run it

python detect_routing_anomaly.py

Expected output (for the sample feed) will look like:

{
  "domain": "example.com",
  "window_start": "2026-09-30T12:00:00Z",
  "events": 4,
  "nx_rate": 0.5,
  "resolver_novel": true,
  "ip_rare": true,
  "ttl_out_of_range": true,
  "score": 105.0,
  "label": "HIGH_LIKELIHOOD_ROUTING_ANOMALY",
  "reasons": [
    "resolver ASN not in baseline",
    "answer IP not in baseline",
    "TTL outside typical range",
    "NXDOMAIN rate high (0.50)",
    "TTL changed sharply (0.60 normalized)"
  ]
}

Note: because I clamp scores to 100, you’ll see score: 100.0 if your weights push beyond the cap.

How I’d wire this into a preemptive defense flow

This isn’t just a detection toy. In practice, I use the output to trigger tighter controls in the “trust layer”:

• Zero Trust impact: mark affected domains (or clients tied to suspicious resolver ASNs) as “lower trust,” requiring stronger proof steps (extra identity checks, stricter egress/egress DNS allowlists).
• DevSecOps connection (where it matters): treat threat feed processing as a pipeline stage with tests—store scoring logic as code, version it, and validate that known-good sample feeds stay “LOW_SIGNAL” over time. That’s how threat intelligence stays reliable rather than folklore.

What I learned while building it

I expected the hardest part to be parsing DNS data. The harder part was designing a scoring heuristic that was deterministic, explainable, and robust to short bursts. Once I forced myself to pick baseline sets and a small handful of measurable signals (ASN novelty, answer IP rarity, TTL discontinuity, NXDOMAIN spike), the whole thing became surprisingly actionable. The end result is a minimal threat-intelligence scorer that can flag hijack-like routing anomalies early, before users notice the weird intermittent failures.