guanghu-ice-heart/server-tools/enterprise-lighthouse/lighthouse.py

#!/usr/bin/env python3
"""Enterprise Lighthouse foundation: registry, preflight, and audit only.

This service is deliberately not a remote shell.  It accepts node admission
requests and validates fixed operation plans; execution remains with a future
GLSV node connector after human authorization.
"""
import hmac
import hashlib
import json
import os
import sqlite3
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from ipaddress import ip_address

DB = os.environ.get("LIGHTHOUSE_DB", "/var/lib/guanghu-enterprise-lighthouse/lighthouse.db")
TOKEN = os.environ.get("LIGHTHOUSE_ADMIN_TOKEN", "")
HOST = os.environ.get("LIGHTHOUSE_BIND", "127.0.0.1")
PORT = int(os.environ.get("LIGHTHOUSE_PORT", "8031"))
NODE_MAP_PATH = os.environ.get("LIGHTHOUSE_NODE_MAP", "/etc/guanghu/navigation-maps/AW-GZ-001.json")
MAP_ACK_TTL = int(os.environ.get("LIGHTHOUSE_MAP_ACK_TTL", "900"))
INTENT_STATE_TTL = int(os.environ.get("LIGHTHOUSE_INTENT_STATE_TTL", "1800"))
AGENT_UNLOCK_TTL = int(os.environ.get("LIGHTHOUSE_AGENT_UNLOCK_TTL", "600"))
FIXED_ACTIONS = {"health_check", "backup", "deploy_release", "restart_service", "rollback"}
REGISTERED_AGENTS = {
    "AW-INSPECTOR": {"health_check"},
    "AW-HLCC-CANDIDATE-DEPLOYER": {"health_check", "backup", "deploy_release", "restart_service", "rollback"},
}
SUBJECT_KINDS = {"human", "persona", "persona_system"}
FORBIDDEN_INTENT_KEYS = {"password", "token", "secret", "private_key", "authorization_code", "credential"}
DOMAINS = {
    "DOMAIN-ZS": "零感域", "DOMAIN-MAIN": "光湖主域", "DOMAIN-SUB": "光湖分域",
    "DOMAIN-ZERO": "光湖零域", "DOMAIN-FIFTH": "第五域",
}
ENTERPRISE_MANAGED_DOMAINS = {"DOMAIN-ZS", "DOMAIN-MAIN", "DOMAIN-SUB", "DOMAIN-ZERO"}
EXTERNAL_FOUNDATION_DOMAINS = {"DOMAIN-FIFTH"}


def now():
    return int(time.time())


def connection():
    os.makedirs(os.path.dirname(DB), exist_ok=True)
    db = sqlite3.connect(DB)
    db.row_factory = sqlite3.Row
    db.executescript("""
      CREATE TABLE IF NOT EXISTS domains (
        id TEXT PRIMARY KEY, name TEXT NOT NULL, state TEXT NOT NULL, created_at INTEGER NOT NULL
      );
      CREATE TABLE IF NOT EXISTS intakes (
        id TEXT PRIMARY KEY, created_at INTEGER NOT NULL, state TEXT NOT NULL,
        human_name TEXT NOT NULL, email TEXT NOT NULL, server_ip TEXT NOT NULL,
        domain_id TEXT NOT NULL, persona_ids TEXT NOT NULL, repository_urls TEXT NOT NULL,
        hosting_mode TEXT NOT NULL, notes TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS nodes (
        id TEXT PRIMARY KEY, domain_id TEXT NOT NULL, intake_id TEXT, state TEXT NOT NULL,
        display_name TEXT NOT NULL, server_ip TEXT NOT NULL, allowed_actions TEXT NOT NULL,
        public_key_fingerprint TEXT, last_heartbeat INTEGER, created_at INTEGER NOT NULL
      );
      CREATE TABLE IF NOT EXISTS audit (
        id TEXT PRIMARY KEY, created_at INTEGER NOT NULL, kind TEXT NOT NULL, payload TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS map_acks (
        principal TEXT PRIMARY KEY, map_hash TEXT NOT NULL, acknowledged_at INTEGER NOT NULL
      );
      CREATE TABLE IF NOT EXISTS intent_states (
        principal TEXT PRIMARY KEY, capsule_hash TEXT NOT NULL, capsule TEXT NOT NULL,
        restored_at INTEGER NOT NULL, expires_at INTEGER NOT NULL
      );
      CREATE TABLE IF NOT EXISTS agent_unlocks (
        id TEXT PRIMARY KEY, principal TEXT NOT NULL, agent_id TEXT NOT NULL,
        action TEXT NOT NULL, target_node_id TEXT NOT NULL, map_hash TEXT NOT NULL,
        capsule_hash TEXT NOT NULL, created_at INTEGER NOT NULL, expires_at INTEGER NOT NULL
      );
    """)
    for domain_id, name in DOMAINS.items():
        state = "EXTERNAL_PRIVATE_FOUNDATION" if domain_id in EXTERNAL_FOUNDATION_DOMAINS else "PENDING_ENTRY_NODE"
        db.execute("INSERT OR IGNORE INTO domains VALUES (?, ?, ?, ?)", (domain_id, name, state, now()))
    db.commit()
    return db


def audit(db, kind, payload):
    db.execute("INSERT INTO audit VALUES (?, ?, ?, ?)", (str(uuid.uuid4()), now(), kind, json.dumps(payload, ensure_ascii=False)))
    db.commit()


def parse_json(handler):
    length = int(handler.headers.get("Content-Length", "0"))
    if not 0 < length <= 50_000:
        raise ValueError("request body must be between 1 and 50000 bytes")
    return json.loads(handler.rfile.read(length).decode("utf-8"))


def valid_email(value):
    return isinstance(value, str) and len(value) <= 254 and value.count("@") == 1


def require_admin(handler):
    received = handler.headers.get("X-Lighthouse-Admin-Token", "")
    return bool(TOKEN) and hmac.compare_digest(received, TOKEN)


def load_node_map():
    with open(NODE_MAP_PATH, encoding="utf-8") as handle:
        node_map = json.load(handle)
    if node_map.get("schema") != "guanghu.navigation-map/v1" or node_map.get("node_id") != "AW-GZ-001":
        raise ValueError("AW-GZ-001 navigation map identity is invalid")
    return node_map


def navigation_map():
    body = {
        "schema": "guanghu.enterprise-global-navigation-map/v2",
        "world_entry": {
            "system_id": "SYS-GLW-ZY-EXEC-0001",
            "architecture": "GLS-0243",
            "rule": "The public Zhuyuan system is an intelligent operations system, not a persona or AI instance.",
            "physical_sovereignty_boundary": "Root may remove the system. Any operation that bypasses this gate is outside the Guanghu system and cannot claim a Guanghu receipt.",
        },
        "node_id": "AW-GZ-001",
        "node_map": load_node_map(),
        "domains": [{"id": key, "name": DOMAINS[key]} for key in sorted(DOMAINS)],
        "fixed_actions": sorted(FIXED_ACTIONS),
        "registered_agents": {agent: sorted(actions) for agent, actions in sorted(REGISTERED_AGENTS.items())},
        "mandatory_order": [
            "read-global-navigation-map",
            "ack-current-map-as-current-subject",
            "restore-intent-state-capsule",
            "unlock-one-registered-agent-for-one-action",
            "execute-registered-action",
            "write-verification-receipt",
        ],
        "forbidden": [
            "raw-shell",
            "unregistered-action",
            "secret-in-repository-or-intent-state",
            "shared-map-ack-between-subjects",
            "agent-unlock-without-current-intent-state",
            "unmapped-human-or-persona-operation",
        ],
    }
    encoded = json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
    return body, hashlib.sha256(encoded).hexdigest()


def request_principal(handler):
    principal_id = handler.headers.get("X-Guanghu-Principal-Id", "").strip()
    subject_kind = handler.headers.get("X-Guanghu-Subject-Kind", "").strip()
    human_anchor = handler.headers.get("X-Guanghu-Human-Anchor", "").strip()
    if not principal_id or subject_kind not in SUBJECT_KINDS or not human_anchor:
        return None
    identity = f"{subject_kind}|{principal_id}|{human_anchor}".encode()
    return {
        "id": principal_id,
        "subject_kind": subject_kind,
        "human_anchor": human_anchor,
        "digest": hmac.new(TOKEN.encode(), identity, hashlib.sha256).hexdigest(),
    }


def has_current_map_ack(db, principal_digest):
    _, current_hash = navigation_map()
    row = db.execute("SELECT map_hash, acknowledged_at FROM map_acks WHERE principal=?", (principal_digest,)).fetchone()
    return bool(
        row
        and hmac.compare_digest(row["map_hash"], current_hash)
        and now() - row["acknowledged_at"] <= MAP_ACK_TTL
    )


def has_forbidden_intent_key(value):
    if isinstance(value, dict):
        return any(str(key).lower() in FORBIDDEN_INTENT_KEYS or has_forbidden_intent_key(item) for key, item in value.items())
    if isinstance(value, list):
        return any(has_forbidden_intent_key(item) for item in value)
    return False


def validate_intent_capsule(payload, principal):
    required = {
        "schema", "human_anchor", "persona_id", "task_intent", "identity_boundary",
        "established_facts", "decisions", "rejected_routes", "authorization_state",
        "current_checkpoint", "next_action", "completion_definition", "evidence",
    }
    if payload.get("schema") != "guanghu.intent-state-capsule/v1" or required.difference(payload):
        raise ValueError("a complete guanghu.intent-state-capsule/v1 is required")
    if payload["human_anchor"] != principal["human_anchor"]:
        raise ValueError("intent-state human anchor does not match the current subject")
    if has_forbidden_intent_key(payload):
        raise ValueError("credentials and secrets are forbidden in intent state")
    for field in ("established_facts", "decisions", "rejected_routes", "completion_definition", "evidence"):
        if not isinstance(payload[field], list):
            raise ValueError(f"{field} must be a list")
    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return encoded, hashlib.sha256(encoded.encode()).hexdigest()


def current_intent_state(db, principal_digest):
    return db.execute(
        "SELECT capsule_hash, expires_at FROM intent_states WHERE principal=? AND expires_at>=?",
        (principal_digest, now()),
    ).fetchone()


def valid_agent_unlock(db, unlock_id, principal_digest, agent_id, action, target_node_id):
    _, map_hash = navigation_map()
    intent = current_intent_state(db, principal_digest)
    if not intent:
        return False
    row = db.execute(
        """SELECT * FROM agent_unlocks
           WHERE id=? AND principal=? AND agent_id=? AND action=? AND target_node_id=? AND expires_at>=?""",
        (unlock_id, principal_digest, agent_id, action, target_node_id, now()),
    ).fetchone()
    return bool(
        row
        and hmac.compare_digest(row["map_hash"], map_hash)
        and hmac.compare_digest(row["capsule_hash"], intent["capsule_hash"])
    )


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

    def log_message(self, fmt, *args):
        print("[lighthouse] " + fmt % args)

    def respond(self, status, body):
        encoded = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(encoded)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(encoded)

    def do_GET(self):
        db = connection()
        try:
            if self.path == "/health":
                try:
                    _, map_hash = navigation_map()
                except (OSError, ValueError, json.JSONDecodeError) as error:
                    return self.respond(503, {"ok": False, "service": "guanghu-enterprise-lighthouse", "error": f"navigation map unavailable: {error}"})
                return self.respond(200, {"ok": True, "service": "guanghu-enterprise-lighthouse", "mode": "map-intent-agent-gate", "map_hash": map_hash, "execution": "disabled"})
            if self.path == "/v1/status":
                counts = {row["state"]: row["count"] for row in db.execute("SELECT state, COUNT(*) AS count FROM nodes GROUP BY state")}
                return self.respond(200, {"ok": True, "domains": [dict(row) for row in db.execute("SELECT id,name,state FROM domains ORDER BY id")], "node_counts": counts, "fixed_actions": sorted(FIXED_ACTIONS), "raw_shell": "rejected"})
            if self.path in {"/v1/navigation-map", "/v2/global-navigation-map"}:
                try:
                    body, map_hash = navigation_map()
                except (OSError, ValueError, json.JSONDecodeError) as error:
                    return self.respond(503, {"ok": False, "error": f"navigation map unavailable: {error}"})
                return self.respond(200, {"ok": True, "map_hash": map_hash, "navigation_map": body})
            if self.path == "/v1/nodes":
                return self.respond(200, {"ok": True, "nodes": [dict(row) for row in db.execute("SELECT id,domain_id,state,display_name,allowed_actions,last_heartbeat,created_at FROM nodes ORDER BY created_at DESC")]})
            return self.respond(404, {"ok": False, "error": "not found"})
        finally:
            db.close()

    def do_POST(self):
        if not require_admin(self):
            return self.respond(401, {"ok": False, "error": "admin authorization required"})
        try:
            payload = parse_json(self)
        except (ValueError, json.JSONDecodeError) as error:
            return self.respond(400, {"ok": False, "error": str(error)})
        if "cmd" in payload or "shell" in payload or "command" in payload:
            return self.respond(400, {"ok": False, "error": "raw commands are never accepted by the lighthouse"})
        principal = request_principal(self)
        if not principal:
            return self.respond(400, {"ok": False, "error": "current human or persona identity and human anchor are required"})
        db = connection()
        try:
            if self.path == "/v1/navigation-map/ack":
                _, current_hash = navigation_map()
                supplied = str(payload.get("map_hash", ""))
                if not hmac.compare_digest(supplied, current_hash):
                    return self.respond(409, {"ok": False, "error": "navigation map changed; read the current map again"})
                db.execute("INSERT OR REPLACE INTO map_acks VALUES (?, ?, ?)", (principal["digest"], current_hash, now()))
                audit(db, "navigation_map_acknowledged", {"map_hash": current_hash, "principal_id": principal["id"], "subject_kind": principal["subject_kind"]})
                return self.respond(200, {"ok": True, "map_hash": current_hash, "expires_in": MAP_ACK_TTL})
            if not has_current_map_ack(db, principal["digest"]):
                return self.respond(423, {"ok": False, "error": "current navigation map must be read and acknowledged before any operation", "required": ["GET /v1/navigation-map", "POST /v1/navigation-map/ack"]})
            if self.path == "/v1/intent-state/restore":
                try:
                    encoded, capsule_hash = validate_intent_capsule(payload, principal)
                except ValueError as error:
                    return self.respond(400, {"ok": False, "error": str(error)})
                restored_at = now()
                db.execute(
                    "INSERT OR REPLACE INTO intent_states VALUES (?, ?, ?, ?, ?)",
                    (principal["digest"], capsule_hash, encoded, restored_at, restored_at + INTENT_STATE_TTL),
                )
                audit(db, "intent_state_restored", {"principal_id": principal["id"], "capsule_hash": capsule_hash})
                return self.respond(200, {"ok": True, "capsule_hash": capsule_hash, "expires_in": INTENT_STATE_TTL})
            if self.path == "/v1/agents/unlock":
                agent_id = str(payload.get("agent_id", ""))
                action = str(payload.get("action", ""))
                target_node_id = str(payload.get("target_node_id", ""))
                intent = current_intent_state(db, principal["digest"])
                if not intent:
                    return self.respond(423, {"ok": False, "error": "current intent state must be restored before an Agent can unlock"})
                if agent_id not in REGISTERED_AGENTS or action not in REGISTERED_AGENTS[agent_id] or target_node_id != "AW-GZ-001":
                    return self.respond(403, {"ok": False, "error": "Agent, action, or target is not registered"})
                _, map_hash = navigation_map()
                unlock_id = "UNLOCK-" + uuid.uuid4().hex.upper()
                created_at = now()
                db.execute(
                    "INSERT INTO agent_unlocks VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                    (unlock_id, principal["digest"], agent_id, action, target_node_id, map_hash, intent["capsule_hash"], created_at, created_at + AGENT_UNLOCK_TTL),
                )
                audit(db, "agent_unlocked", {"principal_id": principal["id"], "agent_id": agent_id, "action": action, "target_node_id": target_node_id, "unlock_id": unlock_id})
                return self.respond(200, {"ok": True, "unlock_id": unlock_id, "agent_id": agent_id, "action": action, "target_node_id": target_node_id, "expires_in": AGENT_UNLOCK_TTL})
            if self.path == "/v1/agents/consume":
                agent_id = str(payload.get("agent_id", ""))
                action = str(payload.get("action", ""))
                target_node_id = str(payload.get("target_node_id", ""))
                unlock_id = str(payload.get("unlock_id", ""))
                if not valid_agent_unlock(db, unlock_id, principal["digest"], agent_id, action, target_node_id):
                    return self.respond(423, {"ok": False, "error": "a current matching Agent unlock is required"})
                db.execute("DELETE FROM agent_unlocks WHERE id=?", (unlock_id,))
                audit(db, "agent_unlock_consumed", {"principal_id": principal["id"], "agent_id": agent_id, "action": action, "target_node_id": target_node_id, "unlock_id": unlock_id})
                return self.respond(200, {"ok": True, "unlock_id": unlock_id, "consumed": True, "execution": "not performed by the lighthouse; pass this receipt to the fixed local connector"})
            if not current_intent_state(db, principal["digest"]):
                return self.respond(423, {"ok": False, "error": "current intent state must be restored before any operation"})
            if self.path == "/v1/intakes":
                required = ("human_name", "email", "server_ip", "domain_id")
                if any(not payload.get(field) for field in required) or payload["domain_id"] not in DOMAINS or not valid_email(payload["email"]):
                    return self.respond(400, {"ok": False, "error": "human_name, valid email, server_ip, and known domain_id are required"})
                if payload["domain_id"] not in ENTERPRISE_MANAGED_DOMAINS:
                    return self.respond(403, {"ok": False, "error": "this domain is not managed by the enterprise lighthouse; a separate explicit sovereign authorization is required"})
                try:
                    ip_address(payload["server_ip"])
                except ValueError:
                    return self.respond(400, {"ok": False, "error": "server_ip must be a valid IP address"})
                intake_id = "INTAKE-" + uuid.uuid4().hex[:12].upper()
                db.execute("INSERT INTO intakes VALUES (?, ?, 'PENDING_REVIEW', ?, ?, ?, ?, ?, ?, ?, ?)", (
                    intake_id, now(), payload["human_name"].strip(), payload["email"].strip(), payload["server_ip"], payload["domain_id"],
                    json.dumps(payload.get("persona_ids", [])), json.dumps(payload.get("repository_urls", [])),
                    payload.get("hosting_mode", "own"), payload.get("notes", ""),
                ))
                audit(db, "intake_created", {"intake_id": intake_id, "domain_id": payload["domain_id"]})
                return self.respond(201, {"ok": True, "intake_id": intake_id, "state": "PENDING_REVIEW", "next": "sovereign approval, human email confirmation, then node connector enrollment"})
            if self.path == "/v1/nodes/bootstrap":
                """Register a manually verified routing node without enabling execution.

                This exists for the controlled migration of an already verified
                domain entry.  It deliberately cannot make a node ACTIVE: the
                GLSV connector, server-local key enrollment, and human email
                confirmation remain required before any operation can proceed.
                """
                required = ("id", "domain_id", "display_name", "server_ip")
                if any(not isinstance(payload.get(field), str) or not payload[field].strip() for field in required):
                    return self.respond(400, {"ok": False, "error": "id, domain_id, display_name, and server_ip are required"})
                if payload["domain_id"] not in ENTERPRISE_MANAGED_DOMAINS:
                    return self.respond(403, {"ok": False, "error": "this domain cannot be registered by the enterprise lighthouse"})
                try:
                    ip_address(payload["server_ip"])
                except ValueError:
                    return self.respond(400, {"ok": False, "error": "server_ip must be a valid IP address"})
                node_id = payload["id"].strip()
                if db.execute("SELECT 1 FROM nodes WHERE id=?", (node_id,)).fetchone():
                    return self.respond(409, {"ok": False, "error": "node id is already registered"})
                db.execute("INSERT INTO nodes VALUES (?, ?, NULL, 'CONNECTED_PENDING_CONNECTOR', ?, ?, ?, NULL, ?, ?)", (
                    node_id, payload["domain_id"], payload["display_name"].strip(), payload["server_ip"],
                    json.dumps(["health_check"]), now(), now(),
                ))
                db.execute("UPDATE domains SET state='ENTRY_NODE_CONNECTED' WHERE id=?", (payload["domain_id"],))
                audit(db, "node_bootstrap_connected", {"node_id": node_id, "domain_id": payload["domain_id"], "mode": "manual_verified_routing_only"})
                return self.respond(201, {"ok": True, "node_id": node_id, "state": "CONNECTED_PENDING_CONNECTOR", "execution": "disabled until GLSV connector enrollment and human authorization"})
            if self.path == "/v1/nodes/revoke":
                node_id = payload.get("node_id")
                if not isinstance(node_id, str) or not node_id.strip():
                    return self.respond(400, {"ok": False, "error": "node_id is required"})
                node = db.execute("SELECT id, domain_id, state FROM nodes WHERE id=?", (node_id.strip(),)).fetchone()
                if not node:
                    return self.respond(404, {"ok": False, "error": "node is not registered"})
                if node["state"] == "ACTIVE":
                    return self.respond(409, {"ok": False, "error": "ACTIVE nodes require a separate migration or retirement procedure"})
                db.execute("DELETE FROM nodes WHERE id=?", (node["id"],))
                if node["domain_id"] in EXTERNAL_FOUNDATION_DOMAINS:
                    db.execute("UPDATE domains SET state='EXTERNAL_PRIVATE_FOUNDATION' WHERE id=?", (node["domain_id"],))
                audit(db, "node_revoked", {"node_id": node["id"], "domain_id": node["domain_id"], "reason": "administrative revocation"})
                return self.respond(200, {"ok": True, "node_id": node["id"], "state": "REVOKED", "execution": "disabled"})
            if self.path == "/v1/preflight":
                action, node_id = payload.get("action"), payload.get("target_node_id")
                if action not in FIXED_ACTIONS or not node_id:
                    return self.respond(400, {"ok": False, "error": "fixed action and target_node_id are required", "allowed_actions": sorted(FIXED_ACTIONS)})
                agent_id = str(payload.get("agent_id", ""))
                unlock_id = str(payload.get("unlock_id", ""))
                if not valid_agent_unlock(db, unlock_id, principal["digest"], agent_id, action, node_id):
                    return self.respond(423, {"ok": False, "decision": "REJECT", "reason": "a current matching Agent unlock is required"})
                node = db.execute("SELECT * FROM nodes WHERE id=?", (node_id,)).fetchone()
                if not node:
                    return self.respond(404, {"ok": False, "decision": "REJECT", "reason": "target node is not registered"})
                allowed = json.loads(node["allowed_actions"])
                reasons = []
                if node["state"] != "ACTIVE": reasons.append("target node is not ACTIVE")
                if action not in allowed: reasons.append("action is not in the node allowlist")
                if action in {"deploy_release", "restart_service", "rollback"} and not payload.get("rollback_plan"): reasons.append("rollback_plan is required")
                if action in {"deploy_release", "restart_service"} and not payload.get("backup_reference"): reasons.append("backup_reference is required")
                decision = "ALLOW_FOR_AUTHORIZATION" if not reasons else "REJECT"
                result = {"ok": not reasons, "decision": decision, "reasons": reasons, "execution": "not performed; GLSV human authorization still required"}
                audit(db, "preflight", {"target_node_id": node_id, "action": action, "decision": decision, "reasons": reasons})
                return self.respond(200, result)
            return self.respond(404, {"ok": False, "error": "not found"})
        finally:
            db.close()


if __name__ == "__main__":
    print(f"Enterprise Lighthouse listening on {HOST}:{PORT}")
    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()