#!/usr/bin/env python3 """Food Safety deployment agent. Standard-library only; Docker is optional at bootstrap.""" import argparse import hashlib import hmac import json import os import platform import re import shutil import socket import subprocess import sys import time import urllib.error import urllib.parse import urllib.request INSTALLER = re.compile(r"^foodsafe-installer-[A-Za-z0-9._-]+-linux-amd64\.run$") def normalize_control_url(value): parsed = urllib.parse.urlparse(value.rstrip("/") + "/") local = parsed.hostname in {"127.0.0.1", "localhost", "::1"} if parsed.scheme != "https" and not (parsed.scheme == "http" and local): raise ValueError("control URL must use HTTPS") if parsed.username or parsed.password or not parsed.hostname: raise ValueError("invalid control URL") return value.rstrip("/") def validate_installer(value): return bool(INSTALLER.fullmatch(str(value or ""))) def host_facts(): os_name = platform.platform() try: values = {} with open("/etc/os-release", encoding="utf-8") as handle: for line in handle: if "=" in line: key, value = line.rstrip().split("=", 1) values[key] = value.strip('"') os_name = values.get("PRETTY_NAME", os_name) except OSError: pass try: ips = sorted({item[4][0] for item in socket.getaddrinfo(socket.gethostname(), None) if item[4][0] not in {"127.0.0.1", "::1"}}) except socket.gaierror: ips = [] return { "hostname": socket.gethostname(), "os": os_name, "arch": platform.machine(), "ips": ips[:12], "docker": shutil.which("docker") is not None, } class Client: def __init__(self, base_url, agent_id=None, token=None): self.base_url = normalize_control_url(base_url) self.agent_id = agent_id self.token = token def request(self, method, path, body=None, timeout=60): headers = {"User-Agent": "foodsafe-agent/0.1", "Accept": "application/json", "ngrok-skip-browser-warning": "true"} if self.token: headers["Authorization"] = "Bearer " + self.token data = None if body is not None: data = json.dumps(body, ensure_ascii=False).encode() headers["Content-Type"] = "application/json" request = urllib.request.Request(self.base_url + "/" + path.lstrip("/"), data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=timeout) as response: if response.status == 204: return None return json.loads(response.read().decode()) except urllib.error.HTTPError as error: message = error.read().decode(errors="replace")[:500] raise RuntimeError("control API HTTP %s: %s" % (error.code, message)) from error def download(self, path, destination): request = urllib.request.Request( self.base_url + "/" + path.lstrip("/"), headers={"Authorization": "Bearer " + self.token, "User-Agent": "foodsafe-agent/0.1", "ngrok-skip-browser-warning": "true"}, ) with urllib.request.urlopen(request, timeout=30 * 60) as response, open(destination, "wb") as output: shutil.copyfileobj(response, output) def write_config(path, config): os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True) temporary = path + ".tmp" with open(temporary, "w", encoding="utf-8") as handle: json.dump(config, handle, ensure_ascii=False, indent=2) handle.write("\n") os.chmod(temporary, 0o600) os.replace(temporary, path) def read_config(path): with open(path, encoding="utf-8") as handle: return json.load(handle) def report_log(client, task_id, message): clean = re.sub(r"(?i)(password|secret|token)(\s*[=:]\s*)\S+", r"\1\2[REDACTED]", message.rstrip()) try: client.request("POST", "api/agents/%s/tasks/%s/logs" % (client.agent_id, task_id), {"message": clean[:4000]}) except Exception: pass def run_task(client, task, state_root): installer = task.get("installer") or {} name = installer.get("name") expected = installer.get("sha256") expected_size = installer.get("size") if (not validate_installer(name) or not re.fullmatch(r"[a-f0-9]{64}", str(expected or "")) or not isinstance(expected_size, int) or expected_size < 1 or expected_size > 10 * 1024 ** 3): raise RuntimeError("invalid installer metadata") job_dir = os.path.join(state_root, "jobs", task["id"]) os.makedirs(job_dir, mode=0o700, exist_ok=True) installer_path = os.path.join(job_dir, name) report_log(client, task["id"], "正在下载并校验安装器") client.download("api/agents/%s/tasks/%s/artifact" % (client.agent_id, task["id"]), installer_path) if os.path.getsize(installer_path) != expected_size: raise RuntimeError("installer size mismatch") digest = hashlib.sha256() with open(installer_path, "rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) if not hmac.compare_digest(digest.hexdigest(), expected): raise RuntimeError("installer SHA256 mismatch") os.chmod(installer_path, 0o700) facts = host_facts() server_ip = next((ip for ip in facts["ips"] if ":" not in ip), facts["hostname"]) command = ["bash", installer_path, "install", "--server-ip", server_ip, "--profile", task.get("profile", "core")] report_log(client, task["id"], "校验通过，开始安装或升级") process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1) for line in process.stdout: report_log(client, task["id"], line) code = process.wait() if code != 0: raise RuntimeError("installer exited with code %s" % code) return {"installerExit": code, "docker": shutil.which("docker") is not None} def enroll(args): client = Client(args.control_url) result = client.request("POST", "api/agents/enroll", {"code": args.enroll_code, "facts": host_facts()}) config = { "control_url": client.base_url, "agent_id": result["agent"]["id"], "token": result["token"], "state_root": args.state_root, } write_config(args.config, config) print("registered agent %s" % result["agent"]["id"]) def daemon(args): config = read_config(args.config) client = Client(config["control_url"], config["agent_id"], config["token"]) while True: try: client.request("POST", "api/agents/%s/heartbeat" % client.agent_id, host_facts()) response = client.request("GET", "api/agents/%s/tasks/next" % client.agent_id, timeout=70) task = response.get("task") if response else None if task: try: checks = run_task(client, task, config.get("state_root", args.state_root)) client.request("POST", "api/agents/%s/tasks/%s/complete" % (client.agent_id, task["id"]), {"success": True, "checks": checks}) except Exception as error: report_log(client, task["id"], "部署失败：%s" % error) client.request("POST", "api/agents/%s/tasks/%s/complete" % (client.agent_id, task["id"]), {"success": False, "error": str(error)[:1000]}) except Exception as error: print("agent warning: %s" % error, file=sys.stderr, flush=True) time.sleep(args.interval) def main(): parser = argparse.ArgumentParser() parser.add_argument("--config", default="/etc/foodsafe-agent/config.json") parser.add_argument("--state-root", default="/var/lib/foodsafe-agent") parser.add_argument("--control-url") parser.add_argument("--enroll-code") parser.add_argument("--interval", type=int, default=8) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: print(json.dumps({ "https": normalize_control_url("https://release.example") == "https://release.example", "rejectsHttp": _rejects_http(), "validInstaller": validate_installer("foodsafe-installer-0.1.8-linux-amd64.run"), })) return if args.enroll_code: if not args.control_url: parser.error("--control-url is required with --enroll-code") enroll(args) else: daemon(args) def _rejects_http(): try: normalize_control_url("http://release.example") return False except ValueError: return True if __name__ == "__main__": main()