Created
January 27, 2026 12:44
-
-
Save lemon24/e622cc8901d70e97ed5e6f80b1675076 to your computer and use it in GitHub Desktop.
local certbot + remote challenge (+ pyinfra) proof of concept
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| local certbot + remote challenge proof of concept, using --manual-auth-hook[1] | |
| a pyinfra operation using this may do something like: | |
| if cert not on server: | |
| if cert not locally: | |
| get_certificate(domain) | |
| yield "copy cert to server" command | |
| of note, the operation is to copy the certificate to server *only*: | |
| * cert generation on local host may or may not happen | |
| * challenge copying to server is not considered an operation | |
| * the server is already configured to serve challenges | |
| also, if the operation is structured like this, how do we roll back? | |
| it might be better to name specific certificates in the server deployment, | |
| and handle generation / renewal etc. separately (maybe even outside pyinfra). | |
| [1]: https://eff-certbot.readthedocs.io/en/stable/using.html#pre-and-post-validation-hooks | |
| """ | |
| import json | |
| import os | |
| import pathlib | |
| import shlex | |
| import subprocess | |
| import tempfile | |
| import threading | |
| import time | |
| def get_certificate(domain): | |
| with Waiter() as auth, Waiter() as cleanup: | |
| # what if there's an error? (thread, certbot, auth hook) | |
| # how do we avoid this waiting forever? | |
| threading.Thread( | |
| target=certonly, | |
| args=(domain, auth.script, cleanup.script) | |
| ).start() | |
| auth_info = auth.wait() | |
| print(f"copying challenge to server: {auth_info['DOMAIN']}") | |
| auth.done() | |
| # technically we can use auth_info instead of cleanup_info, | |
| # since the auth script is used only by this invocation | |
| cleanup_info = cleanup.wait() | |
| print(f"cleaning up challenge from server: {cleanup_info['DOMAIN']}") | |
| cleanup.done() | |
| def certonly(domain, auth, cleanup): | |
| # simulated subprocess.run() call to | |
| # certbot certonly -n --manual --manual-auth-hook $auth ... | |
| env = {'DOMAIN': domain.upper()} | |
| # certbot gets challenge data here | |
| subprocess.run([auth], env=env) | |
| # certbot triggeres the challenge here | |
| subprocess.run([cleanup], env=env) | |
| class Waiter: | |
| def __init__(self): | |
| self._tmpdir = tempfile.TemporaryDirectory() | |
| tmpdir = pathlib.Path(self._tmpdir.name).absolute() | |
| self.script = tmpdir / 'script' | |
| self.script.write_text(WAITER_SCRIPT.format(workdir=shlex.quote(str(tmpdir)))) | |
| self.script.chmod(0o700) | |
| self._pipe = tmpdir / 'pipe' | |
| os.mkfifo(self._pipe, 0o600) | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, *_): | |
| self.close() | |
| def wait(self): | |
| return json.loads(self._pipe.read_text()) | |
| def done(self): | |
| self._pipe.write_text('done\n') | |
| self.close() | |
| def close(self): | |
| self._tmpdir.cleanup() | |
| WAITER_SCRIPT = """\ | |
| #!/bin/sh | |
| python3 -c 'import os, json; print(json.dumps(dict(os.environ)))' > {workdir}/pipe | |
| read < {workdir}/pipe | |
| """ | |
| if __name__ == '__main__': | |
| get_certificate('example.com') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment