| """ |
| Sovereign Node Key — provisioned per authorized node. |
| Generated by authority. Never committed to public repo. |
| Dispose and regenerate on each new authorization cycle. |
| """ |
|
|
| import json |
| import os |
| from pathlib import Path |
|
|
| _SOVEREIGN_DIR = Path(__file__).parent |
| _CAPABILITY_FILE = _SOVEREIGN_DIR / ".capability" |
| _AUTH_FILE = _SOVEREIGN_DIR / "authorization.json" |
|
|
|
|
| def require_node_key() -> str: |
| """ |
| Load and return the node capability token. |
| Raises RuntimeError if not authorized. |
| """ |
| if not _CAPABILITY_FILE.exists(): |
| raise RuntimeError( |
| "No capability token found. " |
| "Contact jessica@collectivekitty.com for access." |
| ) |
| token = _CAPABILITY_FILE.read_text().strip() |
| if not token: |
| raise RuntimeError("Capability token is empty.") |
|
|
| |
| if _AUTH_FILE.exists(): |
| auth = json.loads(_AUTH_FILE.read_text()) |
| status = auth.get("authorization_status", "") |
| if status != "ACTIVE": |
| raise RuntimeError(f"Node authorization status: {status}. Access denied.") |
|
|
| return token |
|
|
|
|
| def node_id() -> str: |
| """Return the provisioned node ID.""" |
| if _AUTH_FILE.exists(): |
| auth = json.loads(_AUTH_FILE.read_text()) |
| return auth.get("node_id", "unknown") |
| return "unknown" |
|
|
|
|
| def is_authorized() -> bool: |
| """Check if this node is authorized without raising.""" |
| try: |
| require_node_key() |
| return True |
| except RuntimeError: |
| return False |
|
|
