#!/usr/bin/env python3 import base64 import fcntl import hashlib import hmac import html import json import os import re import secrets import subprocess import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse 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") REMOTE_URL = os.environ.get("ASLAN_GIT_REMOTE_URL", "git@gitea.haiyihy.com:hy/aslan-server.git") HOST = os.environ.get("ASLAN_OPS_HOST", "0.0.0.0") PORT = int(os.environ.get("ASLAN_OPS_PORT", "18080")) USERNAME = os.environ.get("ASLAN_OPS_USER", "admin") PASSWORD = os.environ.get("ASLAN_OPS_PASSWORD", "") SESSION_SECRET = os.environ.get("ASLAN_OPS_SESSION_SECRET", PASSWORD or secrets.token_hex(32)) ALLOWED_BRANCHES = { branch.strip() for branch in os.environ.get("ASLAN_ALLOWED_BRANCHES", GIT_REF).split(",") if branch.strip() } SERVICES = { "wallet": { "label": "Wallet", "port": "9000", "health": "/actuator/health", }, "other": { "label": "Other", "port": "5800", "health": "/actuator/health", }, "external": { "label": "External", "port": "5200", "health": "/actuator/health", }, "console": { "label": "Console", "port": "5300", "health": "/console/actuator/health", }, } 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 login_failures = {} def now_ts(): return time.strftime("%Y-%m-%d %H:%M:%S") def json_response(handler, status, payload): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Cache-Control", "no-store") handler.send_header("X-Frame-Options", "DENY") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def text_response(handler, status, body, content_type="text/plain; charset=utf-8"): data = body.encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", content_type) handler.send_header("Cache-Control", "no-store") handler.send_header("X-Frame-Options", "DENY") handler.send_header("Content-Length", str(len(data))) handler.end_headers() handler.wfile.write(data) def redirect_response(handler, location, cookies=None): handler.send_response(302) for cookie in cookies or []: handler.send_header("Set-Cookie", cookie) handler.send_header("Location", location) handler.send_header("Content-Length", "0") handler.end_headers() def sign_session(value): mac = hmac.new(SESSION_SECRET.encode("utf-8"), value.encode("utf-8"), hashlib.sha256).hexdigest() return f"{value}.{mac}" def verify_session(signed): if not signed or "." not in signed: return False value, mac = signed.rsplit(".", 1) expected = hmac.new(SESSION_SECRET.encode("utf-8"), value.encode("utf-8"), hashlib.sha256).hexdigest() if not hmac.compare_digest(mac, expected): return False try: username, issued = value.split(":", 1) return username == USERNAME and time.time() - int(issued) < 86400 except ValueError: return False def parse_cookies(header): cookies = {} for part in (header or "").split(";"): if "=" not in part: continue key, value = part.strip().split("=", 1) cookies[key] = value return cookies def too_many_login_failures(ip): now = time.time() window = [ts for ts in login_failures.get(ip, []) if now - ts < 300] login_failures[ip] = window return len(window) >= 5 def record_login_failure(ip): window = login_failures.get(ip, []) window.append(time.time()) login_failures[ip] = window[-10:] def run_json(cmd, timeout=20): result = subprocess.run(cmd, 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 json.loads(result.stdout) def run_text(cmd, timeout=10): result = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout) if result.returncode != 0: return "" 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([ "kubectl", "--kubeconfig", str(KUBECONFIG), "-n", NAMESPACE, "get", "deploy", *names, "-o", "json", ]) items = [] for item in payload.get("items", []): name = item["metadata"]["name"] spec = item.get("spec", {}) status = item.get("status", {}) containers = item.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) image = next((c.get("image", "") for c in containers if c.get("name") == name), containers[0].get("image", "") if containers else "") conditions = {c.get("type"): c.get("status") for c in status.get("conditions", [])} items.append({ "name": name, "label": SERVICES.get(name, {}).get("label", name), "ready": f"{status.get('readyReplicas', 0)}/{spec.get('replicas', 0)}", "updated": status.get("updatedReplicas", 0), "available": status.get("availableReplicas", 0), "image": image, "port": SERVICES.get(name, {}).get("port", ""), "health": SERVICES.get(name, {}).get("health", ""), "healthy": conditions.get("Available") == "True", }) order = {name: idx for idx, name in enumerate(names)} return sorted(items, key=lambda x: order.get(x["name"], 99)) def pod_status(): selector = "app in (" + ",".join(SERVICES) + ")" payload = run_json([ "kubectl", "--kubeconfig", str(KUBECONFIG), "-n", NAMESPACE, "get", "pods", "-l", selector, "-o", "json", ]) pods = [] for item in payload.get("items", []): status = item.get("status", {}) containers = status.get("containerStatuses", []) ready_count = sum(1 for c in containers if c.get("ready")) pods.append({ "name": item["metadata"]["name"], "app": item["metadata"].get("labels", {}).get("app", ""), "ready": f"{ready_count}/{len(containers)}", "phase": status.get("phase", ""), "restarts": sum(c.get("restartCount", 0) for c in containers), "node": item.get("spec", {}).get("nodeName", ""), "ip": status.get("podIP", ""), "age": item["metadata"].get("creationTimestamp", ""), }) return pods def git_status(): head = run_text(["git", "-C", str(SOURCE_DIR), "rev-parse", "HEAD"]) branch = run_text(["git", "-C", str(SOURCE_DIR), "branch", "--show-current"]) short = run_text(["git", "-C", str(SOURCE_DIR), "status", "--short", "--branch"]) return {"head": head, "branch": branch, "status": short} def list_jobs(): RUN_DIR.mkdir(parents=True, exist_ok=True) jobs = [] for meta_path in sorted(RUN_DIR.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)[:20]: try: jobs.append(json.loads(meta_path.read_text())) except json.JSONDecodeError: continue return jobs def save_job(job): RUN_DIR.mkdir(parents=True, exist_ok=True) (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: with log_path.open("a", buffering=1) as log: log.write(f"[{now_ts()}] start {' '.join(cmd)}\n") process = subprocess.Popen( cmd, cwd=str(BASE_DIR), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) job["pid"] = process.pid job["status"] = "running" save_job(job) for line in process.stdout: log.write(line) rc = process.wait() job["status"] = "success" if rc == 0 else "failed" job["returnCode"] = rc job["finishedAt"] = now_ts() save_job(job) log.write(f"[{now_ts()}] finished rc={rc}\n") except Exception as exc: job["status"] = "failed" job["returnCode"] = -1 job["finishedAt"] = now_ts() job["error"] = str(exc) save_job(job) with log_path.open("a", buffering=1) as log: log.write(f"[{now_ts()}] runner error: {exc}\n") finally: global active_job with job_lock: active_job = None def start_deploy(services, branch, fast): global active_job invalid = [svc for svc in services if svc not in SERVICES] if invalid: raise ValueError(f"unsupported services: {', '.join(invalid)}") if not services: raise ValueError("select at least one service") if not BRANCH_RE.match(branch): raise ValueError("invalid branch name") if branch not in ALLOWED_BRANCHES: raise ValueError(f"branch is not allowed: {branch}") if not DEPLOY_SCRIPT.exists(): raise ValueError(f"missing deploy script: {DEPLOY_SCRIPT}") with job_lock: if active_job: raise RuntimeError(f"deploy already running: {active_job['id']}") job_id = time.strftime("%Y%m%d%H%M%S") + "-" + secrets.token_hex(3) job = { "id": job_id, "services": services, "branch": branch, "fast": fast, "status": "queued", "createdAt": now_ts(), "finishedAt": "", "returnCode": None, } active_job = job save_job(job) env = os.environ.copy() env.update({ "USE_GIT_SOURCE": "1", "GIT_REF": branch, "GIT_REMOTE_URL": REMOTE_URL, "MODE": "preload", "KUBECONFIG": str(KUBECONFIG), "NAMESPACE": NAMESPACE, "ASLAN_NAMESPACE": NAMESPACE, "HOME": str(Path.home()), }) if fast: # Fast mode avoids Maven clean so unchanged modules and already downloaded dependencies can reuse local target output. env["MAVEN_GOALS"] = "package" cmd = [str(DEPLOY_SCRIPT), "--mode", "preload", *services] thread = threading.Thread(target=stream_process, args=(job, cmd, env), daemon=True) thread.start() return job def login_page(error=""): message = f"
{html.escape(error)}
" if error else "" return f""" Aslan Ops

Aslan Ops

{message}
""" INDEX_HTML = """ Aslan Ops

Aslan Ops

微服务

部署

固定 preload 到 TKE 节点，不走 registry push

配置

部署日志

无运行任务

      

      

        
Pod

        
名称状态节点

      

    

    

      
最近任务

      
ID服务分支状态时间

    

  

  

"""


class Handler(BaseHTTPRequestHandler):
    server_version = "AslanOps/1.0"

    def authenticated(self):
        cookies = parse_cookies(self.headers.get("Cookie"))
        return verify_session(cookies.get("aslan_ops"))

    def require_auth(self):
        if self.authenticated():
            return True
        redirect_response(self, "/login")
        return False

    def read_body(self, max_bytes=65536):
        size = int(self.headers.get("Content-Length", "0"))
        if size > max_bytes:
            raise ValueError("request body too large")
        return self.rfile.read(size).decode("utf-8")

    def do_GET(self):
        parsed = urlparse(self.path)
        if parsed.path == "/login":
            return text_response(self, 200, login_page(), "text/html; charset=utf-8")
        if parsed.path == "/logout":
            redirect_response(self, "/login", ["aslan_ops=; Max-Age=0; HttpOnly; SameSite=Lax; Path=/"])
            return
        if not self.require_auth():
            return
        if parsed.path == "/":
            return text_response(self, 200, INDEX_HTML, "text/html; charset=utf-8")
        if parsed.path == "/api/status":
            try:
                return json_response(self, 200, {
                    "deployments": deployment_status(),
                    "pods": pod_status(),
                    "git": git_status(),
                    "jobs": list_jobs(),
                })
            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)
            meta_path = RUN_DIR / f"{job_id}.json"
            log_path = RUN_DIR / f"{job_id}.log"
            if not meta_path.exists():
                return json_response(self, 404, {"error": "job not found"})
            job = json.loads(meta_path.read_text())
            log = log_path.read_text(errors="replace")[-120000:] if log_path.exists() else ""
            return json_response(self, 200, {"job": job, "log": log})
        return text_response(self, 404, "not found")

    def do_POST(self):
        parsed = urlparse(self.path)
        if parsed.path == "/login":
            if too_many_login_failures(self.client_address[0]):
                return text_response(self, 429, login_page("登录失败次数过多，稍后再试"), "text/html; charset=utf-8")
            try:
                form = parse_qs(self.read_body(max_bytes=4096))
            except ValueError as exc:
                return text_response(self, 413, str(exc))
            username = form.get("username", [""])[0]
            password = form.get("password", [""])[0]
            if hmac.compare_digest(username, USERNAME) and hmac.compare_digest(password, PASSWORD):
                login_failures.pop(self.client_address[0], None)
                session = sign_session(f"{USERNAME}:{int(time.time())}")
                redirect_response(self, "/", [f"aslan_ops={session}; HttpOnly; SameSite=Lax; Path=/"])
                return
            record_login_failure(self.client_address[0])
            return text_response(self, 401, login_page("账号或密码错误"), "text/html; charset=utf-8")
        if not self.require_auth():
            return
        if parsed.path == "/api/deploy":
            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=4096) or "{}")
                services = payload.get("services") or []
                branch = payload.get("branch") or GIT_REF
                fast = bool(payload.get("fast", True))
                job = start_deploy(services, branch, fast)
                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):
        print(f"[{now_ts()}] {self.address_string()} {fmt % args}", flush=True)


def main():
    if not PASSWORD:
        raise SystemExit("ASLAN_OPS_PASSWORD is required")
    RUN_DIR.mkdir(parents=True, exist_ok=True)
    server = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"aslan ops listening on {HOST}:{PORT}", flush=True)
    server.serve_forever()


if __name__ == "__main__":
    main()