From be31c4136bc0445edc488400fc025399b0283ecc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B0=E6=9C=94?= <565183519@qq.com> Date: Thu, 3 Sep 2026 21:48:04 +0800 Subject: [PATCH] feat(enterprise): add device-bound responsibility gate --- ...enterprise-responsibility-entrance-v1.json | 2 +- .../registries/module-registry.json | 2 +- .../scripts/audit-clean-v1.py | 6 +- .../guanghu-hololake-enterprise-gate.service | 22 ++ .../enterprise-responsibility-gate/nginx.conf | 32 ++ .../enterprise-responsibility-gate/service.py | 300 ++++++++++++++++++ .../test_service.py | 31 ++ 7 files changed, 392 insertions(+), 3 deletions(-) create mode 100644 product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/guanghu-hololake-enterprise-gate.service create mode 100644 product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/nginx.conf create mode 100644 product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/service.py create mode 100644 product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/test_service.py diff --git a/product-source/hololake-clean-desktop/contracts/enterprise-responsibility-entrance-v1.json b/product-source/hololake-clean-desktop/contracts/enterprise-responsibility-entrance-v1.json index 7d0de2f13..7237a1170 100644 --- a/product-source/hololake-clean-desktop/contracts/enterprise-responsibility-entrance-v1.json +++ b/product-source/hololake-clean-desktop/contracts/enterprise-responsibility-entrance-v1.json @@ -1,7 +1,7 @@ { "schema": "hololake.enterprise-responsibility-entrance/v1", "contract_id": "HLP-ENTERPRISE-RESPONSIBILITY-ENTRANCE-0001", - "state": "LOCAL_DEVICE_PROOF_INSTALLED_KEYCHAIN_VERIFIED_SERVER_CHALLENGE_PENDING", + "state": "LOCAL_DEVICE_PROOF_INSTALLED_KEYCHAIN_VERIFIED_SERVER_GATE_SOURCE_TESTED_DEPLOYMENT_PENDING", "surface": "HOLOLAKE_GATED_ENTERPRISE_PORTAL", "enterprise_server_embedded": false, "identities": ["DOMAIN_ID", "RESPONSIBLE_HUMAN_ID", "BOUND_PERSONA_ID", "LOCAL_NODE_ID", "DEVICE_KEY_ID"], diff --git a/product-source/hololake-clean-desktop/registries/module-registry.json b/product-source/hololake-clean-desktop/registries/module-registry.json index ed5e85ee4..36a09a5df 100644 --- a/product-source/hololake-clean-desktop/registries/module-registry.json +++ b/product-source/hololake-clean-desktop/registries/module-registry.json @@ -77,7 +77,7 @@ "kind": "TEAM_ONLY_ENTERPRISE_GATEWAY", "audience": "HUMAN_TEAM", "summary_zh": "为已登记域负责人准备设备、人格与责任范围验证。", - "state": "LOCAL_DEVICE_PROOF_INSTALLED_KEYCHAIN_VERIFIED_SERVER_CHALLENGE_PENDING", + "state": "LOCAL_DEVICE_PROOF_INSTALLED_SERVER_GATE_SOURCE_TESTED_DEPLOYMENT_PENDING", "source": "src-tauri/src/enterprise_entrance.rs", "enterprise_server_embedded": false, "required_proofs": ["DOMAIN_RESPONSIBLE_HUMAN_ID", "BOUND_PERSONA_ID", "LOCAL_DEVICE_KEY", "AUXILIARY_MACHINE_FINGERPRINT", "SERVER_CHALLENGE_SIGNATURE"], diff --git a/product-source/hololake-clean-desktop/scripts/audit-clean-v1.py b/product-source/hololake-clean-desktop/scripts/audit-clean-v1.py index e1d349071..4119dc046 100644 --- a/product-source/hololake-clean-desktop/scripts/audit-clean-v1.py +++ b/product-source/hololake-clean-desktop/scripts/audit-clean-v1.py @@ -41,6 +41,10 @@ for required_file in ( "contracts/clean-v1-execution-baseline.json", "contracts/enterprise-responsibility-entrance-v1.json", "language/HOLOLAKE-CLEAN-V1-DIRECT-LANGUAGE-EXECUTION-BASELINE.tcs", + "server/enterprise-responsibility-gate/service.py", + "server/enterprise-responsibility-gate/test_service.py", + "server/enterprise-responsibility-gate/guanghu-hololake-enterprise-gate.service", + "server/enterprise-responsibility-gate/nginx.conf", ): if not (ROOT / required_file).exists(): errors.append(f"public runtime file missing: {required_file}") @@ -67,7 +71,7 @@ if enterprise.get("device_proof", {}).get("fingerprint_is_sole_credential") is n errors.append("machine fingerprint incorrectly used as sole credential") if enterprise.get("session", {}).get("scope") != "ONE_DOMAIN_ONE_REPOSITORY": errors.append("enterprise session is not responsibility scoped") -if enterprise.get("state") != "LOCAL_DEVICE_PROOF_INSTALLED_KEYCHAIN_VERIFIED_SERVER_CHALLENGE_PENDING": +if enterprise.get("state") != "LOCAL_DEVICE_PROOF_INSTALLED_KEYCHAIN_VERIFIED_SERVER_GATE_SOURCE_TESTED_DEPLOYMENT_PENDING": errors.append("enterprise local device proof implementation state drift") time_source = (ROOT / "src/time.ts").read_text() app_source = (ROOT / "src/App.tsx").read_text() diff --git a/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/guanghu-hololake-enterprise-gate.service b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/guanghu-hololake-enterprise-gate.service new file mode 100644 index 000000000..0631d7420 --- /dev/null +++ b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/guanghu-hololake-enterprise-gate.service @@ -0,0 +1,22 @@ +[Unit] +Description=Guanghu HoloLake Enterprise Responsibility Device Gate +After=network-online.target guanghu-enterprise-identity.service + +[Service] +Type=simple +User=lighthouse +Group=lighthouse +EnvironmentFile=/etc/guanghu/hololake-enterprise-gate.env +ExecStart=/usr/bin/python3 /opt/guanghu-hololake-enterprise-gate/current/service.py +Restart=on-failure +RestartSec=2 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ReadOnlyPaths=/etc/guanghu/enterprise-identity-registry.json +ReadWritePaths=/var/lib/guanghu-hololake-enterprise-gate + +[Install] +WantedBy=multi-user.target + diff --git a/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/nginx.conf b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/nginx.conf new file mode 100644 index 000000000..1d7e3fbd5 --- /dev/null +++ b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/nginx.conf @@ -0,0 +1,32 @@ +location = /api/hololake/enterprise/device-gate/health { + limit_except GET { deny all; } + proxy_pass http://127.0.0.1:8033/health; + proxy_set_header Host $host; + add_header Cache-Control "no-store" always; +} +location = /api/hololake/enterprise/device-gate/enroll { + limit_except POST { deny all; } + proxy_pass http://127.0.0.1:8033/v1/device/enroll; + proxy_set_header Host $host; + proxy_set_header Authorization $http_authorization; + client_max_body_size 16k; +} +location = /api/hololake/enterprise/device-gate/challenges { + limit_except POST { deny all; } + proxy_pass http://127.0.0.1:8033/v1/challenges; + proxy_set_header Host $host; + client_max_body_size 16k; +} +location = /api/hololake/enterprise/device-gate/challenges/verify { + limit_except POST { deny all; } + proxy_pass http://127.0.0.1:8033/v1/challenges/verify; + proxy_set_header Host $host; + client_max_body_size 16k; +} +location = /api/hololake/enterprise/device-gate/session { + limit_except GET { deny all; } + proxy_pass http://127.0.0.1:8033/v1/session; + proxy_set_header Host $host; + proxy_set_header Authorization $http_authorization; + add_header Cache-Control "no-store" always; +} diff --git a/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/service.py b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/service.py new file mode 100644 index 000000000..9300f7728 --- /dev/null +++ b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/service.py @@ -0,0 +1,300 @@ +#!/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() diff --git a/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/test_service.py b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/test_service.py new file mode 100644 index 000000000..4bb064559 --- /dev/null +++ b/product-source/hololake-clean-desktop/server/enterprise-responsibility-gate/test_service.py @@ -0,0 +1,31 @@ +import base64 +import os +import sqlite3 +import tempfile +import unittest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives import serialization + +import service + + +class GateTests(unittest.TestCase): + def test_binding_requires_responsible_domain_and_primary_persona(self): + registry = {"humans": [{"human_number": "TCS-GL-0016∞", "responsibility_domain": "MAIN_DOMAIN", "repository": "awen/guanghu-main-work", "personas": [{"role": "PERSONA_SUBJECT", "current_persona_identity": "PER-AW-ARCH-001"}, {"role": "RELATIONSHIP_CONTINUITY_SUPPORT", "current_persona_identity": "PER-ZQ001"}]}]} + self.assertIsNotNone(service.find_binding(registry, "TCS-GL-0016∞", "PER-AW-ARCH-001", "DOMAIN-MAIN")) + self.assertIsNone(service.find_binding(registry, "TCS-GL-0016∞", "PER-AW-ARCH-001", "DOMAIN-SUB")) + self.assertIsNone(service.find_binding(registry, "TCS-GL-0016∞", "PER-ZQ001", "DOMAIN-MAIN")) + + def test_signature_binds_request_domain_human_persona_challenge_and_expiry(self): + private = Ed25519PrivateKey.generate() + public = private.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + device = {"public_key": base64.b64encode(public).decode(), "request_id": "REQ-1", "domain_id": "DOMAIN-MAIN", "human_number": "TCS-GL-0016∞", "persona_id": "PER-AW-ARCH-001"} + challenge = {"request_id": "REQ-1", "challenge_id": "CH-1", "nonce": "NONCE-1", "expires_at": 2000000000} + signature = private.sign(service.challenge_message(device, challenge)) + self.assertTrue(service.verify_challenge_signature(device, challenge, base64.b64encode(signature).decode())) + altered = dict(challenge, nonce="NONCE-2") + self.assertFalse(service.verify_challenge_signature(device, altered, base64.b64encode(signature).decode())) + + +if __name__ == "__main__": + unittest.main()