From 26d1c909f015be681e6115af16bdf1bbc4cd98ba Mon Sep 17 00:00:00 2001 From: zhx Date: Wed, 8 Jul 2026 19:05:41 +0800 Subject: [PATCH] =?UTF-8?q?=E9=83=A8=E7=BD=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .deploy/prod-deploy/ops/aslan_ops.py | 376 ++++++++++++++++++++++++++- 1 file changed, 374 insertions(+), 2 deletions(-) diff --git a/.deploy/prod-deploy/ops/aslan_ops.py b/.deploy/prod-deploy/ops/aslan_ops.py index e6f90f99..2899482f 100644 --- a/.deploy/prod-deploy/ops/aslan_ops.py +++ b/.deploy/prod-deploy/ops/aslan_ops.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 import base64 +import fcntl import hashlib import hmac import html @@ -19,6 +20,7 @@ BASE_DIR = Path(os.environ.get("ASLAN_DEPLOY_BASE", "/opt/aslan-deploy")) RUN_DIR = Path(os.environ.get("ASLAN_OPS_RUN_DIR", str(BASE_DIR / "ops" / "runs"))) KUBECONFIG = Path(os.environ.get("KUBECONFIG", str(BASE_DIR / "kubeconfig"))) DEPLOY_SCRIPT = Path(os.environ.get("ASLAN_DEPLOY_SCRIPT", str(BASE_DIR / "deploy-likei-services.sh"))) +LOCK_FILE = Path(os.environ.get("ASLAN_DEPLOY_LOCK_FILE", str(BASE_DIR / "deploy.lock"))) SOURCE_DIR = Path(os.environ.get("ASLAN_SOURCE_DIR", str(BASE_DIR / "source" / "likei-services"))) NAMESPACE = os.environ.get("ASLAN_NAMESPACE", "prod") GIT_REF = os.environ.get("ASLAN_GIT_REF", "aslan_prod") @@ -53,6 +55,8 @@ SERVICES = { } BRANCH_RE = re.compile(r"^[A-Za-z0-9._/-]{1,80}$") +ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +SENSITIVE_ENV_RE = re.compile(r"(PASSWORD|SECRET|TOKEN|KEY|CREDENTIAL)", re.IGNORECASE) job_lock = threading.Lock() active_job = None @@ -150,6 +154,255 @@ def run_text(cmd, timeout=10): return result.stdout.strip() +def run_checked_text(cmd, input_text=None, timeout=30): + result = subprocess.run(cmd, input=input_text, text=True, capture_output=True, timeout=timeout) + if result.returncode != 0: + raise RuntimeError(result.stderr.strip() or result.stdout.strip() or f"command failed: {cmd[0]}") + return result.stdout + + +def ensure_service(svc): + if svc not in SERVICES: + raise ValueError(f"unsupported service: {svc}") + return svc + + +def deployment_json(svc): + ensure_service(svc) + return run_json([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "get", + "deployment", + svc, + "-o", + "json", + ]) + + +def deployment_yaml(svc): + ensure_service(svc) + return run_checked_text([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "get", + "deployment", + svc, + "-o", + "yaml", + "--show-managed-fields=false", + ], timeout=20) + + +def service_container(deployment, svc): + containers = deployment.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) + for idx, container in enumerate(containers): + if container.get("name") == svc: + return idx, container + raise ValueError(f"deployment/{svc} does not contain container {svc}") + + +def single_kubernetes_object(payload): + if payload.get("kind") == "List": + items = payload.get("items", []) + if len(items) != 1: + raise ValueError("yaml must contain exactly one object") + return items[0] + return payload + + +def protected_env_unchanged(svc, current_deployment, new_deployment): + _, current_container = service_container(current_deployment, svc) + _, new_container = service_container(new_deployment, svc) + if new_container.get("envFrom", []) != current_container.get("envFrom", []): + raise ValueError("envFrom cannot be changed here") + new_by_name = { + entry.get("name"): entry + for entry in new_container.get("env", []) + if isinstance(entry, dict) + } + for entry in current_container.get("env", []): + if not isinstance(entry, dict): + continue + name = entry.get("name") + if not name: + continue + if ("valueFrom" in entry or SENSITIVE_ENV_RE.search(name)) and new_by_name.get(name) != entry: + raise ValueError(f"protected env {name} cannot be changed here") + + +def verify_deployment_yaml(svc, yaml_text): + ensure_service(svc) + if not yaml_text.strip(): + raise ValueError("yaml is empty") + if len(yaml_text.encode("utf-8")) > 600000: + raise ValueError("yaml is too large") + client_dry_run = run_checked_text([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "apply", + "--dry-run=client", + "-f", + "-", + "-o", + "json", + ], input_text=yaml_text, timeout=30) + obj = single_kubernetes_object(json.loads(client_dry_run)) + metadata = obj.get("metadata", {}) + if obj.get("apiVersion") != "apps/v1" or obj.get("kind") != "Deployment" or metadata.get("name") != svc: + raise ValueError(f"yaml must be deployment/{svc}") + if metadata.get("namespace") not in (None, "", NAMESPACE): + raise ValueError(f"yaml namespace must be {NAMESPACE}") + current = deployment_json(svc) + if metadata.get("resourceVersion") != current.get("metadata", {}).get("resourceVersion"): + raise ValueError("deployment changed since YAML was loaded; reload before saving") + if obj.get("spec", {}).get("selector") != current.get("spec", {}).get("selector"): + raise ValueError("deployment selector cannot be changed") + protected_env_unchanged(svc, current, obj) + run_checked_text([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "apply", + "--dry-run=server", + "-f", + "-", + "-o", + "json", + ], input_text=yaml_text, timeout=30) + return obj + + +def apply_deployment_yaml(svc, yaml_text): + marker = begin_mutation("yaml", svc) + try: + verify_deployment_yaml(svc, yaml_text) + # Apply only after server-side dry-run confirms the YAML still targets the whitelisted deployment. + output = run_checked_text([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "apply", + "-f", + "-", + ], input_text=yaml_text, timeout=60) + return output.strip() + finally: + finish_mutation(marker) + + +def env_config(svc): + deployment = deployment_json(svc) + _, container = service_container(deployment, svc) + return { + "service": svc, + "namespace": NAMESPACE, + "env": container.get("env", []), + "envFrom": container.get("envFrom", []), + } + + +def normalize_env_entries(entries, existing_entries): + if not isinstance(entries, list): + raise ValueError("env must be a list") + existing_by_name = {entry.get("name"): entry for entry in existing_entries if isinstance(entry, dict)} + normalized = [] + for entry in entries: + if not isinstance(entry, dict): + raise ValueError("env entries must be objects") + name = entry.get("name") + if not isinstance(name, str) or not ENV_NAME_RE.match(name): + raise ValueError(f"invalid env name: {name}") + item = {"name": name} + has_value = "value" in entry + has_value_from = "valueFrom" in entry + if has_value and has_value_from: + raise ValueError(f"env {name} cannot contain both value and valueFrom") + if has_value: + if not isinstance(entry["value"], str): + raise ValueError(f"env {name} value must be a string") + item["value"] = entry["value"] + if has_value_from: + if not isinstance(entry["valueFrom"], dict): + raise ValueError(f"env {name} valueFrom must be an object") + item["valueFrom"] = entry["valueFrom"] + if has_value_from and item != existing_by_name.get(name): + raise ValueError(f"env {name} valueFrom cannot be changed here") + if SENSITIVE_ENV_RE.search(name) and item != existing_by_name.get(name): + raise ValueError(f"sensitive env {name} cannot be changed here") + normalized.append(item) + normalized_by_name = {entry["name"]: entry for entry in normalized} + for entry in existing_entries: + if not isinstance(entry, dict): + continue + name = entry.get("name") + if not name: + continue + if ("valueFrom" in entry or SENSITIVE_ENV_RE.search(name)) and normalized_by_name.get(name) != entry: + raise ValueError(f"protected env {name} cannot be removed here") + return normalized + + +def normalize_env_from_entries(entries): + if not isinstance(entries, list): + raise ValueError("envFrom must be a list") + allowed_keys = {"configMapRef", "prefix", "secretRef"} + normalized = [] + for entry in entries: + if not isinstance(entry, dict): + raise ValueError("envFrom entries must be objects") + extra = set(entry) - allowed_keys + if extra: + raise ValueError(f"unsupported envFrom keys: {', '.join(sorted(extra))}") + normalized.append(entry) + return normalized + + +def apply_env_config(svc, payload): + marker = begin_mutation("env", svc) + try: + deployment = deployment_json(svc) + idx, container = service_container(deployment, svc) + existing_env = container.get("env", []) + existing_env_from = container.get("envFrom", []) + env = normalize_env_entries(payload.get("env", []), existing_env) + env_from = normalize_env_from_entries(payload.get("envFrom", existing_env_from)) + if env_from != existing_env_from: + raise ValueError("envFrom cannot be changed here") + patch = [] + env_path = f"/spec/template/spec/containers/{idx}/env" + patch.append({"op": "replace" if "env" in container else "add", "path": env_path, "value": env}) + run_checked_text([ + "kubectl", + "--kubeconfig", + str(KUBECONFIG), + "-n", + NAMESPACE, + "patch", + "deployment", + svc, + "--type=json", + "-p", + json.dumps(patch, ensure_ascii=False), + ], timeout=30) + return env_config(svc) + finally: + finish_mutation(marker) + + def deployment_status(): names = list(SERVICES) payload = run_json([ @@ -188,7 +441,7 @@ def deployment_status(): def pod_status(): - selector = "app in (other,external,console)" + selector = "app in (" + ",".join(SERVICES) + ")" payload = run_json([ "kubectl", "--kubeconfig", @@ -243,6 +496,50 @@ def save_job(job): (RUN_DIR / f"{job['id']}.json").write_text(json.dumps(job, ensure_ascii=False, indent=2)) +def audit_config_change(action, svc, client_ip, result): + RUN_DIR.mkdir(parents=True, exist_ok=True) + line = json.dumps({ + "time": now_ts(), + "ip": client_ip, + "action": action, + "service": svc, + "namespace": NAMESPACE, + "result": result, + }, ensure_ascii=False) + with (RUN_DIR / "config-audit.log").open("a", buffering=1) as log: + log.write(line + "\n") + + +def begin_mutation(operation, svc): + global active_job + LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + lock_handle = LOCK_FILE.open("a") + try: + # Share the same flock file with deploy-likei-services.sh so manual SSH deploys and config edits cannot interleave. + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + lock_handle.close() + raise RuntimeError(f"deploy lock is already held: {LOCK_FILE}") from exc + with job_lock: + if active_job: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) + lock_handle.close() + raise RuntimeError(f"operation already running: {active_job['id']}") + active_job = {"id": f"{operation}-{svc}-{int(time.time())}", "status": "running", "_lock": lock_handle} + return active_job + + +def finish_mutation(marker): + global active_job + with job_lock: + if active_job is marker: + active_job = None + lock_handle = marker.get("_lock") + if lock_handle: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) + lock_handle.close() + + def stream_process(job, cmd, env): log_path = RUN_DIR / f"{job['id']}.log" try: @@ -399,6 +696,9 @@ INDEX_HTML = """ th, td { text-align:left; padding:10px; border-top:1px solid var(--line); vertical-align:top; } th { color:var(--muted); font-weight:600; } pre { margin:0; height:360px; overflow:auto; background:#101820; color:#e6edf3; padding:14px; border-radius:8px; font-size:12px; line-height:1.45; white-space:pre-wrap; } + textarea.editor { width:100%; min-height:420px; resize:vertical; border:1px solid var(--line); border-radius:8px; padding:12px; font-family:ui-monospace,SFMono-Regular,Menlo,monospace; font-size:12px; line-height:1.45; background:#fbfdfe; color:var(--ink); } + select { height:34px; border:1px solid var(--line); border-radius:6px; background:#fff; color:var(--ink); padding:0 10px; } + .config-actions { display:flex; flex-wrap:wrap; gap:10px; align-items:center; } .split { display:grid; grid-template-columns:2fr 1fr; gap:16px; } @media (max-width: 900px) { header { padding:0 14px; } main { padding:14px; } .grid, .split { grid-template-columns:1fr; } .bar { align-items:flex-start; flex-direction:column; } } @@ -422,6 +722,17 @@ INDEX_HTML = """ +
+

配置

+
+ + + + + +
+ +

部署日志

无运行任务
@@ -440,6 +751,7 @@ INDEX_HTML = """ @@ -544,6 +891,15 @@ class Handler(BaseHTTPRequestHandler): }) except Exception as exc: return json_response(self, 500, {"error": str(exc)}) + match = re.match(r"^/api/config/([A-Za-z0-9_-]+)/(yaml|env)$", parsed.path) + if match: + svc, kind = match.groups() + try: + if kind == "yaml": + return json_response(self, 200, {"service": svc, "namespace": NAMESPACE, "yaml": deployment_yaml(svc)}) + return json_response(self, 200, env_config(svc)) + except Exception as exc: + return json_response(self, 400, {"error": str(exc)}) match = re.match(r"^/api/jobs/([A-Za-z0-9-]+)/log$", parsed.path) if match: job_id = match.group(1) @@ -588,6 +944,22 @@ class Handler(BaseHTTPRequestHandler): return json_response(self, 202, {"job": job}) except Exception as exc: return json_response(self, 400, {"error": str(exc)}) + match = re.match(r"^/api/config/([A-Za-z0-9_-]+)/(yaml|env)$", parsed.path) + if match: + svc, kind = match.groups() + try: + if self.headers.get("X-Aslan-Ops") != "1": + return json_response(self, 403, {"error": "missing deploy header"}) + payload = json.loads(self.read_body(max_bytes=900000) or "{}") + if kind == "yaml": + message = apply_deployment_yaml(svc, payload.get("yaml", "")) + audit_config_change("yaml", svc, self.client_address[0], message) + return json_response(self, 200, {"message": message}) + result = apply_env_config(svc, payload) + audit_config_change("env", svc, self.client_address[0], "patched") + return json_response(self, 200, result) + except Exception as exc: + return json_response(self, 400, {"error": str(exc)}) return text_response(self, 404, "not found") def log_message(self, fmt, *args):