key_os/security.py

from __future__ import annotations

import hashlib
import json
import re
from typing import Any


FACILITY_ACCESS_POLICY = {
    "schema": "key_os.facility_access_policy/v1",
    "default_mode": "read_observe_status_log_search_only",
    "safe_default_verbs": ["read", "observe", "status", "log", "search", "inspect"],
    "denied_by_default": [
        "clear_all_data",
        "remote_exec",
        "ssh_control",
        "secret_export",
        "billing_scale",
        "destructive_reset",
    ],
    "delegation_rule": "A delegated runtime must never exceed the parent/operator grant ceiling.",
}


ACTION_CONTRACTS: dict[str, dict[str, Any]] = {
    "route_query": {
        "risk": "inspect",
        "scope": "resource_router",
        "markings": ["operator_visible"],
        "access_mode": "read",
        "requires_confirmation": False,
        "receipt_mode": "optional",
    },
    "facility_register": {
        "risk": "mutating",
        "scope": "facility_registry",
        "markings": ["operator_visible", "registry_change"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "facility_invoke": {
        "risk": "varies",
        "scope": "facility_bus",
        "markings": ["operator_visible", "facility_receipt"],
        "access_mode": "governed",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "continuity_read": {
        "risk": "inspect",
        "scope": "codex_continuity_signal",
        "markings": ["operator_visible", "metadata_only"],
        "access_mode": "read",
        "requires_confirmation": False,
        "receipt_mode": "optional",
    },
    "continuity_checkpoint": {
        "risk": "mutating",
        "scope": "codex_continuity_signal",
        "markings": ["operator_visible", "checkpoint"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "signal_read": {
        "risk": "inspect",
        "scope": "observer_bus_signal_inbox",
        "markings": ["operator_visible", "metadata_only"],
        "access_mode": "read",
        "requires_confirmation": False,
        "receipt_mode": "optional",
    },
    "signal_ack": {
        "risk": "mutating",
        "scope": "observer_bus_signal_inbox",
        "markings": ["operator_visible", "ack_state"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "notepad_add": {
        "risk": "mutating",
        "scope": "research_notepad",
        "markings": ["operator_visible", "research_log"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "notepad_clear": {
        "risk": "destructive",
        "scope": "research_notepad",
        "markings": ["operator_visible", "data_removal"],
        "access_mode": "explicit_only",
        "requires_confirmation": True,
        "receipt_mode": "append",
    },
    "factory_reset": {
        "risk": "destructive",
        "scope": "os_state",
        "markings": ["operator_visible", "factory_reset"],
        "access_mode": "explicit_only",
        "requires_confirmation": True,
        "receipt_mode": "append",
    },
    "dispatch_start": {
        "risk": "mutating",
        "scope": "dispatch_governor",
        "markings": ["operator_visible", "autonomy_boundary"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
    "dispatch_stop": {
        "risk": "mutating",
        "scope": "dispatch_governor",
        "markings": ["operator_visible", "autonomy_boundary"],
        "access_mode": "write",
        "requires_confirmation": False,
        "receipt_mode": "append",
    },
}


SENSITIVE_KEY = re.compile(r"(token|secret|password|authorization|bearer|api[_-]?key)", re.IGNORECASE)


def redact(value: Any) -> Any:
    if isinstance(value, dict):
        redacted: dict[str, Any] = {}
        for key, item in value.items():
            if SENSITIVE_KEY.search(str(key)):
                redacted[str(key)] = "<redacted>"
            else:
                redacted[str(key)] = redact(item)
        return redacted
    if isinstance(value, list):
        return [redact(item) for item in value]
    if isinstance(value, str) and SENSITIVE_KEY.search(value[:80]):
        return "<redacted>"
    return value


def stable_hash(value: Any) -> str:
    raw = json.dumps(redact(value), sort_keys=True, default=str, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def preview(value: Any, limit: int = 900) -> Any:
    redacted = redact(value)
    text = json.dumps(redacted, sort_keys=True, default=str)
    if len(text) <= limit:
        return redacted
    return {"preview": text[:limit], "truncated": True, "chars": len(text)}


def contracts_report() -> dict[str, Any]:
    return {
        "schema": "key_os.security_contracts/v1",
        "policy": FACILITY_ACCESS_POLICY,
        "contracts": ACTION_CONTRACTS,
    }


def require_confirmation(action: str, payload: dict[str, Any]) -> None:
    contract = ACTION_CONTRACTS.get(action, {})
    if not contract.get("requires_confirmation"):
        return
    confirmation = str(payload.get("confirm") or payload.get("confirmation") or "").strip()
    if confirmation != "CONFIRM":
        raise ValueError(f"{action} requires confirmation='CONFIRM'")