#!/usr/bin/env python3 import base64 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-test-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"))) SOURCE_DIR = Path(os.environ.get("ASLAN_SOURCE_DIR", str(BASE_DIR / "source" / "likei-services"))) NAMESPACE = os.environ.get("ASLAN_NAMESPACE", "test") GIT_REF = os.environ.get("ASLAN_GIT_REF", "aslan_test") 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", "18081")) 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 = { "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}$") 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 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 (other,external,console)" 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 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/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)})
        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()