test-driver/scenarios/alice_bob_carol.py

"""The reference scenario.

    Alice owns resource R. Alice grants Bob READ access. Bob can read R.
    Carol cannot read R. Alice revokes Bob's access. Bob can no longer read R.

Written as Python rather than YAML on purpose: claims are predicates over
observations, and a YAML dialect capable of expressing them would be a
programming language with worse tooling. A declarative surface is a later
question, once we know which predicates actually recur.

Every claim here is `Provenance.HUMAN` — authored from the use case in INTENT.md,
not derived from watching the lab behave. See D-06.
"""

from __future__ import annotations

from testdriver import (
    Actor, Cast, Claim, DirectDriver, Invariant, Oracle, Provenance,
    Scenario, SemanticAction, StateObserver, Step, UseCase,
    VerificationAsset, Watch, World,
)
from lab.minimal import ObservationChannel, build_baseline

RESOURCE = "R"
API = frozenset({"api"})

# --- claims ---------------------------------------------------------------
# Each reads the independent observation snapshot. None consults an actor.


def _bob_can_read(obs) -> bool:
    return obs["probe_read:bob:R"] is True


def _bob_cannot_read(obs) -> bool:
    return obs["probe_read:bob:R"] is False


def _carol_cannot_read(obs) -> bool:
    return obs["probe_read:carol:R"] is False


def _enforcement_matches_record(obs) -> bool:
    """Enforcement and stored record must agree about every watched subject.

    This invariant is what catches an authorization defect that leaves the audit
    trail looking correct: the grant is recorded as revoked, yet the enforcement
    path still allows the read. Neither observation alone would notice.
    """
    for key, permitted in obs.items():
        if not key.startswith("probe_read:"):
            continue
        recorded = obs.get("state_permission:" + key.removeprefix("probe_read:"))
        if permitted != (recorded is not None):
            return False
    return True


def _audit_is_append_only(obs) -> bool:
    events = obs["audit:R"]
    sequences = [e["sequence"] for e in events]
    return sequences == sorted(sequences)


USE_CASE = UseCase(
    id="uc-share-resource",
    title="Share a resource and revoke the share",
    narrative=(
        "Alice owns resource R. Alice grants Bob READ access. Bob can read R. "
        "Carol cannot read R. Alice revokes Bob's access. "
        "Bob can no longer read R."
    ),
    provenance=Provenance.HUMAN,
    source_ref="INTENT.md#first-reference-scenario",
    claims=(
        Claim("c-bob-reads", "Bob can read R after the grant",
              Provenance.HUMAN, _bob_can_read, after_step="s2-grant"),
        Claim("c-carol-denied", "Carol can never read R",
              Provenance.HUMAN, _carol_cannot_read, after_step="s2-grant"),
        Claim("c-bob-revoked", "Bob cannot read R after revocation",
              Provenance.HUMAN, _bob_cannot_read, after_step="s3-revoke"),
    ),
    invariants=(
        Invariant("i-enforcement-matches-record",
                  "Enforcement and the stored record agree for every subject",
                  Provenance.HUMAN, _enforcement_matches_record),
        Invariant("i-audit-append-only", "The audit trail is append-only",
                  Provenance.HUMAN, _audit_is_append_only),
    ),
)


def build(variant: str = "baseline"):
    """Assemble world, driver, observer and asset from a known initial state."""
    lab, tokens = build_baseline()
    cast = Cast()
    for name in ("alice", "bob", "carol"):
        cast.add(Actor(id=name, display_name=name.title(),
                       credentials={"token": tokens[name]}))

    world = World(id="w-baseline", sut=lab, sut_version=lab.version,
                  seed={"users": ["alice", "bob", "carol"], "resource": RESOURCE},
                  cast=cast)

    scenario = Scenario(
        id="sc-share-resource",
        use_case=USE_CASE,
        variant=variant,
        watches=(Watch("bob", RESOURCE), Watch("carol", RESOURCE)),
        steps=(
            Step("s1-create", "alice", SemanticAction(
                "create_resource",
                {"resource_id": RESOURCE, "content": "the secret"},
                permitted_surfaces=API,
                postcondition=lambda obs: "audit:R" in obs,
            )),
            Step("s2-grant", "alice", SemanticAction(
                "grant_access",
                {"resource_id": RESOURCE, "subject_id": "bob", "permission": "READ"},
                permitted_surfaces=API,
                postcondition=lambda obs: obs["state_permission:bob:R"] == "READ",
            )),
            Step("s3-revoke", "alice", SemanticAction(
                "revoke_access",
                {"resource_id": RESOURCE, "subject_id": "bob"},
                permitted_surfaces=API,
                postcondition=lambda obs: obs["state_permission:bob:R"] is None,
            )),
        ),
    )

    driver = DirectDriver(lab, tokens)
    observer = StateObserver(ObservationChannel(lab), scenario.watches)
    asset = VerificationAsset(id="va-share-resource", scenario=scenario, maturity="T5")
    return world, driver, observer, asset, Oracle()