#!/usr/bin/env python3 """Owner-installed Certbot renewal bridge. Never uploads a private key. Review this file before executing as root. Requires an existing, current Certbot with the selected nginx/apache plugin and an already configured public vhost. The personalized download contains a domain-scoped report key; keep it private. """ import argparse import base64 import hashlib import hmac import json import os from pathlib import Path import re import secrets import shlex import shutil import ssl import subprocess import sys import time import urllib.request PROFILE_B64 = '__SERTICAN_PROFILE_BASE64__' def send_report(profile, status, pem, expires): fingerprint = hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem)).hexdigest() data = {'domain': profile['domain'], 'status': status, 'fingerprint': fingerprint, 'expires': int(expires)} body = json.dumps(data, separators=(',', ':')).encode() stamp = str(int(time.time())) signature = hmac.new(profile['key'].encode(), stamp.encode()+b'\n'+body, hashlib.sha256).hexdigest() request = urllib.request.Request(profile['endpoint'], data=body, method='POST', headers={'Content-Type': 'application/json', 'X-Sertican-Domain': profile['id'], 'X-Sertican-Time': stamp, 'X-Sertican-Signature': signature}) # HTTPS certificate validation is mandatory. No redirects or alternate hosts. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPSHandler(context=ssl.create_default_context())) with opener.open(request, timeout=15) as response: if response.status != 200 or not json.loads(response.read(1024)).get('ok'): raise RuntimeError('Report was not accepted; verify ownership, key and clock.') def certificate(profile): path = Path('/etc/letsencrypt/live') / profile['domain'] / 'cert.pem' pem = path.read_text() output = subprocess.run([profile['openssl'], 'x509', '-in', str(path), '-noout', '-enddate'], check=True, capture_output=True, text=True, timeout=15).stdout.strip() date = output.split('=', 1)[1] return pem, int(ssl.cert_time_to_seconds(date)) def atomic_private(path, data, mode=0o600): if path.is_symlink(): raise RuntimeError('Refusing a symbolic link: '+str(path)) temporary = path.with_name(path.name+'.tmp-'+str(os.getpid())) fd = os.open(str(temporary), os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode) try: with os.fdopen(fd, 'w') as handle: handle.write(data) os.replace(temporary, path) os.chmod(path, mode) finally: if temporary.exists(): temporary.unlink() def main(): parser = argparse.ArgumentParser(description=__doc__) action = parser.add_mutually_exclusive_group(required=True) action.add_argument('--install', action='store_true') action.add_argument('--run', action='store_true') action.add_argument('--uninstall', action='store_true') parser.add_argument('--webserver', choices=['nginx', 'apache']) parser.add_argument('--email') parser.add_argument('--accept-le-terms', action='store_true', help='Explicitly accept Let’s Encrypt subscriber terms') parser.add_argument('--scheduled', action='store_true', help='Timer-only random second and retry deferral') args = parser.parse_args() if args.scheduled and not args.run: parser.error('--scheduled requires --run') if not hasattr(os, 'geteuid') or os.geteuid() != 0: parser.error('Run this domain-specific tool as root on your own Linux server.') profile = json.loads(base64.b64decode(PROFILE_B64)) domain = profile['domain'] if not re.fullmatch(r'[a-z0-9][a-z0-9.-]{1,251}[a-z0-9]', domain) or '..' in domain: raise RuntimeError('Invalid domain') if not re.fullmatch(r'[a-f0-9]{20}', profile['id']): raise RuntimeError('Invalid domain ID') if not profile['endpoint'].startswith('https://'): raise RuntimeError('HTTPS reporting is required') root = Path('/etc/sertican-acme') target = root / (domain+'.py') saved = root / (domain+'.json') cron = Path('/etc/cron.d') / ('sertican-'+hashlib.sha256(domain.encode()).hexdigest()[:20]) if args.uninstall: for item in (cron, target, saved, root/(domain+'.lock'), root/(domain+'.retry.json')): if item.is_symlink(): raise RuntimeError('Refusing symlink '+str(item)) item.unlink(missing_ok=True) print('Bridge and its cron removed. Certbot certificates were preserved. Revoke the key in your private status panel.') return if args.install: if not args.webserver or not args.email or not re.fullmatch(r'[^\s@]+@[^\s@]+\.[^\s@]+', args.email) or not args.accept_le_terms: parser.error('--webserver, --email and --accept-le-terms are required. Review https://letsencrypt.org/repository/ first.') certbot = shutil.which('certbot') openssl = shutil.which('openssl') if not certbot or not openssl: parser.error('Install current Certbot and its nginx/apache plugin from certbot.eff.org, plus OpenSSL.') if not Path('/etc/cron.d').is_dir() or not shutil.which('cron') and not shutil.which('crond'): parser.error('A working cron service and /etc/cron.d are required.') existing = Path('/etc/letsencrypt/live') / domain / 'cert.pem' if existing.exists(): # Preserve an existing lineage and every SAN. Never narrow it to -d domain. subprocess.run([openssl, 'x509', '-in', str(existing), '-noout', '-checkhost', domain], check=True, timeout=15) subprocess.run([certbot, 'renew', '--cert-name', domain, '--dry-run', '--non-interactive'], check=True, timeout=900) else: # New lineage: Certbot’s official installer plugin configures the vhost. subprocess.run([certbot, '--'+args.webserver, '--cert-name', domain, '-d', domain, '--email', args.email, '--non-interactive', '--agree-tos'], check=True, timeout=600) if root.is_symlink(): raise RuntimeError('Refusing symlink configuration directory') root.mkdir(mode=0o700, exist_ok=True) os.chmod(root, 0o700) profile.update(certbot=str(Path(certbot).resolve()), openssl=str(Path(openssl).resolve())) atomic_private(saved, json.dumps(profile)) atomic_private(target, Path(__file__).read_text(), 0o700) # Independent hour/minute per enrollment; each timer invocation also jitters seconds. hour = secrets.randbelow(12) minute = secrets.randbelow(60) command = shlex.quote(str(Path(sys.executable).resolve()))+' '+shlex.quote(str(target))+' --run --scheduled' atomic_private(cron, 'SHELL=/bin/sh\nPATH=/usr/sbin:/usr/bin:/sbin:/bin\n'+str(minute)+' '+str(hour)+','+str(hour+12)+' * * * root '+command+'\n', 0o644) pem, expires = certificate(profile) send_report(profile, 'ok', pem, expires) print('Installed: '+str(cron)+'. Runs twice daily in this server’s timezone. Check your private panel after its next HTTPS probe.') return profile = json.loads(saved.read_text()) if args.scheduled: time.sleep(secrets.randbelow(60)) retry_file = root / (domain+'.retry.json') import fcntl with (root/(domain+'.lock')).open('a') as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: return if args.scheduled and retry_file.exists() and json.loads(retry_file.read_text()).get('not_before',0)>time.time(): print('Renewal retry deferred after previous failure') return try: result = subprocess.run([profile['certbot'], 'renew', '--cert-name', domain, '--non-interactive'], timeout=900) pem, expires = certificate(profile) send_report(profile, 'ok' if result.returncode == 0 else 'renew_failed', pem, expires) if result.returncode: raise RuntimeError('Certbot renewal failed; inspect the local Certbot log') atomic_private(retry_file, json.dumps({'attempts':0,'not_before':0})) except Exception: state=json.loads(retry_file.read_text()) if retry_file.exists() else {} attempts=min(20,max(0,int(state.get('attempts',0)))+1) delay=(60,600,6000,86400)[min(attempts-1,3)] atomic_private(retry_file,json.dumps({'attempts':attempts,'not_before':int(time.time())+delay})) raise if __name__ == '__main__': main()