railiance-platform/scripts/custody_contract.py

#!/usr/bin/env python3
"""Canonical, value-safe contracts for ephemeral credential projection."""

from __future__ import annotations

import hashlib
import json
import re
import urllib.error
import urllib.parse
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Callable


CONTRACT_INTERFACE = "railiance.custody-projection-contract"
BROKER_INTERFACE = "railiance.custody-broker-readiness"
PROJECTION_INTERFACE = "railiance.custody-projection-receipt"
CLEANUP_INTERFACE = "railiance.custody-cleanup-receipt"
WORKPLAN_ID = "RAILIANCE-WP-0025"
BROKER_OWNER = "whitehat-security"
BROKER_SUBJECT_PREFIX = "WP0025-BROKER-READINESS"
REPO_ROOT = Path(__file__).resolve().parents[1]
INTERFACE_ARTIFACTS = (
    "schemas/custody-projection-contract.schema.json",
    "schemas/custody-broker-readiness.schema.json",
    "schemas/custody-projection-receipt.schema.json",
    "schemas/custody-cleanup-receipt.schema.json",
)
SHA256 = re.compile(r"(?:sha256:)?[0-9a-f]{64}\Z")
GIT_REVISION = re.compile(r"[0-9a-f]{40}\Z")
DNS_LABEL = re.compile(r"[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?\Z")


class ContractError(RuntimeError):
    pass


def canonical_json(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def digest(value: Any) -> str:
    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()


def file_digest(path: Path) -> str:
    result = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            result.update(block)
    return result.hexdigest()


def parse_time(value: Any, field: str) -> datetime:
    if not isinstance(value, str):
        raise ContractError(f"{field} must be an RFC3339 timestamp")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ContractError(f"{field} must be an RFC3339 timestamp") from exc
    if parsed.tzinfo is None:
        raise ContractError(f"{field} must include a timezone")
    return parsed.astimezone(UTC)


def load_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ContractError(f"JSON document is unavailable or invalid: {path}") from exc
    if not isinstance(value, dict):
        raise ContractError("JSON document must be an object")
    return value


def _require_string(value: dict[str, Any], key: str) -> str:
    observed = value.get(key)
    if not isinstance(observed, str) or not observed:
        raise ContractError(f"{key} must be a non-empty string")
    return observed


def _require_sha(value: dict[str, Any], key: str) -> str:
    observed = _require_string(value, key)
    if not SHA256.fullmatch(observed):
        raise ContractError(f"{key} must be a SHA-256 digest")
    return observed.removeprefix("sha256:")


def resource_suffix(engagement_id: str) -> str:
    return hashlib.sha256(engagement_id.encode("utf-8")).hexdigest()[:12]


def interface_artifacts(root: Path = REPO_ROOT) -> dict[str, str]:
    return {path: file_digest(root / path) for path in INTERFACE_ARTIFACTS}


def resource_names(contract: dict[str, Any]) -> dict[str, str]:
    suffix = resource_suffix(contract["engagement_id"])
    return {
        "policy": f"custody-{suffix}",
        "role": f"custody-{suffix}",
        "store": f"custody-{suffix}",
        "external_secret": contract["runner"]["secret_name"],
    }


def validate_projection_contract(value: dict[str, Any]) -> dict[str, Any]:
    if value.get("interface") != CONTRACT_INTERFACE or value.get("version") != 1:
        raise ContractError("unsupported projection contract interface/version")
    if value.get("workplan_id") != WORKPLAN_ID:
        raise ContractError("projection contract has the wrong workplan")
    engagement_id = _require_string(value, "engagement_id")
    if not re.fullmatch(r"[A-Z0-9][A-Z0-9._-]{7,127}", engagement_id):
        raise ContractError("engagement_id has an invalid shape")
    if value.get("status") != "approved":
        raise ContractError("engagement must be approved")

    target = value.get("target")
    runner = value.get("runner")
    window = value.get("window")
    authority = value.get("authority")
    identities = value.get("identities")
    if not all(isinstance(item, dict) for item in (target, runner, window, authority)):
        raise ContractError("target, runner, window and authority must be objects")
    if not isinstance(identities, list) or len(identities) != 2:
        raise ContractError("exactly two identity handles are required")

    target_id = _require_string(target, "id")
    target_namespace = _require_string(target, "namespace")
    target_deployment = _require_string(target, "deployment")
    _require_string(target, "container")
    _require_string(target, "sender_external_secret")
    if not DNS_LABEL.fullmatch(target_namespace) or not DNS_LABEL.fullmatch(target_deployment):
        raise ContractError("target namespace and deployment must be DNS labels")
    revision = _require_string(target, "revision")
    if not GIT_REVISION.fullmatch(revision):
        raise ContractError("target.revision must be a full Git revision")
    image = _require_string(target, "image_digest")
    if not re.fullmatch(r"sha256:[0-9a-f]{64}", image):
        raise ContractError("target.image_digest must be a SHA-256 image digest")
    _require_sha(target, "contract_sha256")
    _require_sha(runner, "manifest_sha256")
    _require_sha(value, "engagement_contract_sha256")

    namespace = _require_string(runner, "namespace")
    secret_name = _require_string(runner, "secret_name")
    _require_string(runner, "service_account")
    mount_root = _require_string(runner, "mount_root").rstrip("/")
    if not DNS_LABEL.fullmatch(namespace) or not DNS_LABEL.fullmatch(secret_name):
        raise ContractError("runner namespace and secret_name must be DNS labels")
    if not mount_root.startswith("/") or ".." in Path(mount_root).parts:
        raise ContractError("runner.mount_root must be an absolute safe path")

    starts = parse_time(window.get("starts_at"), "window.starts_at")
    cutoff = parse_time(window.get("projection_cutoff"), "window.projection_cutoff")
    expires = parse_time(window.get("expires_at"), "window.expires_at")
    if not starts < cutoff < expires:
        raise ContractError("window must satisfy starts_at < projection_cutoff < expires_at")
    if (expires - starts).total_seconds() > 900:
        raise ContractError("ephemeral custody window may not exceed 900 seconds")

    remote = _require_string(authority, "remote")
    if any(character.isspace() for character in remote):
        raise ContractError("authority.remote must be one SSH destination")
    registry_path = _require_string(authority, "registry_path")
    registry_field = _require_string(authority, "registry_field")
    kv_mount = _require_string(authority, "kv_mount")
    kv_prefix = _require_string(authority, "kv_prefix").strip("/")
    _require_string(authority, "eso_service_account")
    _require_string(authority, "eso_namespace")
    if engagement_id not in kv_prefix or target_id not in kv_prefix:
        raise ContractError("authority.kv_prefix must bind engagement_id and target id")
    if "/" in kv_mount or not registry_path or "/" in registry_field:
        raise ContractError("invalid KV mount or registry field")

    expected_handles = {"token-a", "token-b"}
    handles: set[str] = set()
    roles: set[str] = set()
    senders: set[str] = set()
    tenants: set[str] = set()
    for identity in identities:
        if not isinstance(identity, dict):
            raise ContractError("identity entries must be objects")
        handle = _require_string(identity, "handle")
        role = _require_string(identity, "role")
        sender = _require_string(identity, "sender_name")
        tenant = _require_string(identity, "tenant")
        mount_path = _require_string(identity, "mount_path")
        if identity.get("may_read") is not True or identity.get("may_write") is not True:
            raise ContractError("temporary identities require may_read=true and may_write=true")
        if mount_path != f"{mount_root}/{handle}":
            raise ContractError("identity mount_path must be mount_root plus its handle")
        handles.add(handle)
        roles.add(role)
        senders.add(sender)
        tenants.add(tenant)
    if handles != expected_handles or roles != {"attacker", "owner"}:
        raise ContractError("identities must provide token-a/token-b and attacker/owner roles")
    if len(senders) != 2 or len(tenants) != 2:
        raise ContractError("sender names and tenants must be distinct")

    names = resource_names(value)
    if any(not DNS_LABEL.fullmatch(name) for name in names.values()):
        raise ContractError("derived resource name is not a DNS label")
    return value


def contract_digest(contract: dict[str, Any]) -> str:
    validate_projection_contract(contract)
    return digest(contract)


def assert_value_safe(value: Any) -> None:
    forbidden_keys = {
        "token",
        "tokens",
        "password",
        "secret_value",
        "secret_values",
        "bearer",
        "private_key",
    }
    if isinstance(value, dict):
        for key, item in value.items():
            if str(key).lower() in forbidden_keys:
                raise ContractError(f"value-bearing field is forbidden: {key}")
            assert_value_safe(item)
    elif isinstance(value, list):
        for item in value:
            assert_value_safe(item)


def broker_subject(receipt: dict[str, Any]) -> str:
    return "/".join(
        (
            BROKER_SUBJECT_PREFIX,
            "v1",
            receipt["engagement_id"],
            receipt["decision"],
            receipt["projection_contract_digest"],
        )
    )


def validate_broker_receipt(
    receipt: dict[str, Any], contract: dict[str, Any], *, now: datetime | None = None
) -> dict[str, Any]:
    validate_projection_contract(contract)
    assert_value_safe(receipt)
    if receipt.get("interface") != BROKER_INTERFACE or receipt.get("version") != 1:
        raise ContractError("unsupported broker receipt interface/version")
    if receipt.get("workplan_id") != WORKPLAN_ID or receipt.get("owner") != BROKER_OWNER:
        raise ContractError("broker receipt has the wrong workplan or owner")
    if receipt.get("decision") != "approve":
        raise ContractError("broker receipt is not approved")
    if receipt.get("engagement_id") != contract["engagement_id"]:
        raise ContractError("broker receipt engagement does not match")
    if receipt.get("target_id") != contract["target"]["id"]:
        raise ContractError("broker receipt target does not match")
    if receipt.get("projection_contract_digest") != contract_digest(contract):
        raise ContractError("broker receipt projection contract is stale")
    if receipt.get("projection_receipt_interface") != PROJECTION_INTERFACE:
        raise ContractError("broker receipt cannot consume projection receipt v1")
    if receipt.get("interface_artifacts") != interface_artifacts():
        raise ContractError("broker receipt interface artifacts are stale")
    if receipt.get("required_roles") != ["attacker", "owner"]:
        raise ContractError("broker receipt has the wrong identity roles")
    expected_mounts = sorted(identity["mount_path"] for identity in contract["identities"])
    if receipt.get("mount_paths") != expected_mounts:
        raise ContractError("broker receipt mount paths do not match")
    adapter = receipt.get("adapter")
    if not isinstance(adapter, dict):
        raise ContractError("broker receipt requires adapter evidence")
    adapter_path = adapter.get("path")
    if (
        adapter.get("repo") != "whitehat-security"
        or not isinstance(adapter_path, str)
        or not adapter_path
        or Path(adapter_path).is_absolute()
        or ".." in Path(adapter_path).parts
    ):
        raise ContractError("broker adapter repo/path is invalid")
    if not GIT_REVISION.fullmatch(str(adapter.get("revision", ""))):
        raise ContractError("broker adapter revision must be a full Git revision")
    if not SHA256.fullmatch(str(adapter.get("sha256", ""))):
        raise ContractError("broker adapter requires a SHA-256 digest")
    if adapter.get("tests_passed") is not True or receipt.get("cleanup_request_supported") is not True:
        raise ContractError("broker adapter tests and cleanup support are required")
    if receipt.get("secret_values_observed") is not False:
        raise ContractError("broker receipt is not value-safe")
    created = parse_time(receipt.get("created_at"), "broker.created_at")
    current = (now or datetime.now(UTC)).astimezone(UTC)
    starts = parse_time(contract["window"]["starts_at"], "window.starts_at")
    expires = parse_time(contract["window"]["expires_at"], "window.expires_at")
    if created > current or created > expires or created < starts.replace(hour=0, minute=0, second=0, microsecond=0):
        raise ContractError("broker receipt timestamp is outside the engagement day")
    return receipt


def parse_broker_message(
    message: dict[str, Any], contract: dict[str, Any], *, now: datetime | None = None
) -> dict[str, Any] | None:
    if message.get("from_agent") != BROKER_OWNER or message.get("to_agent") != "railiance-platform":
        return None
    body = message.get("body")
    if not isinstance(body, str):
        return None
    try:
        receipt = json.loads(body)
    except json.JSONDecodeError:
        return None
    if not isinstance(receipt, dict):
        return None
    try:
        if message.get("subject") != broker_subject(receipt):
            return None
        return validate_broker_receipt(receipt, contract, now=now)
    except (ContractError, KeyError, TypeError):
        return None


def http_json(
    method: str,
    url: str,
    payload: dict[str, Any] | None = None,
    *,
    opener: Callable[..., Any] = urllib.request.urlopen,
) -> Any:
    data = canonical_json(payload).encode("utf-8") if payload is not None else None
    request = urllib.request.Request(
        url,
        data=data,
        method=method,
        headers={"Content-Type": "application/json"} if data else {},
    )
    try:
        with opener(request, timeout=10) as response:
            return json.loads(response.read().decode("utf-8"))
    except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
        raise ContractError(f"State Hub request failed: {method} {url}") from exc


def current_broker_receipt(
    contract: dict[str, Any], api_base: str, *, now: datetime | None = None
) -> dict[str, Any]:
    query = urllib.parse.urlencode(
        {"to_agent": "railiance-platform", "unread_only": "false"}
    )
    messages = http_json("GET", f"{api_base.rstrip('/')}/messages/?{query}")
    if not isinstance(messages, list):
        raise ContractError("State Hub returned an invalid message list")
    receipts = [
        parsed
        for message in messages
        if isinstance(message, dict)
        and (parsed := parse_broker_message(message, contract, now=now)) is not None
    ]
    if not receipts:
        raise ContractError("no current Whitehat broker-readiness receipt exists")
    return max(receipts, key=lambda item: item["created_at"])


def validate_projection_receipt(
    receipt: dict[str, Any], contract: dict[str, Any]
) -> dict[str, Any]:
    validate_projection_contract(contract)
    assert_value_safe(receipt)
    if receipt.get("interface") != PROJECTION_INTERFACE or receipt.get("version") != 1:
        raise ContractError("unsupported projection receipt interface/version")
    if receipt.get("workplan_id") != WORKPLAN_ID or receipt.get("state") != "projected":
        raise ContractError("projection receipt has the wrong workplan or state")
    if receipt.get("engagement_id") != contract["engagement_id"]:
        raise ContractError("projection receipt engagement does not match")
    if receipt.get("projection_contract_digest") != contract_digest(contract):
        raise ContractError("projection receipt contract is stale")
    expected_target = {
        "id": contract["target"]["id"],
        "revision": contract["target"]["revision"],
        "image_digest": contract["target"]["image_digest"],
    }
    if receipt.get("target") != expected_target:
        raise ContractError("projection receipt target does not match")
    lease_id = receipt.get("lease_id")
    if not isinstance(lease_id, str) or not re.fullmatch(r"custody:[0-9a-f]{32}", lease_id):
        raise ContractError("projection receipt lease_id is invalid")
    if receipt.get("secret_values_observed") is not False:
        raise ContractError("projection receipt is not value-safe")
    projected = parse_time(receipt.get("projected_at"), "projection.projected_at")
    expires = parse_time(receipt.get("expires_at"), "projection.expires_at")
    contract_expires = parse_time(contract["window"]["expires_at"], "window.expires_at")
    if expires != contract_expires or projected >= expires:
        raise ContractError("projection receipt has invalid projection/expiry times")
    identities = receipt.get("identities")
    expected_identities = sorted(
        (
            {
                "handle": item["handle"],
                "role": item["role"],
                "sender_name": item["sender_name"],
                "mount_path": item["mount_path"],
            }
            for item in contract["identities"]
        ),
        key=lambda item: item["handle"],
    )
    if identities != expected_identities:
        raise ContractError("projection receipt identity handles do not match")
    resources = receipt.get("resources")
    if not isinstance(resources, dict) or resources.get("names") != resource_names(contract):
        raise ContractError("projection receipt resources do not match")
    uids = resources.get("uids")
    if not isinstance(uids, dict) or set(uids) != {"store", "external_secret", "mounted_secret"}:
        raise ContractError("projection receipt requires exact Kubernetes UIDs")
    if any(not isinstance(uid, str) or not uid for uid in uids.values()):
        raise ContractError("projection receipt contains an invalid Kubernetes UID")
    base = dict(receipt)
    observed_id = base.pop("receipt_id", None)
    if observed_id != f"sha256:{digest(base)}":
        raise ContractError("projection receipt id does not match its canonical content")
    return receipt


def validate_cleanup_receipt(
    receipt: dict[str, Any], projection: dict[str, Any], contract: dict[str, Any]
) -> dict[str, Any]:
    validate_projection_receipt(projection, contract)
    assert_value_safe(receipt)
    if receipt.get("interface") != CLEANUP_INTERFACE or receipt.get("version") != 1:
        raise ContractError("unsupported cleanup receipt interface/version")
    if receipt.get("workplan_id") != WORKPLAN_ID or receipt.get("state") != "cleaned":
        raise ContractError("cleanup receipt has the wrong workplan or state")
    if receipt.get("engagement_id") != contract["engagement_id"]:
        raise ContractError("cleanup receipt engagement does not match")
    if receipt.get("lease_id") != projection["lease_id"]:
        raise ContractError("cleanup receipt lease does not match")
    if receipt.get("projection_receipt_id") != projection["receipt_id"]:
        raise ContractError("cleanup receipt projection id does not match")
    parse_time(receipt.get("cleaned_at"), "cleanup.cleaned_at")
    if receipt.get("target_ready") is not True or receipt.get("secret_values_observed") is not False:
        raise ContractError("cleanup receipt does not prove safe target recovery")
    return receipt