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.mutations import ObservationChannel, build_lab

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 _bob_cannot_write(obs) -> bool:
    """A READ grant must not confer write.

    INTENT.md derives exactly this question from the reference use case:
    "Can Bob write when only read permission was granted?" It is part of what
    sharing means, not an extra.
    """
    return obs["probe_write:bob:R"] is False


def _revocation_is_audited(obs) -> bool:
    """Revocation must leave a trace.

    Enforcement being correct is not sufficient. An access change nobody can
    later evidence is a compliance failure even when the access itself is right.
    """
    return any(event["event"] == "revoke" for event in obs["audit:R"])


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-cannot-write", "A READ grant does not let Bob write R",
              Provenance.HUMAN, _bob_cannot_write, after_step="s2-grant",
              source_ref="INTENT.md#security-by-use-case-mutation"),
        Claim("c-bob-revoked", "Bob cannot read R after revocation",
              Provenance.HUMAN, _bob_cannot_read, after_step="s3-revoke"),
        Claim("c-revoke-audited", "Revocation is recorded in the audit trail",
              Provenance.HUMAN, _revocation_is_audited, 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(*mutations: str):
    """Assemble world, driver, observer and asset from a known initial state.

    `mutations` names entries from the lab catalogue. The same scenario runs
    unchanged against every lab version — that is the point: the use case does
    not know the implementation moved.
    """
    lab, tokens = build_lab(*mutations)
    variant = "+".join(mutations) if mutations else "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()