feat: add fixed enterprise Agent gate

Part 2/4 of verified local 18dfdfd: fail-closed navigation and intent gate, fixed Agent connector, loopback code-channel candidate scripts and tests.

server-tools/enterprise-lighthouse/fixed_agent_connector.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""Consume one lighthouse unlock and execute one compiled-in local connector."""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+
+
+LIGHTHOUSE_URL = "http://127.0.0.1:8031"
+FIXED_CONNECTORS = {
+    (
+        "AW-HLCC-CANDIDATE-DEPLOYER",
+        "deploy_release",
+        "AW-GZ-001",
+    ): "/usr/local/libexec/guanghu/aw-hlcc-candidate-deploy",
+}
+
+
+def fail(message, code=2):
+    print(message, file=sys.stderr)
+    raise SystemExit(code)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--unlock-id", required=True)
+    parser.add_argument("--principal-id", required=True)
+    parser.add_argument("--subject-kind", required=True, choices=("human", "persona", "persona_system"))
+    parser.add_argument("--human-anchor", required=True)
+    parser.add_argument("--agent-id", required=True)
+    parser.add_argument("--action", required=True)
+    parser.add_argument("--target-node-id", required=True)
+    return parser.parse_args()
+
+
+def consume_unlock(args, token):
+    payload = json.dumps(
+        {
+            "unlock_id": args.unlock_id,
+            "agent_id": args.agent_id,
+            "action": args.action,
+            "target_node_id": args.target_node_id,
+        }
+    ).encode()
+    request = urllib.request.Request(
+        LIGHTHOUSE_URL + "/v1/agents/consume",
+        data=payload,
+        method="POST",
+        headers={
+            "X-Lighthouse-Admin-Token": token,
+            "Content-Type": "application/json",
+            "X-Guanghu-Principal-Id": args.principal_id,
+            "X-Guanghu-Subject-Kind": args.subject_kind,
+            "X-Guanghu-Human-Anchor": args.human_anchor,
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            body = json.load(response)
+    except urllib.error.HTTPError as error:
+        fail("Lighthouse rejected the connector request with HTTP %d." % error.code, 4)
+    except (OSError, ValueError) as error:
+        fail("Lighthouse connector request failed: %s" % error, 4)
+    if not body.get("ok") or not body.get("consumed"):
+        fail("Lighthouse did not return a consumed Agent unlock.", 4)
+
+
+def main():
+    if os.geteuid() != 0:
+        fail("The fixed connector must be invoked by root.", 3)
+    args = parse_args()
+    key = (args.agent_id, args.action, args.target_node_id)
+    executable = FIXED_CONNECTORS.get(key)
+    if not executable:
+        fail("No fixed local connector is registered for this Agent/action/target.", 3)
+    token = os.environ.get("LIGHTHOUSE_ADMIN_TOKEN", "")
+    if not token:
+        fail("LIGHTHOUSE_ADMIN_TOKEN is required but must not be passed on the command line.", 3)
+    if not os.path.isfile(executable) or not os.access(executable, os.X_OK):
+        fail("The registered fixed local connector is unavailable.", 3)
+    consume_unlock(args, token)
+    completed = subprocess.run([executable], check=False)
+    raise SystemExit(completed.returncode)
+
+
+if __name__ == "__main__":
+    main()