chimera-gfx-Public/tools/phase10w_shsrv_client_policy.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Offline policy model for a future one-shot shsrv client.

There is intentionally no transport, CLI, socket, DNS lookup, file output or
clock acquisition. Tests supply synthetic records and an explicit host time.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
import re
from typing import Any

from phase10v_shsrv_collector_model import (
    CollectorError as CollectorModelError,
    OfflineCollector,
)


COLLECTOR_SHA256 = \
    "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c"
SOURCE_BOUND_PORT = 2323
MAX_DEADLINE_SECONDS = 10
MAX_APPROVAL_LIFETIME = timedelta(minutes=15)
WINDOW_COMMANDS = {
    "T2_GREETING_AND_HELP": ("help",),
    "T3_ONE_EXACT_PATH": ("stat", "sum"),
}
ACTIVATION_FIELDS = {
    "active", "policy_sha256", "collector_sha256", "run_id",
    "target_address", "target_port", "window", "exact_literal_path",
    "commands", "deadline_seconds", "expires_at",
}
REQUIRED_TRUE_FIELDS = {
    "ps5_connection_authorized", "device_request_authorized",
    "result_receive_authorized", "spawned_shell_effects_accepted",
    "automatic_serial_query_accepted", "automatic_telemetry_query_accepted",
    "sanitized_output_only_accepted",
    "physical_memory_erasure_unproven_accepted",
}
REQUIRED_FALSE_FIELDS = {
    "target_build_authorized", "device_transfer_authorized",
    "device_execution_authorized", "installation_authorized",
    "autoload_authorized", "device_write_authorized", "automatic_retry",
    "reconnect_authorized", "resume_authorized", "fallback_authorized",
}
APPROVAL_FIELDS = ACTIVATION_FIELDS | REQUIRED_TRUE_FIELDS | \
    REQUIRED_FALSE_FIELDS | {"attested", "listener_already_running_attested"}


class PolicyError(RuntimeError):
    """Fail-closed policy error without record values."""


@dataclass(frozen=True)
class SessionPlan:
    """Immutable plan only; it has no method capable of network I/O."""

    run_id: str
    target_address: str
    target_port: int
    window: str
    exact_literal_path: str | None
    commands: tuple[str, ...]
    deadline_seconds: int
    expires_at: datetime


def inactive_record_is_inert(record: dict[str, Any]) -> bool:
    return record == {
        "active": False,
        "policy_sha256": None,
        "collector_sha256": None,
        "run_id": None,
        "target_address": None,
        "target_port": None,
        "window": None,
        "exact_literal_path": None,
        "commands": [],
        "deadline_seconds": None,
        "expires_at": None,
    }


def _parse_expiry(value: Any, now: datetime) -> datetime:
    if not isinstance(value, str):
        raise PolicyError("expiry is missing")
    try:
        expiry = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise PolicyError("expiry format is invalid") from error
    if now.tzinfo is None or expiry.tzinfo is None:
        raise PolicyError("expiry must be timezone aware")
    if not now < expiry <= now + MAX_APPROVAL_LIFETIME:
        raise PolicyError("approval is expired or too long")
    return expiry


def _validate_target(value: Any) -> str:
    if not isinstance(value, str) or not re.fullmatch(
            r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?", value):
        raise PolicyError("target syntax is invalid")
    if ".." in value:
        raise PolicyError("target syntax is invalid")
    return value


def _validate_authority(record: dict[str, Any]) -> None:
    if not all(record.get(field) is True for field in REQUIRED_TRUE_FIELDS):
        raise PolicyError("required authority or effect acceptance is missing")
    if not all(record.get(field) is False for field in REQUIRED_FALSE_FIELDS):
        raise PolicyError("forbidden authority is active")


def build_session_plan(
    activation: dict[str, Any], approval: dict[str, Any], now: datetime,
) -> SessionPlan:
    """Validate dual synthetic records and return an immutable offline plan."""
    if set(activation) != ACTIVATION_FIELDS or set(approval) != APPROVAL_FIELDS:
        raise PolicyError("record shape is not exact")
    if activation.get("active") is not True or approval.get("active") is not True:
        raise PolicyError("activation is inactive")
    if approval.get("attested") is not True:
        raise PolicyError("operator attestation is missing")
    if approval.get("listener_already_running_attested") is not True:
        raise PolicyError("listener attestation is missing")
    _validate_authority(approval)
    matching_fields = (
        "policy_sha256", "collector_sha256", "run_id", "target_address", "target_port",
        "window", "exact_literal_path", "commands", "deadline_seconds",
        "expires_at",
    )
    if any(activation.get(field) != approval.get(field)
           for field in matching_fields):
        raise PolicyError("activation and approval do not match")
    if activation.get("collector_sha256") != COLLECTOR_SHA256:
        raise PolicyError("collector identity mismatch")
    if not isinstance(activation.get("policy_sha256"), str) or \
            not re.fullmatch(r"[0-9a-f]{64}", activation["policy_sha256"]):
        raise PolicyError("policy identity syntax is invalid")
    run_id = activation.get("run_id")
    if not isinstance(run_id, str) or not re.fullmatch(
            r"[A-Za-z0-9_-]{8,64}", run_id):
        raise PolicyError("run identifier is invalid")
    target = _validate_target(activation.get("target_address"))
    if activation.get("target_port") != SOURCE_BOUND_PORT:
        raise PolicyError("target port mismatch")
    window = activation.get("window")
    if window not in WINDOW_COMMANDS:
        raise PolicyError("window is not allowlisted")
    commands = activation.get("commands")
    if commands != list(WINDOW_COMMANDS[window]):
        raise PolicyError("command sequence mismatch")
    path = activation.get("exact_literal_path")
    if window == "T2_GREETING_AND_HELP" and path is not None:
        raise PolicyError("help window cannot include a path")
    if window == "T3_ONE_EXACT_PATH":
        if not isinstance(path, str):
            raise PolicyError("exact path is missing")
        try:
            OfflineCollector._validate_expected_paths({path})
        except CollectorModelError as error:
            raise PolicyError("exact path is invalid") from error
    deadline = activation.get("deadline_seconds")
    if not isinstance(deadline, int) or isinstance(deadline, bool) or \
            not 1 <= deadline <= MAX_DEADLINE_SECONDS:
        raise PolicyError("deadline is invalid")
    expiry = _parse_expiry(activation.get("expires_at"), now)
    return SessionPlan(
        run_id=run_id,
        target_address=target,
        target_port=SOURCE_BOUND_PORT,
        window=window,
        exact_literal_path=path,
        commands=WINDOW_COMMANDS[window],
        deadline_seconds=deadline,
        expires_at=expiry,
    )