#!/usr/bin/env python3 """UUID外置存储守护：自动发现容器、校验挂载视图、协调服务。""" import argparse import fcntl import json import logging import os from pathlib import Path import re import signal import stat import subprocess import time import tempfile LOG = logging.getLogger("usb-storage-guard") CONFIG = "/etc/usb-storage-guard.json" def run(args): result = subprocess.run(args, capture_output=True, text=True, timeout=35, check=False) if result.returncode: # 不转储Docker输出，避免环境、凭据或业务日志泄漏。 raise RuntimeError(f"命令失败({result.returncode}): {args[0]} {args[1]}") return result.stdout def unescape(value): return re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), value) def mounts(path="/proc/self/mountinfo"): result = [] for line in Path(path).read_text().splitlines(): left, right = line.split(" - ", 1) a, b = left.split(), right.split() result.append(dict(id=a[0], dev=a[2], root=unescape(a[3]), target=unescape(a[4]), options=a[5].split(","), fs=b[0])) return result def inside(path, parent): return path == parent or path.startswith(parent.rstrip("/") + "/") def load_config(path): cfg = json.loads(Path(path).read_text()) if not 2 <= cfg.get("poll_seconds", 10) <= 300: raise ValueError("检查间隔必须为2至300秒") if not 10 <= cfg.get("retry_seconds", 30) <= 3600: raise ValueError("重试间隔必须为10至3600秒") if not str(cfg["state_dir"]).startswith("/var/lib/"): raise ValueError("状态目录必须在系统盘/var/lib下") names, targets, services = set(), [], set() for disk in cfg["disks"]: name = disk["name"] if not re.fullmatch(r"[a-zA-Z0-9_-]+", name) or name in names: raise ValueError("存储标识非法或重复") names.add(name) if not re.fullmatch(r"[a-zA-Z0-9-]+", disk["uuid"]): raise ValueError("UUID格式非法") target = disk["mountpoint"] if not target.startswith("/") or target == "/" or os.path.normpath(target) != target: raise ValueError("挂载点必须是规范绝对路径且不能为根目录") if any(inside(target, t) or inside(t, target) for t in targets): raise ValueError("挂载点不能重复或嵌套") targets.append(target) expected_unit = run(["systemd-escape", "--path", "--suffix=mount", target]).strip() if disk["mount_unit"] != expected_unit: raise ValueError("挂载单元与挂载点不匹配") for service in disk.get("services", []): if not re.fullmatch(r"[a-zA-Z0-9@_.-]+\.service", service) or service in services: raise ValueError("服务单元名称非法或重复") services.add(service) for path in disk.get("required_paths", []): if not inside(path, target) or os.path.normpath(path) != path: raise ValueError("业务目录必须位于挂载点下") return cfg def probe(disk, table=None): """只读内核挂载表与udev设备链接，不扫描硬盘目录。""" table = mounts() if table is None else table matches = [m for m in table if m["target"] == disk["mountpoint"] and m["fs"] != "autofs"] if not matches: return None m = matches[-1] try: device = os.stat("/dev/disk/by-uuid/" + disk["uuid"]) major, minor = map(int, m["dev"].split(":")) if not stat.S_ISBLK(device.st_mode) or device.st_rdev != os.makedev(major, minor): return None if not Path(f"/sys/dev/block/{m['dev']}").exists(): return None except OSError: return None if m["fs"] != disk["fstype"] or "rw" not in m["options"] or m["root"] != "/": return None return m["id"], m["dev"] def paths_ready(disk, extra=(), include_required=True): expected = os.stat("/dev/disk/by-uuid/" + disk["uuid"]).st_rdev for value in [*(disk.get("required_paths", []) if include_required else []), *extra]: p = Path(value) if not inside(str(p.resolve()), disk["mountpoint"]): raise RuntimeError("业务路径经符号链接指向外盘之外") if not p.exists() or p.stat().st_dev != expected: raise RuntimeError("业务路径不存在或不是目标硬盘") def inspect_containers(): ids = run(["docker", "ps", "-aq"]).split() result = [] # 分批避免命令行长度限制；inspect内容仅在内存，不写入日志/归档。 for offset in range(0, len(ids), 50): result.extend(json.loads(run(["docker", "inspect", *ids[offset:offset + 50]]))) return result def external_binds(container, disk): return [m for m in container["Mounts"] if m["Type"] == "bind" and inside(m["Source"], disk["mountpoint"])] def dependencies(container, disks): labels = container["Config"].get("Labels") or {} if labels.get("com.docker.compose.oneoff", "false").lower() == "true": return [] if labels.get("usb-storage-guard.mode") == "ignore": return [] return [d for d in disks if external_binds(container, d)] def container_view_ready(container, disk, token): if not container["State"]["Running"] or container["State"].get("Restarting"): return False try: table = mounts(f"/proc/{container['State']['Pid']}/mountinfo") bindings = external_binds(container, disk) return bool(bindings) and all(any(m["target"] == b["Destination"] and m["dev"] == token[1] and m["fs"] == disk["fstype"] and m["root"] == (b["Source"][len(disk["mountpoint"]):] or "/") for m in table) for b in bindings) except OSError: return False def atomic_json(path, data): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) fd, temp = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=path.parent) with os.fdopen(fd, "w", encoding="utf-8") as f: os.fchmod(f.fileno(), 0o600) json.dump(data, f, ensure_ascii=False, indent=2) f.flush() os.fsync(f.fileno()) os.replace(temp, path) # 目录同步确保改名后的停止意图在掉电/重启后仍持久存在。 fd = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) def policy_arg(policy): name = policy["Name"] maximum = policy.get("MaximumRetryCount", 0) return f"{name}:{maximum}" if name == "on-failure" and maximum else name def docker_root(): # 本适配器只支持本机Unix socket；禁止将远端Docker状态和本机文件混用。 endpoint = os.environ.get("DOCKER_HOST") if not endpoint: endpoint = run(["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"]).strip() if not endpoint.startswith("unix://"): raise RuntimeError("手动停止保护仅支持本机Docker Unix socket") root = run(["docker", "info", "--format", "{{.DockerRootDir}}"]).strip() if not root.startswith("/") or os.path.normpath(root) != root or root == "/": raise RuntimeError("Docker数据根目录无效，拒绝猜测停止状态") return Path(root) def docker_control(root, ident): """只读本机持久元数据；绝不修改Docker内部文件或输出Config中的凭据。""" if not re.fullmatch(r"[0-9a-f]{64}", ident): raise RuntimeError("读取停止状态需要完整容器ID") data = json.loads((root / "containers" / ident / "config.v2.json").read_text()) if data.get("ID") != ident or type(data.get("HasBeenManuallyStopped")) is not bool: raise RuntimeError("Docker停止元数据不兼容，拒绝自动启动") state = data.get("State", {}) if (type(state.get("Running")) is not bool or type(state.get("Restarting")) is not bool or not isinstance(state.get("StartedAt"), str) or not isinstance(state.get("FinishedAt"), str)): raise RuntimeError("Docker运行元数据不兼容，拒绝自动启动") return dict(manual=data["HasBeenManuallyStopped"], running=state["Running"], restarting=state.get("Restarting", False), started=state["StartedAt"], finished=state["FinishedAt"]) def service_state(service): fields = ["ActiveState", "SubState", "InvocationID", "Result"] return dict(line.split("=", 1) for line in run(["systemctl", "show", service, *[arg for field in fields for arg in ("-p", field)]]).splitlines() if "=" in line) def service_intent_path(cfg, service): if not re.fullmatch(r"[a-zA-Z0-9@_.-]+\.service", service): raise RuntimeError("服务名称非法") return Path(cfg["state_dir"], service + ".intent.json") def read_service_intent(cfg, service): path = service_intent_path(cfg, service) return json.loads(path.read_text()) if path.exists() else {} def service_hook(cfg, service, starting): """systemd同步钩子：即使守护已停止或用户随即重启，也先落盘停服意图。""" disks = [d for d in cfg["disks"] if service in d.get("services", [])] if not disks: raise RuntimeError("服务未登记，拒绝修改停止意图") path = service_intent_path(cfg, service) intent = read_service_intent(cfg, service) if starting: atomic_json(path, dict(manual=False)) return if run(["systemctl", "show", "-p", "SystemState", "--value"]).strip() == "stopping": return if intent.get("guard_invocation") and intent["guard_invocation"] == os.environ.get("INVOCATION_ID"): return # mount绑定导致的停止不属于用户停服；失败由原Restart策略/守护恢复。 jobs = run(["systemctl", "list-jobs", "--no-legend", "--plain"]).splitlines() stopping = any(len(fields := line.split()) >= 3 and fields[1:3] == [service, "stop"] for line in jobs) if not all(probe(d) for d in disks): return if not stopping and os.environ.get("SERVICE_RESULT") != "success": return atomic_json(path, dict(manual=True, reason="正常停止，保持停服至显式启动")) def prepare_services(cfg): """升级迁移：只读取当前状态，已有停服意图不覆盖，不启动任何服务。""" for disk in cfg["disks"]: for service in disk.get("services", []): path = service_intent_path(cfg, service) if not path.exists(): state = service_state(service)["ActiveState"] atomic_json(path, dict(manual=state in ("inactive", "deactivating"), reason="升级时保留现状")) class Guard: def __init__(self, cfg): self.cfg = cfg self.tokens = {} self.retry = {} self.messages = {} self.docker_root = None self.registry_path = Path(cfg["state_dir"], "containers.json") self.registry = json.loads(self.registry_path.read_text()) if self.registry_path.exists() else {} def message(self, name, message): if self.messages.get(name) != message: LOG.info("%s: %s", name, message) self.messages[name] = message def save(self): atomic_json(self.registry_path, self.registry) def paused(self, name): return Path(self.cfg["state_dir"], name + ".paused").exists() def may_retry(self, key): now = time.monotonic() if now < self.retry.get(key, 0): return False self.retry[key] = now + self.cfg.get("retry_seconds", 30) return True def reconcile_service(self, disk, token, old): available = token is not None and not self.paused(disk["name"]) for service in disk.get("services", []): state = service_state(service) intent = read_service_intent(self.cfg, service) # 已有服务管理器的锁止仍是最高优先级，绝不反复提交被条件拒绝的启动。 blocked = Path("/var/lib/local-service-manager", service + ".stopped").exists() if intent.get("manual") or blocked or self.paused(service): self.message(service, "手动停止，保持停服（跨开机保留）") continue if state["ActiveState"] == "deactivating": # ExecStopPost尚未持久化意图，绝不能排队反向start。 continue active = state["ActiveState"] in ("active", "activating", "reloading") if not available: if active: atomic_json(service_intent_path(self.cfg, service), dict(manual=False, guard_invocation=state["InvocationID"])) run(["systemctl", "stop", "--no-block", service]) elif not active or (old and old != token): if self.may_retry(service): paths_ready(disk) if probe(disk) != token: raise RuntimeError("服务启动前硬盘状态变化") # 再读一次持久意图，避免目录检查期间用户刚停止服务被重新拉起。 if (read_service_intent(self.cfg, service).get("manual") or self.paused(service) or Path("/var/lib/local-service-manager", service + ".stopped").exists() or service_state(service)["ActiveState"] == "deactivating"): continue if active: atomic_json(service_intent_path(self.cfg, service), dict(manual=False, guard_invocation=state["InvocationID"])) run(["systemctl", "restart" if active else "start", "--no-block", service]) def enroll_and_clean(self, containers): current = {c["Id"]: c for c in containers} for ident in list(self.registry): c = current.get(ident) if c is None or not dependencies(c, self.cfg["disks"]): if c and c["HostConfig"]["RestartPolicy"]["Name"] == "no": run(["docker", "update", "--restart=" + policy_arg(self.registry[ident]["original_policy"]), ident]) del self.registry[ident] Path(self.cfg["state_dir"], ident + ".paused").unlink(missing_ok=True) Path(self.cfg["state_dir"], ident + ".resume.json").unlink(missing_ok=True) self.save() self.messages.pop(ident, None) self.retry.pop(ident, None) self.message("registry", "已清理删除或不再依赖外盘的容器记录") for c in containers: deps = dependencies(c, self.cfg["disks"]) if not deps: continue ident = c["Id"] policy = c["HostConfig"]["RestartPolicy"] if ident not in self.registry: self.registry[ident] = dict(name=c["Name"], original_policy=policy, disks=[d["name"] for d in deps]) self.save() # 必须先保存原策略，再关闭Docker抢跑自启。 LOG.info("自动纳入外盘容器 %s", c["Name"]) elif (self.registry[ident]["disks"] != [d["name"] for d in deps] or self.registry[ident]["name"] != c["Name"]): self.registry[ident]["disks"] = [d["name"] for d in deps] self.registry[ident]["name"] = c["Name"] self.save() if policy["Name"] != "no": run(["docker", "update", "--restart=no", ident]) def control(self, ident): if self.docker_root is None: self.docker_root = docker_root() return docker_control(self.docker_root, ident) def observe_control(self, c): ident = c["Id"] control = self.control(ident) record = self.registry.setdefault(ident, {}) owner = record.get("guard_stop", {}) owned = bool(owner) and owner.get("started") == control["started"] and ( owner.get("finished") is None or owner["finished"] == control["finished"]) request = Path(self.cfg["state_dir"], ident + ".resume.json") resume = request.exists() and json.loads(request.read_text()) == control # 手工start会清除Docker的manual标志；启动后立即异常退出也能正确恢复。 manual = control["manual"] and not owned and not resume if record.get("manual_stopped") != manual: record["manual_stopped"] = manual self.save() if not control["manual"] and owner: record.pop("guard_stop", None) self.save() return control, manual def stop_container(self, c, control): # 先落盘责任归属，崩溃发生在stop调用内也不会把守护停止误认成用户停止。 ident = c["Id"] self.registry[ident]["guard_stop"] = dict(started=control["started"], finished=None) self.save() run(["docker", "stop", "-t", "20", ident]) stopped = self.control(ident) self.registry[ident]["guard_stop"] = dict(started=stopped["started"], finished=stopped["finished"]) self.save() return stopped def reconcile_container(self, c, tokens, old_tokens): deps = dependencies(c, self.cfg["disks"]) if not deps: return ident = c["Id"] control, manual = self.observe_control(c) if manual: self.message(ident, "手动停止，保持停止（跨开机保留）") return available = all(tokens[d["name"]] and not self.paused(d["name"]) for d in deps) available = available and not self.paused(ident) and not self.paused(c["Name"].lstrip("/")) running = control["running"] or control["restarting"] # Docker可能在本轮inspect后被手动start，留到下一轮刷新PID再检查视图。 if running and not c["State"]["Running"]: return if not available: if running: self.stop_container(c, control) self.message(ident, "外盘未就绪或维护暂停，容器保持停止") return last_mounts = self.registry.get(ident, {}).get("last_mounts", {}) changed = any((old_tokens.get(d["name"]) and old_tokens[d["name"]] != tokens[d["name"]]) or (last_mounts.get(d["name"]) and tuple(last_mounts[d["name"]]) != tokens[d["name"]]) for d in deps) correct = not changed and all(container_view_ready(c, d, tokens[d["name"]]) for d in deps) if correct: current_mounts = {d["name"]: list(tokens[d["name"]]) for d in deps} if ident in self.registry and self.registry[ident].get("last_mounts") != current_mounts: self.registry[ident]["last_mounts"] = current_mounts self.save() self.message(ident, "外盘挂载视图正常") return if not self.may_retry(ident): return # 视图不正确先停止，路径缺失时也不能让它继续访问旧目录。 if running: fresh, manual = self.observe_control(c) if manual or fresh != control: return control = self.stop_container(c, control) for d in deps: paths_ready(d, [b["Source"] for b in external_binds(c, d)], include_required=False) if probe(d) != tokens[d["name"]]: raise RuntimeError("容器启动前硬盘状态变化") # 路径检查可能耗时；再次检查手停、暂停与快照变化后才提交start。 fresh, manual = self.observe_control(c) if manual or fresh != control or self.paused(ident) or self.paused(c["Name"].lstrip("/")): return if any(self.paused(d["name"]) or probe(d) != tokens[d["name"]] for d in deps): return run(["docker", "start", ident]) Path(self.cfg["state_dir"], ident + ".resume.json").unlink(missing_ok=True) self.registry[ident].pop("guard_stop", None) self.registry[ident]["manual_stopped"] = False if ident in self.registry: self.registry[ident]["last_mounts"] = {d["name"]: list(tokens[d["name"]]) for d in deps} self.save() self.message(ident, "已重新建立容器外盘绑定 " + c["Name"]) def tick(self): table = mounts() tokens = {d["name"]: probe(d, table) for d in self.cfg["disks"]} for disk in self.cfg["disks"]: name, token = disk["name"], tokens[disk["name"]] try: self.reconcile_service(disk, token, self.tokens.get(name)) self.message(name, "维护暂停" if self.paused(name) else ("硬盘就绪" if token else "硬盘未就绪，相关服务停止")) if not token and not self.paused(name) and self.may_retry("mount:" + name): if Path("/dev/disk/by-uuid", disk["uuid"]).exists(): # 后台排队；设备不存在时绝不提交等待设备的挂载作业。 run(["systemctl", "start", "--no-block", disk["mount_unit"]]) except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as exc: self.message(name, f"协调失败，稍后重试: {exc}") try: containers = inspect_containers() self.enroll_and_clean(containers) for c in containers: try: self.reconcile_container(c, tokens, self.tokens) except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as exc: self.message(c["Id"], f"容器协调失败，稍后重试: {exc}") except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as exc: self.message("docker", f"Docker暂不可用，不影响宿主机服务处理: {exc}") self.tokens = tokens def release(self): """回滚接管：仅还原仍然存在且当前策略仍为no的容器。""" containers = {c["Id"]: c for c in inspect_containers()} for ident, record in list(self.registry.items()): if ident in containers and containers[ident]["HostConfig"]["RestartPolicy"]["Name"] == "no": run(["docker", "update", "--restart=" + policy_arg(record["original_policy"]), ident]) del self.registry[ident] self.save() def main(): parser = argparse.ArgumentParser(description="外置存储识别与服务自动恢复") parser.add_argument("--config", default=CONFIG) parser.add_argument("action", choices=["run", "once", "check", "pause", "resume", "status", "release", "service-start", "service-stop", "prepare-services"]) parser.add_argument("name", nargs="?") args = parser.parse_args() cfg = load_config(args.config) logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") if args.action in ("service-start", "service-stop"): service_hook(cfg, args.name, args.action == "service-start") return if args.action == "prepare-services": prepare_services(cfg) return if args.action in ("pause", "resume"): if not args.name or not re.fullmatch(r"[a-zA-Z0-9_.-]+", args.name): parser.error("请提供存储标识、容器名或容器ID") Path(cfg["state_dir"]).mkdir(parents=True, exist_ok=True, mode=0o700) marker = Path(cfg["state_dir"], args.name + ".paused") if args.action == "pause": marker.touch(mode=0o600) else: marker.unlink(missing_ok=True) # 恢复命令绑定确切停止快照；后来的用户stop不会被旧resume覆盖。 if args.name.endswith(".service"): if not any(args.name in d.get("services", []) for d in cfg["disks"]): parser.error("该服务未登记到USB守护") atomic_json(service_intent_path(cfg, args.name), dict(manual=False)) else: matches = [c for c in inspect_containers() if c["Name"].lstrip("/") == args.name or c["Id"] == args.name] if matches: control = docker_control(docker_root(), matches[0]["Id"]) atomic_json(Path(cfg["state_dir"], matches[0]["Id"] + ".resume.json"), control) print("已记录状态，守护将在下一个检查周期处理") return if args.action in ("check", "status"): failed = False for disk in cfg["disks"]: token = probe(disk) print(disk["name"], "就绪" if token else "未就绪", token) if args.action == "check": if token: paths_ready(disk) else: failed = True if args.action == "status": path = Path(cfg["state_dir"], "containers.json") print(path.read_text() if path.exists() else "尚无容器记录") for disk in cfg["disks"]: for service in disk.get("services", []): print(service, read_service_intent(cfg, service)) raise SystemExit(1 if failed else 0) Path(cfg["state_dir"]).mkdir(parents=True, exist_ok=True, mode=0o700) lock = Path(cfg["state_dir"], "guard.lock").open("a") try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: parser.error("该状态目录已有守护运行，请勿并发执行once/run/release") guard = Guard(cfg) if args.action == "release": if run(["systemctl", "show", "usb-storage-guard.service", "-p", "ActiveState", "--value"]).strip() in ("active", "activating"): parser.error("先停止守护服务，再解除接管") guard.release() return if args.action == "once": guard.tick() return stopping = False def stop(signum, frame): nonlocal stopping stopping = True signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) while not stopping: guard.tick() for _ in range(cfg.get("poll_seconds", 10)): if stopping: break time.sleep(1) if __name__ == "__main__": main()