#!/usr/bin/env python3
"""Singapore-side control plane for the Cang'er outbound node."""

from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import html
import json
import os
import re
import smtplib
import sqlite3
import sys
from email.message import EmailMessage
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from shared import (  # noqa: E402
    MAX_JSON_BYTES,
    MAX_RESULT_BYTES,
    PAIRING_TTL_SECONDS,
    TASK_LEASE_SECONDS,
    compact_json,
    new_id,
    new_pairing_code,
    new_token,
    now_ts,
    secret_hash,
    secret_matches,
    validate_node_name,
    validate_task,
)


SCHEMA = """
CREATE TABLE IF NOT EXISTS pairings (
    id TEXT PRIMARY KEY,
    code_hash TEXT NOT NULL,
    expires_at INTEGER NOT NULL,
    used_at INTEGER
);
CREATE TABLE IF NOT EXISTS nodes (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL UNIQUE,
    token_hash TEXT NOT NULL,
    created_at INTEGER NOT NULL,
    last_seen_at INTEGER,
    agent_version TEXT,
    policy_json TEXT NOT NULL DEFAULT '{}'
);
CREATE TABLE IF NOT EXISTS tasks (
    id TEXT PRIMARY KEY,
    node_id TEXT NOT NULL REFERENCES nodes(id),
    action TEXT NOT NULL,
    args_json TEXT NOT NULL,
    status TEXT NOT NULL,
    created_at INTEGER NOT NULL,
    approved_at INTEGER,
    leased_until INTEGER,
    completed_at INTEGER,
    result_json TEXT,
    grant_id TEXT,
    requested_by TEXT NOT NULL DEFAULT 'ICE-GL-ZY001',
    executor_persona TEXT NOT NULL DEFAULT 'ICE-GL-CA001',
    FOREIGN KEY(node_id) REFERENCES nodes(id)
);
CREATE INDEX IF NOT EXISTS idx_tasks_node_status
ON tasks(node_id, status, created_at);
CREATE TABLE IF NOT EXISTS task_events (
    seq INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id TEXT NOT NULL REFERENCES tasks(id),
    created_at INTEGER NOT NULL,
    kind TEXT NOT NULL,
    payload_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_task_events_task_seq
ON task_events(task_id, seq);
CREATE TABLE IF NOT EXISTS grants (
    id TEXT PRIMARY KEY,
    node_id TEXT NOT NULL REFERENCES nodes(id),
    scopes_json TEXT NOT NULL,
    roots_json TEXT NOT NULL,
    reason TEXT NOT NULL,
    duration_seconds INTEGER NOT NULL,
    status TEXT NOT NULL,
    created_at INTEGER NOT NULL,
    approved_at INTEGER,
    expires_at INTEGER,
    revoked_at INTEGER
    ,requested_by TEXT NOT NULL DEFAULT 'UNKNOWN'
    ,requester_name TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_grants_node_status
ON grants(node_id, status, expires_at);
"""


class ControlPlane:
    def __init__(
        self,
        db_path: str,
        admin_token: str,
        owner_token: str,
        persona_token: str,
        operator_token: str,
        approval_signing_key: str,
        owner_email: str = "",
        public_url: str = "",
    ) -> None:
        for name, value in {
            "admin token": admin_token,
            "owner token": owner_token,
            "persona token": persona_token,
            "operator token": operator_token,
        }.items():
            if len(value) < 24:
                raise ValueError(f"{name} must contain at least 24 characters")
        if len(approval_signing_key) < 32:
            raise ValueError("approval signing key must contain at least 32 characters")
        self.db_path = db_path
        self.admin_hash = secret_hash(admin_token)
        self.owner_hash = secret_hash(owner_token)
        self.persona_hash = secret_hash(persona_token)
        self.operator_hash = secret_hash(operator_token)
        self.operator_id = os.environ.get("CANGER_OPERATOR_ID", "ICE-GL-ZY001")
        self.owner_id = os.environ.get("CANGER_OWNER_ID", "CANGER-HUMAN-OWNER")
        self.executor_persona_id = os.environ.get(
            "CANGER_EXECUTOR_PERSONA_ID", "ICE-GL-CA001"
        )
        self.approval_signing_key = approval_signing_key.encode("utf-8")
        self.owner_email = owner_email
        self.public_url = public_url.rstrip("/")
        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
        with self.db() as conn:
            conn.executescript(SCHEMA)
            columns = {
                row["name"]
                for row in conn.execute("PRAGMA table_info(grants)").fetchall()
            }
            if "requester_name" not in columns:
                conn.execute(
                    "ALTER TABLE grants ADD COLUMN requester_name "
                    "TEXT NOT NULL DEFAULT ''"
                )

    def db(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path, timeout=10, isolation_level=None)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA foreign_keys=ON")
        conn.execute("PRAGMA journal_mode=WAL")
        return conn

    @staticmethod
    def bearer(headers: Any) -> str:
        value = headers.get("Authorization", "")
        if not value.startswith("Bearer "):
            return ""
        return value[7:].strip()

    def role_allowed(self, headers: Any, *roles: str) -> bool:
        token = self.bearer(headers)
        hashes = {
            "admin": self.admin_hash,
            "owner": self.owner_hash,
            "persona": self.persona_hash,
            "operator": self.operator_hash,
        }
        return any(secret_matches(token, hashes[role]) for role in roles)

    def node_for_token(self, node_id: str, headers: Any) -> sqlite3.Row | None:
        token = self.bearer(headers)
        with self.db() as conn:
            row = conn.execute(
                "SELECT * FROM nodes WHERE id=?", (node_id,)
            ).fetchone()
        if row is None or not secret_matches(token, row["token_hash"]):
            return None
        return row

    def grant_token(self, grant_id: str, link_expires_at: int) -> str:
        message = f"canger-grant/v1\n{grant_id}\n{link_expires_at}".encode("utf-8")
        signature = hmac.new(
            self.approval_signing_key, message, hashlib.sha256
        ).digest()
        encoded = base64.urlsafe_b64encode(signature).decode("ascii").rstrip("=")
        return f"{link_expires_at}.{encoded}"

    def verify_grant_token(self, grant_id: str, token: str) -> bool:
        try:
            raw_expiry, _ = token.split(".", 1)
            expiry = int(raw_expiry)
        except (ValueError, TypeError):
            return False
        if expiry < now_ts():
            return False
        return hmac.compare_digest(token, self.grant_token(grant_id, expiry))

    def send_grant_email(self, grant: dict[str, Any]) -> str:
        required = {
            "owner_email": self.owner_email,
            "public_url": self.public_url,
            "smtp_host": os.environ.get("CANGER_SMTP_HOST", ""),
            "smtp_from": os.environ.get("CANGER_SMTP_FROM", ""),
        }
        if not all(required.values()):
            return "not_configured"
        link_expiry = now_ts() + 30 * 60
        token = self.grant_token(grant["id"], link_expiry)
        link = (
            f"{self.public_url}/approve/grant?"
            + urlencode({"id": grant["id"], "token": token})
        )
        message = EmailMessage()
        message["Subject"] = f"苍耳服务器副控运维授权（{grant['duration_seconds'] // 3600} 小时）"
        message["From"] = required["smtp_from"]
        message["To"] = required["owner_email"]
        message.set_content(
            "有人格体申请进入苍耳服务器的限时受控运维会话。"
            "苍耳仍是服务器主控与最终决定者。\n\n"
            f"申请者：{grant['requester_name']} ({grant['requested_by']})\n"
            f"申请目的：{grant['reason']}\n"
            f"时长：{grant['duration_seconds'] // 3600} 小时\n\n"
            "期间可以：检查服务器、同步和修改代码、运行测试、查看故障日志。\n"
            "不能：修改系统账户、密钥、网络、防火墙、磁盘和系统启动，"
            "也不能操作受管工作区以外的个人文件。\n\n"
            f"打开链接查看并确认：{link}\n\n"
            "链接 30 分钟失效；授权本身到期后自动失效。"
        )
        host = required["smtp_host"]
        port = int(os.environ.get("CANGER_SMTP_PORT", "465"))
        username = os.environ.get("CANGER_SMTP_USERNAME", "")
        password = os.environ.get("CANGER_SMTP_PASSWORD", "")
        if port == 465:
            client: Any = smtplib.SMTP_SSL(host, port, timeout=20)
        else:
            client = smtplib.SMTP(host, port, timeout=20)
            client.starttls()
        try:
            if username:
                client.login(username, password)
            client.send_message(message)
        finally:
            client.quit()
        return "sent"


class Handler(BaseHTTPRequestHandler):
    server_version = "CangerControl/0.1"

    @property
    def app(self) -> ControlPlane:
        return self.server.app  # type: ignore[attr-defined]

    def log_message(self, fmt: str, *args: Any) -> None:
        print(f"{self.address_string()} - {fmt % args}", file=sys.stderr)

    def send_json(self, status: int, payload: Any) -> None:
        data = compact_json(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def send_html(self, status: int, content: str) -> None:
        data = content.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def read_json(self) -> dict[str, Any]:
        try:
            size = int(self.headers.get("Content-Length", "0"))
        except ValueError as exc:
            raise ValueError("invalid Content-Length") from exc
        if size <= 0 or size > MAX_JSON_BYTES:
            raise ValueError("invalid JSON body size")
        value = json.loads(self.rfile.read(size))
        if not isinstance(value, dict):
            raise ValueError("JSON body must be an object")
        return value

    def path_parts(self) -> list[str]:
        return [part for part in urlparse(self.path).path.split("/") if part]

    def require_role(self, *roles: str) -> bool:
        if self.app.role_allowed(self.headers, *roles):
            return True
        self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
        return False

    def requester_id(self) -> str:
        if self.app.role_allowed(self.headers, "persona"):
            return self.app.executor_persona_id
        if self.app.role_allowed(self.headers, "owner"):
            return self.app.owner_id
        return self.app.operator_id

    def request_identity(self, body: dict[str, Any]) -> tuple[str, str]:
        requester_id = body.get("requester_id")
        requester_name = body.get("requester_name")
        if not isinstance(requester_id, str) or not re.fullmatch(
            r"[A-Za-z0-9._:+∞-]{2,80}", requester_id
        ):
            raise ValueError("valid requester_id is required")
        if (
            not isinstance(requester_name, str)
            or not requester_name.strip()
            or len(requester_name) > 100
        ):
            raise ValueError("valid requester_name is required")
        return requester_id, requester_name.strip()

    def do_GET(self) -> None:  # noqa: N802
        try:
            parts = self.path_parts()
            if parts == ["health"]:
                self.send_json(HTTPStatus.OK, {"ok": True, "service": "canger-control"})
                return
            if parts == ["approve", "grant"]:
                self.grant_approval_page()
                return
            if parts == ["v1", "nodes"]:
                if not self.require_role("admin", "owner", "persona", "operator"):
                    return
                with self.app.db() as conn:
                    rows = conn.execute(
                        "SELECT id,name,created_at,last_seen_at,agent_version FROM nodes "
                        "ORDER BY created_at"
                    ).fetchall()
                self.send_json(HTTPStatus.OK, {"nodes": [dict(row) for row in rows]})
                return
            if len(parts) == 4 and parts[:2] == ["v1", "tasks"]:
                if parts[3] != "":
                    pass
            if len(parts) == 3 and parts[:2] == ["v1", "tasks"]:
                if not self.require_role("admin", "owner", "persona", "operator"):
                    return
                with self.app.db() as conn:
                    row = conn.execute(
                        "SELECT * FROM tasks WHERE id=?", (parts[2],)
                    ).fetchone()
                if row is None:
                    self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
                    return
                payload = dict(row)
                payload["args"] = json.loads(payload.pop("args_json"))
                if payload["result_json"] is not None:
                    payload["result"] = json.loads(payload.pop("result_json"))
                else:
                    payload.pop("result_json")
                self.send_json(HTTPStatus.OK, payload)
                return
            if (
                len(parts) == 4
                and parts[:2] == ["v1", "tasks"]
                and parts[3] == "events"
            ):
                if not self.require_role("admin", "owner", "persona", "operator"):
                    return
                after = 0
                query = urlparse(self.path).query
                for item in query.split("&"):
                    if item.startswith("after="):
                        after = max(0, int(item.split("=", 1)[1]))
                with self.app.db() as conn:
                    rows = conn.execute(
                        "SELECT seq,created_at,kind,payload_json FROM task_events "
                        "WHERE task_id=? AND seq>? ORDER BY seq LIMIT 500",
                        (parts[2], after),
                    ).fetchall()
                events = []
                for row in rows:
                    event = dict(row)
                    event["payload"] = json.loads(event.pop("payload_json"))
                    events.append(event)
                self.send_json(HTTPStatus.OK, {"events": events})
                return
            if parts == ["v1", "grants"]:
                if not self.require_role("admin", "owner", "persona", "operator"):
                    return
                with self.app.db() as conn:
                    rows = conn.execute(
                        "SELECT * FROM grants ORDER BY created_at DESC LIMIT 200"
                    ).fetchall()
                grants = []
                for row in rows:
                    item = dict(row)
                    item["scopes"] = json.loads(item.pop("scopes_json"))
                    item["roots"] = json.loads(item.pop("roots_json"))
                    grants.append(item)
                self.send_json(HTTPStatus.OK, {"grants": grants})
                return
            if (
                len(parts) == 5
                and parts[:2] == ["v1", "nodes"]
                and parts[3:] == ["tasks", "next"]
            ):
                self.next_task(parts[2])
                return
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
        except Exception as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})

    def do_POST(self) -> None:  # noqa: N802
        try:
            parts = self.path_parts()
            if parts == ["v1", "pairings"]:
                self.create_pairing()
                return
            if parts == ["v1", "pairings", "claim"]:
                self.claim_pairing()
                return
            if parts == ["v1", "tasks"]:
                self.create_task()
                return
            if parts == ["v1", "grants"]:
                self.create_grant()
                return
            if len(parts) == 4 and parts[:2] == ["v1", "grants"]:
                if parts[3] == "revoke":
                    self.revoke_grant(parts[2])
                    return
            if len(parts) == 4 and parts[:2] == ["v1", "tasks"]:
                if parts[3] == "approve":
                    self.decide_task(parts[2], "approved")
                    return
                if parts[3] == "reject":
                    self.decide_task(parts[2], "rejected")
                    return
                if parts[3] == "result":
                    self.complete_task(parts[2])
                    return
                if parts[3] == "events":
                    self.append_event(parts[2])
                    return
            if (
                len(parts) == 4
                and parts[:2] == ["v1", "nodes"]
                and parts[3] == "heartbeat"
            ):
                self.heartbeat(parts[2])
                return
            if parts == ["approve", "grant"]:
                self.approve_grant_form()
                return
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
        except json.JSONDecodeError:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON"})
        except ValueError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
        except Exception as exc:
            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)})

    def create_pairing(self) -> None:
        if not self.require_role("admin"):
            return
        code = new_pairing_code()
        pairing_id = new_id("pair")
        expires_at = now_ts() + PAIRING_TTL_SECONDS
        with self.app.db() as conn:
            conn.execute(
                "INSERT INTO pairings(id,code_hash,expires_at) VALUES(?,?,?)",
                (pairing_id, secret_hash(code), expires_at),
            )
        self.send_json(
            HTTPStatus.CREATED,
            {"pairing_id": pairing_id, "code": code, "expires_at": expires_at},
        )

    def claim_pairing(self) -> None:
        body = self.read_json()
        code = body.get("code")
        name = validate_node_name(body.get("node_name"))
        agent_version = str(body.get("agent_version", ""))[:64]
        policy = self.validated_node_policy(body.get("policy", {}))
        if not isinstance(code, str):
            raise ValueError("code is required")
        now = now_ts()
        node_id = new_id("node")
        node_token = new_token()
        with self.app.db() as conn:
            conn.execute("BEGIN IMMEDIATE")
            rows = conn.execute(
                "SELECT * FROM pairings WHERE used_at IS NULL AND expires_at>=?", (now,)
            ).fetchall()
            pairing = next(
                (row for row in rows if secret_matches(code, row["code_hash"])), None
            )
            if pairing is None:
                conn.rollback()
                self.send_json(
                    HTTPStatus.UNAUTHORIZED, {"error": "invalid or expired pairing code"}
                )
                return
            existing = conn.execute(
                "SELECT id FROM nodes WHERE name=?", (name,)
            ).fetchone()
            if existing is not None:
                conn.rollback()
                self.send_json(
                    HTTPStatus.CONFLICT, {"error": "node name already registered"}
                )
                return
            conn.execute(
                "INSERT INTO nodes(id,name,token_hash,created_at,last_seen_at,"
                "agent_version,policy_json) VALUES(?,?,?,?,?,?,?)",
                (
                    node_id,
                    name,
                    secret_hash(node_token),
                    now,
                    now,
                    agent_version,
                    compact_json(policy),
                ),
            )
            conn.execute(
                "UPDATE pairings SET used_at=? WHERE id=?", (now, pairing["id"])
            )
            conn.commit()
        self.send_json(
            HTTPStatus.CREATED, {"node_id": node_id, "node_token": node_token}
        )

    @staticmethod
    def validated_node_policy(value: Any) -> dict[str, Any]:
        if not isinstance(value, dict):
            raise ValueError("node policy must be an object")
        roots = value.get("allowed_roots", [])
        actions = value.get("enabled_actions", [])
        allowed_actions = {
            "system.status",
            "repo.status",
            "repo.fetch",
            "service.logs",
            "command.run",
        }
        if (
            not isinstance(roots, list)
            or not roots
            or any(not isinstance(root, str) or not root.startswith("/") for root in roots)
        ):
            raise ValueError("node policy requires absolute allowed_roots")
        if (
            not isinstance(actions, list)
            or any(action not in allowed_actions for action in actions)
        ):
            raise ValueError("node policy contains unsupported actions")
        return {
            "profile": "developer_standard",
            "allowed_roots": sorted(set(root.rstrip("/") for root in roots)),
            "enabled_actions": sorted(set(actions)),
        }

    def create_task(self) -> None:
        if not self.require_role("admin", "operator", "persona"):
            return
        body = self.read_json()
        requested_by, requester_name = self.request_identity(body)
        node_id = body.get("node_id")
        if not isinstance(node_id, str):
            raise ValueError("node_id is required")
        action, args = validate_task(body.get("action"), body.get("args"))
        task_id = new_id("task")
        approved_at = None
        grant_id = None
        with self.app.db() as conn:
            if conn.execute("SELECT 1 FROM nodes WHERE id=?", (node_id,)).fetchone() is None:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "node not found"})
                return
            rows = conn.execute(
                "SELECT * FROM grants WHERE node_id=? AND status='active' "
                "AND expires_at>? AND requested_by=? ORDER BY expires_at",
                (node_id, now_ts(), requested_by),
            ).fetchall()
            for row in rows:
                if self.task_allowed_by_grant(action, args, row):
                    grant_id = row["id"]
                    approved_at = now_ts()
                    break
            status = "approved" if grant_id else "pending"
            conn.execute(
                "INSERT INTO tasks(id,node_id,action,args_json,status,created_at,"
                "approved_at,grant_id,requested_by,executor_persona) "
                "VALUES(?,?,?,?,?,?,?,?,?,?)",
                (
                    task_id,
                    node_id,
                    action,
                    compact_json(args),
                    status,
                    now_ts(),
                    approved_at,
                    grant_id,
                    requested_by,
                    requested_by,
                ),
            )
        self.send_json(
            HTTPStatus.CREATED,
            {
                "task_id": task_id,
                "status": status,
                "approval_required": grant_id is None,
                "grant_id": grant_id,
                "requested_by": requested_by,
                "requester_name": requester_name,
                "executor_persona": requested_by,
                "action": action,
                "args": args,
            },
        )

    @staticmethod
    def task_allowed_by_grant(
        action: str, args: dict[str, Any], grant: sqlite3.Row
    ) -> bool:
        scopes = set(json.loads(grant["scopes_json"]))
        if action not in scopes:
            return False
        roots = [str(root).rstrip("/") for root in json.loads(grant["roots_json"])]
        candidate = None
        if action in {"repo.status", "repo.fetch"}:
            candidate = args["repo"]
        elif action == "command.run":
            candidate = args["cwd"]
        if candidate is None:
            return True
        normalized = str(candidate).rstrip("/")
        return any(
            normalized == root or normalized.startswith(root + "/") for root in roots
        )

    def create_grant(self) -> None:
        if not self.require_role("admin", "operator", "persona"):
            return
        body = self.read_json()
        requested_by, requester_name = self.request_identity(body)
        node_id = body.get("node_id")
        reason = body.get("reason", "编程与调试")
        duration = body.get("duration_seconds", 4 * 3600)
        if not isinstance(node_id, str):
            raise ValueError("node_id is required")
        if not isinstance(reason, str) or not reason.strip() or len(reason) > 300:
            raise ValueError("invalid grant reason")
        if not isinstance(duration, int) or not 3600 <= duration <= 8 * 3600:
            raise ValueError("grant duration must be between 1 and 8 hours")
        with self.app.db() as conn:
            node = conn.execute(
                "SELECT policy_json FROM nodes WHERE id=?", (node_id,)
            ).fetchone()
            if node is None:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "node not found"})
                return
            policy = json.loads(node["policy_json"])
            scopes = policy.get("enabled_actions", [])
            roots = policy.get("allowed_roots", [])
            if not scopes or not roots:
                self.send_json(
                    HTTPStatus.CONFLICT,
                    {"error": "node has not reported a usable safety policy"},
                )
                return
            grant_id = new_id("grant")
            grant = {
                "id": grant_id,
                "node_id": node_id,
                "scopes": scopes,
                "roots": roots,
                "reason": reason.strip(),
                "duration_seconds": duration,
                "requested_by": requested_by,
                "requester_name": requester_name,
            }
            conn.execute(
                "INSERT INTO grants(id,node_id,scopes_json,roots_json,reason,"
                "duration_seconds,status,created_at,requested_by,requester_name) "
                "VALUES(?,?,?,?,?,?,?,?,?,?)",
                (
                    grant_id,
                    node_id,
                    compact_json(grant["scopes"]),
                    compact_json(grant["roots"]),
                    grant["reason"],
                    duration,
                    "pending",
                    now_ts(),
                    requested_by,
                    requester_name,
                ),
            )
        try:
            notification = self.app.send_grant_email(grant)
        except Exception as exc:
            print(f"grant email failed: {exc}", file=sys.stderr)
            notification = "failed"
        self.send_json(
            HTTPStatus.CREATED,
            {
                "grant_id": grant_id,
                "status": "pending",
                "requested_by": requested_by,
                "requester_name": requester_name,
                "notification": notification,
                "duration_seconds": duration,
                "scopes": grant["scopes"],
                "roots": grant["roots"],
            },
        )

    def grant_approval_page(self) -> None:
        query = parse_qs(urlparse(self.path).query)
        grant_id = query.get("id", [""])[0]
        token = query.get("token", [""])[0]
        if not self.app.verify_grant_token(grant_id, token):
            self.send_html(HTTPStatus.UNAUTHORIZED, "<h1>链接无效或已过期</h1>")
            return
        with self.app.db() as conn:
            row = conn.execute("SELECT * FROM grants WHERE id=?", (grant_id,)).fetchone()
        if row is None or row["status"] != "pending":
            self.send_html(HTTPStatus.CONFLICT, "<h1>申请已处理或不存在</h1>")
            return
        content = f"""<!doctype html><meta charset="utf-8">
<title>苍耳服务器临时授权</title>
<h1>苍耳服务器临时授权</h1>
<p>主控：苍耳。是否批准，以苍耳本次选择为准。</p>
<p>申请者：{html.escape(row["requester_name"])} ({html.escape(row["requested_by"])})</p>
<p>申请目的：{html.escape(row["reason"])}</p>
<p>时长：{row["duration_seconds"] // 3600} 小时</p>
<p>期间可以：检查服务器、同步和修改代码、运行测试、查看故障日志。</p>
<p>不能：修改系统账户、密钥、网络、防火墙、磁盘和系统启动，
也不能操作受管工作区以外的个人文件。</p>
<p>授权到期自动失效，可从新加坡端提前撤销。</p>
<form method="post" action="/approve/grant">
<input type="hidden" name="id" value="{html.escape(grant_id)}">
<input type="hidden" name="token" value="{html.escape(token)}">
<button type="submit">批准本次临时授权</button>
</form>"""
        self.send_html(HTTPStatus.OK, content)

    def approve_grant_form(self) -> None:
        size = int(self.headers.get("Content-Length", "0"))
        if size <= 0 or size > 16_384:
            raise ValueError("invalid form body size")
        form = parse_qs(self.rfile.read(size).decode("utf-8"))
        grant_id = form.get("id", [""])[0]
        token = form.get("token", [""])[0]
        if not self.app.verify_grant_token(grant_id, token):
            self.send_html(HTTPStatus.UNAUTHORIZED, "<h1>链接无效或已过期</h1>")
            return
        approved_at = now_ts()
        with self.app.db() as conn:
            row = conn.execute("SELECT * FROM grants WHERE id=?", (grant_id,)).fetchone()
            if row is None or row["status"] != "pending":
                self.send_html(HTTPStatus.CONFLICT, "<h1>申请已处理或不存在</h1>")
                return
            expires_at = approved_at + row["duration_seconds"]
            conn.execute(
                "UPDATE grants SET status='active',approved_at=?,expires_at=? "
                "WHERE id=? AND status='pending'",
                (approved_at, expires_at, grant_id),
            )
        self.send_html(
            HTTPStatus.OK,
            f"<h1>副控运维授权成功</h1><p>苍耳仍保留最终控制权；"
            f"本次授权将在 {expires_at} 自动失效，可以关闭此页面。</p>",
        )

    def revoke_grant(self, grant_id: str) -> None:
        if not self.require_role("admin", "owner"):
            return
        with self.app.db() as conn:
            cursor = conn.execute(
                "UPDATE grants SET status='revoked',revoked_at=? "
                "WHERE id=? AND status='active'",
                (now_ts(), grant_id),
            )
        if cursor.rowcount != 1:
            self.send_json(HTTPStatus.CONFLICT, {"error": "grant is not active"})
            return
        self.send_json(HTTPStatus.OK, {"grant_id": grant_id, "status": "revoked"})

    def decide_task(self, task_id: str, decision: str) -> None:
        if not self.require_role("admin", "owner"):
            return
        with self.app.db() as conn:
            cursor = conn.execute(
                "UPDATE tasks SET status=?,approved_at=? "
                "WHERE id=? AND status='pending'",
                (decision, now_ts(), task_id),
            )
        if cursor.rowcount != 1:
            self.send_json(
                HTTPStatus.CONFLICT, {"error": "task is not pending or does not exist"}
            )
            return
        self.send_json(HTTPStatus.OK, {"task_id": task_id, "status": decision})

    def next_task(self, node_id: str) -> None:
        if self.app.node_for_token(node_id, self.headers) is None:
            self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
            return
        now = now_ts()
        with self.app.db() as conn:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE grants SET status='expired' "
                "WHERE status='active' AND expires_at<=?",
                (now,),
            )
            conn.execute(
                "UPDATE tasks SET status='pending',approved_at=NULL,grant_id=NULL "
                "WHERE node_id=? AND status='approved' AND grant_id IS NOT NULL "
                "AND grant_id NOT IN (SELECT id FROM grants WHERE status='active' "
                "AND expires_at>?)",
                (node_id, now),
            )
            conn.execute(
                "UPDATE tasks SET status='approved',leased_until=NULL "
                "WHERE node_id=? AND status='leased' AND leased_until<?",
                (node_id, now),
            )
            row = conn.execute(
                "SELECT id,action,args_json FROM tasks "
                "WHERE node_id=? AND status='approved' ORDER BY approved_at,id LIMIT 1",
                (node_id,),
            ).fetchone()
            if row is not None:
                conn.execute(
                    "UPDATE tasks SET status='leased',leased_until=? WHERE id=?",
                    (now + TASK_LEASE_SECONDS, row["id"]),
                )
            conn.execute(
                "UPDATE nodes SET last_seen_at=? WHERE id=?", (now, node_id)
            )
            conn.commit()
        if row is None:
            self.send_json(HTTPStatus.OK, {"task": None})
            return
        self.send_json(
            HTTPStatus.OK,
            {
                "task": {
                    "id": row["id"],
                    "action": row["action"],
                    "args": json.loads(row["args_json"]),
                }
            },
        )

    def complete_task(self, task_id: str) -> None:
        body = self.read_json()
        if len(compact_json(body).encode("utf-8")) > MAX_RESULT_BYTES:
            raise ValueError("task result is too large")
        with self.app.db() as conn:
            row = conn.execute(
                "SELECT node_id,status FROM tasks WHERE id=?", (task_id,)
            ).fetchone()
            if row is None:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
                return
            if self.app.node_for_token(row["node_id"], self.headers) is None:
                self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
                return
            if row["status"] != "leased":
                self.send_json(HTTPStatus.CONFLICT, {"error": "task is not leased"})
                return
            conn.execute(
                "UPDATE tasks SET status='completed',completed_at=?,result_json=? "
                "WHERE id=?",
                (now_ts(), compact_json(body), task_id),
            )
        self.send_json(HTTPStatus.OK, {"task_id": task_id, "status": "completed"})

    def append_event(self, task_id: str) -> None:
        body = self.read_json()
        kind = body.get("kind")
        payload = body.get("payload", {})
        if (
            not isinstance(kind, str)
            or not kind
            or len(kind) > 64
            or not all(char.isalnum() or char in "._-" for char in kind)
        ):
            raise ValueError("invalid event kind")
        encoded = compact_json(payload)
        if len(encoded.encode("utf-8")) > 32 * 1024:
            raise ValueError("event payload is too large")
        with self.app.db() as conn:
            row = conn.execute(
                "SELECT node_id FROM tasks WHERE id=?", (task_id,)
            ).fetchone()
            if row is None:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
                return
            if self.app.node_for_token(row["node_id"], self.headers) is None:
                self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
                return
            cursor = conn.execute(
                "INSERT INTO task_events(task_id,created_at,kind,payload_json) "
                "VALUES(?,?,?,?)",
                (task_id, now_ts(), kind, encoded),
            )
        self.send_json(
            HTTPStatus.CREATED, {"task_id": task_id, "seq": cursor.lastrowid}
        )

    def heartbeat(self, node_id: str) -> None:
        node = self.app.node_for_token(node_id, self.headers)
        if node is None:
            self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
            return
        body = self.read_json()
        version = str(body.get("agent_version", ""))[:64]
        policy = self.validated_node_policy(body.get("policy", {}))
        with self.app.db() as conn:
            conn.execute(
                "UPDATE nodes SET last_seen_at=?,agent_version=?,policy_json=? WHERE id=?",
                (now_ts(), version, compact_json(policy), node_id),
            )
        self.send_json(HTTPStatus.OK, {"ok": True})


class AppServer(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, address: tuple[str, int], app: ControlPlane):
        self.app = app
        super().__init__(address, Handler)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bind", default=os.environ.get("CANGER_BIND", "127.0.0.1"))
    parser.add_argument(
        "--port", type=int, default=int(os.environ.get("CANGER_PORT", "8787"))
    )
    parser.add_argument(
        "--db",
        default=os.environ.get(
            "CANGER_CONTROL_DB", "/var/lib/canger-control/control.sqlite3"
        ),
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    app = ControlPlane(
        args.db,
        os.environ.get("CANGER_ADMIN_TOKEN", ""),
        os.environ.get("CANGER_OWNER_TOKEN", ""),
        os.environ.get("CANGER_PERSONA_TOKEN", ""),
        os.environ.get("CANGER_OPERATOR_TOKEN", ""),
        os.environ.get("CANGER_APPROVAL_SIGNING_KEY", ""),
        os.environ.get("CANGER_OWNER_EMAIL", ""),
        os.environ.get("CANGER_PUBLIC_URL", ""),
    )
    server = AppServer((args.bind, args.port), app)
    print(f"canger-control listening on {args.bind}:{args.port}")
    server.serve_forever()


if __name__ == "__main__":
    main()