Created
April 8, 2021 05:07
-
-
Save Caligatio/f97ba88dad586ed2a0423f34ce67106b to your computer and use it in GitHub Desktop.
This is a simplistic Python 3.8+ script that will download the latest available version of Caddy (with plugins!), compare to the existing Caddy, and will replace/upgrade if the downloaded version is different. It also has rudimentary systemd support to stop and restart the service after upgrade.
This file contains 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
#!/usr/bin/env python3 | |
""" | |
caddy_autoupgrade | |
This is a simplistic Python 3.8+ script that will download the latest available version of Caddy (with plugins!), | |
compare to the existing Caddy, and will replace/upgrade if the downloaded version is different. It also has rudimentary | |
systemd support to stop and restart the service after upgrade. | |
It was written by Brian Turek (https://github.com/Caligatio) and released under the Unlicense (https://unlicense.org/). | |
""" | |
from argparse import ArgumentParser | |
from hashlib import sha256 | |
from os import close | |
from pathlib import Path | |
from re import search | |
from shutil import copy | |
from subprocess import run | |
from sys import exit | |
from tempfile import mkstemp | |
from typing import Final, Optional | |
from urllib.request import urlopen | |
CHUNK_SIZE: Final = 8 * 1024 | |
def download_file(url: str) -> Path: | |
fd, file_name = mkstemp() | |
close(fd) | |
temp_file = Path(file_name) | |
resp = urlopen(url) | |
if resp.code != 200: | |
raise RuntimeError(f"Could not download remote file, HTTP error code: {resp.code}") | |
try: | |
with temp_file.open("wb") as f_out: | |
while data := resp.read(CHUNK_SIZE): | |
f_out.write(data) | |
except Exception: | |
temp_file.unlink() | |
raise | |
temp_file.chmod(0o755) | |
return temp_file | |
def is_caddy_running(systemd_svc: str) -> bool: | |
proc = run(["/usr/bin/systemctl", "status", systemd_svc], capture_output=True, check=True) | |
if search(b"Active: active \(running\)", proc.stdout): | |
return True | |
else: | |
return False | |
def hash_file(src_file: Path) -> bytes: | |
file_hash = sha256() | |
with src_file.open("rb") as f_in: | |
while data := f_in.read(CHUNK_SIZE): | |
file_hash.update(data) | |
return file_hash.digest() | |
def main(caddy_url: str, caddy_exec: Path, systemd_svc: Optional[str] = None) -> int: | |
if not caddy_exec.is_file(): | |
print("Cannot find existing Caddy install, exiting") | |
return 1 | |
if systemd_svc and not is_caddy_running(systemd_svc): | |
print("Cannot confirm Caddy is running before upgrade, exiting") | |
return 2 | |
new_caddy = download_file(caddy_url) | |
try: | |
if hash_file(caddy_exec) == hash_file(new_caddy): | |
print("Caddy already up-to-date") | |
return 0 | |
if systemd_svc: | |
backup_file = caddy_exec.parent / "{}.bak".format(caddy_exec.name) | |
run(["/usr/bin/systemctl", "stop", systemd_svc], check=True) | |
caddy_exec.rename(backup_file) | |
copy(new_caddy, caddy_exec) | |
run(["/usr/bin/systemctl", "start", systemd_svc], check=True) | |
if not is_caddy_running(systemd_svc): | |
backup_file.rename(caddy_exec) | |
run(["/usr/bin/systemctl", "start", systemd_svc], check=True) | |
print("Upgrade failed, reverted to previous version") | |
return 3 | |
else: | |
backup_file.unlink() | |
else: | |
copy(new_caddy, caddy_exec) | |
finally: | |
new_caddy.unlink() | |
print("Caddy successfully updated") | |
return 0 | |
def cli() -> None: | |
parser = ArgumentParser(description="Caddy auto-update script") | |
parser.add_argument( | |
"--url", | |
default="https://caddyserver.com/api/download?os=linux&arch=amd64&p=github.com/caddy-dns/cloudflare", | |
help="URL to Caddy binary to use", | |
) | |
parser.add_argument("--file", default=Path("/usr/local/bin/caddy"), type=Path, help="Path to Caddy binary on disk") | |
parser.add_argument("--service", default=None, help="systemd service to restart/monitor on upgrade") | |
args = parser.parse_args() | |
exit(main(args.url, args.file, args.service)) | |
if __name__ == "__main__": | |
cli() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment