mirror of
https://github.com/saymrwulf/CertTransparencySearch.git
synced 2026-05-14 20:37:52 +00:00
502 lines
30 KiB
Markdown
502 lines
30 KiB
Markdown
|
# ct_dns_utils.py
|
||
|
|
|
||
|
|
Source file: [`ct_dns_utils.py`](../ct_dns_utils.py)
|
||
|
|
|
||
|
|
Public DNS scanner. This file runs dig, follows alias chains, finds public addresses, and collapses raw DNS evidence into readable delivery labels.
|
||
|
|
|
||
|
|
Main flow in one line: `DNS name -> dig answers -> normalized observation -> provider hints -> delivery label`
|
||
|
|
|
||
|
|
How to read this page:
|
||
|
|
|
||
|
|
- left side: the actual source code block
|
||
|
|
- right side: a plain-English explanation for a beginner
|
||
|
|
- read from top to bottom because later blocks depend on earlier ones
|
||
|
|
|
||
|
|
## Module setup
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">#!/usr/bin/env python3
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import ipaddress
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
import subprocess
|
||
|
|
import time
|
||
|
|
from dataclasses import asdict, dataclass
|
||
|
|
from datetime import UTC, datetime
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import ct_scan</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Shared DNS scanning helpers, cache helpers, and the logic that turns raw DNS answers into platform clues.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Nothing yet; this is the starting point. → <strong>Module setup</strong> → The later DNS helpers all reuse these imports and small shared helpers.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## DnsObservation
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">@dataclass
|
||
|
|
class DnsObservation:
|
||
|
|
original_name: str
|
||
|
|
original_status: str
|
||
|
|
cname_chain: list[str]
|
||
|
|
terminal_name: str
|
||
|
|
terminal_status: str
|
||
|
|
a_records: list[str]
|
||
|
|
aaaa_records: list[str]
|
||
|
|
ptr_records: list[str]
|
||
|
|
classification: str
|
||
|
|
stack_signature: str
|
||
|
|
provider_hints: list[str]</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>One complete DNS observation for one hostname.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>DnsObservation</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## normalize_name
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def normalize_name(name: str) -> str:
|
||
|
|
return name.rstrip(".").lower()</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This block makes values consistent so matching and grouping do not get confused by superficial differences.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>normalize_name</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## cache_key
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def cache_key(value: str) -> str:
|
||
|
|
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
|
||
|
|
slug = re.sub(r"[^a-z0-9.-]+", "-", value.lower()).strip("-")
|
||
|
|
slug = slug[:80] or "item"
|
||
|
|
return f"v1-{slug}-{digest}.json"</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>cache_key</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## load_json_cache
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def load_json_cache(cache_dir: Path, key: str, ttl_seconds: int) -> dict[str, Any] | None:
|
||
|
|
path = cache_dir / key
|
||
|
|
if not path.exists():
|
||
|
|
return None
|
||
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
|
|
cached_at = datetime.fromisoformat(payload["cached_at"].replace("Z", "+00:00"))
|
||
|
|
age = time.time() - cached_at.astimezone(UTC).timestamp()
|
||
|
|
if age > ttl_seconds:
|
||
|
|
return None
|
||
|
|
return payload</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This block loads data from disk, cache, or an earlier stage so later code can work with it.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>load_json_cache</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## store_json_cache
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def store_json_cache(cache_dir: Path, key: str, payload: dict[str, Any]) -> None:
|
||
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
enriched = dict(payload)
|
||
|
|
enriched["cached_at"] = ct_scan.utc_iso(datetime.now(UTC))
|
||
|
|
(cache_dir / key).write_text(json.dumps(enriched, indent=2, sort_keys=True), encoding="utf-8")</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This block saves an intermediate result so the next run can reuse it instead of recomputing everything.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>store_json_cache</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## run_dig
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def run_dig(name: str, rrtype: str, short: bool) -> str:
|
||
|
|
cmd = ["dig", "+time=2", "+tries=1"]
|
||
|
|
if short:
|
||
|
|
cmd.append("+short")
|
||
|
|
else:
|
||
|
|
cmd.extend(["+noall", "+comments", "+answer"])
|
||
|
|
cmd.extend([name, rrtype])
|
||
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
|
||
|
|
return result.stdout</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>A hostname and record type. → <strong>run_dig</strong> → `scan_name_live`, `dig_status`, `dig_short`, and `ptr_lookup` all rely on this.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## dig_status
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def dig_status(name: str, rrtype: str = "A") -> str:
|
||
|
|
output = run_dig(name, rrtype, short=False)
|
||
|
|
match = re.search(r"status:\s*([A-Z]+)", output)
|
||
|
|
if match:
|
||
|
|
return match.group(1)
|
||
|
|
if output.strip():
|
||
|
|
return "NOERROR"
|
||
|
|
return "UNKNOWN"</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>dig_status</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## dig_short
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def dig_short(name: str, rrtype: str) -> list[str]:
|
||
|
|
output = run_dig(name, rrtype, short=True)
|
||
|
|
return [normalize_name(line) for line in output.splitlines() if line.strip()]</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>dig_short</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## parse_answer_section
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def parse_answer_section(output: str) -> list[tuple[str, str]]:
|
||
|
|
in_answer = False
|
||
|
|
parsed: list[tuple[str, str]] = []
|
||
|
|
for raw_line in output.splitlines():
|
||
|
|
line = raw_line.strip()
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
if line.startswith(";; ANSWER SECTION:"):
|
||
|
|
in_answer = True
|
||
|
|
continue
|
||
|
|
if not in_answer or line.startswith(";;"):
|
||
|
|
continue
|
||
|
|
match = re.match(r"^\S+\s+\d+\s+IN\s+(\S+)\s+(.+)$", line)
|
||
|
|
if not match:
|
||
|
|
continue
|
||
|
|
rrtype, rdata = match.groups()
|
||
|
|
parsed.append((rrtype.upper(), normalize_name(rdata)))
|
||
|
|
return parsed</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>parse_answer_section</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## is_ip_address
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def is_ip_address(value: str) -> bool:
|
||
|
|
try:
|
||
|
|
ipaddress.ip_address(value)
|
||
|
|
return True
|
||
|
|
except ValueError:
|
||
|
|
return False</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>is_ip_address</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## classify_observation
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def classify_observation(chain: list[str], terminal_status: str, a_records: list[str], aaaa_records: list[str]) -> str:
|
||
|
|
has_addresses = bool(a_records or aaaa_records)
|
||
|
|
if chain and has_addresses:
|
||
|
|
return "cname_to_address"
|
||
|
|
if chain and not has_addresses:
|
||
|
|
return "dangling_cname"
|
||
|
|
if has_addresses:
|
||
|
|
return "direct_address"
|
||
|
|
if terminal_status == "NXDOMAIN":
|
||
|
|
return "nxdomain"
|
||
|
|
if terminal_status == "NOERROR":
|
||
|
|
return "no_data"
|
||
|
|
return "other"</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This block applies rules and chooses a category label.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>classify_observation</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## infer_provider_hints
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def infer_provider_hints(observation: DnsObservation) -> list[str]:
|
||
|
|
text = " ".join(
|
||
|
|
[
|
||
|
|
observation.original_name,
|
||
|
|
*observation.cname_chain,
|
||
|
|
observation.terminal_name,
|
||
|
|
*observation.ptr_records,
|
||
|
|
]
|
||
|
|
).lower()
|
||
|
|
hints: list[str] = []
|
||
|
|
if "campaign.adobe.com" in text:
|
||
|
|
hints.append("Adobe Campaign")
|
||
|
|
if "cloudfront.net" in text:
|
||
|
|
hints.append("AWS CloudFront")
|
||
|
|
if "elb.amazonaws.com" in text or "compute.amazonaws.com" in text:
|
||
|
|
hints.append("AWS")
|
||
|
|
if "apigee.net" in text or "googleusercontent.com" in text:
|
||
|
|
hints.append("Google Apigee")
|
||
|
|
if "pegacloud.net" in text or ".pega.net" in text:
|
||
|
|
hints.append("Pega Cloud")
|
||
|
|
if "useinfinite.io" in text:
|
||
|
|
hints.append("Infinite / agency alias")
|
||
|
|
if any(ip.startswith("13.107.") for ip in observation.a_records) or any(ip.startswith("2620:1ec:") for ip in observation.aaaa_records):
|
||
|
|
hints.append("Microsoft Edge")
|
||
|
|
if not hints:
|
||
|
|
hints.append("Unclassified")
|
||
|
|
return hints</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Reads the raw DNS trail and pulls out likely platform or vendor clues.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>One normalized DNS observation. → <strong>infer_provider_hints</strong> → `infer_stack_signature` and the report layers use the hints it produces.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## infer_stack_signature
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def infer_stack_signature(observation: DnsObservation) -> str:
|
||
|
|
hints = infer_provider_hints(observation)
|
||
|
|
if observation.classification == "nxdomain":
|
||
|
|
return "No public DNS (NXDOMAIN)"
|
||
|
|
if observation.classification == "no_data":
|
||
|
|
return "No public address data"
|
||
|
|
if "Adobe Campaign" in hints and "AWS CloudFront" in hints:
|
||
|
|
return "Adobe Campaign -> AWS CloudFront"
|
||
|
|
if "Adobe Campaign" in hints and "AWS" in hints:
|
||
|
|
return "Adobe Campaign -> AWS ALB"
|
||
|
|
if "Adobe Campaign" in hints and observation.a_records:
|
||
|
|
return "Adobe Campaign direct IP"
|
||
|
|
if "AWS CloudFront" in hints:
|
||
|
|
return "AWS CloudFront"
|
||
|
|
if "Google Apigee" in hints:
|
||
|
|
return "Google Apigee"
|
||
|
|
if "Pega Cloud" in hints and "AWS" in hints:
|
||
|
|
return "Pega Cloud -> AWS ALB"
|
||
|
|
if "Infinite / agency alias" in hints and observation.classification == "dangling_cname":
|
||
|
|
return "Dangling agency alias"
|
||
|
|
if "Microsoft Edge" in hints:
|
||
|
|
return "Direct Microsoft edge"
|
||
|
|
if "AWS" in hints:
|
||
|
|
return "Direct AWS"
|
||
|
|
if observation.classification == "direct_address":
|
||
|
|
return "Direct address (provider unclear)"
|
||
|
|
if observation.classification == "cname_to_address":
|
||
|
|
return "CNAME to address (provider unclear)"
|
||
|
|
return hints[0]</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Collapses several low-level DNS clues into one human-readable delivery label.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>One DNS observation plus provider clues. → <strong>infer_stack_signature</strong> → `ct_master_report` uses the resulting label in naming and DNS chapters.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## scan_name_live
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def scan_name_live(name: str) -> DnsObservation:
|
||
|
|
name = normalize_name(name)
|
||
|
|
a_output = run_dig(name, "A", short=False)
|
||
|
|
aaaa_output = run_dig(name, "AAAA", short=False)
|
||
|
|
original_status = dig_status(name, "A")
|
||
|
|
a_answers = parse_answer_section(a_output)
|
||
|
|
aaaa_answers = parse_answer_section(aaaa_output)
|
||
|
|
chain: list[str] = []
|
||
|
|
for rrtype, rdata in a_answers + aaaa_answers:
|
||
|
|
if rrtype == "CNAME" and rdata not in chain:
|
||
|
|
chain.append(rdata)
|
||
|
|
a_records = sorted({rdata for rrtype, rdata in a_answers if rrtype == "A" and is_ip_address(rdata)})
|
||
|
|
aaaa_records = sorted({rdata for rrtype, rdata in aaaa_answers if rrtype == "AAAA" and is_ip_address(rdata)})
|
||
|
|
terminal_name = chain[-1] if chain else name
|
||
|
|
terminal_status = original_status
|
||
|
|
observation = DnsObservation(
|
||
|
|
original_name=name,
|
||
|
|
original_status=original_status,
|
||
|
|
cname_chain=chain,
|
||
|
|
terminal_name=terminal_name,
|
||
|
|
terminal_status=terminal_status,
|
||
|
|
a_records=a_records,
|
||
|
|
aaaa_records=aaaa_records,
|
||
|
|
ptr_records=[],
|
||
|
|
classification=classify_observation(chain, terminal_status, a_records, aaaa_records),
|
||
|
|
stack_signature="",
|
||
|
|
provider_hints=[],
|
||
|
|
)
|
||
|
|
observation.provider_hints = infer_provider_hints(observation)
|
||
|
|
observation.stack_signature = infer_stack_signature(observation)
|
||
|
|
return observation</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Runs the live DNS walk for one hostname.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>One DNS name from a SAN entry. → <strong>scan_name_live</strong> → `scan_name_cached` returns this result shape to higher-level analytics.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## scan_name_cached
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def scan_name_cached(name: str, cache_dir: Path, ttl_seconds: int) -> DnsObservation:
|
||
|
|
key = cache_key(name)
|
||
|
|
cached = load_json_cache(cache_dir, key, ttl_seconds)
|
||
|
|
if cached is not None:
|
||
|
|
payload = dict(cached)
|
||
|
|
payload.pop("cached_at", None)
|
||
|
|
return DnsObservation(**payload)
|
||
|
|
observation = scan_name_live(name)
|
||
|
|
store_json_cache(cache_dir, key, asdict(observation))
|
||
|
|
return observation</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Reuses a recent DNS result if possible, otherwise performs the live scan.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>A DNS name plus cache settings. → <strong>scan_name_cached</strong> → `ct_master_report.enrich_dns` uses this for every SAN name in the current corpus.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## ptr_lookup
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def ptr_lookup(ip: str, cache_dir: Path, ttl_seconds: int) -> list[str]:
|
||
|
|
key = cache_key(f"ptr-{ip}")
|
||
|
|
cached = load_json_cache(cache_dir, key, ttl_seconds)
|
||
|
|
if cached is not None:
|
||
|
|
return list(cached.get("answers", []))
|
||
|
|
output = subprocess.run(
|
||
|
|
["dig", "+time=2", "+tries=1", "+short", "-x", ip, "PTR"],
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
check=False,
|
||
|
|
).stdout
|
||
|
|
answers = [normalize_name(line) for line in output.splitlines() if line.strip()]
|
||
|
|
store_json_cache(cache_dir, key, {"answers": answers})
|
||
|
|
return answers</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>This function is one of the building blocks inside `ct_dns_utils.py`. It exists so the file can do one narrow job at a time instead of one giant unreadable routine.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>Earlier blocks or operator input feed this block. → <strong>ptr_lookup</strong> → Later blocks in the same file or in the next analytical stage consume its output.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|
||
|
|
## provider_explanations
|
||
|
|
|
||
|
|
<table style="width:100%; table-layout:fixed; border-collapse:collapse;">
|
||
|
|
<tr>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<pre style="margin:0; padding:14px; overflow-x:auto; background:#111827; color:#e5e7eb; border-radius:10px; border:1px solid #374151; font-size:12px; line-height:1.45;"><code class="language-python">def provider_explanations() -> dict[str, str]:
|
||
|
|
return {
|
||
|
|
"Adobe Campaign": "A marketing and communication platform often used to send customer messages, email journeys, and campaign traffic. In DNS terms, it can sit in front of cloud infrastructure rather than hosting the final application by itself.",
|
||
|
|
"AWS": "Amazon Web Services, a large public cloud platform. In this report it usually means the endpoint ultimately lands on Amazon-hosted compute or load-balancing infrastructure.",
|
||
|
|
"AWS ALB": "AWS Application Load Balancer. A traffic-distribution front door that sends incoming web requests to one or more backend services.",
|
||
|
|
"AWS CloudFront": "Amazon's global content-delivery and edge network. It is often used to front websites, APIs, and static assets close to users.",
|
||
|
|
"Google Apigee": "An API gateway and API-management layer. If a hostname lands here, it usually means the public endpoint is being governed as an API product rather than being exposed directly from an application server.",
|
||
|
|
"Pega Cloud": "A managed hosting platform for Pega applications and workflow systems. It often fronts case-management or process-heavy applications.",
|
||
|
|
"Microsoft Edge": "Microsoft-operated edge infrastructure. In DNS this usually means the public name lands on Microsoft's front-door network rather than directly on a private application host.",
|
||
|
|
"Infinite / agency alias": "A third-party aliasing pattern typically used by an agency or service intermediary. It points traffic onward to the actual delivery platform.",
|
||
|
|
"CNAME": "A DNS alias record. It says one hostname is really another hostname, rather than directly mapping to an IP address.",
|
||
|
|
"A record": "A DNS record that maps a hostname to an IPv4 address.",
|
||
|
|
"AAAA record": "A DNS record that maps a hostname to an IPv6 address.",
|
||
|
|
"PTR record": "A reverse-DNS record. It maps an IP address back to a hostname and is useful as a provider clue, not as proof of ownership.",
|
||
|
|
"NXDOMAIN": "A DNS response meaning the name does not exist publicly.",
|
||
|
|
}</code></pre>
|
||
|
|
</td>
|
||
|
|
<td style="width:50%; vertical-align:top; padding:8px;">
|
||
|
|
<p><strong>What this block is doing</strong></p><p>Supplies the glossary text used later in the reports.</p>
|
||
|
|
<p><strong>Flow arrows</strong></p><p>The delivery labels used by the report. → <strong>provider_explanations</strong> → The monograph glossary uses these explanations directly.</p>
|
||
|
|
<p><strong>How to think about it</strong></p><p>Treat this block as one small station in a pipeline. Ask: what comes in here, what gets changed here, and what comes out for the next block?</p>
|
||
|
|
</td>
|
||
|
|
</tr>
|
||
|
|
</table>
|
||
|
|
|