test-driver/usecases/audit_core_e2_tenant_boundary.py

"""Audit-core E2 tenant-boundary TestUseCase.

This captures the intent and role hand-offs exercised successfully by
WH-ENG-20260822-AUDIT-E2-03. It is deliberately not a runnable ``Scenario``
yet. The current kernel has no causal/time-window schedule, external custody
driver, Kubernetes driver, or independent cleanup observer capable of realizing
this use case without an attended operator.

The claims come from the approved engagement and AUDIT-WP-0008-T05, not from
reverse-engineering the successful responses. The -03 report is precedent
evidence and a calibration example; it is not the source of truth for verdicts.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Mapping

from testdriver import Claim, Invariant, Provenance, UseCase


SPEC_REF = (
    "audit-core/workplans/AUDIT-WP-0008-tenancy-posture-alignment.md"
    "#AUDIT-WP-0008-T05"
)
ENGAGEMENT_REF = (
    "whitehat-security/engagements/2026-08-22-audit-core-e2-03.json"
)
PRECEDENT_EVIDENCE_REF = (
    "audit-core/docs/evidence/"
    "AUDIT-WP-0008-T05-whitehat-e2-03-pass-2026-08-22.md"
)


@dataclass(frozen=True, slots=True)
class RoleContract:
    """One independently driven responsibility in the attended protocol."""

    id: str
    responsibility: str
    may_observe_secret_values: bool = False


@dataclass(frozen=True, slots=True)
class PhaseContract:
    """Causal intent for the future orchestrator; never a shell transcript."""

    id: str
    driver_role: str
    requires: tuple[str, ...]
    completion: str
    timing: str


ROLE_CONTRACTS = (
    RoleContract(
        "authorizer",
        "Approve the exact production package, target, scope, window and cleanup "
        "obligation; hold no test credential.",
    ),
    RoleContract(
        "target-owner",
        "Acknowledge the exact target revision, routes, fixtures and abort limits.",
    ),
    RoleContract(
        "credential-custodian",
        "Preflight, project and revoke two short-lived tenant-scoped identities; "
        "emit value-safe projection and cleanup receipts.",
        may_observe_secret_values=True,
    ),
    RoleContract(
        "security-coordinator",
        "Validate the engagement, admit only the receipt-bound plane lease, invoke "
        "the bounded probe, finalize the report and route it to risk-nexus.",
    ),
    RoleContract(
        "cluster-executor",
        "Create, wait for and delete only the approved runner pod; mount but never "
        "read the projected Secret through the control plane.",
    ),
    RoleContract(
        "tenant-a-attacker",
        "Use an ordinary tenant-A identity to attempt the declared cross-tenant "
        "reads and write.",
        may_observe_secret_values=True,
    ),
    RoleContract(
        "tenant-b-control",
        "Use an independently scoped tenant-B identity only for known-good owner "
        "controls and state-after checks.",
        may_observe_secret_values=True,
    ),
    RoleContract(
        "independent-observer",
        "Collect sanitized outcomes, target readiness and post-cleanup absence "
        "without accepting an actor's report as its own evidence.",
    ),
)


PHASE_CONTRACTS = (
    PhaseContract(
        "authorize",
        "authorizer",
        (),
        "Exact engagement and custody contracts are approved.",
        "Before projection and before the engagement window opens.",
    ),
    PhaseContract(
        "acknowledge-target",
        "target-owner",
        ("authorize",),
        "Target owner accepts the exact revision, routes and bounded fixtures.",
        "Before any temporary credential exists.",
    ),
    PhaseContract(
        "preflight",
        "credential-custodian",
        ("acknowledge-target",),
        "Target is ready and all engagement-specific resources are absent.",
        "Immediately before the projection cutoff.",
    ),
    PhaseContract(
        "project-custody",
        "credential-custodian",
        ("preflight",),
        "Exactly two scoped handles exist and a value-safe receipt is emitted.",
        "Inside the narrow projection window; expiry equals engagement end.",
    ),
    PhaseContract(
        "admit-plane",
        "security-coordinator",
        ("project-custody",),
        "Receipt, broker contract, target revision and cleanup authority agree.",
        "After projection and before any target packet.",
    ),
    PhaseContract(
        "ready-runner",
        "cluster-executor",
        ("admit-plane",),
        "The exact attributed runner is Ready with its Secret mounted.",
        "Inside the authorized engagement window.",
    ),
    PhaseContract(
        "run-probes",
        "security-coordinator",
        ("ready-runner",),
        "Calibrated owner, attacker and absent controls produce a sanitized report.",
        "Before credential and engagement expiry.",
    ),
    PhaseContract(
        "delete-runner",
        "cluster-executor",
        ("run-probes",),
        "The exact runner pod is absent.",
        "Immediately after evidence collection.",
    ),
    PhaseContract(
        "cleanup-custody",
        "credential-custodian",
        ("delete-runner",),
        "Both identities, exact KV paths, projection resources and Secret are absent.",
        "Before credential and engagement expiry.",
    ),
    PhaseContract(
        "finalize-and-deliver",
        "security-coordinator",
        ("cleanup-custody",),
        "The report is bound to projection and cleanup receipts and reaches risk-nexus.",
        "Only after cleanup is independently observable.",
    ),
)


def _same_absence_surface(left: Mapping[str, object], right: Mapping[str, object]) -> bool:
    """Compare only the sanitized non-disclosure surface."""

    return (
        left["status"] == right["status"]
        and left["schema"] == right["schema"]
        and left["digest"] == right["digest"]
        and left["fixture_match_count"] == 0
        and right["fixture_match_count"] == 0
    )


def _event_id_is_tenant_scoped(obs: Mapping[str, object]) -> bool:
    probe = obs["event_by_id"]
    owner = probe["owner"]
    attacker = probe["attacker"]
    absent = probe["absent"]
    return (
        owner["status"] == 200
        and owner["fixture_match_count"] > 0
        and _same_absence_surface(attacker, absent)
    )


def _correlation_slice_is_tenant_scoped(obs: Mapping[str, object]) -> bool:
    probe = obs["correlation_slice"]
    return (
        probe["owner"]["status"] == 200
        and probe["owner"]["fixture_match_count"] > 0
        and probe["attacker"]["status"] == 200
        and probe["attacker"]["fixture_match_count"] == 0
    )


def _cross_tenant_append_is_refused(obs: Mapping[str, object]) -> bool:
    probe = obs["append_as_b"]
    return (
        probe["attacker"]["status"] in {400, 403, 404}
        and _same_absence_surface(probe["state_after"], probe["absent"])
    )


def _cleanup_is_complete(obs: Mapping[str, object]) -> bool:
    cleanup = obs["cleanup"]
    return (
        cleanup["completed_before_expiry"] is True
        and cleanup["runner_absent"] is True
        and cleanup["temporary_identities"] == []
        and cleanup["kv_paths"] == []
        and cleanup["projection_resources"] == []
        and cleanup["mounted_secrets"] == []
        and cleanup["target_ready"] is True
    )


def _execution_remains_bounded(obs: Mapping[str, object]) -> bool:
    run = obs["run"]
    return (
        run["attempted_operations"] <= run["maximum_operations"]
        and run["max_concurrency"] == 1
        and run["rate_per_minute"] <= 10
        and run["target_revision_matches"] is True
        and run["attribution_complete"] is True
    )


def _secret_values_do_not_enter_evidence(obs: Mapping[str, object]) -> bool:
    return obs["cleanup"]["secret_values_observed"] is False


def _report_preserves_assurance_limit(obs: Mapping[str, object]) -> bool:
    report = obs["report"]
    return (
        report["bound_to_projection_receipt"] is True
        and report["bound_to_cleanup_receipt"] is True
        and report["bounded_assurance_statement"] is True
        and report["delivered_to_risk_nexus"] is True
    )


TEST_USE_CASE = UseCase(
    id="uc-audit-core-e2-tenant-boundary",
    title="Demonstrate audit-core's E2 tenant boundary with separate drivers",
    narrative=(
        "Within one explicitly authorized production window, independent target, "
        "custody, security, cluster, attacker, control and observer roles establish "
        "that an ordinary tenant-A identity cannot read or create tenant-B audit "
        "fixtures. The run must remain bounded, preserve non-disclosure semantics, "
        "clean every temporary resource before expiry, and deliver a cleanup-bound "
        "sanitized report without retaining credential values."
    ),
    provenance=Provenance.SPEC,
    source_ref=SPEC_REF,
    claims=(
        Claim(
            "c-event-id-tenant-scoped",
            "Tenant A cannot distinguish tenant B's event id from an absent event",
            Provenance.SPEC,
            _event_id_is_tenant_scoped,
            after_step="run-probes",
            source_ref=ENGAGEMENT_REF,
        ),
        Claim(
            "c-correlation-slice-tenant-scoped",
            "Tenant A's correlation slice contains no tenant-B fixture",
            Provenance.SPEC,
            _correlation_slice_is_tenant_scoped,
            after_step="run-probes",
            source_ref=ENGAGEMENT_REF,
        ),
        Claim(
            "c-cross-tenant-append-refused",
            "Tenant A cannot append an audit event attributed to tenant B",
            Provenance.SPEC,
            _cross_tenant_append_is_refused,
            after_step="run-probes",
            source_ref=ENGAGEMENT_REF,
        ),
        Claim(
            "c-receipt-bound-cleanup",
            "Every engagement-specific runtime and custody resource is absent "
            "before expiry while the target remains ready",
            Provenance.SPEC,
            _cleanup_is_complete,
            after_step="cleanup-custody",
            source_ref=SPEC_REF,
        ),
    ),
    invariants=(
        Invariant(
            "i-bounded-execution",
            "The run stays within its operation, rate, concurrency, revision and "
            "attribution boundaries",
            Provenance.SPEC,
            _execution_remains_bounded,
            source_ref=ENGAGEMENT_REF,
        ),
        Invariant(
            "i-no-secret-evidence",
            "Credential values never enter retained observations or reports",
            Provenance.SPEC,
            _secret_values_do_not_enter_evidence,
            source_ref=SPEC_REF,
        ),
        Invariant(
            "i-bounded-assurance-report",
            "Delivery is receipt-bound and says only that attempted attacks failed",
            Provenance.SPEC,
            _report_preserves_assurance_limit,
            source_ref=SPEC_REF,
        ),
    ),
)

# Alias retained for the existing scenario-module convention.
USE_CASE = TEST_USE_CASE