117 lines
5.0 KiB
Python
117 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Certbot DNS hook: only one approved delegation, exact TXT IDs, IPv4 API."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
|
|
CONFIG = pathlib.Path('/etc/white-proxy-dns/config.json')
|
|
TOKEN = pathlib.Path('/etc/white-proxy-dns/cf-token')
|
|
STATE = pathlib.Path('/var/lib/white-proxy-dns')
|
|
DOMAIN = 'tomshell.com.de'
|
|
TARGET = 'old-com-de.acme.tomshell.online'
|
|
AUTH_NS = ('jaziel.ns.cloudflare.com', 'nena.ns.cloudflare.com')
|
|
|
|
|
|
def check_private(path):
|
|
if path.stat().st_uid != 0 or path.stat().st_mode & 0o077:
|
|
raise RuntimeError('DNS credentials must be root-owned and private')
|
|
|
|
|
|
def api(method, suffix, payload=None):
|
|
check_private(TOKEN)
|
|
check_private(CONFIG)
|
|
cfg = json.loads(CONFIG.read_text())
|
|
zone = cfg['zone_id']
|
|
if not re.fullmatch(r'[a-f0-9]{32}', zone) or cfg['target'] != TARGET:
|
|
raise RuntimeError('Unapproved DNS configuration')
|
|
token = TOKEN.read_text().strip()
|
|
if not re.fullmatch(r'[A-Za-z0-9_-]{40}', token):
|
|
raise RuntimeError('Invalid token format')
|
|
with tempfile.TemporaryDirectory(prefix='acme-api-', dir=STATE) as folder:
|
|
body = pathlib.Path(folder) / 'request.json'
|
|
args = ['curl', '-4', '--silent', '--show-error', '--fail-with-body',
|
|
'--connect-timeout', '10', '--max-time', '30', '--config', '-',
|
|
'--request', method,
|
|
'https://api.cloudflare.com/client/v4/zones/' + zone + '/dns_records' + suffix]
|
|
if payload is not None:
|
|
body.write_text(json.dumps(payload))
|
|
args.extend(['--data-binary', '@' + str(body)])
|
|
config = 'header = "Authorization: Bearer ' + token + '"\nheader = "Content-Type: application/json"\n'
|
|
result = subprocess.run(args, input=config, capture_output=True, text=True)
|
|
if result.returncode:
|
|
raise RuntimeError('Cloudflare DNS API request failed; response intentionally not logged')
|
|
value = json.loads(result.stdout)
|
|
if not value.get('success'):
|
|
raise RuntimeError('Cloudflare DNS API rejected request')
|
|
return value['result']
|
|
|
|
|
|
def dns(name, kind, server):
|
|
return subprocess.check_output(['dig', '-4', '+time=3', '+tries=1', '+short',
|
|
'@' + server, name, kind], text=True).strip()
|
|
|
|
|
|
def matches(record, validation):
|
|
return record.get('type') == 'TXT' and record.get('name') == TARGET and record.get('content') == validation
|
|
|
|
|
|
def main():
|
|
if os.geteuid() != 0 or len(sys.argv) != 2 or sys.argv[1] not in ('auth', 'cleanup'):
|
|
raise RuntimeError('Root auth/cleanup mode required')
|
|
os.umask(0o077)
|
|
STATE.mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
domain = os.environ['CERTBOT_DOMAIN'].removeprefix('*.')
|
|
validation = os.environ['CERTBOT_VALIDATION']
|
|
if domain != DOMAIN or not re.fullmatch(r'[A-Za-z0-9_-]{43}', validation):
|
|
raise RuntimeError('Challenge outside approved domain')
|
|
state = STATE / (hashlib.sha256(validation.encode()).hexdigest() + '.json')
|
|
if sys.argv[1] == 'cleanup':
|
|
if not state.exists():
|
|
return
|
|
record_id = json.loads(state.read_text())['id']
|
|
if not re.fullmatch(r'[a-f0-9]{32}', record_id):
|
|
raise RuntimeError('Invalid TXT record ID')
|
|
record = api('GET', '/' + record_id)
|
|
if not matches(record, validation):
|
|
raise RuntimeError('TXT record changed; refusing deletion')
|
|
api('DELETE', '/' + record_id)
|
|
state.unlink()
|
|
return
|
|
for ns in ('ns1.dns-parking.com', 'ns2.dns-parking.com'):
|
|
if dns('_acme-challenge.' + DOMAIN, 'CNAME', ns).rstrip('.') != TARGET:
|
|
raise RuntimeError('Approved CNAME delegation not published')
|
|
if state.exists():
|
|
raise RuntimeError('Existing challenge state; inspect before retry')
|
|
record = api('POST', '', {'type': 'TXT', 'name': TARGET, 'content': validation, 'ttl': 60,
|
|
'comment': 'Isolated legacy white-proxy ACME challenge'})
|
|
state.write_text(json.dumps({'id': record['id']}))
|
|
try:
|
|
deadline = time.monotonic() + 600
|
|
while time.monotonic() < deadline:
|
|
if all('"' + validation + '"' in dns(TARGET, 'TXT', ns).splitlines() for ns in AUTH_NS):
|
|
# Also allow recursors to replace a cached negative answer before validation.
|
|
for _ in range(24):
|
|
if all('"' + validation + '"' in dns('_acme-challenge.' + DOMAIN, 'TXT', ns).splitlines()
|
|
for ns in ('1.1.1.1', '8.8.8.8')):
|
|
return
|
|
time.sleep(5)
|
|
raise RuntimeError('Challenge not visible through public DNS recursors')
|
|
time.sleep(5)
|
|
raise RuntimeError('Challenge propagation timed out')
|
|
except Exception:
|
|
if matches(api('GET', '/' + record['id']), validation):
|
|
api('DELETE', '/' + record['id'])
|
|
state.unlink()
|
|
raise
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|