1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
#!/usr/bin/env python3
import argparse
import dns.name
import n.gssapi
import n.kerberos
import n.log
import os
import subprocess
import sys
import time
'''
certbot certonly \
--manual \
--manual-auth-hook /usr/local/nullroute/certbot_manual_hook.py \
--manual-cleanup-hook /usr/local/nullroute/certbot_manual_hook.py \
--preferred-challenges dns \
--debug-challenges \
-d ldap.nullroute.lt
'''
def replace_domain_suffix(name, old_suffix, new_suffix):
name = ".%s" % name.strip(".")
old_suffix = ".%s" % old_suffix.strip(".")
new_suffix = ".%s" % new_suffix.strip(".")
if name == old_suffix:
name = new_suffix
elif name.endswith(old_suffix):
name = name[:-len(old_suffix)] + new_suffix
else:
raise ValueError("name %r not underneath %r" % (name.lstrip("."), old_suffix))
return name.lstrip(".")
parser = argparse.ArgumentParser()
parser.add_argument("-d", "--domain",
required=True,
help="domain or subdomain being validated")
parser.add_argument("-c", "--challenge",
help="ACME challenge to insert")
parser.add_argument("-N", "--use-nsupdate",
action="store_true",
help="use BIND 'nsupdate' instead of dnspython")
parser.add_argument("-v", "--verbose",
action="store_true",
help="show operation details")
args = parser.parse_args()
logger = n.log.init("dns_acme_update", args.verbose)
domain = args.domain
challenge = args.challenge
use_nsupdate = args.use_nsupdate
if os.getuid() == 0:
# Ensure we always identify as host@ and not just whatever happens to be
# first in the keytab.
n.kerberos.use_system_keytab()
n.gssapi.host_credentials()
# Currently we only support this single domain.
raw_zone = "nullroute.lt"
dyn_zone = "dyn.nullroute.lt"
fqdn = "_acme-challenge.%s" % replace_domain_suffix(domain, raw_zone, dyn_zone)
fqdn = dns.name.from_text(fqdn)
if challenge:
print(f"Inserting challenge {challenge!r} at [{fqdn}]", flush=True)
else:
print(f"Removing all challenges at [{fqdn}]", flush=True)
if use_nsupdate:
if challenge:
cmds = [f"zone {dyn_zone}",
f"del {fqdn}",
f"add {fqdn} 60 TXT \"{challenge}\"",
f"send"]
else:
cmds = [f"zone {dyn_zone}",
f"del {fqdn}",
f"send"]
cmds = "".join(f"{c}\n" for c in cmds)
subprocess.run(["nsupdate", "-g"],
input=cmds.encode(),
check=True)
else:
import n.dns.gssapi
import dns.update
msg = dns.update.UpdateMessage(zone=dyn_zone)
msg.delete(fqdn)
if challenge:
msg.add(fqdn, 60, "TXT", challenge)
if args.verbose:
print(msg, file=sys.stderr)
n.dns.gssapi.gss_tsig_update(dyn_zone, msg)
|
