111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
|
"""Framework isolation and frozen schedule identity are acceptance boundaries."""
|
||
|
|
from copy import deepcopy
|
||
|
|
from dataclasses import replace
|
||
|
|
import json
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from scenarios.alice_bob_carol import build
|
||
|
|
from testdriver import Runner, Verdict
|
||
|
|
from testdriver.actions import SemanticAction
|
||
|
|
from testdriver.classification import classify
|
||
|
|
from testdriver.crystallization import CrystallizedDriver, Trajectory, assess_stability
|
||
|
|
from testdriver.drivers import CompositeDriver
|
||
|
|
from testdriver.scenario import Step
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('leak_at', [None, 0, 1, 2])
|
||
|
|
def test_isolation_breach_stops_run_and_cannot_be_accepted_or_frozen(leak_at):
|
||
|
|
world, driver, observer, asset, oracle = build()
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def leak():
|
||
|
|
world.cast['bob'].remember('overheard', world.cast['alice'].canary)
|
||
|
|
|
||
|
|
class LeakingDriver:
|
||
|
|
def realize(self, actor, action):
|
||
|
|
result = driver.realize(actor, action)
|
||
|
|
calls.append(action.name)
|
||
|
|
if len(calls) - 1 == leak_at:
|
||
|
|
leak()
|
||
|
|
return result
|
||
|
|
|
||
|
|
if leak_at is None:
|
||
|
|
leak()
|
||
|
|
result = Runner(world, LeakingDriver(), observer, oracle).run(asset)
|
||
|
|
assert result.verdict is Verdict.INCONCLUSIVE
|
||
|
|
assert len(calls) == (0 if leak_at is None else leak_at + 1)
|
||
|
|
assert any(j.verdict is Verdict.INCONCLUSIVE for j in result.judgments)
|
||
|
|
pack = json.loads(result.evidence.to_json())
|
||
|
|
assert any(o['data']['violations'] for o in pack['observations']
|
||
|
|
if o['kind'] == 'actor_isolation')
|
||
|
|
assert not classify(pack, pack).safe_to_accept
|
||
|
|
assert not assess_stability([pack, deepcopy(pack), deepcopy(pack)]).stable
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('damage', ['violation', 'missing', 'stratum', 'duplicate', 'malformed'])
|
||
|
|
def test_passing_product_verdicts_cannot_hide_invalid_isolation_evidence(damage):
|
||
|
|
packs = []
|
||
|
|
for _ in range(3):
|
||
|
|
world, driver, observer, asset, oracle = build()
|
||
|
|
pack = json.loads(Runner(world, driver, observer, oracle).run(asset).evidence.to_json())
|
||
|
|
guard = [o for o in pack['observations'] if o['kind'] == 'actor_isolation'][-1]
|
||
|
|
if damage == 'violation':
|
||
|
|
guard['data']['violations'] = ['actor leaked']
|
||
|
|
elif damage == 'missing':
|
||
|
|
pack['observations'].remove(guard)
|
||
|
|
elif damage == 'stratum':
|
||
|
|
guard['stratum'] = 'S1'
|
||
|
|
elif damage == 'duplicate':
|
||
|
|
pack['observations'].append(deepcopy(guard))
|
||
|
|
else:
|
||
|
|
guard['data'].pop('violations')
|
||
|
|
assert all(v['verdict'] == 'PASS' for v in pack['verdicts'])
|
||
|
|
assert not classify(pack, pack).safe_to_accept
|
||
|
|
packs.append(pack)
|
||
|
|
assert not assess_stability(packs).stable
|
||
|
|
|
||
|
|
|
||
|
|
def trajectories():
|
||
|
|
return [Trajectory('a', 'grant_access', 'browser', '/resources/A/grant', ('subject_id',)),
|
||
|
|
Trajectory('b', 'grant_access', 'browser', '/resources/B/grant', ('subject_id',))]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('composite', [False, True])
|
||
|
|
def test_runner_replays_repeated_actions_by_step_and_can_reuse_driver(composite):
|
||
|
|
world, _, observer, asset, oracle = build()
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
class Session:
|
||
|
|
def post_form(self, path, fields):
|
||
|
|
calls.append((path, fields))
|
||
|
|
return 200, ''
|
||
|
|
|
||
|
|
driver = CrystallizedDriver(trajectories(), lambda actor: Session())
|
||
|
|
if composite:
|
||
|
|
driver = CompositeDriver({'browser': driver}, 'browser')
|
||
|
|
asset.scenario = replace(asset.scenario, steps=tuple(
|
||
|
|
Step(step_id, 'alice', SemanticAction('grant_access', {'subject_id': subject},
|
||
|
|
frozenset({'browser'})))
|
||
|
|
for step_id, subject in [('a', 'bob'), ('b', 'carol')]
|
||
|
|
))
|
||
|
|
for _ in range(2):
|
||
|
|
Runner(world, driver, observer, oracle).run(asset)
|
||
|
|
assert calls == [('/resources/A/grant', {'subject_id': 'bob'}),
|
||
|
|
('/resources/B/grant', {'subject_id': 'carol'})] * 2
|
||
|
|
|
||
|
|
|
||
|
|
def test_ambiguous_unknown_and_mismatched_frozen_actions_do_not_send_requests():
|
||
|
|
world, _, _, _, _ = build()
|
||
|
|
|
||
|
|
def unexpected_session(actor):
|
||
|
|
pytest.fail('invalid frozen dispatch must not open a session')
|
||
|
|
|
||
|
|
driver = CrystallizedDriver(trajectories(), unexpected_session)
|
||
|
|
action = SemanticAction('grant_access', {}, frozenset({'browser'}))
|
||
|
|
assert driver.realize(world.cast['alice'], action).raised
|
||
|
|
assert driver.realize_step(world.cast['alice'], 'unknown', action).raised
|
||
|
|
assert driver.realize_step(world.cast['alice'], 'a', replace(action, name='revoke_access')).raised
|
||
|
|
with pytest.raises(ValueError, match='duplicate'):
|
||
|
|
CrystallizedDriver([trajectories()[0]] * 2, unexpected_session)
|