#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import logging import os import shlex import shutil import subprocess import sys from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from urllib import error, request logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") LOGGER = logging.getLogger("pullio") LABEL_PREFIX = "org.hotio.pullio" @dataclass(frozen=True) class Config: compose_binary: str docker_binary: str cache_location: Path tag: str parallel: int compose_type: str script_hash: str telegram_bot_token: str telegram_chat_id: str def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Pull and optionally update Docker Compose containers based on labels.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--tag", default="") parser.add_argument("--debug", action="store_true") parser.add_argument("--parallel", type=int, default=1) args = parser.parse_args() if args.parallel < 1: parser.error("--parallel must be >= 1") return args def run_command( command: list[str], *, check: bool = True, env: dict[str, str] | None = None, input_text: str | None = None, ) -> str: completed = subprocess.run( command, check=check, capture_output=True, text=True, env=env, input=input_text, ) return completed.stdout.strip() def detect_compose_type(compose_binary: str, docker_binary: str) -> str: if compose_binary: return "V1" if docker_binary: try: run_command([docker_binary, "compose", "version"]) return "V2" except subprocess.CalledProcessError: return "NONE" return "NONE" def docker_inspect_value(docker_binary: str, name: str, template: str) -> str: try: value = run_command([docker_binary, "inspect", f"--format={template}", name]) except subprocess.CalledProcessError: return "" if value == "": return "" return value def docker_image_inspect_value(docker_binary: str, image: str, template: str) -> str: try: value = run_command( [docker_binary, "image", "inspect", f"--format={template}", image] ) except subprocess.CalledProcessError: return "" if value == "": return "" return value def compose_pull(config: Config, workdir: str, service: str) -> bool: if config.compose_type == "V1": cmd = [config.compose_binary, "pull", service] elif config.compose_type == "V2": cmd = [config.docker_binary, "compose", "pull", service] elif config.docker_binary: cmd = [ config.docker_binary, "run", "--rm", "-v", "/var/run/docker.sock:/var/run/docker.sock", "-v", f"{workdir}:{workdir}", f"-w={workdir}", "linuxserver/docker-compose", "pull", service, ] else: LOGGER.error( "Neither Docker Compose nor Docker binary is available. Cannot pull." ) return False try: run_command(cmd) return True except subprocess.CalledProcessError: return False def compose_up(config: Config, workdir: str, service: str) -> bool: if config.compose_type == "V1": cmd = [config.compose_binary, "up", "-d", "--always-recreate-deps", service] elif config.compose_type == "V2": cmd = [ config.docker_binary, "compose", "up", "-d", "--always-recreate-deps", service, ] elif config.docker_binary: cmd = [ config.docker_binary, "run", "--rm", "-v", "/var/run/docker.sock:/var/run/docker.sock", "-v", f"{workdir}:{workdir}", f"-w={workdir}", "linuxserver/docker-compose", "up", "-d", "--always-recreate-deps", service, ] else: LOGGER.error( "Neither Docker Compose nor Docker binary is available. Cannot bring up services." ) return False try: run_command(cmd) return True except subprocess.CalledProcessError: return False def post_json(url: str, payload: dict[str, object]) -> None: body = json.dumps(payload).encode("utf-8") req = request.Request( url=url, data=body, headers={ "User-Agent": "Pullio", "Content-Type": "application/json", }, method="POST", ) try: with request.urlopen(req) as response: response.read() except (error.HTTPError, error.URLError) as exc: LOGGER.warning("Webhook request failed: %s", exc) def now_iso_utc() -> str: return ( datetime.now(timezone.utc) .isoformat(timespec="milliseconds") .replace("+00:00", "Z") ) def parse_script_command(value: str) -> list[str]: return shlex.split(value) if value else [] def prepare_script_env( *, container: str, image: str, avatar: str, old_image_id: str, new_image_id: str, old_version: str, new_version: str, old_revision: str, new_revision: str, compose_service: str, compose_workdir: str, author_url: str, ) -> dict[str, str]: env = os.environ.copy() env.update( { "PULLIO_CONTAINER": container, "PULLIO_IMAGE": image, "PULLIO_AVATAR": avatar, "PULLIO_OLD_IMAGE_ID": old_image_id, "PULLIO_NEW_IMAGE_ID": new_image_id, "PULLIO_OLD_VERSION": old_version, "PULLIO_NEW_VERSION": new_version, "PULLIO_OLD_REVISION": old_revision, "PULLIO_NEW_REVISION": new_revision, "PULLIO_COMPOSE_SERVICE": compose_service, "PULLIO_COMPOSE_WORKDIR": compose_workdir, "PULLIO_AUTHOR_URL": author_url, } ) return env def send_telegram_notification( *, status: str, container_name: str, old_version: str, new_version: str, image_name: str, bot_token: str, chat_id: str, old_revision: str, new_revision: str, old_image_id: str, new_image_id: str, color: int, author_avatar: str, author_url: str, ) -> None: version_indicator = "=" if old_version == new_version else ">" revision_indicator = "=" if old_revision == new_revision else ">" digest_indicator = "=" if old_image_id == new_image_id else ">" lines = [ f"{container_name}", status.replace("\\n", " "), f"Image: {image_name}", f"Image ID: {old_image_id[:11]} {digest_indicator} {new_image_id[:11]}", ] if old_version and new_version: lines.append(f"Version: {old_version} {version_indicator} {new_version}") if old_revision and new_revision: lines.append( f"Revision: {old_revision[:6]} {revision_indicator} {new_revision[:6]}" ) if author_url: lines.append(f"URL: {author_url}") if author_avatar: lines.append(f"Avatar: {author_avatar}") lines.append(f"Color: {color}") lines.append(f"Time: {now_iso_utc()}") payload = { "chat_id": chat_id, "text": "\n".join(lines), "disable_web_page_preview": True, } post_json(f"https://api.telegram.org/bot{bot_token}/sendMessage", payload) def send_generic_webhook( *, status_generic: str, container_name: str, old_version: str, new_version: str, image_name: str, webhook: str, old_revision: str, new_revision: str, old_image_id: str, new_image_id: str, avatar: str, author_url: str, ) -> None: payload = { "container": container_name, "image": image_name, "avatar": avatar, "old_image_id": old_image_id, "new_image_id": new_image_id, "old_version": old_version, "new_version": new_version, "old_revision": old_revision, "new_revision": new_revision, "type": status_generic, "url": author_url, "timestamp": now_iso_utc(), } post_json(webhook, payload) def process_container(config: Config, container_name: str) -> None: LOGGER.info("%s: Checking...", container_name) image_name = docker_inspect_value( config.docker_binary, container_name, "{{.Config.Image}}" ) container_image_digest = docker_inspect_value( config.docker_binary, container_name, "{{.Image}}" ) docker_compose_service = docker_inspect_value( config.docker_binary, container_name, '{{ index .Config.Labels "com.docker.compose.service" }}', ) docker_compose_version = docker_inspect_value( config.docker_binary, container_name, '{{ index .Config.Labels "com.docker.compose.version" }}', ) docker_compose_workdir = docker_inspect_value( config.docker_binary, container_name, '{{ index .Config.Labels "com.docker.compose.project.working_dir" }}', ) old_version = docker_inspect_value( config.docker_binary, container_name, '{{ index .Config.Labels "org.opencontainers.image.version" }}', ) old_revision = docker_inspect_value( config.docker_binary, container_name, '{{ index .Config.Labels "org.opencontainers.image.revision" }}', ) pullio_update = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.update" }}}}', ) pullio_notify = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.notify" }}}}', ) pullio_telegram_bot_token = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.telegram.bot_token" }}}}', ) pullio_telegram_chat_id = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.telegram.chat_id" }}}}', ) pullio_generic_webhook = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.generic.webhook" }}}}', ) pullio_script_update = parse_script_command( docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.script.update" }}}}', ) ) pullio_script_notify = parse_script_command( docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.script.notify" }}}}', ) ) pullio_registry_authfile = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.registry.authfile" }}}}', ) pullio_author_avatar = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.author.avatar" }}}}', ) pullio_author_url = docker_inspect_value( config.docker_binary, container_name, f'{{{{ index .Config.Labels "{LABEL_PREFIX}{config.tag}.author.url" }}}}', ) if not docker_compose_version or ( pullio_update != "true" and pullio_notify != "true" ): return if pullio_registry_authfile and Path(pullio_registry_authfile).is_file(): LOGGER.info("%s: Registry login...", container_name) try: auth = json.loads( Path(pullio_registry_authfile).read_text(encoding="utf-8") ) run_command( [ config.docker_binary, "login", "--username", str(auth.get("username", "")), "--password-stdin", str(auth.get("registry", "")), ], input_text=str(auth.get("password", "")), ) except (json.JSONDecodeError, OSError, subprocess.CalledProcessError) as exc: LOGGER.warning("%s: Registry login failed: %s", container_name, exc) LOGGER.info("%s: Pulling image...", container_name) if not compose_pull(config, docker_compose_workdir, docker_compose_service): LOGGER.error("%s: Pulling failed!", container_name) image_digest = docker_image_inspect_value(config.docker_binary, image_name, "{{.Id}}") new_version = docker_image_inspect_value( config.docker_binary, image_name, '{{ index .Config.Labels "org.opencontainers.image.version" }}', ) new_revision = docker_image_inspect_value( config.docker_binary, image_name, '{{ index .Config.Labels "org.opencontainers.image.revision" }}', ) status = "I've got an update waiting for me.\nGive it to me, please." status_generic = "update_available" color = 768753 if image_digest != container_image_digest and pullio_update == "true": script_env = prepare_script_env( container=container_name, image=image_name, avatar=pullio_author_avatar, old_image_id=container_image_digest.removeprefix("sha256:"), new_image_id=image_digest.removeprefix("sha256:"), old_version=old_version, new_version=new_version, old_revision=old_revision, new_revision=new_revision, compose_service=docker_compose_service, compose_workdir=docker_compose_workdir, author_url=pullio_author_url, ) if pullio_script_update: LOGGER.info("%s: Stopping container...", container_name) try: run_command([config.docker_binary, "stop", container_name]) except subprocess.CalledProcessError: LOGGER.warning( "%s: Failed to stop container before update script.", container_name ) LOGGER.info("%s: Executing update script...", container_name) try: subprocess.run(pullio_script_update, env=script_env, check=False) except OSError as exc: LOGGER.warning( "%s: Update script failed to start: %s", container_name, exc ) LOGGER.info("%s: Updating container...", container_name) if compose_up(config, docker_compose_workdir, docker_compose_service): status = "I just updated myself.\nFeeling brand spanking new again!" status_generic = "update_success" color = 3066993 else: LOGGER.error("%s: Updating container failed!", container_name) status = ( "I tried to update myself.\nIt didn't work out, I might need some help." ) status_generic = "update_failure" color = 15158332 notified_path = ( config.cache_location / f"{config.script_hash}-{container_name}.notified" ) try: notified_path.unlink(missing_ok=True) except OSError: LOGGER.warning("%s: Failed to clear notify cache file.", container_name) if image_digest != container_image_digest and pullio_notify == "true": notified_path = ( config.cache_location / f"{config.script_hash}-{container_name}.notified" ) try: notified_path.touch(exist_ok=True) notified_digest = notified_path.read_text(encoding="utf-8").strip() except OSError: notified_digest = "" if notified_digest != image_digest: script_env = prepare_script_env( container=container_name, image=image_name, avatar=pullio_author_avatar, old_image_id=container_image_digest.removeprefix("sha256:"), new_image_id=image_digest.removeprefix("sha256:"), old_version=old_version, new_version=new_version, old_revision=old_revision, new_revision=new_revision, compose_service=docker_compose_service, compose_workdir=docker_compose_workdir, author_url=pullio_author_url, ) if pullio_script_notify: LOGGER.info("%s: Executing notify script...", container_name) try: subprocess.run(pullio_script_notify, env=script_env, check=False) except OSError as exc: LOGGER.warning( "%s: Notify script failed to start: %s", container_name, exc ) old_digest_short = container_image_digest.removeprefix("sha256:") new_digest_short = image_digest.removeprefix("sha256:") effective_telegram_bot_token = ( pullio_telegram_bot_token or config.telegram_bot_token ) effective_telegram_chat_id = ( pullio_telegram_chat_id or config.telegram_chat_id ) if effective_telegram_bot_token and effective_telegram_chat_id: LOGGER.info("%s: Sending telegram notification...", container_name) send_telegram_notification( status=status, container_name=container_name, old_version=old_version, new_version=new_version, image_name=image_name, bot_token=effective_telegram_bot_token, chat_id=effective_telegram_chat_id, old_revision=old_revision, new_revision=new_revision, old_image_id=old_digest_short, new_image_id=new_digest_short, color=color, author_avatar=pullio_author_avatar, author_url=pullio_author_url, ) if pullio_generic_webhook: LOGGER.info("%s: Sending generic webhook...", container_name) send_generic_webhook( status_generic=status_generic, container_name=container_name, old_version=old_version, new_version=new_version, image_name=image_name, webhook=pullio_generic_webhook, old_revision=old_revision, new_revision=new_revision, old_image_id=old_digest_short, new_image_id=new_digest_short, avatar=pullio_author_avatar, author_url=pullio_author_url, ) try: notified_path.write_text(image_digest, encoding="utf-8") except OSError: LOGGER.warning("%s: Failed to write notify cache file.", container_name) def main() -> int: args = parse_args() if args.debug: logging.getLogger().setLevel(logging.DEBUG) compose_binary = os.getenv("COMPOSE_BINARY") or ( shutil.which("docker-compose") or "" ) docker_binary = os.getenv("DOCKER_BINARY") or (shutil.which("docker") or "") telegram_bot_token = os.getenv("TELEGRAM_BOT_TOKEN", "") telegram_chat_id = os.getenv("TELEGRAM_CHAT_ID", "") cache_location = Path("/tmp") tag = f".{args.tag}" if args.tag else "" compose_type = detect_compose_type(compose_binary, docker_binary) if not docker_binary: LOGGER.error("Docker binary not found.") return 1 try: script_hash = hashlib.sha1(Path(__file__).read_bytes()).hexdigest() except OSError: script_hash = hashlib.sha1( str(Path(__file__).resolve()).encode("utf-8") ).hexdigest() config = Config( compose_binary=compose_binary, docker_binary=docker_binary, cache_location=cache_location, tag=tag, parallel=args.parallel, compose_type=compose_type, script_hash=script_hash, telegram_bot_token=telegram_bot_token, telegram_chat_id=telegram_chat_id, ) LOGGER.info( 'Running with "DEBUG=%s", "TAG=%s", and "PARALLEL=%s".', args.debug, tag, args.parallel, ) try: raw_containers = run_command([docker_binary, "ps", "--format", "{{.Names}}"]) except subprocess.CalledProcessError as exc: LOGGER.error("Failed to list running containers: %s", exc) return 1 containers = sorted([line for line in raw_containers.splitlines() if line]) LOGGER.info( "Processing %s containers with parallelism of %s", len(containers), args.parallel, ) try: if args.parallel > 1: with ThreadPoolExecutor(max_workers=args.parallel) as executor: list( executor.map( lambda name: process_container(config, name), containers ) ) else: for container_name in containers: process_container(config, container_name) except KeyboardInterrupt: return 130 LOGGER.info("Pruning docker images...") try: run_command([docker_binary, "image", "prune", "--force"]) except subprocess.CalledProcessError as exc: LOGGER.warning("Image prune failed: %s", exc) return 0 if __name__ == "__main__": sys.exit(main())