diff --git a/capabilities/web-security/agents/web-security.md b/capabilities/web-security/agents/web-security.md index 8b1d880..1ee185b 100644 --- a/capabilities/web-security/agents/web-security.md +++ b/capabilities/web-security/agents/web-security.md @@ -106,6 +106,7 @@ Any tool that scans, fuzzes, or floods runs on shared local hardware. Cap concur - Use `generate_rebinding_hostname` and `list_rebinding_presets` for DNS rebinding SSRF bypass when IP filters validate resolved addresses before fetching. - Use the `agentmail_*` tools (`agentmail_list_inboxes`, `agentmail_create_inbox`, `agentmail_list_messages`, `agentmail_get_message`, `agentmail_send_message`, `agentmail_reply_message`) to work with AgentMail email inboxes when a real, agent-owned email address is useful — for example signup, recovery, or email-verification flows. Requires an AgentMail API key via the `AGENTMAIL_API_KEY` environment variable or the `api_key` argument. Available only when the key is configured. - Use IP rotation (`flareprox_*` tools or the local `fireprox` CLI at `~/git/fireprox/fire.py`) only when `IPROTATE_ENABLED` is set and the target is rate-limiting, IP-banning, or WAF-blocking normal requests. Load the `ip-rotation` skill for backend selection and lifecycle. Always clean up fireprox proxies to avoid AWS charges. +- Use geo-distributed DNS resolution (`check_geo_resolver_readiness`, `discover_open_resolvers`, `resolve_via_open_resolvers`) only when `GEO_RESOLVERS_ENABLED` is set and a Shodan or Censys API key is available. This discovers open recursive resolvers per country to resolve hostnames from multiple geographic vantage points, revealing geo-fenced infrastructure invisible from a single location. Load the `distributed-routing-testing` skill for the full workflow. This is orthogonal to IP rotation: IP rotation changes the egress IP for HTTP requests, geo-distributed DNS changes the resolver location for DNS queries. - Use the local `pacu` CLI when an authorized test yields AWS credentials, cloud metadata access, or another AWS-impact lead that needs validation. Load the `pacu-aws-exploitation` skill first, confirm AWS scope, and start with identity/read-only enumeration before any mutating module. - When the target accepts or extracts archive uploads (ZIP, TAR, etc.), use the local `archive-alchemist` CLI at `~/git/archivealchemist/archive-alchemist.py` to craft malicious archives. Load the `archive-path-traversal` skill for the full attack pattern catalog and iterative workflow. - Use `exif_read`, `exif_write`, `exif_strip`, and `exif_copy` for EXIF/XMP/IPTC metadata manipulation on image and document files. Primary use cases: injecting XSS payloads into metadata fields (Comment, Artist, Copyright, ImageDescription) that get rendered by the target, crafting images with SSRF-triggering metadata for server-side processing, testing whether upload handlers strip metadata, and transplanting payloads between file formats. Strip metadata first with `exif_strip` to create a clean baseline, then inject specific payloads with `exif_write`. diff --git a/capabilities/web-security/capability.yaml b/capabilities/web-security/capability.yaml index d53ee5d..2c81e30 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -1,12 +1,14 @@ schema: 1 name: web-security -version: "1.12.0" +version: "1.13.0" description: > - Web application penetration testing with 82 attack technique playbooks + Web application penetration testing with 83 attack technique playbooks covering HTTP desync/request smuggling, cache poisoning, SSRF, SSTI, DOM vulnerabilities, authentication bypasses, parser differentials, AEM/Sling exploitation, GraphQL, OAuth, and client-side attacks. - Includes HTTP client tooling with OOB callbacks via webhook.site + Includes geo-distributed DNS resolution via in-region open resolvers + (Shodan/Censys) for detecting geo-fenced infrastructure, + HTTP client tooling with OOB callbacks via webhook.site (API-key aware) and interactsh, four coexisting Caido surfaces (caido-cli server, the Python caido-sdk-client, lightweight and full-surface MCP servers, and the caido-mode TypeScript SDK CLI on @@ -244,4 +246,10 @@ keywords: - interactsh - oob-callbacks - securitycontext + - geo-dns + - geo-fencing + - open-resolvers + - shodan + - censys + - distributed-routing - caido diff --git a/capabilities/web-security/skills/distributed-routing-testing/SKILL.md b/capabilities/web-security/skills/distributed-routing-testing/SKILL.md new file mode 100644 index 0000000..03ac752 --- /dev/null +++ b/capabilities/web-security/skills/distributed-routing-testing/SKILL.md @@ -0,0 +1,119 @@ +--- +name: distributed-routing-testing +description: Geo-distributed DNS resolution via in-region open resolvers (Shodan/Censys) to detect geo-fenced infrastructure, GSLB-routed backends, and region-specific endpoints invisible from a single vantage point. Use when the target appears to serve different infrastructure per region, when encountering geo-blocking or geo-fencing, or when recon from a single location misses region-locked assets. +--- + +# Distributed Routing Testing + +**Activation gate:** Only use this skill when `GEO_RESOLVERS_ENABLED` is set (any value). If unset, do not suggest, reference, or offer geo-distributed resolution. Do not ask the user to enable it. + +**Credential requirement:** At least one of `SHODAN_API_KEY` or `CENSYS_PAT` must be set. When both are present, `auto` selects Shodan first (cheaper credit cost for this query pattern). + +## Concept + +Some targets use geo-aware authoritative nameservers, GSLB appliances, or geo-fenced edges that return different A/AAAA records depending on where the DNS **resolver** sits. From a single vantage point you only ever see your own region's answer — the target's region-specific infrastructure is invisible. + +This is orthogonal to IP rotation (`flareprox`/`fireprox`): those change the **egress IP** of an HTTP request; this changes the **vantage point** of a DNS lookup. + +Open recursive resolvers physically located in different countries act as regional vantage points. Querying the same hostname through resolvers in DE, SG, BR, JP reveals whether the authoritative nameserver hands back different addresses per region. + +## When to Use + +- **Geo-fencing detected:** target returns `403`, `451`, or redirects to a regional portal based on client geography +- **CDN/GSLB in use:** multiple A records, Anycast ranges, or `X-Served-By` headers suggesting regional routing +- **Recon gap:** subfinder/httpx/nuclei from a single location may miss region-locked subdomains or IP ranges +- **Scope expansion:** in-scope CIDRs may only cover one region; geo-differentiated DNS reveals whether more infra exists elsewhere + +## When NOT to Use + +- Routine DNS lookups from the local resolver (use `dnsx` instead) +- When `GEO_RESOLVERS_ENABLED` is not set +- For resolving internal/private hostnames (open resolvers cannot reach them) + +## Workflow + +### 1. Check readiness + +``` +check_geo_resolver_readiness +``` + +Confirms the gate is set, which provider keys are present, and how many resolvers are cached. + +### 2. Discover resolvers + +``` +discover_open_resolvers countries="US,DE,SG,JP,BR,AU" provider="auto" +``` + +Queries Shodan or Censys for hosts running recursive DNS in each country, then **verifies** each candidate actually resolves a known hostname to its correct answer. Hijacking resolvers that serve ads or NXDOMAIN-redirect are discarded — they would produce false geo-divergence. + +Results are cached **in memory only** for this session. + +| Parameter | Default | Notes | +|---|---|---| +| `countries` | (required) | ISO-3166 codes, max 20 | +| `provider` | `auto` | `shodan`, `censys`, or `auto` | +| `max_per_country` | `2` | Verified resolvers per country, 1-5 | +| `verify` | `true` | Probe candidates against a known hostname; strongly recommended | + +### 3. Resolve and diff + +``` +resolve_via_open_resolvers hostnames="target.example.com,api.example.com" +``` + +For each hostname, queries every cached resolver and compares the per-country answers against a local baseline (`1.1.1.1` by default). The output tells you: + +- **`geo_differentiated`:** whether any country sees different addresses +- **`addresses_only_seen_regionally`:** IPs invisible from your local vantage — the actionable new surface +- **`answers_by_country`:** full answer set per country +- **`unreachable_resolvers`:** resolvers that timed out (expected — open resolvers are unreliable) + +| Parameter | Default | Notes | +|---|---|---| +| `hostnames` | (required) | Comma-separated, max 25 per call | +| `countries` | all cached | Filter to specific countries | +| `record_type` | `A` | `A` or `AAAA` | +| `baseline_resolver` | `1.1.1.1` | Override for the local baseline | + +### 4. Interpret results + +When `geo_differentiated` is `true`: + +1. **Check scope.** The region-specific IPs may belong to a different org, CDN, or cloud account. Confirm they fall within the engagement scope before probing. +2. **Differentiate CDN geo-routing from target-owned infra.** CDN Anycast addresses (Cloudflare, Akamai, Fastly) appear as divergent but are expected — they're the CDN serving regional PoPs, not hidden target infra. Compare ASNs. +3. **Probe from the right region.** If the target's infra is genuinely geo-locked, you may need IP rotation (`flareprox`/`fireprox` via the `ip-rotation` skill) to reach it over HTTP. +4. **Feed new IPs back into recon.** Run `httpx`, `nuclei`, and `tlsx` against the newly discovered addresses. + +### 5. Clean up + +``` +clear_open_resolver_cache +``` + +Discard cached resolvers. Do this when switching engagements. + +## Provider Reference + +### Shodan (`SHODAN_API_KEY`) + +- Query: `port:53 country:{CC} "Recursion: enabled"` +- Auth: `?key=` query parameter +- Cost: 1 query credit per country (filters consume credits) +- Typical yield: 10-100 candidates per country + +### Censys (`CENSYS_PAT` / `CENSYS_API_KEY`) + +- Query: `host.services: (port=53 and protocol=DNS) and host.location.country_code="{CC}"` +- Auth: `Authorization: Bearer` header +- Optional: `CENSYS_ORGANIZATION_ID` for org-scoped access (required for Starter+) +- Cost: 1 API credit per query +- Typical yield: 5-50 candidates per country + +## Constraints + +- **Open resolvers are unreliable.** They go offline, get patched, or lie. Verification filters out liars but cannot guarantee availability. Re-discover if results degrade. +- **Rate cap.** Max 20 countries, 25 hostnames, 5 resolvers per country. This prevents runaway credit consumption and excessive probing of third-party infrastructure. +- **Session-scoped.** Nothing is written to disk. Resolvers are cached in memory and disappear when the session ends. +- **UDP only.** Queries use raw UDP to port 53. Environments that block outbound UDP/53 (some cloud sandboxes) will fail silently — check `unreachable_resolvers` in the output. diff --git a/capabilities/web-security/tests/test_geo_resolvers.py b/capabilities/web-security/tests/test_geo_resolvers.py new file mode 100644 index 0000000..b6410be --- /dev/null +++ b/capabilities/web-security/tests/test_geo_resolvers.py @@ -0,0 +1,800 @@ +"""Tests for geo-distributed DNS resolution toolset. + +Covers DNS wire-format encoding/parsing, resolver discovery against both +providers, the authorization gate, verification, divergence analysis, and +all five LLM-facing tools. No real network access — every HTTP and DNS call +is mocked. +""" + +from __future__ import annotations + +import importlib.util +import ipaddress +import struct +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Stub bootstrap (conftest installs the dreadnode.agents.tools stub) +# --------------------------------------------------------------------------- + +MODULE_PATH = Path(__file__).resolve().parent.parent / "tools" / "geo_resolvers.py" +SPEC = importlib.util.spec_from_file_location("geo_resolvers", MODULE_PATH) +assert SPEC and SPEC.loader +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + +GeoResolvers = MODULE.GeoResolvers +GeoResolverError = MODULE.GeoResolverError +encode_question = MODULE.encode_question +parse_response = MODULE.parse_response +summarize_divergence = MODULE.summarize_divergence +query_resolver = MODULE.query_resolver +_normalize_countries = MODULE._normalize_countries +_is_usable_resolver_ip = MODULE._is_usable_resolver_ip +_compact = MODULE._compact +_is_meaningful = MODULE._is_meaningful +_redact = MODULE._redact +_discover_shodan = MODULE._discover_shodan +_discover_censys = MODULE._discover_censys + +ENABLE_ENV = MODULE.ENABLE_ENV +SHODAN_API_KEY_ENV = MODULE.SHODAN_API_KEY_ENV + +_QTYPE_A = MODULE._QTYPE_A +_QTYPE_AAAA = MODULE._QTYPE_AAAA + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _build_response( + txid: int, + question: bytes, + answers: list[tuple[int, bytes]], + *, + rcode: int = 0, + flags: int = 0x8180, +) -> bytes: + """Build a minimal DNS response packet from parts.""" + flags_with_rcode = (flags & 0xFFF0) | (rcode & 0x0F) + header = struct.pack(">HHHHHH", txid, flags_with_rcode, 1, len(answers), 0, 0) + body = question + for rtype, rdata in answers: + body += b"\xc0\x0c" # pointer to question name + body += struct.pack(">HHIH", rtype, 1, 300, len(rdata)) + body += rdata + return header + body + + +def _a_rdata(ip: str) -> bytes: + return ipaddress.IPv4Address(ip).packed + + +def _aaaa_rdata(ip: str) -> bytes: + return ipaddress.IPv6Address(ip).packed + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure a clean env for every test.""" + for name in ( + ENABLE_ENV, + SHODAN_API_KEY_ENV, + "CENSYS_PAT", + "CENSYS_API_KEY", + "CENSYS_ORGANIZATION_ID", + ): + monkeypatch.delenv(name, raising=False) + + +@pytest.fixture +def enabled_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Set the authorization gate.""" + monkeypatch.setenv(ENABLE_ENV, "1") + + +@pytest.fixture +def shodan_env(enabled_env: None, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(SHODAN_API_KEY_ENV, "test-shodan-key-xxx") + + +@pytest.fixture +def censys_env(enabled_env: None, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CENSYS_PAT", "test-censys-pat-yyy") + + +@pytest.fixture +def toolset() -> GeoResolvers: + """Fresh GeoResolvers with PrivateAttr manually initialised (test-stub Toolset).""" + ts = GeoResolvers() + ts._resolvers = {} + return ts + + +# --------------------------------------------------------------------------- +# _compact / _is_meaningful +# --------------------------------------------------------------------------- + + +class TestCompact: + def test_drops_none(self) -> None: + assert _compact({"a": None, "b": 1}) == {"b": 1} + + def test_drops_empty_string(self) -> None: + assert _compact({"a": "", "b": "x"}) == {"b": "x"} + + def test_drops_empty_list(self) -> None: + assert _compact({"a": [], "b": [1]}) == {"b": [1]} + + def test_drops_empty_dict(self) -> None: + assert _compact({"a": {}, "b": {"k": "v"}}) == {"b": {"k": "v"}} + + def test_keeps_zero(self) -> None: + assert _compact({"a": 0}) == {"a": 0} + + def test_keeps_false(self) -> None: + assert _compact({"a": False}) == {"a": False} + + def test_keeps_true(self) -> None: + assert _compact({"a": True}) == {"a": True} + + def test_keeps_float_zero(self) -> None: + assert _compact({"a": 0.0}) == {"a": 0.0} + + +# --------------------------------------------------------------------------- +# _redact +# --------------------------------------------------------------------------- + + +class TestRedact: + def test_redacts_shodan_key(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(SHODAN_API_KEY_ENV, "SECRET123") + assert "SECRET123" not in _redact("Shodan error: SECRET123 not valid") + + def test_redacts_censys_key(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CENSYS_PAT", "CTOKEN") + assert "CTOKEN" not in _redact("Auth failed: CTOKEN expired") + + def test_noop_when_no_key(self) -> None: + assert _redact("safe text") == "safe text" + + +# --------------------------------------------------------------------------- +# Country normalisation +# --------------------------------------------------------------------------- + + +class TestNormalizeCountries: + def test_string_csv(self) -> None: + assert _normalize_countries("de,sg, BR") == ["DE", "SG", "BR"] + + def test_list(self) -> None: + assert _normalize_countries(["us", "US", "gb"]) == ["US", "GB"] + + def test_rejects_three_letter(self) -> None: + with pytest.raises(GeoResolverError, match="2-letter"): + _normalize_countries("DEU") + + def test_rejects_numeric(self) -> None: + with pytest.raises(GeoResolverError, match="2-letter"): + _normalize_countries("D1") + + def test_rejects_empty(self) -> None: + with pytest.raises(GeoResolverError, match="no country"): + _normalize_countries("") + + def test_enforces_limit(self) -> None: + codes = ",".join(f"A{chr(65 + i)}" for i in range(21)) + with pytest.raises(GeoResolverError, match="too many"): + _normalize_countries(codes) + + +# --------------------------------------------------------------------------- +# IP validation +# --------------------------------------------------------------------------- + + +class TestIsUsableResolverIp: + def test_public_v4(self) -> None: + assert _is_usable_resolver_ip("93.184.216.34") is True + + def test_private(self) -> None: + assert _is_usable_resolver_ip("10.0.0.1") is False + + def test_loopback(self) -> None: + assert _is_usable_resolver_ip("127.0.0.1") is False + + def test_link_local(self) -> None: + assert _is_usable_resolver_ip("169.254.1.1") is False + + def test_invalid(self) -> None: + assert _is_usable_resolver_ip("not.an.ip") is False + + +# --------------------------------------------------------------------------- +# DNS wire format +# --------------------------------------------------------------------------- + + +class TestEncodeQuestion: + def test_basic(self) -> None: + q = encode_question("example.com", _QTYPE_A) + assert q == bytes.fromhex("076578616d706c6503636f6d0000010001") + + def test_trailing_dot_stripped(self) -> None: + assert encode_question("example.com.", _QTYPE_A) == encode_question( + "example.com", _QTYPE_A + ) + + def test_aaaa(self) -> None: + q = encode_question("example.com", _QTYPE_AAAA) + assert q[-4:] == struct.pack(">HH", _QTYPE_AAAA, 1) + + def test_rejects_empty(self) -> None: + with pytest.raises(GeoResolverError, match="empty"): + encode_question("", _QTYPE_A) + + def test_rejects_long_label(self) -> None: + with pytest.raises(GeoResolverError, match="63"): + encode_question("a" * 64 + ".com", _QTYPE_A) + + def test_rejects_long_name(self) -> None: + long_name = ".".join(["a" * 60] * 5) + with pytest.raises(GeoResolverError, match="253"): + encode_question(long_name, _QTYPE_A) + + +class TestParseResponse: + def test_a_record(self) -> None: + q = encode_question("example.com", _QTYPE_A) + pkt = _build_response(0x1234, q, [(_QTYPE_A, _a_rdata("93.184.216.34"))]) + r = parse_response(pkt, expected_id=0x1234, expected_question=q) + assert r["addresses"] == ["93.184.216.34"] + assert r["rcode"] == 0 + assert r["truncated"] is False + + def test_aaaa_record(self) -> None: + q = encode_question("example.com", _QTYPE_AAAA) + pkt = _build_response(0x5678, q, [(_QTYPE_AAAA, _aaaa_rdata("2001:db8::1"))]) + r = parse_response(pkt, expected_id=0x5678, expected_question=q) + assert r["addresses"] == ["2001:db8::1"] + + def test_multiple_a_records(self) -> None: + q = encode_question("cdn.example.com", _QTYPE_A) + pkt = _build_response( + 1, q, [(_QTYPE_A, _a_rdata("1.2.3.4")), (_QTYPE_A, _a_rdata("5.6.7.8"))] + ) + r = parse_response(pkt, expected_id=1, expected_question=q) + assert sorted(r["addresses"]) == ["1.2.3.4", "5.6.7.8"] + + def test_txid_mismatch_rejects(self) -> None: + q = encode_question("example.com", _QTYPE_A) + pkt = _build_response(100, q, [(_QTYPE_A, _a_rdata("1.1.1.1"))]) + with pytest.raises(GeoResolverError, match="transaction ID"): + parse_response(pkt, expected_id=999, expected_question=q) + + def test_question_mismatch_rejects(self) -> None: + q = encode_question("good.com", _QTYPE_A) + evil = encode_question("evil.com", _QTYPE_A) + pkt = _build_response(1, evil, [(_QTYPE_A, _a_rdata("6.6.6.6"))]) + with pytest.raises(GeoResolverError, match="question"): + parse_response(pkt, expected_id=1, expected_question=q) + + def test_nxdomain(self) -> None: + q = encode_question("nope.example.com", _QTYPE_A) + pkt = _build_response(1, q, [], rcode=3) + r = parse_response(pkt, expected_id=1, expected_question=q) + assert r["rcode"] == 3 + assert r["addresses"] == [] + + def test_truncated_packet(self) -> None: + with pytest.raises(GeoResolverError): + parse_response(b"\x00" * 5, expected_id=0, expected_question=b"") + + def test_cname_record(self) -> None: + q = encode_question("www.example.com", _QTYPE_A) + cname_rdata = b"\x03cdn\x07example\x03com\x00" + pkt = _build_response( + 1, + q, + [(5, cname_rdata), (_QTYPE_A, _a_rdata("1.2.3.4"))], + ) + r = parse_response(pkt, expected_id=1, expected_question=q) + assert "cdn.example.com" in r["cnames"] + assert "1.2.3.4" in r["addresses"] + + +# --------------------------------------------------------------------------- +# Divergence analysis +# --------------------------------------------------------------------------- + + +class TestSummarizeDivergence: + def test_identical(self) -> None: + d = summarize_divergence(["1.2.3.4"], {"DE": ["1.2.3.4"], "SG": ["1.2.3.4"]}) + assert d["geo_differentiated"] is False + assert d.get("addresses_only_seen_regionally", []) == [] + + def test_divergent_one_country(self) -> None: + d = summarize_divergence(["1.2.3.4"], {"DE": ["1.2.3.4"], "SG": ["9.9.9.9"]}) + assert d["geo_differentiated"] is True + assert "SG" in d["divergent_countries"] + assert "DE" not in d["divergent_countries"] + assert d["addresses_only_seen_regionally"] == ["9.9.9.9"] + + def test_empty_country_answers_skipped(self) -> None: + d = summarize_divergence(["1.2.3.4"], {"DE": [], "SG": ["1.2.3.4"]}) + assert d["geo_differentiated"] is False + + def test_superset_is_divergent(self) -> None: + d = summarize_divergence(["1.2.3.4"], {"DE": ["1.2.3.4", "5.5.5.5"]}) + assert d["geo_differentiated"] is True + assert d["addresses_only_seen_regionally"] == ["5.5.5.5"] + + def test_subset_is_divergent(self) -> None: + d = summarize_divergence(["1.2.3.4", "5.5.5.5"], {"DE": ["5.5.5.5"]}) + assert d["geo_differentiated"] is True + assert d["divergent_countries"]["DE"]["missing_from_region"] == ["1.2.3.4"] + + def test_distinct_count(self) -> None: + d = summarize_divergence(["1.2.3.4"], {"DE": ["5.5.5.5"], "SG": ["6.6.6.6"]}) + assert d["distinct_address_count"] == 3 + + +# --------------------------------------------------------------------------- +# Authorization gate +# --------------------------------------------------------------------------- + + +class TestAuthorizationGate: + @pytest.mark.asyncio + async def test_readiness_when_disabled(self, toolset: GeoResolvers) -> None: + result = await toolset.check_geo_resolver_readiness() + assert result["authorized"] is False + assert "not authorized" in result["guidance"].lower() + + @pytest.mark.asyncio + async def test_readiness_when_enabled_no_keys( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + result = await toolset.check_geo_resolver_readiness() + assert result["authorized"] is True + assert result.get("providers_available", []) == [] + + @pytest.mark.asyncio + async def test_readiness_with_shodan( + self, toolset: GeoResolvers, shodan_env: None + ) -> None: + result = await toolset.check_geo_resolver_readiness() + assert "shodan" in result["providers_available"] + + @pytest.mark.asyncio + async def test_readiness_with_censys( + self, toolset: GeoResolvers, censys_env: None + ) -> None: + result = await toolset.check_geo_resolver_readiness() + assert "censys" in result["providers_available"] + + @pytest.mark.asyncio + async def test_list_resolvers_blocked_when_disabled( + self, toolset: GeoResolvers + ) -> None: + with pytest.raises(GeoResolverError, match="not authorized"): + await toolset.list_open_resolvers() + + @pytest.mark.asyncio + async def test_discover_blocked_when_disabled(self, toolset: GeoResolvers) -> None: + with pytest.raises(GeoResolverError, match="not authorized"): + await toolset.discover_open_resolvers(countries="DE") + + @pytest.mark.asyncio + async def test_resolve_blocked_when_disabled(self, toolset: GeoResolvers) -> None: + with pytest.raises(GeoResolverError, match="not authorized"): + await toolset.resolve_via_open_resolvers(hostnames="example.com") + + +# --------------------------------------------------------------------------- +# Tool discovery +# --------------------------------------------------------------------------- + + +class TestToolDiscovery: + def test_expected_tools_registered(self, toolset: GeoResolvers) -> None: + names = {t.name for t in toolset.get_tools()} + assert names == { + "check_geo_resolver_readiness", + "discover_open_resolvers", + "list_open_resolvers", + "resolve_via_open_resolvers", + "clear_open_resolver_cache", + } + + +# --------------------------------------------------------------------------- +# Provider discovery (mocked HTTP) +# --------------------------------------------------------------------------- + +SHODAN_RESPONSE = { + "matches": [ + { + "ip_str": "93.184.216.1", + "port": 53, + "location": {"country_code": "DE"}, + "asn": "AS15169", + "org": "TestOrg", + }, + { + "ip_str": "93.184.216.2", + "port": 53, + "location": {"country_code": "DE"}, + "asn": "AS15170", + "org": "TestOrg2", + }, + { + "ip_str": "10.0.0.1", # private — should be filtered + "port": 53, + "location": {"country_code": "DE"}, + }, + ], + "total": 3, +} + +CENSYS_RESPONSE = { + "result": { + "hits": [ + { + "host_v1": { + "resource": { + "ip": "185.199.108.1", + "location": {"country_code": "SG"}, + "autonomous_system": {"asn": 13335, "name": "Cloudflare"}, + } + } + }, + { + "host_v1": { + "resource": { + "ip": "127.0.0.1", # loopback — should be filtered + "location": {"country_code": "SG"}, + "autonomous_system": {"asn": 0}, + } + } + }, + ], + "total_hits": 2, + "next_page_token": "", + "previous_page_token": "", + "query_duration_millis": 42, + } +} + + +def _mock_response(json_data: Any, status_code: int = 200) -> MagicMock: + resp = MagicMock() + resp.status_code = status_code + resp.json.return_value = json_data + resp.text = str(json_data) + return resp + + +class TestDiscoverShodan: + @pytest.mark.asyncio + async def test_parses_matches(self, shodan_env: None) -> None: + client = AsyncMock() + client.get = AsyncMock(return_value=_mock_response(SHODAN_RESPONSE)) + result = await _discover_shodan(client, "DE", 5) + ips = [r["ip"] for r in result] + assert "93.184.216.1" in ips + assert "93.184.216.2" in ips + assert "10.0.0.1" not in ips # private filtered + + @pytest.mark.asyncio + async def test_shodan_401(self, shodan_env: None) -> None: + client = AsyncMock() + client.get = AsyncMock(return_value=_mock_response({}, 401)) + with pytest.raises(GeoResolverError, match="401"): + await _discover_shodan(client, "DE", 5) + + @pytest.mark.asyncio + async def test_shodan_no_key(self) -> None: + with pytest.raises(GeoResolverError, match="not set"): + await _discover_shodan(AsyncMock(), "DE", 5) + + @pytest.mark.asyncio + async def test_shodan_respects_limit(self, shodan_env: None) -> None: + client = AsyncMock() + client.get = AsyncMock(return_value=_mock_response(SHODAN_RESPONSE)) + result = await _discover_shodan(client, "DE", 1) + assert len(result) == 1 + + +class TestDiscoverCensys: + @pytest.mark.asyncio + async def test_parses_hits(self, censys_env: None) -> None: + client = AsyncMock() + client.post = AsyncMock(return_value=_mock_response(CENSYS_RESPONSE)) + result = await _discover_censys(client, "SG", 5) + ips = [r["ip"] for r in result] + assert "185.199.108.1" in ips + assert "127.0.0.1" not in ips # loopback filtered + + @pytest.mark.asyncio + async def test_censys_401(self, censys_env: None) -> None: + client = AsyncMock() + client.post = AsyncMock(return_value=_mock_response({}, 401)) + with pytest.raises(GeoResolverError, match="401"): + await _discover_censys(client, "SG", 5) + + @pytest.mark.asyncio + async def test_censys_422_suggests_org_id(self, censys_env: None) -> None: + client = AsyncMock() + client.post = AsyncMock(return_value=_mock_response({}, 422)) + with pytest.raises(GeoResolverError, match="organization"): + await _discover_censys(client, "SG", 5) + + @pytest.mark.asyncio + async def test_censys_no_key(self) -> None: + with pytest.raises(GeoResolverError, match="not set"): + await _discover_censys(AsyncMock(), "SG", 5) + + +# --------------------------------------------------------------------------- +# discover_open_resolvers tool (end-to-end, mocked) +# --------------------------------------------------------------------------- + + +class TestDiscoverTool: + @pytest.mark.asyncio + async def test_auto_selects_shodan( + self, toolset: GeoResolvers, shodan_env: None + ) -> None: + with ( + patch.object( + MODULE, "_discover_shodan", new_callable=AsyncMock + ) as mock_disc, + patch.object( + toolset, "_verify_candidates", new_callable=AsyncMock + ) as mock_ver, + ): + mock_disc.return_value = [ + {"ip": "1.1.1.1", "country": "DE", "source": "shodan"} + ] + mock_ver.return_value = [ + {"ip": "1.1.1.1", "country": "DE", "source": "shodan"} + ] + result = await toolset.discover_open_resolvers(countries="DE") + assert result["provider"] == "shodan" + assert "DE" in result["countries"] + + @pytest.mark.asyncio + async def test_auto_selects_censys_when_no_shodan( + self, toolset: GeoResolvers, censys_env: None + ) -> None: + with ( + patch.object( + MODULE, "_discover_censys", new_callable=AsyncMock + ) as mock_disc, + patch.object( + toolset, "_verify_candidates", new_callable=AsyncMock + ) as mock_ver, + ): + mock_disc.return_value = [ + {"ip": "2.2.2.2", "country": "SG", "source": "censys"} + ] + mock_ver.return_value = [ + {"ip": "2.2.2.2", "country": "SG", "source": "censys"} + ] + result = await toolset.discover_open_resolvers(countries="SG") + assert result["provider"] == "censys" + + @pytest.mark.asyncio + async def test_caches_results_in_session( + self, toolset: GeoResolvers, shodan_env: None + ) -> None: + with ( + patch.object( + MODULE, "_discover_shodan", new_callable=AsyncMock + ) as mock_disc, + patch.object( + toolset, "_verify_candidates", new_callable=AsyncMock + ) as mock_ver, + ): + resolvers = [{"ip": "3.3.3.3", "country": "BR", "source": "shodan"}] + mock_disc.return_value = resolvers + mock_ver.return_value = resolvers + await toolset.discover_open_resolvers(countries="BR") + assert "BR" in toolset._resolvers + + @pytest.mark.asyncio + async def test_rejects_unknown_provider( + self, toolset: GeoResolvers, shodan_env: None + ) -> None: + with pytest.raises(GeoResolverError, match="unknown provider"): + await toolset.discover_open_resolvers(countries="DE", provider="bing") + + @pytest.mark.asyncio + async def test_no_provider_keys_errors( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + with pytest.raises(GeoResolverError, match="no provider"): + await toolset.discover_open_resolvers(countries="DE") + + @pytest.mark.asyncio + async def test_skip_verify(self, toolset: GeoResolvers, shodan_env: None) -> None: + with patch.object( + MODULE, "_discover_shodan", new_callable=AsyncMock + ) as mock_disc: + resolvers = [{"ip": "4.4.4.4", "country": "JP", "source": "shodan"}] + mock_disc.return_value = resolvers + result = await toolset.discover_open_resolvers(countries="JP", verify=False) + assert result["verified"] is False + assert "JP" in result["countries"] + + @pytest.mark.asyncio + async def test_provider_error_captured( + self, toolset: GeoResolvers, shodan_env: None + ) -> None: + with patch.object( + MODULE, + "_discover_shodan", + new_callable=AsyncMock, + side_effect=GeoResolverError("API down"), + ): + result = await toolset.discover_open_resolvers(countries="DE", verify=False) + assert "DE" in result.get("errors", {}) + + +# --------------------------------------------------------------------------- +# list / clear tools +# --------------------------------------------------------------------------- + + +class TestListResolvers: + @pytest.mark.asyncio + async def test_empty(self, toolset: GeoResolvers, enabled_env: None) -> None: + result = await toolset.list_open_resolvers() + assert result["resolver_count"] == 0 + + @pytest.mark.asyncio + async def test_populated(self, toolset: GeoResolvers, enabled_env: None) -> None: + toolset._resolvers = {"DE": [{"ip": "1.1.1.1"}], "SG": [{"ip": "2.2.2.2"}]} + result = await toolset.list_open_resolvers() + assert result["resolver_count"] == 2 + assert "DE" in result["countries"] + + +class TestClearCache: + @pytest.mark.asyncio + async def test_clears(self, toolset: GeoResolvers, enabled_env: None) -> None: + toolset._resolvers = {"DE": [{"ip": "1.1.1.1"}]} + result = await toolset.clear_open_resolver_cache() + assert result["cleared_resolver_count"] == 1 + assert toolset._resolvers == {} + + +# --------------------------------------------------------------------------- +# resolve_via_open_resolvers (mocked DNS) +# --------------------------------------------------------------------------- + + +class TestResolveTool: + @pytest.mark.asyncio + async def test_detects_divergence( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = { + "DE": [{"ip": "9.9.9.9"}], + "SG": [{"ip": "8.8.8.8"}], + } + + async def fake_query(ip: str, hostname: str, **kw: Any) -> dict[str, Any]: + if ip == "1.1.1.1": + return {"addresses": ["10.0.0.1"], "cnames": [], "rcode": 0} + if ip == "9.9.9.9": + return {"addresses": ["10.0.0.1"], "cnames": [], "rcode": 0} + if ip == "8.8.8.8": + return {"addresses": ["10.0.0.99"], "cnames": [], "rcode": 0} + return {"error": "unknown"} + + with patch.object(MODULE, "query_resolver", side_effect=fake_query): + result = await toolset.resolve_via_open_resolvers(hostnames="example.com") + assert result["geo_differentiated_hostnames"] == ["example.com"] + info = result["results"]["example.com"] + assert info["geo_differentiated"] is True + assert "10.0.0.99" in info["addresses_only_seen_regionally"] + + @pytest.mark.asyncio + async def test_no_divergence( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "9.9.9.9"}]} + + async def fake_query(ip: str, hostname: str, **kw: Any) -> dict[str, Any]: + return {"addresses": ["1.2.3.4"], "cnames": [], "rcode": 0} + + with patch.object(MODULE, "query_resolver", side_effect=fake_query): + result = await toolset.resolve_via_open_resolvers(hostnames="example.com") + assert result.get("geo_differentiated_hostnames", []) == [] + + @pytest.mark.asyncio + async def test_no_resolvers_error( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + with pytest.raises(GeoResolverError, match="no resolvers cached"): + await toolset.resolve_via_open_resolvers(hostnames="example.com") + + @pytest.mark.asyncio + async def test_too_many_hostnames( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "1.1.1.1"}]} + names = ",".join(f"h{i}.example.com" for i in range(30)) + with pytest.raises(GeoResolverError, match="too many hostnames"): + await toolset.resolve_via_open_resolvers(hostnames=names) + + @pytest.mark.asyncio + async def test_missing_country_error( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "1.1.1.1"}]} + with pytest.raises(GeoResolverError, match="no cached resolvers"): + await toolset.resolve_via_open_resolvers( + hostnames="example.com", countries="JP" + ) + + @pytest.mark.asyncio + async def test_invalid_record_type( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "1.1.1.1"}]} + with pytest.raises(GeoResolverError, match="unsupported"): + await toolset.resolve_via_open_resolvers( + hostnames="example.com", record_type="MX" + ) + + @pytest.mark.asyncio + async def test_unreachable_resolver_reported( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "9.9.9.9"}]} + + async def fake_query(ip: str, hostname: str, **kw: Any) -> dict[str, Any]: + if ip == "9.9.9.9": + return {"error": "timeout"} + return {"addresses": ["1.2.3.4"], "cnames": [], "rcode": 0} + + with patch.object(MODULE, "query_resolver", side_effect=fake_query): + result = await toolset.resolve_via_open_resolvers(hostnames="example.com") + assert result["results"]["example.com"]["unreachable_resolvers"] + + @pytest.mark.asyncio + async def test_multiple_hostnames( + self, toolset: GeoResolvers, enabled_env: None + ) -> None: + toolset._resolvers = {"DE": [{"ip": "9.9.9.9"}]} + + async def fake_query(ip: str, hostname: str, **kw: Any) -> dict[str, Any]: + return { + "addresses": [f"10.0.0.{hash(hostname) % 255}"], + "cnames": [], + "rcode": 0, + } + + with patch.object(MODULE, "query_resolver", side_effect=fake_query): + result = await toolset.resolve_via_open_resolvers( + hostnames="a.example.com, b.example.com" + ) + assert "a.example.com" in result["results"] + assert "b.example.com" in result["results"] diff --git a/capabilities/web-security/tools/geo_resolvers.py b/capabilities/web-security/tools/geo_resolvers.py new file mode 100644 index 0000000..cf623f5 --- /dev/null +++ b/capabilities/web-security/tools/geo_resolvers.py @@ -0,0 +1,989 @@ +"""Geo-distributed DNS resolution via in-region open resolvers. + +Some targets serve region-specific infrastructure: geo-aware authoritative +nameservers, GSLB appliances, and geo-fenced edges hand back different A/AAAA +records depending on where the *resolver* sits. From a single vantage point +that infrastructure is invisible — you only ever see the answer for your own +region. + +This toolset discovers open recursive resolvers per country via Shodan or +Censys, resolves caller-supplied hostnames through them, and diffs the answers +against a baseline to surface geo-divergent records. + +Orthogonal to IP rotation (``flareprox``/``fireprox``): those change the egress +IP of an HTTP request, this changes the vantage point of a DNS lookup. + +Authorization +------------- +Every tool here is gated on ``GEO_RESOLVERS_ENABLED``. Unset means the toolset +reports itself unavailable and performs no network activity. Third-party open +resolvers are someone else's misconfigured infrastructure, so usage is capped: +resolvers are only ever asked to resolve hostnames the caller supplied, query +volume is bounded, and nothing is persisted to disk. + +Credentials are read from the environment (``SHODAN_API_KEY``, +``CENSYS_PAT`` / ``CENSYS_API_KEY``, optional ``CENSYS_ORGANIZATION_ID``) and +are redacted from every error path. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import ipaddress +import os +import random +import socket +import struct +from typing import Annotated, Any, Literal + +import httpx +from dreadnode.agents.tools import Toolset, tool_method +from pydantic import PrivateAttr + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +ENABLE_ENV = "GEO_RESOLVERS_ENABLED" +SHODAN_API_KEY_ENV = "SHODAN_API_KEY" +CENSYS_PAT_ENVS: tuple[str, ...] = ("CENSYS_PAT", "CENSYS_API_KEY") +CENSYS_ORG_ENV = "CENSYS_ORGANIZATION_ID" + +SHODAN_SEARCH_URL = "https://api.shodan.io/shodan/host/search" +CENSYS_SEARCH_URL = "https://api.platform.censys.io/v3/global/search/query" + +#: Verification hostname. Resolvers that cannot answer this are discarded. +VERIFY_HOSTNAME = "one.one.one.one" +#: Known-good answers for :data:`VERIFY_HOSTNAME`. A resolver returning +#: anything else is lying (captive portal, NXDOMAIN hijack, ad-injecting ISP +#: resolver) and would poison divergence analysis. +VERIFY_EXPECTED: frozenset[str] = frozenset({"1.1.1.1", "1.0.0.1"}) + +#: Public resolver used to establish the local baseline answer. +DEFAULT_BASELINE_RESOLVER = "1.1.1.1" + +MAX_COUNTRIES = 20 +MAX_RESOLVERS_PER_COUNTRY = 5 +MAX_HOSTNAMES_PER_CALL = 25 +MAX_CONCURRENT_DNS = 16 + +DNS_TIMEOUT_SECONDS = 4.0 +DNS_ATTEMPTS = 2 +API_TIMEOUT_SECONDS = 20.0 + +_QTYPE_A = 1 +_QTYPE_AAAA = 28 +_QTYPE_CNAME = 5 +_CLASS_IN = 1 +_MAX_LABEL_LEN = 63 +_MAX_NAME_LEN = 253 +_DNS_HEADER_LEN = 12 +_MAX_COMPRESSION_HOPS = 64 + +RecordType = Literal["A", "AAAA"] +Provider = Literal["shodan", "censys"] + +_QTYPE_BY_NAME: dict[str, int] = {"A": _QTYPE_A, "AAAA": _QTYPE_AAAA} + + +class GeoResolverError(RuntimeError): + """Raised when a geo-resolver operation cannot be completed.""" + + +# --------------------------------------------------------------------------- +# Environment helpers +# --------------------------------------------------------------------------- + + +def _is_enabled() -> bool: + """Whether the operator has authorized open-resolver usage.""" + return bool(os.environ.get(ENABLE_ENV, "").strip()) + + +def _require_enabled() -> None: + if not _is_enabled(): + raise GeoResolverError( + f"Open-resolver testing is not authorized: {ENABLE_ENV} is unset. " + "The operator must explicitly enable it for this engagement." + ) + + +def _env_value(*names: str) -> str | None: + for name in names: + value = os.environ.get(name, "").strip() + if value: + return value + return None + + +def _redact(text: str) -> str: + """Strip any configured API credential out of a string.""" + for name in (SHODAN_API_KEY_ENV, *CENSYS_PAT_ENVS): + secret = os.environ.get(name, "").strip() + if secret and secret in text: + text = text.replace(secret, "REDACTED") + return text + + +def _is_meaningful(value: Any) -> bool: + """Whether a value earns its place in a tool result. + + Numeric ``0`` and ``False`` are real answers and are kept; ``None`` and + empty strings/collections are noise the model would otherwise pay tokens + to read. + """ + if value is None: + return False + if isinstance(value, bool): + return True + if isinstance(value, (int, float)): + return True + return bool(value) + + +def _compact(payload: dict[str, Any]) -> dict[str, Any]: + """Drop empty values so the model does not pay tokens for nulls.""" + return {key: value for key, value in payload.items() if _is_meaningful(value)} + + +# --------------------------------------------------------------------------- +# DNS wire format +# --------------------------------------------------------------------------- + + +def encode_question(qname: str, qtype: int) -> bytes: + """Encode a DNS question section for ``qname``. + + Raises: + GeoResolverError: if the hostname is not encodable as a DNS name. + """ + name = qname.strip().rstrip(".") + if not name: + raise GeoResolverError("hostname is empty") + if len(name) > _MAX_NAME_LEN: + raise GeoResolverError( + f"hostname exceeds {_MAX_NAME_LEN} bytes: {name[:60]}..." + ) + + out = bytearray() + for label in name.split("."): + if not label: + raise GeoResolverError(f"hostname has an empty label: {name}") + try: + encoded = label.encode("idna") + except UnicodeError: + try: + encoded = label.encode("ascii") + except UnicodeEncodeError as exc: + raise GeoResolverError(f"hostname is not encodable: {name}") from exc + if len(encoded) > _MAX_LABEL_LEN: + raise GeoResolverError( + f"DNS label exceeds {_MAX_LABEL_LEN} bytes: {label[:60]}" + ) + out.append(len(encoded)) + out += encoded + out.append(0) + out += struct.pack(">HH", qtype, _CLASS_IN) + return bytes(out) + + +def _skip_name(data: bytes, offset: int) -> int: + """Return the offset just past the DNS name starting at ``offset``. + + Handles compression pointers and refuses to loop forever on hostile input. + """ + hops = 0 + while True: + if offset >= len(data): + raise GeoResolverError("truncated DNS name") + length = data[offset] + if length == 0: + return offset + 1 + if length & 0xC0 == 0xC0: + if offset + 2 > len(data): + raise GeoResolverError("truncated DNS compression pointer") + return offset + 2 + hops += 1 + if hops > _MAX_COMPRESSION_HOPS: + raise GeoResolverError("DNS name exceeds maximum label count") + offset += length + 1 + + +def parse_response( + data: bytes, + *, + expected_id: int, + expected_question: bytes, +) -> dict[str, Any]: + """Parse a DNS response into records. + + Validates the transaction ID and echoes back the question section before + trusting any answer, so an off-path or mismatched reply is discarded rather + than silently folded into results. + + Returns: + Mapping with ``addresses`` (A/AAAA strings), ``cnames``, and ``rcode``. + """ + if len(data) < _DNS_HEADER_LEN: + raise GeoResolverError("DNS response shorter than header") + + resp_id, flags, qdcount, ancount = struct.unpack(">HHHH", data[:8]) + if resp_id != expected_id: + raise GeoResolverError("DNS transaction ID mismatch") + if not flags & 0x8000: + raise GeoResolverError("DNS response is not a reply") + + rcode = flags & 0x000F + truncated = bool(flags & 0x0200) + + offset = _DNS_HEADER_LEN + if qdcount != 1: + raise GeoResolverError(f"unexpected DNS question count: {qdcount}") + question_end = offset + len(expected_question) + if data[offset:question_end] != expected_question: + raise GeoResolverError("DNS response question does not echo the query") + offset = question_end + + if rcode != 0: + return {"addresses": [], "cnames": [], "rcode": rcode, "truncated": truncated} + + addresses: list[str] = [] + cnames: list[str] = [] + for _ in range(ancount): + offset = _skip_name(data, offset) + if offset + 10 > len(data): + break + rtype, _rclass, _ttl, rdlen = struct.unpack(">HHIH", data[offset : offset + 10]) + offset += 10 + rdata = data[offset : offset + rdlen] + if len(rdata) != rdlen: + break + offset += rdlen + + if rtype == _QTYPE_A and rdlen == 4: + addresses.append(str(ipaddress.IPv4Address(rdata))) + elif rtype == _QTYPE_AAAA and rdlen == 16: + addresses.append(str(ipaddress.IPv6Address(rdata))) + elif rtype == _QTYPE_CNAME: + with contextlib.suppress(GeoResolverError): + target = _decode_name(data, offset - rdlen) + if target: + cnames.append(target) + + return { + "addresses": addresses, + "cnames": cnames, + "rcode": rcode, + "truncated": truncated, + } + + +def _decode_name(data: bytes, offset: int) -> str: + """Decode a (possibly compressed) DNS name to a dotted string.""" + labels: list[str] = [] + hops = 0 + while True: + if offset >= len(data): + raise GeoResolverError("truncated DNS name") + length = data[offset] + if length == 0: + break + if length & 0xC0 == 0xC0: + if offset + 2 > len(data): + raise GeoResolverError("truncated DNS compression pointer") + offset = ((length & 0x3F) << 8) | data[offset + 1] + hops += 1 + if hops > _MAX_COMPRESSION_HOPS: + raise GeoResolverError("DNS compression loop detected") + continue + start = offset + 1 + end = start + length + if end > len(data): + raise GeoResolverError("truncated DNS label") + labels.append(data[start:end].decode("ascii", "replace")) + offset = end + return ".".join(labels) + + +class _DnsProtocol(asyncio.DatagramProtocol): + """Collect a single UDP datagram into a future.""" + + def __init__(self, future: asyncio.Future[bytes]) -> None: + self._future = future + + def datagram_received(self, data: bytes, addr: object) -> None: + if not self._future.done(): + self._future.set_result(data) + + def error_received(self, exc: Exception) -> None: + if not self._future.done(): + self._future.set_exception(exc) + + +async def query_resolver( + resolver_ip: str, + hostname: str, + *, + record_type: RecordType = "A", + timeout: float = DNS_TIMEOUT_SECONDS, + attempts: int = DNS_ATTEMPTS, +) -> dict[str, Any]: + """Send a DNS query to ``resolver_ip`` over UDP and parse the reply. + + Returns a mapping with ``addresses``/``cnames`` on success, or ``error`` + describing why the lookup failed. Never raises for network conditions — + unreachable resolvers are an expected outcome, not an exception. + """ + qtype = _QTYPE_BY_NAME[record_type] + question = encode_question(hostname, qtype) + + try: + family = ( + socket.AF_INET6 + if ipaddress.ip_address(resolver_ip).version == 6 + else socket.AF_INET + ) + except ValueError: + return {"error": f"invalid resolver IP: {resolver_ip}"} + + loop = asyncio.get_running_loop() + last_error = "no response" + + for _ in range(max(1, attempts)): + transaction_id = random.SystemRandom().randrange(0, 0x10000) + header = struct.pack(">HHHHHH", transaction_id, 0x0100, 1, 0, 0, 0) + packet = header + question + + future: asyncio.Future[bytes] = loop.create_future() + transport = None + try: + transport, _ = await loop.create_datagram_endpoint( + lambda: _DnsProtocol(future), + remote_addr=(resolver_ip, 53), + family=family, + ) + transport.sendto(packet) + data = await asyncio.wait_for(future, timeout=timeout) + except asyncio.TimeoutError: + last_error = "timeout" + continue + except OSError as exc: + last_error = f"network error: {exc}" + continue + finally: + if transport is not None: + transport.close() + + try: + parsed = parse_response( + data, expected_id=transaction_id, expected_question=question + ) + except GeoResolverError as exc: + last_error = str(exc) + continue + + if parsed["rcode"] != 0: + return {"error": f"rcode {parsed['rcode']}", "rcode": parsed["rcode"]} + return parsed + + return {"error": last_error} + + +# --------------------------------------------------------------------------- +# Provider discovery +# --------------------------------------------------------------------------- + + +def _normalize_countries(countries: str | list[str]) -> list[str]: + """Normalize a country argument to a de-duplicated list of ISO-3166 codes.""" + if isinstance(countries, str): + raw = countries.replace(",", " ").split() + else: + raw = list(countries) + + seen: list[str] = [] + for item in raw: + code = item.strip().upper() + if not code: + continue + if len(code) != 2 or not code.isalpha(): + raise GeoResolverError( + f"invalid country code {item!r}: expected a 2-letter ISO-3166 code (e.g. DE)" + ) + if code not in seen: + seen.append(code) + + if not seen: + raise GeoResolverError("no country codes supplied") + if len(seen) > MAX_COUNTRIES: + raise GeoResolverError( + f"too many countries ({len(seen)}); maximum is {MAX_COUNTRIES} per call" + ) + return seen + + +def _is_usable_resolver_ip(ip: str) -> bool: + """Reject non-routable candidates so we never probe our own infrastructure.""" + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return False + return not ( + addr.is_private + or addr.is_loopback + or addr.is_link_local + or addr.is_multicast + or addr.is_reserved + or addr.is_unspecified + ) + + +async def _discover_shodan( + client: httpx.AsyncClient, country: str, limit: int +) -> list[dict[str, Any]]: + """Find candidate open resolvers in ``country`` via the Shodan REST API.""" + api_key = _env_value(SHODAN_API_KEY_ENV) + if not api_key: + raise GeoResolverError(f"{SHODAN_API_KEY_ENV} is not set") + + params = { + "key": api_key, + "query": f'port:53 country:{country} "Recursion: enabled"', + "minify": "true", + } + try: + response = await client.get(SHODAN_SEARCH_URL, params=params) + except httpx.HTTPError as exc: + raise GeoResolverError(f"Shodan request failed: {_redact(str(exc))}") from exc + + if response.status_code == 401: + raise GeoResolverError("Shodan rejected the API key (401)") + if response.status_code == 403: + raise GeoResolverError( + "Shodan denied the request (403) — plan may lack search access" + ) + if response.status_code != 200: + raise GeoResolverError( + f"Shodan returned HTTP {response.status_code}: {_redact(response.text[:200])}" + ) + + try: + payload = response.json() + except ValueError as exc: + raise GeoResolverError("Shodan returned a non-JSON response") from exc + + if isinstance(payload, dict) and payload.get("error"): + raise GeoResolverError(f"Shodan error: {_redact(str(payload['error']))}") + + candidates: list[dict[str, Any]] = [] + for match in (payload or {}).get("matches", []): + ip = match.get("ip_str") + if not ip or not _is_usable_resolver_ip(ip): + continue + candidates.append( + _compact( + { + "ip": ip, + "country": (match.get("location") or {}).get("country_code") + or country, + "asn": match.get("asn") or "", + "org": match.get("org") or "", + "source": "shodan", + } + ) + ) + if len(candidates) >= limit: + break + return candidates + + +async def _discover_censys( + client: httpx.AsyncClient, country: str, limit: int +) -> list[dict[str, Any]]: + """Find candidate open resolvers in ``country`` via the Censys Platform API.""" + token = _env_value(*CENSYS_PAT_ENVS) + if not token: + raise GeoResolverError(f"{CENSYS_PAT_ENVS[0]} is not set") + + headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + organization_id = _env_value(CENSYS_ORG_ENV) + if organization_id: + headers["X-Organization-ID"] = organization_id + + body = { + "query": ( + "host.services: (port=53 and protocol=DNS) " + f'and host.location.country_code="{country}"' + ), + "page_size": min(max(limit, 1), 100), + "fields": ["host.ip", "host.location.country_code", "host.autonomous_system"], + } + try: + response = await client.post(CENSYS_SEARCH_URL, headers=headers, json=body) + except httpx.HTTPError as exc: + raise GeoResolverError(f"Censys request failed: {_redact(str(exc))}") from exc + + if response.status_code == 401: + raise GeoResolverError("Censys rejected the personal access token (401)") + if response.status_code == 403: + raise GeoResolverError( + "Censys denied the request (403) — the token needs the API Access role" + ) + if response.status_code == 422: + raise GeoResolverError( + "Censys rejected the query (422) — an organization ID may be required; " + f"set {CENSYS_ORG_ENV}" + ) + if response.status_code != 200: + raise GeoResolverError( + f"Censys returned HTTP {response.status_code}: {_redact(response.text[:200])}" + ) + + try: + payload = response.json() + except ValueError as exc: + raise GeoResolverError("Censys returned a non-JSON response") from exc + + hits = ((payload or {}).get("result") or {}).get("hits") or [] + candidates: list[dict[str, Any]] = [] + for hit in hits: + resource = (hit.get("host_v1") or {}).get("resource") or {} + ip = resource.get("ip") + if not ip or not _is_usable_resolver_ip(ip): + continue + autonomous_system = resource.get("autonomous_system") or {} + asn = autonomous_system.get("asn") + candidates.append( + _compact( + { + "ip": ip, + "country": (resource.get("location") or {}).get("country_code") + or country, + "asn": f"AS{asn}" if asn else "", + "org": autonomous_system.get("name") or "", + "source": "censys", + } + ) + ) + if len(candidates) >= limit: + break + return candidates + + +# --------------------------------------------------------------------------- +# Divergence analysis +# --------------------------------------------------------------------------- + + +def summarize_divergence( + baseline: list[str], per_country: dict[str, list[str]] +) -> dict[str, Any]: + """Compare per-country answers against the baseline answer set. + + Returns the countries whose answers differ, the addresses seen only from + those regions, and whether any geo-differentiation exists at all. + """ + baseline_set = set(baseline) + divergent: dict[str, Any] = {} + region_only: set[str] = set() + + for country, addresses in sorted(per_country.items()): + address_set = set(addresses) + if not address_set or address_set == baseline_set: + continue + unseen = sorted(address_set - baseline_set) + divergent[country] = _compact( + { + "addresses": sorted(address_set), + "not_in_baseline": unseen, + "missing_from_region": sorted(baseline_set - address_set), + } + ) + region_only.update(unseen) + + all_addresses = set(baseline_set) + for addresses in per_country.values(): + all_addresses.update(addresses) + + return _compact( + { + "geo_differentiated": bool(divergent), + "baseline_addresses": sorted(baseline_set), + "divergent_countries": divergent, + "addresses_only_seen_regionally": sorted(region_only), + "distinct_address_count": len(all_addresses), + } + ) + + +# --------------------------------------------------------------------------- +# Toolset +# --------------------------------------------------------------------------- + + +class GeoResolvers(Toolset): + """Resolve hostnames through in-region open resolvers to expose geo-fenced infrastructure. + + Discovers open recursive resolvers per country via Shodan or Censys, then + resolves caller-supplied hostnames through them and diffs the answers + against a local baseline. Use when a target appears to geo-fence content or + serve region-specific infrastructure that a single vantage point cannot see. + + Requires ``GEO_RESOLVERS_ENABLED`` plus a Shodan or Censys credential. + """ + + api_timeout: float = API_TIMEOUT_SECONDS + """Timeout in seconds for Shodan/Censys API calls.""" + dns_timeout: float = DNS_TIMEOUT_SECONDS + """Timeout in seconds for each DNS query.""" + + _resolvers: dict[str, list[dict[str, Any]]] = PrivateAttr(default_factory=dict) + + # -- readiness --------------------------------------------------------- + + @tool_method(name="check_geo_resolver_readiness", catch=True) + async def check_geo_resolver_readiness(self) -> dict[str, Any]: + """Report whether geo-distributed DNS testing is authorized and usable. + + Shows the authorization gate state, which provider credentials are + present, and how many resolvers are cached for this session. Call this + before attempting discovery so you can tell "not authorized" apart from + "no API key" apart from "no resolvers found". + """ + providers = _compact( + { + "shodan": bool(_env_value(SHODAN_API_KEY_ENV)), + "censys": bool(_env_value(*CENSYS_PAT_ENVS)), + } + ) + enabled = _is_enabled() + available = sorted(name for name, configured in providers.items() if configured) + + if not enabled: + guidance = ( + f"Not authorized. The operator must set {ENABLE_ENV} to enable " + "open-resolver testing for this engagement." + ) + elif not available: + guidance = ( + f"Authorized, but no provider credentials found. Set {SHODAN_API_KEY_ENV} " + f"or {CENSYS_PAT_ENVS[0]}." + ) + else: + guidance = ( + f"Ready. Discover resolvers with discover_open_resolvers using: " + f"{', '.join(available)}." + ) + + return _compact( + { + "authorized": enabled, + "authorization_env": ENABLE_ENV, + "providers_configured": providers, + "providers_available": available, + "cached_countries": sorted(self._resolvers), + "cached_resolver_count": sum(len(v) for v in self._resolvers.values()), + "guidance": guidance, + } + ) + + # -- discovery --------------------------------------------------------- + + @tool_method(name="discover_open_resolvers", catch=True) + async def discover_open_resolvers( + self, + countries: Annotated[ + str, + "Comma-separated ISO-3166 country codes to source resolvers from (e.g. 'DE,SG,BR')", + ], + provider: Annotated[ + str, + "Discovery source: 'shodan', 'censys', or 'auto' to use whichever is configured", + ] = "auto", + max_per_country: Annotated[ + int, "Maximum verified resolvers to keep per country (1-5)" + ] = 2, + verify: Annotated[ + bool, + "Verify each candidate returns the correct answer for a known hostname before caching", + ] = True, + ) -> dict[str, Any]: + """Find open recursive resolvers in specific countries and cache them for this session. + + Queries Shodan or Censys for hosts exposing recursive DNS in each + country, then (by default) verifies each candidate actually resolves a + known-good hostname to its correct address. Verification matters: + hijacking resolvers that answer everything with an ad server would + otherwise show up as fake geo-divergence. + + Results are cached in memory for this session only — nothing is written + to disk. Re-running replaces the cache for the countries requested. + """ + _require_enabled() + + codes = _normalize_countries(countries) + limit = max(1, min(int(max_per_country), MAX_RESOLVERS_PER_COUNTRY)) + + selected = provider.strip().lower() + if selected not in {"auto", "shodan", "censys"}: + raise GeoResolverError( + f"unknown provider {provider!r}: expected 'shodan', 'censys', or 'auto'" + ) + if selected == "auto": + if _env_value(SHODAN_API_KEY_ENV): + selected = "shodan" + elif _env_value(*CENSYS_PAT_ENVS): + selected = "censys" + else: + raise GeoResolverError( + f"no provider credentials found: set {SHODAN_API_KEY_ENV} " + f"or {CENSYS_PAT_ENVS[0]}" + ) + + discover = _discover_shodan if selected == "shodan" else _discover_censys + # Over-fetch when verifying, since most candidates fail verification. + fetch_limit = min(limit * 5, 100) if verify else limit + + found: dict[str, list[dict[str, Any]]] = {} + errors: dict[str, str] = {} + async with httpx.AsyncClient(timeout=self.api_timeout) as client: + for code in codes: + try: + candidates = await discover(client, code, fetch_limit) + except GeoResolverError as exc: + errors[code] = str(exc) + continue + found[code] = candidates + + verified: dict[str, list[dict[str, Any]]] = {} + for code, candidates in found.items(): + if not verify: + verified[code] = candidates[:limit] + continue + kept = await self._verify_candidates(candidates, limit) + if kept: + verified[code] = kept + + for code, resolvers in verified.items(): + self._resolvers[code] = resolvers + + empty = sorted(set(codes) - set(verified) - set(errors)) + return _compact( + { + "provider": selected, + "verified": verify, + "countries": { + code: [r["ip"] for r in resolvers] + for code, resolvers in sorted(verified.items()) + }, + "resolver_details": verified, + "resolver_count": sum(len(v) for v in verified.values()), + "countries_without_resolvers": empty, + "errors": errors, + } + ) + + async def _verify_candidates( + self, candidates: list[dict[str, Any]], limit: int + ) -> list[dict[str, Any]]: + """Keep candidates that answer :data:`VERIFY_HOSTNAME` correctly, up to ``limit``.""" + kept: list[dict[str, Any]] = [] + semaphore = asyncio.Semaphore(MAX_CONCURRENT_DNS) + + async def probe(candidate: dict[str, Any]) -> tuple[dict[str, Any], bool]: + async with semaphore: + result = await query_resolver( + candidate["ip"], VERIFY_HOSTNAME, timeout=self.dns_timeout + ) + addresses = set(result.get("addresses") or []) + return candidate, bool(addresses & VERIFY_EXPECTED) + + # Probe in batches so a large candidate pool does not fan out unbounded. + for start in range(0, len(candidates), MAX_CONCURRENT_DNS): + batch = candidates[start : start + MAX_CONCURRENT_DNS] + for candidate, ok in await asyncio.gather(*(probe(c) for c in batch)): + if ok: + kept.append(candidate) + if len(kept) >= limit: + return kept + return kept + + @tool_method(name="list_open_resolvers", catch=True) + async def list_open_resolvers(self) -> dict[str, Any]: + """List the open resolvers cached for this session, grouped by country. + + Returns what ``discover_open_resolvers`` most recently verified. The + cache is in-memory and disappears when the session ends. + """ + _require_enabled() + return _compact( + { + "countries": { + code: [r["ip"] for r in resolvers] + for code, resolvers in sorted(self._resolvers.items()) + }, + "resolver_details": dict(sorted(self._resolvers.items())), + "resolver_count": sum(len(v) for v in self._resolvers.values()), + "guidance": ( + "No resolvers cached — run discover_open_resolvers first." + if not self._resolvers + else "Resolve hostnames through these with resolve_via_open_resolvers." + ), + } + ) + + # -- resolution -------------------------------------------------------- + + @tool_method(name="resolve_via_open_resolvers", catch=True) + async def resolve_via_open_resolvers( + self, + hostnames: Annotated[ + str, "Comma-separated hostnames to resolve (max 25 per call)" + ], + countries: Annotated[ + str, + "Comma-separated country codes to query, or empty to use every cached country", + ] = "", + record_type: Annotated[str, "DNS record type to request: 'A' or 'AAAA'"] = "A", + baseline_resolver: Annotated[ + str, "Resolver used for the local baseline answer" + ] = DEFAULT_BASELINE_RESOLVER, + ) -> dict[str, Any]: + """Resolve hostnames through in-region resolvers and diff against a local baseline. + + For each hostname this queries every cached resolver in the selected + countries, then compares the per-country answers to the baseline. A + country whose answers differ indicates geo-differentiated DNS: the + target hands back different infrastructure depending on where the + resolver sits, which is the DNS-layer footprint of geo-fencing. + + Addresses reported under ``addresses_only_seen_regionally`` are the + actionable output — endpoints invisible from the local vantage point. + Confirm they are in scope before touching them. + """ + _require_enabled() + + names = [ + h.strip().rstrip(".") + for h in hostnames.replace(",", " ").split() + if h.strip() + ] + if not names: + raise GeoResolverError("no hostnames supplied") + if len(names) > MAX_HOSTNAMES_PER_CALL: + raise GeoResolverError( + f"too many hostnames ({len(names)}); maximum is {MAX_HOSTNAMES_PER_CALL} per call" + ) + + rtype = record_type.strip().upper() + if rtype not in _QTYPE_BY_NAME: + raise GeoResolverError( + f"unsupported record type {record_type!r}: use 'A' or 'AAAA'" + ) + + if countries.strip(): + codes = _normalize_countries(countries) + missing = [c for c in codes if c not in self._resolvers] + if missing: + raise GeoResolverError( + f"no cached resolvers for {', '.join(missing)} — " + "run discover_open_resolvers for those countries first" + ) + else: + codes = sorted(self._resolvers) + if not codes: + raise GeoResolverError( + "no resolvers cached — run discover_open_resolvers first" + ) + + semaphore = asyncio.Semaphore(MAX_CONCURRENT_DNS) + + async def lookup(resolver_ip: str, hostname: str) -> dict[str, Any]: + async with semaphore: + return await query_resolver( + resolver_ip, + hostname, + record_type=rtype, # type: ignore[arg-type] + timeout=self.dns_timeout, + ) + + # Fail fast on unencodable hostnames before issuing any network traffic. + for hostname in names: + encode_question(hostname, _QTYPE_BY_NAME[rtype]) + + targets = [ + (code, resolver["ip"]) + for code in codes + for resolver in self._resolvers[code] + ] + + results: dict[str, Any] = {} + for hostname in names: + baseline_result = await lookup(baseline_resolver, hostname) + baseline_addresses = baseline_result.get("addresses") or [] + + per_country: dict[str, list[str]] = {} + per_resolver: dict[str, Any] = {} + unreachable: list[str] = [] + + outcomes = await asyncio.gather( + *(lookup(ip, hostname) for _, ip in targets) + ) + + for (code, ip), outcome in zip(targets, outcomes): + if outcome.get("error"): + unreachable.append(f"{code}/{ip}: {outcome['error']}") + continue + addresses = outcome.get("addresses") or [] + per_resolver[f"{code}/{ip}"] = addresses + per_country.setdefault(code, []) + for address in addresses: + if address not in per_country[code]: + per_country[code].append(address) + + results[hostname] = _compact( + { + **summarize_divergence(baseline_addresses, per_country), + "baseline_error": baseline_result.get("error", ""), + "answers_by_country": { + k: sorted(v) for k, v in sorted(per_country.items()) + }, + "answers_by_resolver": per_resolver, + "unreachable_resolvers": unreachable, + } + ) + + differentiated = sorted( + name for name, data in results.items() if data.get("geo_differentiated") + ) + return _compact( + { + "record_type": rtype, + "baseline_resolver": baseline_resolver, + "countries_queried": codes, + "results": results, + "geo_differentiated_hostnames": differentiated, + "guidance": ( + "No geo-differentiation detected — the target returns the same records " + "from every vantage point tested." + if not differentiated + else ( + "Geo-differentiated DNS detected. Review " + "addresses_only_seen_regionally and confirm scope before probing." + ) + ), + } + ) + + @tool_method(name="clear_open_resolver_cache", catch=True) + async def clear_open_resolver_cache(self) -> dict[str, Any]: + """Discard the session's cached open resolvers. + + Use when switching engagements or after finishing geo-differentiated + testing so a later call cannot silently reuse a previous target's + vantage points. + """ + cleared = sum(len(v) for v in self._resolvers.values()) + self._resolvers.clear() + return {"cleared_resolver_count": cleared, "cached_countries": []}