How to Build an Attack-Surface Inventory from Certificate Transparency
A domain inventory maintained in a spreadsheet starts drifting as soon as another team creates a staging host or replaces a certificate. I built a small Python job that discovers names from Certificate Transparency, records what the domain exposes today, and reports only the changes worth reviewing.
The Domain Intelligence Suite provides the WHOIS, DNS, TLS, and certificate-derived subdomain data. Python turns that response into a durable inventory rather than another JSON file nobody checks.
Treat Certificate Transparency as discovery, not proof
Public certificate authorities submit certificates to Certificate Transparency logs. Searching those logs can reveal names such as api.example.com, staging.example.com, and wildcard entries that were included in issued certificates.
That makes CT useful for discovery, but it is not a perfect inventory. A name in a certificate might be retired, internal, misspelled, or no longer resolve. A service using a private certificate will not appear. The right workflow is:
- Discover candidate names from CT.
- Normalize and deduplicate them.
- Resolve or verify candidates you intend to monitor.
- Compare the latest set with the previous snapshot.
- Ask an owner to classify unexpected additions.
Do not turn a new certificate name directly into an incident. Treat it as a lead that needs context.
Fetch the domain report in one request
Apify's synchronous dataset endpoint waits for the Actor and returns its dataset items in the response. This works well for a scheduled inventory job because there is no run ID or polling loop to maintain.
import os
from typing import Any
import requests
APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~domain-intelligence-suite"
RUN_URL = (
f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
"run-sync-get-dataset-items"
)
def fetch_domain_report(domain: str) -> dict[str, Any]:
response = requests.post(
RUN_URL,
params={"token": APIFY_TOKEN},
json={
"domain": domain,
"modules": ["dns", "ssl", "subdomains"],
"dnsRecordTypes": ["A", "AAAA", "CNAME", "MX", "NS"],
"maxSubdomains": 1000,
"includeWildcardSubdomains": True,
},
timeout=120,
)
response.raise_for_status()
items = response.json()
if not isinstance(items, list) or len(items) != 1:
raise RuntimeError(f"Expected one domain result, got {len(items)}")
return items[0]
Keep APIFY_TOKEN in a secret store for cron or CI. YOUR_APIFY_TOKEN shows the required value without placing a real credential in the script.
The Actor runs its modules independently. One module can contain an error while others still return useful data, so check module results instead of treating any partial failure as an empty inventory.
Normalize the names before storing them
CT data often includes wildcard names, duplicates with different casing, and entries that do not belong under the requested suffix after loose matching. Normalize the hostname and apply a strict domain boundary.
import re
HOSTNAME = re.compile(
r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$"
)
def normalize_hostname(value: str, root_domain: str) -> str | None:
name = value.strip().lower().rstrip(".")
if name.startswith("*."):
name = name[2:]
root = root_domain.lower().rstrip(".")
if name != root and not name.endswith(f".{root}"):
return None
if not HOSTNAME.fullmatch(name):
return None
return name
def discovered_names(report: dict, root_domain: str) -> set[str]:
module = report.get("subdomains", {})
if module.get("error"):
raise RuntimeError(f"CT discovery failed: {module['error']}")
candidates = module.get("subdomains", [])
normalized = {
name
for candidate in candidates
if (name := normalize_hostname(str(candidate), root_domain))
}
normalized.add(root_domain.lower().rstrip("."))
return normalized
Removing *. does not mean every possible wildcard hostname exists. It records the certificate scope as the base name, which is enough to flag the wildcard for review without inventing thousands of hosts.
Save immutable snapshots in SQLite
An inventory becomes useful when it can answer “what changed?” SQLite is enough for a domain portfolio and keeps each scan as an immutable snapshot.
import json
import sqlite3
from datetime import datetime, timezone
def open_inventory(path: str = "attack-surface.sqlite3") -> sqlite3.Connection:
connection = sqlite3.connect(path)
connection.execute("""
CREATE TABLE IF NOT EXISTS scans (
id INTEGER PRIMARY KEY,
root_domain TEXT NOT NULL,
scanned_at TEXT NOT NULL,
names_json TEXT NOT NULL,
raw_report_json TEXT NOT NULL
)
""")
return connection
def save_scan(connection: sqlite3.Connection, domain: str, report: dict) -> int:
names = sorted(discovered_names(report, domain))
cursor = connection.execute(
"""
INSERT INTO scans (root_domain, scanned_at, names_json, raw_report_json)
VALUES (?, ?, ?, ?)
""",
(
domain,
datetime.now(timezone.utc).isoformat(),
json.dumps(names),
json.dumps(report, sort_keys=True),
),
)
connection.commit()
return int(cursor.lastrowid)
Saving the raw response matters. If a parser changes later, the original evidence still shows what DNS, TLS, and CT modules returned during that scan. Apply a retention policy if the portfolio is large rather than silently overwriting history.
Compare the two latest successful snapshots
A daily list of every hostname creates noise. The useful output is additions and removals between two successful scans.
def latest_name_sets(
connection: sqlite3.Connection,
domain: str,
) -> tuple[set[str], set[str]] | None:
rows = connection.execute(
"""
SELECT names_json
FROM scans
WHERE root_domain = ?
ORDER BY scanned_at DESC
LIMIT 2
""",
(domain,),
).fetchall()
if len(rows) < 2:
return None
current = set(json.loads(rows[0][0]))
previous = set(json.loads(rows[1][0]))
return current, previous
def inventory_changes(connection: sqlite3.Connection, domain: str) -> dict:
snapshots = latest_name_sets(connection, domain)
if snapshots is None:
return {"added": [], "removed": [], "baseline_created": True}
current, previous = snapshots
return {
"added": sorted(current - previous),
"removed": sorted(previous - current),
"baseline_created": False,
}
Only save a scan when the subdomain module succeeded. Saving an empty list after a timeout would make every known name appear removed and every name reappear on the next run.
Add context before paging anyone
Not every new hostname deserves the same response. A name like staging.example.com might be expected, while vpn-old.example.com could expose a forgotten service. Start with deterministic labels rather than a mysterious risk score.
SENSITIVE_LABELS = {
"admin", "auth", "billing", "internal", "jenkins",
"old", "staging", "test", "vpn",
}
def classify_name(hostname: str, root_domain: str) -> dict:
relative = hostname.removesuffix(f".{root_domain}")
labels = set(relative.split("."))
matched = sorted(labels & SENSITIVE_LABELS)
return {
"hostname": hostname,
"review": bool(matched),
"matched_labels": matched,
}
A label match means “review this first,” not “this system is vulnerable.” Confirm DNS resolution, ownership, intended exposure, authentication, and patch status through authorized processes. Do not scan ports or attempt access unless that activity is explicitly approved.
Cost and operating limits
The Domain Intelligence Suite is currently listed from $5 per 1,000 domain intelligence results, or $0.005 per domain result in Actor charges. Apify platform usage may also apply, and the live Actor page should be checked before increasing the schedule or domain count.
CT sources can be delayed or rate-limited, and certificate names describe issuance rather than current reachability. Run the inventory daily or weekly, retry transient module errors once with backoff, and keep the previous successful snapshot when discovery fails.
For each added name, record an owner, environment, expected exposure, and review date in the system your team already uses. The SQLite scan remains evidence of discovery; the ownership record explains whether the hostname is expected and who must remove it when it is no longer needed.
Top comments (0)