Skip to content

Instantly share code, notes, and snippets.

@Caligatio
Created April 8, 2021 05:07
Show Gist options
  • Save Caligatio/f97ba88dad586ed2a0423f34ce67106b to your computer and use it in GitHub Desktop.
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.
#!/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