Archive discontinued selective SRS implementation
This commit is contained in:
@@ -1,195 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Read-only, fail-closed public-DNS preflight for the approved SRS DNS profile.
|
||||
|
||||
Checks the authoritative nameservers, bypassing local hosts/systemd-resolved.
|
||||
This deliberately validates only the approved single-host DNS profile:
|
||||
exactly one MX pointing to this host and one SPF record `v=spf1 mx -all`.
|
||||
It is NOT a general-purpose SPF evaluator.
|
||||
"""
|
||||
import argparse
|
||||
import ipaddress
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
# A public resolver is used only to discover the delegated authoritative NS and
|
||||
# their numeric IPs. The actual TXT/MX/A/AAAA values are read from each NS.
|
||||
BOOTSTRAP_RESOLVERS = ('1.1.1.1', '9.9.9.9')
|
||||
DNS_TIMEOUT = 5
|
||||
|
||||
|
||||
def dig(server, name, kind, authoritative=False):
|
||||
command = ['dig', '@' + server, '+time=2', '+tries=1', '+noall', '+comments', '+answer']
|
||||
if authoritative:
|
||||
command.append('+norecurse')
|
||||
command.extend([name, kind])
|
||||
try:
|
||||
proc = subprocess.run(command, capture_output=True, text=True, timeout=DNS_TIMEOUT, check=False)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise ValueError(f'DNS request failed ({server}, {name}, {kind}): {exc}') from exc
|
||||
if proc.returncode != 0:
|
||||
raise ValueError(f'DNS request failed ({server}, {name}, {kind}): {proc.stderr.strip() or proc.returncode}')
|
||||
header = re.search(r'HEADER.*?status:\s*(\w+)', proc.stdout)
|
||||
if not header or header.group(1) != 'NOERROR':
|
||||
raise ValueError(f'DNS returned {header.group(1) if header else "no status"}: {server} {name} {kind}')
|
||||
flags = re.search(r'^;; flags:\s*(.*?);\s*QUERY:', proc.stdout, re.M)
|
||||
if authoritative and (not flags or 'aa' not in flags.group(1).split()):
|
||||
raise ValueError(f'Authoritative response missing AA flag: {server} {name} {kind}')
|
||||
rows = []
|
||||
for line in proc.stdout.splitlines():
|
||||
parts = line.split(None, 4)
|
||||
if len(parts) != 5 or parts[2].upper() != 'IN':
|
||||
continue
|
||||
owner, _ttl, _cls, rr_type, data = parts
|
||||
if rr_type.upper() == kind.upper() and owner.rstrip('.').lower() == name.rstrip('.').lower():
|
||||
rows.append(data.strip())
|
||||
return rows
|
||||
|
||||
|
||||
def public_lookup(name, kind):
|
||||
failures = []
|
||||
for server in BOOTSTRAP_RESOLVERS:
|
||||
try:
|
||||
return dig(server, name, kind)
|
||||
except ValueError as exc:
|
||||
failures.append(str(exc))
|
||||
raise ValueError('Public NS bootstrap failed: ' + ' | '.join(failures))
|
||||
|
||||
|
||||
def discover_nameservers(domain):
|
||||
labels = domain.rstrip('.').split('.')
|
||||
# Pick closest delegated zone. A zone without its own delegation has no NS.
|
||||
for i in range(len(labels) - 1):
|
||||
zone = '.'.join(labels[i:])
|
||||
ns = public_lookup(zone, 'NS')
|
||||
if ns:
|
||||
names = sorted({n.rstrip('.').lower() for n in ns})
|
||||
return zone, names
|
||||
raise ValueError(f'Cannot discover authoritative nameservers for {domain}')
|
||||
|
||||
|
||||
def addr_of_ns(ns):
|
||||
for kind in ('A', 'AAAA'):
|
||||
for value in public_lookup(ns, kind):
|
||||
try:
|
||||
return str(ipaddress.ip_address(value))
|
||||
except ValueError:
|
||||
continue
|
||||
raise ValueError(f'Cannot resolve nameserver {ns} to a public IP address')
|
||||
|
||||
|
||||
def parse_txt(value):
|
||||
# DNS may split a single TXT resource record into multiple quoted strings.
|
||||
parts = re.findall(r'"((?:\\.|[^"\\])*)"', value)
|
||||
if not parts:
|
||||
raise ValueError('Unexpected TXT presentation format: ' + value)
|
||||
return ''.join(parts).replace('\\"', '"')
|
||||
|
||||
|
||||
def check_server(ns_name, ns_ip, domain, hostname, bind4, bind6):
|
||||
errors = []
|
||||
def query(name, kind):
|
||||
try:
|
||||
return dig(ns_ip, name, kind, authoritative=True)
|
||||
except ValueError as exc:
|
||||
errors.append(str(exc))
|
||||
return []
|
||||
|
||||
spf_records = []
|
||||
for raw in query(domain, 'TXT'):
|
||||
try:
|
||||
text = parse_txt(raw)
|
||||
if text.lower().startswith('v=spf1'):
|
||||
spf_records.append(text)
|
||||
except ValueError as exc:
|
||||
errors.append(str(exc))
|
||||
spf_ok = (
|
||||
len(spf_records) == 1
|
||||
and spf_records[0].lower().split() == ['v=spf1', 'mx', '-all']
|
||||
)
|
||||
if not spf_ok:
|
||||
errors.append('SPF is not the approved single-host profile: v=spf1 mx -all')
|
||||
|
||||
mx = query(domain, 'MX')
|
||||
if len(mx) != 1:
|
||||
errors.append(f'Expected exactly one MX pointing to {hostname}; found {mx}')
|
||||
else:
|
||||
parts = mx[0].split()
|
||||
if len(parts) != 2 or not parts[0].isdigit() or parts[1].rstrip('.').lower() != hostname:
|
||||
errors.append(f'MX is not the expected mailserver {hostname}: {mx[0]}')
|
||||
|
||||
for kind, configured in [('A', bind4), ('AAAA', bind6)]:
|
||||
answers = query(hostname, kind)
|
||||
normalized = []
|
||||
for entry in answers:
|
||||
try:
|
||||
normalized.append(str(ipaddress.ip_address(entry)))
|
||||
except ValueError:
|
||||
errors.append(f'Invalid {kind} address returned for {hostname}: {entry}')
|
||||
if not configured:
|
||||
errors.append(f'Postfix smtp_bind_address{("6" if kind == "AAAA" else "")} is empty; cannot verify outbound IP')
|
||||
elif configured not in normalized:
|
||||
errors.append(f'Postfix outbound {kind} address {configured} is not published for {hostname}: {normalized}')
|
||||
|
||||
print(f' [ {"ok" if not errors else "failed"} ] {ns_name} ({ns_ip})')
|
||||
for err in errors:
|
||||
print(f' [ error ] {err}')
|
||||
return not errors, spf_ok
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('--domain', required=True)
|
||||
ap.add_argument('--hostname', required=True)
|
||||
ap.add_argument('--ipv4', required=True)
|
||||
ap.add_argument('--ipv6', required=True)
|
||||
args = ap.parse_args()
|
||||
domain = args.domain.lower().rstrip('.')
|
||||
hostname = args.hostname.lower().rstrip('.')
|
||||
if not shutil.which('dig'):
|
||||
print(' [ error ] dig is required to verify public DNS')
|
||||
return 1
|
||||
try:
|
||||
bind4 = str(ipaddress.ip_address(args.ipv4)) if args.ipv4 else ''
|
||||
bind6 = str(ipaddress.ip_address(args.ipv6)) if args.ipv6 else ''
|
||||
zone, nameservers = discover_nameservers(domain)
|
||||
print(f' DNS delegation zone................: {zone}')
|
||||
print(f' Expected MX host...................: {hostname}')
|
||||
print(f' Expected SMTP IPv4.................: {bind4 or "unknown"}')
|
||||
print(f' Expected SMTP IPv6.................: {bind6 or "unknown"}')
|
||||
print(' Required SPF record...............: v=spf1 mx -all')
|
||||
if len(nameservers) > 8:
|
||||
raise ValueError('More than eight authoritative nameservers; manual review required')
|
||||
print()
|
||||
print(' Authoritative nameservers:')
|
||||
ok = True
|
||||
spf_ok_all = True
|
||||
for ns in nameservers:
|
||||
try:
|
||||
ip = addr_of_ns(ns)
|
||||
server_ok, server_spf_ok = check_server(
|
||||
ns, ip, domain, hostname, bind4, bind6
|
||||
)
|
||||
if not server_ok:
|
||||
ok = False
|
||||
if not server_spf_ok:
|
||||
spf_ok_all = False
|
||||
except ValueError as exc:
|
||||
print(f' [ failed ] {ns}')
|
||||
print(f' [ error ] {exc}')
|
||||
ok = False
|
||||
spf_ok_all = False
|
||||
print()
|
||||
print(' SPF validation.....................: ' +
|
||||
('OK' if spf_ok_all else 'FAILED / NOT VERIFIED'))
|
||||
print(' Public SRS DNS preflight...........: ' + ('OK' if ok else 'FAILED'))
|
||||
return 0 if ok else 1
|
||||
except ValueError as exc:
|
||||
print(f' [ error ] {exc}')
|
||||
print(' Public SRS DNS preflight...........: FAILED')
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user