diff --git a/hooks/AGENTS.md b/hooks/AGENTS.md index 3f58b35..b47dcc7 100644 --- a/hooks/AGENTS.md +++ b/hooks/AGENTS.md @@ -48,7 +48,8 @@ head -1 hooks/*.py Iterate there until the import is clean, then commit. - **Hooks have access to the full repo (`repo`, `node`, `nodes`).** Don't make them block on network unless that's the explicit purpose - (e.g. `test_ptr_records.py` does `dig` against `9.9.9.9`). + (e.g. `test_ptr_records.py` resolves via DNS-over-HTTPS, since the + agent sandbox allows no UDP). - **Order is not guaranteed across hook files.** Two hooks that both define `apply_start` will both fire; don't assume which runs first. diff --git a/hooks/test_ptr_records.py b/hooks/test_ptr_records.py index a868c60..4b4d6dc 100644 --- a/hooks/test_ptr_records.py +++ b/hooks/test_ptr_records.py @@ -1,20 +1,36 @@ -"""test_ptr_records: bw-test gate verifying live A/PTR DNS records for the `mailserver` group via `dig @9.9.9.9`.""" +"""test_ptr_records: bw-test gate verifying live A/PTR DNS records for the `mailserver` group via DNS-over-HTTPS.""" -from subprocess import check_output +import json +from ipaddress import ip_address +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +# DNS-over-HTTPS instead of dig: the agent sandbox allows no UDP, only HTTPS through its proxy +DOH_URL = 'https://dns.google/resolve' +DOH_TIMEOUT = 10 -def test_node(repo, node, **kwargs): +def resolve(name, rtype): + query = urlencode({'name': name, 'type': rtype}) + with urlopen(Request(f'{DOH_URL}?{query}', headers={'accept': 'application/dns-json'}), timeout=DOH_TIMEOUT) as response: + answer = json.load(response) + if answer.get('Status') != 0: + raise AssertionError(f"{rtype} lookup for {name} failed with DNS status {answer.get('Status')}") + return [record['data'] for record in answer.get('Answer', []) if record['type'] == {'A': 1, 'PTR': 12}[rtype]] + + +def test(repo, **kwargs): for node in repo.nodes_in_group('mailserver'): domain = node.metadata.get('mailserver/hostname') expected_ptr_record = f"{domain}." expected_a_record = node.hostname # check A record - actual_a_record = check_output(['dig', '+short', 'A', domain, '@9.9.9.9'], text=True).strip() - if actual_a_record != expected_a_record: - raise AssertionError(f"A record for {expected_a_record} on node {node.name} is {actual_a_record}, expected {expected_a_record}") + actual_a_records = resolve(domain, 'A') + if actual_a_records != [expected_a_record]: + raise AssertionError(f"A record for {domain} on node {node.name} is {actual_a_records}, expected {expected_a_record}") - # check otr record - actual_ptr_record = check_output(['dig', '+short', '-x', expected_a_record, '@9.9.9.9'], text=True).strip() - if actual_ptr_record != expected_ptr_record: - raise AssertionError(f"PTR record for {expected_a_record} on node {node.name} is {actual_ptr_record}, expected {expected_ptr_record}") + # check PTR record + actual_ptr_records = resolve(ip_address(expected_a_record).reverse_pointer, 'PTR') + if actual_ptr_records != [expected_ptr_record]: + raise AssertionError(f"PTR record for {expected_a_record} on node {node.name} is {actual_ptr_records}, expected {expected_ptr_record}")