From def4deb77d8ef4154a52da4c07383642fd2140d8 Mon Sep 17 00:00:00 2001 From: incendiary Date: Fri, 14 Aug 2026 13:18:11 +0800 Subject: [PATCH] feat: publish run observations and manifest --- README.md | 7 +- ROADMAP.md | 15 ++- classes/run_contract.py | 191 +++++++++++++++++++++++++++ docs/ALLOCATOR-CONTRACT.md | 6 + resolver.py | 74 ++++++++++- tests/test_resolver.py | 34 ++++- tests/test_run_contract_publisher.py | 127 ++++++++++++++++++ 7 files changed, 446 insertions(+), 8 deletions(-) create mode 100644 classes/run_contract.py create mode 100644 tests/test_run_contract_publisher.py diff --git a/README.md b/README.md index 92901e1..df0258a 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,8 @@ Each run creates a timestamped subdirectory under the output directory containin | `takeover_candidates_*.txt` | `DANGLING\|origin\|target\|category\|recommendation\|evidence\|hops\|chain` — the chain records the full CNAME path (`a -> b -> c`), so the claimable hop is visible without re-resolving. Plus `NS_TAKEOVER\|` lines for unresolvable nameservers | | `csp_matches_*.txt` | One handoff record per provider-published attribution: `domain\|ip\|provider\|region\|service\|prefix\|border_group`. One address can produce several records. Prefixed `WILDCARD\|` when the resolution was a catch-all. See [Cloud IP attribution](#cloud-ip-attribution) | | `allocator-targets-v1.json` | Versioned provider-aware allocator handoff. Groups service, prefix, and border-group metadata per provider/hostname/address/region and excludes wildcard observations; one address may produce multiple records when a provider publishes multiple regions. | +| `dns-observations-v1.json` | Versioned excluded-observation feed for wildcard, dangling CNAME, NS, unresolved, and provider-catalogue-incomplete results. | +| `run-manifest-v1.json` | Final completion marker recording run status, provider catalogue provenance, and the published contract documents. | | `environment_results_*.json` | Run metadata (command, external IP, Docker status) | | `provider_catalogues.json` | AWS, GCP, and Azure catalogue status, source, retrieval time, snapshot identifier, and any failure reason | | `{provider}_ip_ranges.json` | Validated provider ranges and provenance used by this run, from either a live source or a fresh cache | @@ -145,6 +147,7 @@ lambda_handler.py — Lambda entry point → run(env_manager) ├── DomainProcessingContext — per-domain state (domain name, resolver, CSP IPs) ├── CSPIPAddresses — value object holding fetched AWS/GCP/Azure IP ranges ├── DomainCategoriser — regex-based classification of dangling CNAME targets +├── run_contract.py — excluded-observation and final run-manifest publisher └── domain_processor.py — orchestrates DNS → CSP checks per domain ``` @@ -157,8 +160,8 @@ to the unresolved output only after both resolvers fail on the final configured ## Cloud IP attribution -The current text files are the v2 CLI output. The versioned JSON handoff planned -for provider-aware allocators is defined separately in the +The current text files are the v2 CLI output. The published versioned JSON +handoff for provider-aware allocators is defined separately in the [allocator contract](docs/ALLOCATOR-CONTRACT.md); its schemas, examples, and documentation field tables are validated together in CI. diff --git a/ROADMAP.md b/ROADMAP.md index 73b1d30..a5cf0f3 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -33,8 +33,8 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg | 6 | Pin Lambda input to the triggering object version | Merged, PR 165 | The reference handler reads the exact S3 object version from the event and rejects incomplete version information. This does not reopen Lambda expansion. | | 7 | Preserve every provider attribution | Merged, PR 169 | Duplicate services, overlapping prefixes, identical CIDRs across providers, and multiple provider-published regions survive catalogue parsing, matching, pipe output, and JSON publication without changing the schema. Legacy scalar cache snapshots remain readable. | | 8 | Replace linear cloud-range matching with an indexed matcher | Deferred until measured need | A deterministic benchmark is defined before implementation; outputs are byte-for-byte equivalent to item 7; measured runtime and memory are reported at representative scale. | -| 9 | Fail explicitly on required output errors | Implemented on `fix/fail-on-output-errors`; PR pending | Required writes cannot be swallowed; partial runs cannot leave a stale actionable document; injected open/write/replace failures produce a nonzero, explicit failure with regression tests. | -| 10 | Complete observation/run-manifest publication | Deferred; not an initial-workability gate | The checked observation and manifest contracts are emitted by real runs; manifest state reflects provider completeness and publication outcome; actionable output is null for incomplete/failed runs. | +| 9 | Fail explicitly on required output errors | Merged, PR 170 | Required writes cannot be swallowed; partial runs cannot leave a stale actionable document; injected open/write/replace failures produce a nonzero, explicit failure with regression tests. | +| 10 | Publish observations and the run manifest | Implemented on `feat/publish-run-contracts`; PR pending | The checked observation and manifest contracts are emitted by real runs; manifest state reflects provider completeness and publication outcome; actionable output is null for incomplete/failed runs. | | 11 | Validate the real allocator consumer end to end | Planned, cross-repository | The AWS consumer ingests a current DNSResolver document unchanged; GCP/Azure route only to provider-aware implementations or are explicitly rejected; a synthetic authorized fixture proves no provider is misrouted. | | 12 | Measure and calibrate large-run behavior | Planned last | A repeatable representative benchmark replaces the unmeasured README scalability claim; resource limits and operational guidance reflect measured results. | @@ -54,8 +54,8 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg ### Delivery state — 2026-08-14 -Item 7 was squash-merged in PR 169. Item 9 is implemented on -`fix/fail-on-output-errors` and ready for its focused pull request. +Items 7 and 9 were squash-merged in PRs 169 and 170. Item 10 is implemented on +`feat/publish-run-contracts` and ready for its focused pull request. - Focused suite: 91 tests passed. - Full local CI: Ruff and format passed; baseline 5 passed; full suite 310 passed @@ -81,6 +81,13 @@ Item 7 was squash-merged in PR 169. Item 9 is implemented on system-resolver run fetched all three current catalogues, resolved both public inputs, emitted 45 attribution records, and published 20 schema-valid targets without leaving a temporary allocator document. +- Item 10 focused suite: 32 tests passed. Full local CI: 320 tests passed at 95% + coverage for `classes` and `imports`. A production system-resolver run published + 20 actionable targets, an empty but schema-valid observation document, and a + schema-valid complete manifest referencing both documents; no temporary run + document remained. A reserved `example.com` negative control separately + published one schema-valid unresolved observation, an empty actionable array, + and a complete manifest. > **On its derivation.** The plan below came from [`REVIEW.md`](REVIEW.md) (2026-07-15), which > assessed the tool against a misread goal — cloud attribution treated as a supporting attribute diff --git a/classes/run_contract.py b/classes/run_contract.py new file mode 100644 index 0000000..fcb4281 --- /dev/null +++ b/classes/run_contract.py @@ -0,0 +1,191 @@ +"""Publish non-actionable DNS observations and the run completion manifest.""" + +import json +import os +from pathlib import Path + +from classes.custom_exceptions import OutputWriteError + +CONTRACT_VERSION = "1.0" +OBSERVATIONS_FILENAME = "dns-observations-v1.json" +MANIFEST_FILENAME = "run-manifest-v1.json" +TARGETS_FILENAME = "allocator-targets-v1.json" + + +def clear_run_documents(output_dir): + """Remove stale run documents and their temporary files.""" + for filename in (OBSERVATIONS_FILENAME, MANIFEST_FILENAME): + destination = Path(output_dir) / filename + _remove_output(destination) + _remove_output(destination.with_suffix(destination.suffix + ".tmp")) + + +def publish_dns_observations( + output_files, + output_dir, + provider_failures=None, + hostnames=None, +): + """Build the excluded-observation contract from the run's text outputs.""" + grouped = {} + standard = output_files.get("standard", {}) + + for line in _lines(standard.get("resolved")): + if line.startswith("WILDCARD|"): + _add_observation( + grouped, + line, + "WILDCARD", + "wildcard", + "DNS answer matched the zone wildcard probe", + ) + elif line.startswith("WILDCARD_ZONE|"): + _add_observation( + grouped, + line, + "WILDCARD_ZONE", + "wildcard_zone", + "Parent zone answered random-label probes", + ) + + for line in _lines(standard.get("takeover")): + fields = line.split("|") + if line.startswith("DANGLING|") and len(fields) >= 3: + _merge_observation( + grouped, + fields[1], + "dangling_cname", + [], + [f"CNAME target did not resolve: {fields[2]}"], + ) + elif line.startswith("NS_TAKEOVER|") and len(fields) >= 3: + _merge_observation( + grouped, + fields[1], + "ns_takeover", + [], + [f"Nameserver did not resolve: {fields[2]}"], + ) + else: + raise ValueError("Malformed takeover observation output") + + unresolved_prefix = "DNS resolution error for " + for line in _lines(standard.get("unresolved")): + if not line.startswith(unresolved_prefix): + raise ValueError("Malformed unresolved observation output") + hostname, separator, reason = line[len(unresolved_prefix) :].partition(": ") + if not separator or not hostname or not reason: + raise ValueError("Malformed unresolved observation output") + _merge_observation(grouped, hostname, "unresolved", [], [reason]) + + if provider_failures: + reasons = [ + f"{provider} provider catalogue unusable: {reason}" + for provider, reason in sorted(provider_failures.items()) + ] + for hostname in sorted(set(hostnames or [])): + _merge_observation( + grouped, + hostname, + "provider_catalog_incomplete", + [], + reasons, + ) + + observations = [] + for (hostname, kind), entry in sorted(grouped.items()): + observations.append( + { + "contract_version": CONTRACT_VERSION, + "hostname": hostname, + "kind": kind, + "ips": sorted(entry["ips"]), + "reasons": sorted(entry["reasons"]), + "actionability": "excluded", + } + ) + + _write_json_atomic(observations, Path(output_dir) / OBSERVATIONS_FILENAME) + return observations + + +def publish_run_manifest( + output_dir, + run_id, + status, + started_at, + completed_at, + catalogues, +): + """Publish the final run state after its referenced outputs are settled.""" + if status not in {"complete", "incomplete", "failed"}: + raise ValueError(f"Unknown run status: {status}") + if set(catalogues) != {"aws", "gcp", "azure"}: + raise ValueError("Run manifest requires AWS, GCP, and Azure catalogues") + + manifest = { + "contract_version": CONTRACT_VERSION, + "run_id": run_id, + "status": status, + "started_at": started_at, + "completed_at": completed_at, + "provider_catalogs": { + provider: catalogues[provider].manifest_entry() + for provider in ("aws", "gcp", "azure") + }, + "outputs": { + "actionable_targets": TARGETS_FILENAME if status == "complete" else None, + "observations": OBSERVATIONS_FILENAME, + }, + } + _write_json_atomic(manifest, Path(output_dir) / MANIFEST_FILENAME) + return manifest + + +def _add_observation(grouped, line, marker, kind, reason): + fields = line.split("|") + if len(fields) < 3 or fields[0] != marker or not fields[1]: + raise ValueError(f"Malformed {kind} observation output") + _merge_observation(grouped, fields[1], kind, fields[2:], [reason]) + + +def _merge_observation(grouped, hostname, kind, ips, reasons): + entry = grouped.setdefault((hostname, kind), {"ips": set(), "reasons": set()}) + entry["ips"].update(ip for ip in ips if ip) + entry["reasons"].update(reason for reason in reasons if reason) + + +def _lines(path): + if not path: + return [] + try: + with open(path, encoding="utf-8") as handle: + return [line.strip() for line in handle if line.strip()] + except OSError as error: + raise ValueError( + f"Unable to read observation output {path}: {error}" + ) from error + + +def _write_json_atomic(document, destination): + temporary = destination.with_suffix(destination.suffix + ".tmp") + _remove_output(destination) + _remove_output(temporary) + try: + with open(temporary, "w", encoding="utf-8") as handle: + json.dump(document, handle, indent=2) + handle.write("\n") + os.replace(temporary, destination) + except OSError as error: + raise OutputWriteError( + f"Unable to publish run document {destination}: {error}" + ) from error + finally: + _remove_output(temporary) + + +def _remove_output(path): + try: + path.unlink(missing_ok=True) + except OSError as error: + raise OutputWriteError(f"Unable to remove output {path}: {error}") from error diff --git a/docs/ALLOCATOR-CONTRACT.md b/docs/ALLOCATOR-CONTRACT.md index 7b2f1cf..311892e 100644 --- a/docs/ALLOCATOR-CONTRACT.md +++ b/docs/ALLOCATOR-CONTRACT.md @@ -28,6 +28,12 @@ provider-published region, and excludes unusable, or a CSP record is malformed, no allocator-target document is published. +Every run also publishes `dns-observations-v1.json` and writes +`run-manifest-v1.json` last as its completion marker. A complete manifest points +to the actionable document. Incomplete or failed manifests set that path to +`null`; a failed-state manifest is best effort when the underlying failure also +prevents output writes. + ## Documents | Document | Schema | Purpose | diff --git a/resolver.py b/resolver.py index 898b1a7..8a13c5d 100644 --- a/resolver.py +++ b/resolver.py @@ -1,6 +1,8 @@ import asyncio import json +from datetime import datetime, timezone from pathlib import Path +from uuid import uuid4 from tqdm import tqdm @@ -12,6 +14,11 @@ from classes.custom_exceptions import OutputWriteError, ProviderCatalogueError from classes.dns_handler import DNSHandler from classes.environment_manager import EnvironmentManager +from classes.run_contract import ( + clear_run_documents, + publish_dns_observations, + publish_run_manifest, +) from classes.run_summary import RunSummary from imports.cloud_ip_ranges import ( fetch_aws_ip_ranges, @@ -28,14 +35,47 @@ async def run(env_manager): both the CLI entrypoint (resolver.py) and the Lambda entrypoint (lambda_handler.py) can share the same logic. """ + run_id = str(uuid4()) + started_at = _timestamp() + catalogues = {} clear_allocator_targets(env_manager.output_dir) - catalogues = { + clear_run_documents(env_manager.output_dir) + try: + await _run(env_manager, run_id, started_at, catalogues) + except ProviderCatalogueError: + raise + except Exception: + try: + clear_allocator_targets(env_manager.output_dir) + publish_dns_observations( + env_manager.output_files, + env_manager.output_dir, + ) + if set(catalogues) == {"aws", "gcp", "azure"}: + publish_run_manifest( + env_manager.output_dir, + run_id=run_id, + status="failed", + started_at=started_at, + completed_at=_timestamp(), + catalogues=catalogues, + ) + except Exception as contract_error: + env_manager.log_error( + "Unable to publish failed run state: %s", contract_error + ) + raise + + +async def _run(env_manager, run_id, started_at, catalogues): + fetched_catalogues = { "gcp": fetch_google_cloud_ip_ranges( env_manager.output_dir, env_manager.extreme ), "aws": fetch_aws_ip_ranges(env_manager.output_dir, env_manager.extreme), "azure": fetch_azure_ip_ranges(env_manager.output_dir, env_manager.extreme), } + catalogues.update(fetched_catalogues) status_path = Path(env_manager.output_dir) / "provider_catalogues.json" with open(status_path, "w", encoding="utf-8") as handle: json.dump( @@ -49,10 +89,28 @@ async def run(env_manager): unusable = [name for name, catalogue in catalogues.items() if not catalogue.usable] if unusable: + env_manager.set_domains() details = "; ".join( f"{name}: {catalogues[name].error or catalogues[name].status}" for name in unusable ) + publish_dns_observations( + env_manager.output_files, + env_manager.output_dir, + provider_failures={ + name: catalogues[name].error or catalogues[name].status + for name in unusable + }, + hostnames=env_manager.domains, + ) + publish_run_manifest( + env_manager.output_dir, + run_id=run_id, + status="incomplete", + started_at=started_at, + completed_at=_timestamp(), + catalogues=catalogues, + ) raise ProviderCatalogueError( "Required provider catalogue(s) unavailable; no domains were processed " f"and no actionable results were published. {details}. Status: {status_path}" @@ -143,6 +201,7 @@ async def bounded_process(domain, pbar, final_retry): "Published %d actionable target(s) to allocator-targets-v1.json", len(targets), ) + publish_dns_observations(env_manager.output_files, env_manager.output_dir) RunSummary(env_manager.output_files, env_manager.output_dir, __version__).display( len(env_manager.domains), len(domains_to_process) @@ -160,6 +219,19 @@ async def bounded_process(domain, pbar, final_retry): env_manager.log_info("Azure IPv4 Ranges: %s", csp_ip_addresses.get_azure_ipv4()) env_manager.log_info("Azure IPv6 Ranges: %s", csp_ip_addresses.get_azure_ipv6()) + publish_run_manifest( + env_manager.output_dir, + run_id=run_id, + status="complete", + started_at=started_at, + completed_at=_timestamp(), + catalogues=catalogues, + ) + + +def _timestamp(): + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + async def main_async(): print(f"DNSResolver v{__version__}") diff --git a/tests/test_resolver.py b/tests/test_resolver.py index 4576880..760a5d4 100644 --- a/tests/test_resolver.py +++ b/tests/test_resolver.py @@ -66,6 +66,17 @@ def catalogue(provider): ) == [] ) + assert ( + json.loads( + (Path(run_environment.output_dir) / "dns-observations-v1.json").read_text() + ) + == [] + ) + manifest = json.loads( + (Path(run_environment.output_dir) / "run-manifest-v1.json").read_text() + ) + assert manifest["status"] == "complete" + assert manifest["outputs"]["actionable_targets"] == "allocator-targets-v1.json" async def test_run_fails_closed_before_processing_domains(run_environment): @@ -102,7 +113,9 @@ async def test_run_fails_closed_before_processing_domains(run_environment): ): await resolver.run(run_environment) - run_environment.set_domains.assert_not_called() + # Input names are loaded only so each excluded hostname can carry the + # provider-catalogue reason; no DNS processing begins. + run_environment.set_domains.assert_called_once_with() process.assert_not_awaited() status = json.loads( (Path(run_environment.output_dir) / "provider_catalogues.json").read_text() @@ -110,11 +123,24 @@ async def test_run_fails_closed_before_processing_domains(run_environment): assert status["gcp"]["usable"] is False assert status["gcp"]["error"] == "TLS failure" assert not (Path(run_environment.output_dir) / "allocator-targets-v1.json").exists() + observations = json.loads( + (Path(run_environment.output_dir) / "dns-observations-v1.json").read_text() + ) + assert observations[0]["kind"] == "provider_catalog_incomplete" + manifest = json.loads( + (Path(run_environment.output_dir) / "run-manifest-v1.json").read_text() + ) + assert manifest["status"] == "incomplete" + assert manifest["outputs"]["actionable_targets"] is None async def test_run_aborts_before_publication_on_output_failure(run_environment): stale_targets = Path(run_environment.output_dir) / "allocator-targets-v1.json" stale_targets.write_text('[{"stale": true}]', encoding="utf-8") + stale_observations = Path(run_environment.output_dir) / "dns-observations-v1.json" + stale_observations.write_text('[{"stale": true}]', encoding="utf-8") + stale_manifest = Path(run_environment.output_dir) / "run-manifest-v1.json" + stale_manifest.write_text('{"status": "complete"}', encoding="utf-8") failure = OutputWriteError("disk full") def catalogue(provider): @@ -152,3 +178,9 @@ def catalogue(provider): "example.com", failure, ) + manifest = json.loads( + (Path(run_environment.output_dir) / "run-manifest-v1.json").read_text() + ) + assert manifest["status"] == "failed" + assert manifest["outputs"]["actionable_targets"] is None + assert json.loads(stale_observations.read_text()) == [] diff --git a/tests/test_run_contract_publisher.py b/tests/test_run_contract_publisher.py new file mode 100644 index 0000000..a8623ca --- /dev/null +++ b/tests/test_run_contract_publisher.py @@ -0,0 +1,127 @@ +import json +from pathlib import Path +from unittest.mock import patch + +import pytest +from jsonschema import Draft202012Validator, FormatChecker + +from classes.custom_exceptions import OutputWriteError +from classes.run_contract import publish_dns_observations, publish_run_manifest +from imports.cloud_ip_ranges import ProviderCatalogue + +REPO_ROOT = Path(__file__).resolve().parents[1] +OBSERVATION_SCHEMA = json.loads( + (REPO_ROOT / "contracts" / "dns-observations-v1.schema.json").read_text() +) +MANIFEST_SCHEMA = json.loads( + (REPO_ROOT / "contracts" / "run-manifest-v1.schema.json").read_text() +) + + +def _output_files(tmp_path): + files = { + "resolved": ( + "WILDCARD|wild.example.com|192.0.2.10\n" + "WILDCARD_ZONE|rotating.example.com|192.0.2.20|2001:db8::20\n" + "ordinary.example.com|192.0.2.30\n" + ), + "unresolved": ( + "DNS resolution error for missing.example.com: query timed out\n" + ), + "takeover": ( + "DANGLING|old.example.com|missing.example.net|unknown|Review it|N/A|1|old.example.com -> missing.example.net\n" + "NS_TAKEOVER|delegated.example.com|ns1.missing.example.net\n" + ), + "csp": "", + } + paths = {} + for name, content in files.items(): + path = tmp_path / f"{name}.txt" + path.write_text(content, encoding="utf-8") + paths[name] = str(path) + return {"standard": paths} + + +def _catalogue(provider, usable=True): + return ProviderCatalogue( + provider=provider, + ipv4_ranges=["192.0.2.0/24"] if usable else [], + ipv6_ranges=[], + metadata={}, + status="complete" if usable else "failed", + usable=usable, + source_url=f"https://example.com/{provider}.json", + retrieved_at="2026-01-01T00:00:00Z" if usable else None, + snapshot_id=f"{provider}-1" if usable else None, + error=None if usable else "catalogue unavailable", + ) + + +def test_observation_publisher_retains_every_excluded_kind(tmp_path): + observations = publish_dns_observations( + _output_files(tmp_path), + tmp_path, + provider_failures={"gcp": "catalogue unavailable"}, + hostnames=["ordinary.example.com"], + ) + + Draft202012Validator(OBSERVATION_SCHEMA, format_checker=FormatChecker()).validate( + observations + ) + assert {item["kind"] for item in observations} == { + "wildcard", + "wildcard_zone", + "dangling_cname", + "ns_takeover", + "unresolved", + "provider_catalog_incomplete", + } + rotating = next( + item for item in observations if item["hostname"] == "rotating.example.com" + ) + assert rotating["ips"] == ["192.0.2.20", "2001:db8::20"] + assert all(item["actionability"] == "excluded" for item in observations) + assert ( + json.loads((tmp_path / "dns-observations-v1.json").read_text()) == observations + ) + + +@pytest.mark.parametrize("status", ["complete", "incomplete", "failed"]) +def test_manifest_publisher_matches_checked_contract(tmp_path, status): + catalogues = { + provider: _catalogue( + provider, usable=not (status == "incomplete" and provider == "gcp") + ) + for provider in ("aws", "gcp", "azure") + } + + manifest = publish_run_manifest( + tmp_path, + run_id=f"{status}-run", + status=status, + started_at="2026-01-01T00:00:00Z", + completed_at="2026-01-01T00:01:00Z", + catalogues=catalogues, + ) + + Draft202012Validator(MANIFEST_SCHEMA, format_checker=FormatChecker()).validate( + manifest + ) + expected_target = "allocator-targets-v1.json" if status == "complete" else None + assert manifest["outputs"]["actionable_targets"] == expected_target + assert manifest["outputs"]["observations"] == "dns-observations-v1.json" + + +def test_observation_publication_failure_removes_stale_and_temporary_files(tmp_path): + destination = tmp_path / "dns-observations-v1.json" + temporary = tmp_path / "dns-observations-v1.json.tmp" + destination.write_text('[{"stale": true}]', encoding="utf-8") + + with ( + patch("classes.run_contract.os.replace", side_effect=OSError("read only")), + pytest.raises(OutputWriteError), + ): + publish_dns_observations(_output_files(tmp_path), tmp_path) + + assert not destination.exists() + assert not temporary.exists()