#!/usr/bin/env python3 """ mount-guard.py — Pre-flight check for bind mount paths before `docker stack deploy`. Usage: mount-guard.py Reads the FINAL rendered compose YAML (after env substitution, right before it would be handed to `docker stack deploy -c -`) and checks: 1. MISSING PATHS — every bind-mount source path referenced by any service actually exists on disk. Swarm rejects the task at schedule time if not (see: "bind source path does not exist"), but catching it here is faster and clearer, and gives the option to create it on the spot. 2. SUSPICIOUS POSTGRES DATA DIRS — for any bind mount whose *target* looks like a Postgres data directory, warn if the *source* exists but is empty or missing a PG_VERSION file. This is the class of near-miss from the 2026-08-26 Immich incident: a wrong-but-existing empty path would have let Postgres silently initialize a brand-new database while the real data sat orphaned elsewhere, with no error at all. Exit codes: 0 = safe to deploy, 1 = blocked / aborted. Interactive sessions get a prompt with remediation options. Non-interactive sessions (Woodpecker, cron, CI) NEVER auto-proceed past a finding here — this is a data-safety check, not a convenience autofix. """ import sys import os import re import yaml def is_bind_like(path): return isinstance(path, str) and ( path.startswith('/') or path.startswith('./') or path.startswith('../') ) def gather_bind_mounts(doc): """Return list of (service_name, target, source) for host-path bind mounts. Handles both compose-file syntaxes that can appear in a rendered stack: - short form: "hostpath:containerpath[:mode]" (raw single-file path) - long form: {type: bind, source: ..., target: ...} (post `docker compose config`) Named volumes (source has no leading '/', './', '../') are intentionally skipped. """ results = [] services = doc.get('services') or {} for svc_name, svc in services.items(): if not isinstance(svc, dict): continue for v in (svc.get('volumes') or []): if isinstance(v, str): parts = v.split(':') if len(parts) >= 2 and is_bind_like(parts[0]): results.append((svc_name, parts[1], parts[0])) elif isinstance(v, dict): if v.get('type') == 'bind': src = v.get('source') tgt = v.get('target') if is_bind_like(src): results.append((svc_name, tgt, src)) # devices: short form "host:container[:mode]" (rarely used in this repo — # devices are conventionally expressed as bind-mounted volumes instead, # for Docker Swarm / DDM compatibility — but handle it if present). for d in (svc.get('devices') or []): if isinstance(d, str): parts = d.split(':') if parts and is_bind_like(parts[0]): tgt = parts[1] if len(parts) > 1 else parts[0] results.append((svc_name, tgt, parts[0])) return results _PG_TARGET_RE = re.compile(r'postgres', re.IGNORECASE) def looks_like_postgres_target(target): return bool(target) and bool(_PG_TARGET_RE.search(target)) def check_postgres_data(path): """Return a warning string if `path` looks like an empty/uninitialized Postgres data directory. Returns None if it looks fine. Caller must ensure `path` already exists before calling this.""" if not os.path.isdir(path): # Exists but isn't a directory (e.g. a file) — a different, separate # problem; the missing-path style check doesn't apply, but this is # clearly wrong too. return "exists but is not a directory" try: entries = os.listdir(path) except PermissionError: return "cannot list directory contents (permission denied) — unable to verify" if not entries: return "directory exists but is EMPTY — looks like an uninitialized/wrong Postgres data dir" if 'PG_VERSION' not in entries: return "directory exists and is non-empty but has no PG_VERSION file — does not look like a real Postgres data directory" return None def prompt(question, choices): """choices: dict of key -> description. Returns the chosen key (lowercase).""" print("Options:") for k, desc in choices.items(): print(f" [{k}] {desc}") while True: ans = input(f"{question} [{'/'.join(choices.keys())}]: ").strip().lower() if ans in choices: return ans print(f"Please enter one of: {', '.join(choices.keys())}") def main(): if len(sys.argv) != 2: print("Usage: mount-guard.py ", file=sys.stderr) return 1 rendered_path = sys.argv[1] with open(rendered_path) as f: doc = yaml.safe_load(f) if not doc or 'services' not in doc: print("!! mount-guard: rendered YAML has no 'services' key — refusing to guess, blocking.") return 1 mounts = gather_bind_mounts(doc) missing = [] pg_warnings = [] for svc_name, target, src in mounts: if not os.path.exists(src): missing.append((svc_name, target, src)) continue if looks_like_postgres_target(target): warn = check_postgres_data(src) if warn: pg_warnings.append((svc_name, target, src, warn)) if not missing and not pg_warnings: print(f"==> mount-guard: {len(mounts)} bind mount path(s) checked, all present and sane. OK to deploy.") return 0 print("!! mount-guard found potential issues with bind mount paths:\n") if missing: print("MISSING PATHS (Swarm will reject the task if these aren't created first):") for svc_name, target, src in missing: print(f" - service '{svc_name}': {src} (mounted at {target})") print() if pg_warnings: print("SUSPICIOUS POSTGRES DATA DIRECTORIES:") print(" Path exists, but contents don't look like a real Postgres data dir.") print(" This is the exact shape of the 2026-08-26 Immich near-miss: a wrong") print(" bind path that HAPPENED to exist (empty) would have let Postgres") print(" silently init a new empty DB while the real data sat orphaned") print(" elsewhere — with no error or warning from Docker at all.") for svc_name, target, src, warn in pg_warnings: print(f" - service '{svc_name}': {src} (mounted at {target})") print(f" {warn}") print() interactive = sys.stdin.isatty() if not interactive: print("(non-interactive session — mount-guard will NOT auto-proceed on any finding above)") print("Re-run interactively to review and decide, or fix the paths and retry.") return 1 if missing: choices = { 'a': 'Abort deploy (recommended if any path above is unexpected)', 'm': 'mkdir -p the missing path(s) now, then continue', 'c': 'Continue anyway without creating them (will likely fail at Swarm scheduling)', } ans = prompt("Missing bind mount paths found. Choice", choices) if ans == 'a': print("Aborting deploy.") return 1 if ans == 'm': for _, _, src in missing: print(f" mkdir -p {src}") os.makedirs(src, exist_ok=True) print("==> Created missing path(s). Continuing.") # 'c' falls through and continues without creating if pg_warnings: choices = { 'a': 'Abort deploy (recommended unless you intended a fresh Postgres init here)', 'c': 'Continue anyway (I have verified this is expected, e.g. legitimate first-time init)', } ans = prompt("Suspicious Postgres data directory found. Choice", choices) if ans == 'a': print("Aborting deploy.") return 1 print("==> Continuing deploy despite Postgres data warning, per operator confirmation.") return 0 if __name__ == '__main__': sys.exit(main())