#!/usr/bin/env python3 """Device-bound HoloLake entrance to one enterprise responsibility repository.""" from __future__ import annotations import base64 import hashlib import hmac import json import os import re import secrets import sqlite3 import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey BIND = os.environ.get("GH_HOLOLAKE_ENTERPRISE_GATE_BIND", "127.0.0.1") PORT = int(os.environ.get("GH_HOLOLAKE_ENTERPRISE_GATE_PORT", "8033")) DB_PATH = os.environ.get("GH_HOLOLAKE_ENTERPRISE_GATE_DB", "/var/lib/guanghu-hololake-enterprise-gate/gate.sqlite3") REGISTRY_PATH = os.environ.get("GH_ENTERPRISE_IDENTITY_REGISTRY", "/etc/guanghu/enterprise-identity-registry.json") FORGEJO_USER_API = os.environ.get("GH_ENTERPRISE_FORGEJO_USER_API", "http://127.0.0.1:3341/api/v1/user") FORGEJO_WEB_BASE = os.environ.get("GH_ENTERPRISE_FORGEJO_WEB_BASE", "https://guanghu.chat/code").rstrip("/") RECEIPT_KEY = os.environ.get("GH_HOLOLAKE_ENTERPRISE_GATE_RECEIPT_KEY", "") MAX_BODY = 16_384 IDENTITY = re.compile(r"^[A-Za-z0-9._∞-]{3,80}$") HEX64 = re.compile(r"^[0-9a-f]{64}$") DOMAIN_MAP = { "DOMAIN-MAIN": "MAIN_DOMAIN", "DOMAIN-SUB": "BRANCH_DOMAIN", "DOMAIN-ZERO": "ZERO_DOMAIN", "DOMAIN-ZS": "ZERO_SENSE_DOMAIN", } def now() -> int: return int(time.time()) def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() def load_registry() -> dict: value = json.loads(Path(REGISTRY_PATH).read_text(encoding="utf-8")) if value.get("schema") != "guanghu.enterprise-identity-registry/v1" or not value.get("humans"): raise ValueError("enterprise identity registry invalid") return value def database() -> sqlite3.Connection: path = Path(DB_PATH) path.parent.mkdir(parents=True, exist_ok=True) db = sqlite3.connect(path) db.row_factory = sqlite3.Row db.executescript( """ PRAGMA journal_mode=WAL; CREATE TABLE IF NOT EXISTS devices ( key_id TEXT PRIMARY KEY, node_id TEXT NOT NULL UNIQUE, request_id TEXT NOT NULL, human_number TEXT NOT NULL, persona_id TEXT NOT NULL, domain_id TEXT NOT NULL, repository TEXT NOT NULL, public_key TEXT NOT NULL, fingerprint_sha256 TEXT NOT NULL, enrolled_at INTEGER NOT NULL, revoked_at INTEGER ); CREATE TABLE IF NOT EXISTS challenges ( challenge_id TEXT PRIMARY KEY, key_id TEXT NOT NULL, request_id TEXT NOT NULL, nonce TEXT NOT NULL, expires_at INTEGER NOT NULL, used_at INTEGER ); CREATE TABLE IF NOT EXISTS sessions ( session_hash TEXT PRIMARY KEY, key_id TEXT NOT NULL, human_number TEXT NOT NULL, persona_id TEXT NOT NULL, domain_id TEXT NOT NULL, repository TEXT NOT NULL, issued_at INTEGER NOT NULL, expires_at INTEGER NOT NULL, revoked_at INTEGER ); CREATE TABLE IF NOT EXISTS audit ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, observed_at INTEGER NOT NULL, kind TEXT NOT NULL, subject_hash TEXT NOT NULL, object_id TEXT NOT NULL ); """ ) return db def find_binding(registry: dict, human_number: str, persona_id: str, domain_id: str) -> dict | None: expected_domain = DOMAIN_MAP.get(domain_id) for human in registry["humans"]: if not hmac.compare_digest( str(human.get("human_number", "")).encode("utf-8"), human_number.encode("utf-8"), ): continue if human.get("responsibility_domain") != expected_domain: return None persona = next( ( item for item in human.get("personas", []) if item.get("role") == "PERSONA_SUBJECT" and hmac.compare_digest( str(item.get("current_persona_identity", "")).encode("utf-8"), persona_id.encode("utf-8"), ) ), None, ) return human if persona else None return None def parse_basic(header: str) -> tuple[str, str] | None: if not header.startswith("Basic "): return None try: username, password = base64.b64decode(header[6:], validate=True).decode().split(":", 1) except (ValueError, UnicodeDecodeError): return None if not re.fullmatch(r"[A-Za-z0-9_-]{1,40}", username) or not password or len(password) > 512: return None return username, password def verify_forgejo(username: str, password: str) -> bool: request = urllib.request.Request(FORGEJO_USER_API) request.add_header("Authorization", "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode()) request.add_header("Accept", "application/json") try: with urllib.request.urlopen(request, timeout=10) as response: value = json.load(response) return response.status == 200 and hmac.compare_digest(str(value.get("login", "")), username) except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, ValueError): return False def receipt(payload: dict) -> dict: if len(RECEIPT_KEY) < 32: raise RuntimeError("receipt signing key unavailable") body = canonical(payload) return { **payload, "receipt_hash": hashlib.sha256(body).hexdigest(), "receipt_signature": hmac.new(RECEIPT_KEY.encode(), body, hashlib.sha256).hexdigest(), } def challenge_message(device: sqlite3.Row, challenge: sqlite3.Row) -> bytes: return ( "HLP-ENTERPRISE-RESPONSIBILITY-ENTRANCE-0001\n" f"{challenge['request_id']}\n{device['domain_id']}\n{device['human_number']}\n" f"{device['persona_id']}\n{challenge['challenge_id']}\n" f"{challenge['nonce']}:{challenge['expires_at'] * 1000}" ).encode() def verify_challenge_signature(device: sqlite3.Row, challenge: sqlite3.Row, encoded_signature: str) -> bool: try: public = base64.b64decode(device["public_key"], validate=True) signature = base64.b64decode(encoded_signature, validate=True) if len(public) != 32 or len(signature) != 64: return False Ed25519PublicKey.from_public_bytes(public).verify(signature, challenge_message(device, challenge)) return True except (ValueError, InvalidSignature): return False class Handler(BaseHTTPRequestHandler): server_version = "GuanghuHoloLakeEnterpriseGate/1.0" def log_message(self, fmt: str, *args: object) -> None: print("[hololake-enterprise-gate] " + fmt % args) def respond(self, status: int, body: dict) -> None: encoded = json.dumps(body, ensure_ascii=False).encode() 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.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(encoded) def body(self) -> dict: length = int(self.headers.get("Content-Length", "0")) if length < 1 or length > MAX_BODY: raise ValueError("request body size invalid") value = json.loads(self.rfile.read(length)) if not isinstance(value, dict): raise ValueError("JSON object required") return value def do_GET(self) -> None: if self.path == "/health": try: registry = load_registry() db = database() counts = { "devices": db.execute("SELECT count(*) FROM devices WHERE revoked_at IS NULL").fetchone()[0], "active_sessions": db.execute("SELECT count(*) FROM sessions WHERE revoked_at IS NULL AND expires_at>?", (now(),)).fetchone()[0], } db.close() return self.respond(200, {"ok": True, "service": "guanghu-hololake-enterprise-gate", "registry_version": registry["version"], **counts}) except Exception: return self.respond(503, {"ok": False, "error": "enterprise gate unavailable"}) if self.path == "/v1/session": token = self.headers.get("Authorization", "").removeprefix("Bearer ") if len(token) < 32: return self.respond(401, {"ok": False, "error": "session required"}) db = database() row = db.execute("SELECT * FROM sessions WHERE session_hash=? AND revoked_at IS NULL AND expires_at>?", (hashlib.sha256(token.encode()).hexdigest(), now())).fetchone() db.close() if not row: return self.respond(401, {"ok": False, "error": "session invalid or expired"}) return self.respond(200, {"ok": True, "session": {"human_number": row["human_number"], "persona_id": row["persona_id"], "domain_id": row["domain_id"], "repository": row["repository"], "expires_at": row["expires_at"], "repository_url": f"{FORGEJO_WEB_BASE}/{row['repository']}"}}) return self.respond(404, {"ok": False, "error": "not found"}) def do_POST(self) -> None: try: payload = self.body() registry = load_registry() except (ValueError, OSError, json.JSONDecodeError): return self.respond(400, {"ok": False, "error": "request invalid"}) if self.path == "/v1/device/enroll": fields = {name: str(payload.get(name, "")) for name in ("request_id", "human_number", "persona_id", "domain_id", "node_id", "key_id", "public_key", "fingerprint_sha256", "idempotency_key")} if any(not IDENTITY.fullmatch(fields[name]) for name in ("request_id", "human_number", "persona_id", "domain_id", "node_id", "key_id", "idempotency_key")) or not HEX64.fullmatch(fields["fingerprint_sha256"]): return self.respond(400, {"ok": False, "error": "device enrollment fields invalid"}) human = find_binding(registry, fields["human_number"], fields["persona_id"], fields["domain_id"]) credentials = parse_basic(self.headers.get("Authorization", "")) if not human or not credentials or credentials[0] != human["username"] or not verify_forgejo(*credentials): return self.respond(401, {"ok": False, "error": "enterprise responsibility authentication failed"}) try: public = base64.b64decode(fields["public_key"], validate=True) except ValueError: public = b"" if len(public) != 32: return self.respond(400, {"ok": False, "error": "device public key invalid"}) db = database() existing = db.execute("SELECT * FROM devices WHERE key_id=? OR node_id=?", (fields["key_id"], fields["node_id"])).fetchone() if existing: matches = all(hmac.compare_digest(str(existing[name]), fields[name]) for name in ("request_id", "human_number", "persona_id", "domain_id", "node_id", "key_id", "public_key", "fingerprint_sha256")) and existing["revoked_at"] is None db.close() return self.respond(200 if matches else 409, {"ok": matches, "idempotent": matches, "error": None if matches else "device identity conflict"}) observed = now() db.execute("INSERT INTO devices VALUES (?,?,?,?,?,?,?,?,?,?,NULL)", (fields["key_id"], fields["node_id"], fields["request_id"], fields["human_number"], fields["persona_id"], fields["domain_id"], human["repository"], fields["public_key"], fields["fingerprint_sha256"], observed)) db.execute("INSERT INTO audit(observed_at,kind,subject_hash,object_id) VALUES (?,?,?,?)", (observed, "DEVICE_ENROLLED", hashlib.sha256(fields["human_number"].encode()).hexdigest(), fields["key_id"])) db.commit(); db.close() return self.respond(201, {"ok": True, "receipt": receipt({"state": "DEVICE_ENROLLED", "key_id": fields["key_id"], "node_id": fields["node_id"], "domain_id": fields["domain_id"], "repository": human["repository"], "observed_at": observed})}) if self.path == "/v1/challenges": key_id = str(payload.get("key_id", "")); request_id = str(payload.get("request_id", "")) db = database(); device = db.execute("SELECT * FROM devices WHERE key_id=? AND request_id=? AND revoked_at IS NULL", (key_id, request_id)).fetchone() if not device: db.close(); return self.respond(404, {"ok": False, "error": "registered device required"}) challenge_id = "HL-ENT-CH-" + secrets.token_hex(12).upper(); nonce = secrets.token_urlsafe(24); expires = now() + 120 db.execute("INSERT INTO challenges VALUES (?,?,?,?,?,NULL)", (challenge_id, key_id, request_id, nonce, expires)); db.commit(); db.close() return self.respond(201, {"ok": True, "challenge": {"request_id": request_id, "challenge_id": challenge_id, "nonce": nonce, "expires_unix_ms": expires * 1000}}) if self.path == "/v1/challenges/verify": challenge_id = str(payload.get("challenge_id", "")); signature = str(payload.get("signature", "")) db = database(); challenge = db.execute("SELECT * FROM challenges WHERE challenge_id=?", (challenge_id,)).fetchone() if not challenge or challenge["used_at"] is not None or challenge["expires_at"] <= now(): db.close(); return self.respond(401, {"ok": False, "error": "challenge invalid expired or replayed"}) device = db.execute("SELECT * FROM devices WHERE key_id=? AND revoked_at IS NULL", (challenge["key_id"],)).fetchone() if not device or not verify_challenge_signature(device, challenge, signature): db.close(); return self.respond(401, {"ok": False, "error": "device signature invalid"}) observed = now(); token = secrets.token_urlsafe(48); expires = observed + 600 db.execute("UPDATE challenges SET used_at=? WHERE challenge_id=?", (observed, challenge_id)) db.execute("INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?,NULL)", (hashlib.sha256(token.encode()).hexdigest(), device["key_id"], device["human_number"], device["persona_id"], device["domain_id"], device["repository"], observed, expires)) db.execute("INSERT INTO audit(observed_at,kind,subject_hash,object_id) VALUES (?,?,?,?)", (observed, "SESSION_ISSUED", hashlib.sha256(device["human_number"].encode()).hexdigest(), challenge_id)) db.commit(); db.close() return self.respond(200, {"ok": True, "session": {"token": token, "expires_unix_ms": expires * 1000, "domain_id": device["domain_id"], "repository": device["repository"], "repository_url": f"{FORGEJO_WEB_BASE}/{device['repository']}", "scope": "ONE_DOMAIN_ONE_REPOSITORY"}}) return self.respond(404, {"ok": False, "error": "not found"}) if __name__ == "__main__": load_registry() if len(RECEIPT_KEY) < 32: raise SystemExit("GH_HOLOLAKE_ENTERPRISE_GATE_RECEIPT_KEY must contain at least 32 characters") ThreadingHTTPServer((BIND, PORT), Handler).serve_forever()