NK-WP-0048-T27W: wrap bind-env as a full Warden Adapter
Source-digest, descriptor, case and acceptance pins; owner-published and authenticated evidence; parent-session revocation in finally. Existing reference-only refusals and created-only rollback kept. Synthetic tests only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 903097@bnt-lap001 Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23
This commit is contained in:
parent
ece7b0a2a8
commit
7442d212ec
3 changed files with 282 additions and 68 deletions
|
|
@ -14,13 +14,27 @@ bound to anything else (including a literal), is refused.
|
|||
and `allow_unimplemented` are refused.
|
||||
- `read`/`write` are injected capabilities; `write` is compare-and-swap on the
|
||||
observed `resourceVersion`. No kubectl, credential or secret value is used.
|
||||
- `build(...)` returns a closed Warden v1 `Adapter` (T13 pattern). It pins the
|
||||
owner source digest over `bind-env-adapter.py` + `keycape-client.py`, the
|
||||
descriptor digest (`Pins.adapter`), case source and acceptance digests, and the
|
||||
admitted invocation (step `bind-env`, actor `human`, lane `key-cape-oidc-login`,
|
||||
created object `keycape-env-binding`, not retry-safe, `parent-session` scope).
|
||||
Every attempt is rebound to that invocation and re-checks source drift.
|
||||
- `publish`, `authenticate` and `revoke_session` are mandatory injected
|
||||
capabilities. The parent session is revoked in `finally`, also on refusal; a
|
||||
missing revocation or invalid/reused evidence identifier is never success.
|
||||
Evidence payload is closed metadata (`netkingdom.keycape-bind-env.attempt.v1`:
|
||||
invocation, source, outcome, env name, before/after revision) with no values.
|
||||
- An identical existing binding is `no-change` and is not rollbackable.
|
||||
- Rollback removes only the entry this process created, and only while the
|
||||
- Rollback removes only the entry this process created, whose evidence still
|
||||
authenticates (a fresh process has no creation authority), and only while the
|
||||
Deployment is still at the revision that creation produced; otherwise owner
|
||||
reconciliation is required.
|
||||
|
||||
Tests: `python3 -m pytest -q test_bind_env_adapter.py` (synthetic only).
|
||||
Tests: `PYTHONPATH=~/ops-warden/src ~/ops-warden/.venv/bin/python -m pytest -q
|
||||
test_bind_env_adapter.py` (synthetic only).
|
||||
|
||||
Not supplied: Warden `Adapter`/descriptor wrapping and evidence publication (see
|
||||
T13 `keycape-registration-adapter.py`), NetKingdom runtime owner source review
|
||||
before pinning, and any native action. Those remain separate gates.
|
||||
Not supplied: kernel-level admission/dispatch tests with fixture trust (as in T13),
|
||||
owner confirmation of the proposed step/lane/schema/object names, NetKingdom
|
||||
runtime owner source review before pinning, the real owner evidence and
|
||||
revocation capabilities, and any native action. Those remain separate gates.
|
||||
|
|
|
|||
|
|
@ -1,29 +1,36 @@
|
|||
"""Unexercised source candidate: bind one issuer env reference, never a value (NK-WP-0048-T27).
|
||||
"""Unexercised, injection-only bind-env seam for Warden's v1 kernel (NK-WP-0048-T27).
|
||||
|
||||
Given an admitted oidc-browser-relying-party case (canonical 1.1 by default via the
|
||||
maintained helper), add exactly one `env` entry to the `keycape` container of the
|
||||
issuer Deployment: name from the helper, valueFrom.secretKeyRef to the delivered
|
||||
secret *reference*. No CLI, kubectl, credential, or value handling lives here;
|
||||
read/write are injected capabilities with compare-and-swap on resourceVersion.
|
||||
Only an entry this process created, still byte-identical, may be rolled back.
|
||||
Only an entry this process created, authenticated and still byte-identical, may
|
||||
be rolled back. Owner publication/authentication and parent-session revocation
|
||||
are mandatory injected capabilities; no evidence trust root is installed here.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from dataclasses import asdict, replace
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER
|
||||
from warden.case_receipt_state import Outcome, Refused
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
REFERENCE = 'net-kingdom/keycape-bind-env@v1'
|
||||
SOURCE_FILES = ('bind-env-adapter.py', 'keycape-client.py')
|
||||
CREATED = 'keycape-env-binding'
|
||||
_spec = importlib.util.spec_from_file_location('bind_env_owner_helper', ROOT / 'keycape-client.py')
|
||||
helper = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(helper)
|
||||
|
||||
|
||||
class Refused(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _container(deployment):
|
||||
if (deployment.get('kind') != 'Deployment'
|
||||
or deployment.get('metadata', {}).get('name') != 'keycape'
|
||||
|
|
@ -66,45 +73,145 @@ def plan(deployment, binding):
|
|||
return new, True
|
||||
|
||||
|
||||
def build(*, case, expected_resource_version, read, write, allow_unimplemented=False):
|
||||
"""read()->deployment dict; write(deployment, expected_rv)->new rv (CAS, raises on conflict)."""
|
||||
binding = binding_for_case(case, allow_unimplemented=allow_unimplemented)
|
||||
if not (callable(read) and callable(write)):
|
||||
raise Refused('read/write capabilities required')
|
||||
if not isinstance(expected_resource_version, str) or not expected_resource_version:
|
||||
raise Refused('observed resourceVersion required')
|
||||
state = {'created_rv': None}
|
||||
def source_digest():
|
||||
"""Pin adapter plus its local helper import closure, in filename order."""
|
||||
composition = {'files': {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES}}
|
||||
return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode())
|
||||
|
||||
def run():
|
||||
|
||||
def descriptor():
|
||||
return Descriptor('warden.case.execution.v1', 'bind-env', 'net-kingdom', REFERENCE,
|
||||
source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change',
|
||||
(CREATED,), False, ('parent-session',))
|
||||
|
||||
|
||||
def build(*, admitted: Invocation, case_source: bytes, expected_resource_version: str,
|
||||
case_acceptance: str, read, write, publish, authenticate, revoke_session,
|
||||
allow_unimplemented=False):
|
||||
"""Bind one admitted case and observed revision; return a closed kernel Adapter.
|
||||
|
||||
read()->deployment dict; write(deployment, expected_rv)->new rv (CAS, raises
|
||||
on conflict). publish(bytes)->identifier retains only the closed metadata
|
||||
payload; authenticate(identifier, bytes)->True verifies owner custody;
|
||||
revoke_session(invocation)->identifier revokes this attempt's parent session.
|
||||
Process loss, readback ambiguity and publication/revocation failure require
|
||||
owner reconciliation; none are converted to retry-safe failures.
|
||||
"""
|
||||
reviewed_source = source_digest()
|
||||
desc = descriptor()
|
||||
if type(admitted) is not Invocation or not all(
|
||||
isinstance(value, str) and IDENTIFIER.fullmatch(value)
|
||||
for value in (admitted.step, admitted.owner, admitted.target,
|
||||
admitted.actor, admitted.principal, admitted.lane)):
|
||||
raise Refused('bounded invocation required')
|
||||
if (admitted.step, admitted.owner, admitted.actor, admitted.lane) != (
|
||||
desc.step, desc.owner, desc.actor, desc.lane):
|
||||
raise Refused('bind-env invocation boundary differs')
|
||||
if admitted.operation != 'admission' or admitted.attempt_id:
|
||||
raise Refused('expected admission invocation')
|
||||
if type(case_source) is not bytes or len(case_source) > 262144 or digest(case_source) != admitted.pins.case:
|
||||
raise Refused('case source pin differs')
|
||||
# Pins.adapter is the descriptor digest, not the owner source digest.
|
||||
descriptor_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode()
|
||||
if admitted.pins.adapter != digest(descriptor_bytes):
|
||||
raise Refused('descriptor pin differs')
|
||||
if not all(isinstance(x, str) and 0 < len(x) <= 192
|
||||
for x in (expected_resource_version, case_acceptance)):
|
||||
raise Refused('observed resourceVersion and acceptance required')
|
||||
if digest(case_acceptance.encode()) != admitted.pins.acceptance:
|
||||
raise Refused('acceptance evidence binding differs')
|
||||
if not all(callable(x) for x in (read, write, publish, authenticate, revoke_session)):
|
||||
raise Refused('read/write, owner evidence and revocation capabilities required')
|
||||
with contained_output():
|
||||
case = yaml.safe_load(case_source)
|
||||
client = helper.client_from_case(case)
|
||||
binding = binding_for_case(case, allow_unimplemented=allow_unimplemented)
|
||||
if admitted.target != client['clientId']:
|
||||
raise Refused('case target differs')
|
||||
records = {}
|
||||
creation = None
|
||||
|
||||
def bound(inv, operation):
|
||||
if (type(inv) is not Invocation
|
||||
or replace(inv, attempt_id='', operation='admission') != admitted
|
||||
or inv.operation != operation or not IDENTIFIER.fullmatch(inv.attempt_id)):
|
||||
raise Refused('attempt binding differs')
|
||||
if source_digest() != reviewed_source:
|
||||
raise Refused('owner source changed')
|
||||
|
||||
def add():
|
||||
current = read()
|
||||
rv = current['metadata'].get('resourceVersion')
|
||||
if rv != expected_resource_version:
|
||||
before = current['metadata'].get('resourceVersion')
|
||||
if before != expected_resource_version:
|
||||
raise Refused('deployment revision differs from observed')
|
||||
new, created = plan(current, binding)
|
||||
if not created:
|
||||
return {'outcome': 'no-change', 'created': (), 'values_emitted': False}
|
||||
after = write(new, rv)
|
||||
return Outcome.NO_CHANGE, (), before, before
|
||||
after = write(new, before)
|
||||
back = read()
|
||||
if back['metadata'].get('resourceVersion') != after or plan(back, binding)[1]:
|
||||
if (not isinstance(after, str) or not 0 < len(after) <= 192
|
||||
or back['metadata'].get('resourceVersion') != after or plan(back, binding)[1]):
|
||||
raise Refused('readback ambiguous; owner reconciliation required')
|
||||
state['created_rv'] = after
|
||||
return {'outcome': 'success', 'created': (binding['name'],), 'values_emitted': False}
|
||||
return Outcome.SUCCESS, (CREATED,), before, after
|
||||
|
||||
def rollback():
|
||||
if state['created_rv'] is None:
|
||||
raise Refused('no case-created binding to roll back')
|
||||
def remove(created):
|
||||
if created != (CREATED,) or creation is None or not verify(*creation):
|
||||
raise Refused('no authenticated case-created binding')
|
||||
created_rv = json.loads(records[creation[1].evidence][1])['after']
|
||||
current = read()
|
||||
rv = current['metadata'].get('resourceVersion')
|
||||
if rv != state['created_rv']:
|
||||
before = current['metadata'].get('resourceVersion')
|
||||
if before != created_rv:
|
||||
raise Refused('deployment changed since creation; owner reconciliation required')
|
||||
new = copy.deepcopy(current)
|
||||
env = _container(new)['env']
|
||||
env = _container(new).get('env', [])
|
||||
matches = [i for i, e in enumerate(env) if e == binding]
|
||||
if len(matches) != 1:
|
||||
if len(matches) != 1 or [e for e in env if e.get('name') == binding['name']] != [binding]:
|
||||
raise Refused('created binding not found unchanged')
|
||||
del env[matches[0]]
|
||||
write(new, rv)
|
||||
state['created_rv'] = None
|
||||
return {'outcome': 'rolled-back', 'created': (), 'values_emitted': False}
|
||||
after = write(new, before)
|
||||
back = read()
|
||||
if (not isinstance(after, str) or back['metadata'].get('resourceVersion') != after
|
||||
or any(e.get('name') == binding['name'] for e in _container(back).get('env', []))):
|
||||
raise Refused('rollback readback ambiguous; owner reconciliation required')
|
||||
return Outcome.ROLLED_BACK, (), before, after
|
||||
|
||||
return REFERENCE, run, rollback
|
||||
def execute(inv, operation, created=()):
|
||||
nonlocal creation
|
||||
try:
|
||||
with contained_output():
|
||||
bound(inv, 'run' if operation == 'add' else 'rollback')
|
||||
outcome, objects, before, after = add() if operation == 'add' else remove(created)
|
||||
finally:
|
||||
# Even failure must end the parent session; failures propagate to the
|
||||
# kernel as uncertain, never safe to replay.
|
||||
with contained_output():
|
||||
revocation = revoke_session(inv)
|
||||
if not isinstance(revocation, str) or not IDENTIFIER.fullmatch(revocation):
|
||||
raise Refused('session revocation evidence missing')
|
||||
payload = json.dumps({'schema': 'netkingdom.keycape-bind-env.attempt.v1',
|
||||
'invocation': asdict(inv), 'source': reviewed_source, 'outcome': outcome.value,
|
||||
'created': objects, 'env_name': binding['name'], 'before': before, 'after': after,
|
||||
'values_emitted': False, 'session_revocation': revocation},
|
||||
sort_keys=True, separators=(',', ':')).encode()
|
||||
with contained_output():
|
||||
evidence = publish(payload)
|
||||
if not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or evidence in records:
|
||||
raise Refused('owner evidence identifier invalid or reused')
|
||||
result = OwnerResult(outcome, evidence, objects, True, revocation,
|
||||
attempt_id=inv.attempt_id, operation=inv.operation)
|
||||
records[evidence] = (inv, payload, result)
|
||||
if objects:
|
||||
creation = (inv, result)
|
||||
elif operation == 'remove':
|
||||
creation = None
|
||||
return result
|
||||
|
||||
def verify(inv, result):
|
||||
record = records.get(result.evidence) if type(result) is OwnerResult else None
|
||||
if record is None or record[0] != inv or record[2] != result:
|
||||
return False
|
||||
with contained_output():
|
||||
return authenticate(result.evidence, record[1]) is True
|
||||
|
||||
return Adapter(REFERENCE, reviewed_source, lambda inv: execute(inv, 'add'),
|
||||
verify, lambda inv, created: execute(inv, 'remove', created))
|
||||
|
|
|
|||
|
|
@ -1,8 +1,15 @@
|
|||
"""Synthetic-only tests for the T27 bind-env candidate; no cluster, no values."""
|
||||
import copy
|
||||
from dataclasses import asdict, replace
|
||||
import json
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from warden.case_run import Invocation, OwnerResult, digest
|
||||
from warden.case_receipt_state import Outcome, Pins, Refused
|
||||
import yaml
|
||||
|
||||
ROOT = Path(__file__).parent
|
||||
spec = importlib.util.spec_from_file_location('bind_env_adapter', ROOT / 'bind-env-adapter.py')
|
||||
|
|
@ -38,65 +45,151 @@ class Store:
|
|||
|
||||
|
||||
class BindEnvTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.case = yaml.safe_dump(CASE).encode()
|
||||
desc_bytes = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode()
|
||||
pins = Pins(digest(self.case), digest(b'type'), digest(desc_bytes),
|
||||
digest(b'catalog'), digest(b'case-acceptance:test'), digest(b'run sheet'))
|
||||
self.admitted = Invocation('bind-env', 'net-kingdom', 'forgejo', 'human',
|
||||
'founder', 'key-cape-oidc-login', pins)
|
||||
self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run')
|
||||
self.undo = replace(self.admitted, attempt_id='attempt-two', operation='rollback')
|
||||
self.evidence, self.revocations = {}, []
|
||||
|
||||
def publish(self, payload):
|
||||
self.assertNotIn(b'synthetic', payload)
|
||||
ref = 'owner-evidence:attempt-' + str(len(self.evidence))
|
||||
self.evidence[ref] = payload
|
||||
return ref
|
||||
|
||||
def revoke(self, inv):
|
||||
self.revocations.append(inv)
|
||||
return 'revocation:' + inv.attempt_id
|
||||
|
||||
def adapter(self, store, **kw):
|
||||
args = dict(admitted=self.admitted, case_source=self.case, expected_resource_version='1',
|
||||
case_acceptance='case-acceptance:test', read=store.read, write=store.write,
|
||||
publish=self.publish, authenticate=lambda r, p: self.evidence.get(r) == p,
|
||||
revoke_session=self.revoke)
|
||||
args.update(kw)
|
||||
return a.build(**args)
|
||||
|
||||
def env(self, store):
|
||||
return store.d['spec']['template']['spec']['containers'][0]['env']
|
||||
|
||||
def test_binding_is_reference_only(self):
|
||||
b = a.binding_for_case(CASE)
|
||||
self.assertEqual(b['name'], 'KEYCAPE_FORGEJO_CLIENT_SECRET')
|
||||
self.assertEqual(b['valueFrom']['secretKeyRef'], {'name': 'keycape-forgejo-client', 'key': 'client-secret'})
|
||||
self.assertNotIn('value', b)
|
||||
|
||||
def test_run_then_created_only_rollback(self):
|
||||
def test_run_verify_then_created_only_rollback(self):
|
||||
s = Store(deployment())
|
||||
ref, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
||||
self.assertEqual(ref, a.REFERENCE)
|
||||
r = run()
|
||||
self.assertEqual((r['outcome'], r['created'], r['values_emitted']), ('success', ('KEYCAPE_FORGEJO_CLIENT_SECRET',), False))
|
||||
self.assertEqual(len(s.d['spec']['template']['spec']['containers'][0]['env']), 2)
|
||||
self.assertEqual(rollback()['outcome'], 'rolled-back')
|
||||
self.assertEqual(s.d['spec']['template']['spec']['containers'][0]['env'], deployment()['spec']['template']['spec']['containers'][0]['env'])
|
||||
with self.assertRaises(a.Refused):
|
||||
rollback()
|
||||
ad = self.adapter(s)
|
||||
self.assertEqual((ad.reference, ad.source), (a.REFERENCE, a.source_digest()))
|
||||
r = ad.invoke(self.inv)
|
||||
self.assertEqual((r.outcome, r.created, r.session_revocation_evidence), (Outcome.SUCCESS, (a.CREATED,), 'revocation:attempt-one'))
|
||||
self.assertTrue(ad.verify(self.inv, r))
|
||||
payload = json.loads(self.evidence[r.evidence])
|
||||
self.assertEqual((payload['before'], payload['after'], payload['values_emitted']), ('1', '2', False))
|
||||
self.assertEqual(len(self.env(s)), 2)
|
||||
u = ad.rollback(self.undo, (a.CREATED,))
|
||||
self.assertEqual(u.outcome, Outcome.ROLLED_BACK)
|
||||
self.assertEqual(self.env(s), deployment()['spec']['template']['spec']['containers'][0]['env'])
|
||||
self.assertEqual(self.revocations, [self.inv, self.undo])
|
||||
with self.assertRaises(Refused):
|
||||
ad.rollback(replace(self.undo, attempt_id='attempt-three'), (a.CREATED,))
|
||||
self.assertEqual(len(self.revocations), 3)
|
||||
|
||||
def test_existing_identical_binding_is_no_change_and_not_rollbackable(self):
|
||||
s = Store(deployment([a.binding_for_case(CASE)]))
|
||||
_, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
||||
self.assertEqual(run()['outcome'], 'no-change')
|
||||
ad = self.adapter(s)
|
||||
self.assertEqual(ad.invoke(self.inv).outcome, Outcome.NO_CHANGE)
|
||||
self.assertEqual(s.writes, [])
|
||||
with self.assertRaises(a.Refused):
|
||||
rollback()
|
||||
with self.assertRaises(Refused):
|
||||
ad.rollback(self.undo, (a.CREATED,))
|
||||
|
||||
def test_refusals(self):
|
||||
def test_refusals_still_revoke_session(self):
|
||||
other = dict(a.binding_for_case(CASE), valueFrom={'secretKeyRef': {'name': 'x', 'key': 'client-secret'}})
|
||||
literal = {'name': 'KEYCAPE_FORGEJO_CLIENT_SECRET', 'value': 'synthetic'}
|
||||
for env in ([other], [literal]):
|
||||
s = Store(deployment(env))
|
||||
_, run, _ = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
||||
with self.assertRaises(a.Refused):
|
||||
run()
|
||||
with self.assertRaises(Refused):
|
||||
self.adapter(s).invoke(self.inv)
|
||||
self.assertEqual(s.writes, [])
|
||||
s = Store(deployment())
|
||||
_, run, _ = a.build(case=CASE, expected_resource_version='0', read=s.read, write=s.write)
|
||||
with self.assertRaises(a.Refused):
|
||||
run()
|
||||
with self.assertRaises(a.Refused):
|
||||
a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write, allow_unimplemented=True)
|
||||
with self.assertRaises(Refused):
|
||||
self.adapter(s, expected_resource_version='0').invoke(self.inv)
|
||||
self.assertEqual((s.writes, len(self.revocations), self.evidence), ([], 3, {}))
|
||||
public = {'spec': {'type': 'oidc-browser-relying-party@1.1',
|
||||
'parameters': dict(f.FORGEJO_PARAMS, client_type='public', mfa_required='false')}}
|
||||
with self.assertRaises(a.Refused):
|
||||
with self.assertRaises(Refused):
|
||||
a.binding_for_case(public)
|
||||
with self.assertRaises(Exception):
|
||||
a.binding_for_case({'spec': {'type': 'other@1', 'parameters': f.FORGEJO_PARAMS}})
|
||||
with self.assertRaises(a.Refused):
|
||||
with self.assertRaises(Refused):
|
||||
a.plan(dict(deployment(), metadata={'name': 'keycape', 'namespace': 'other'}), a.binding_for_case(CASE))
|
||||
with self.assertRaises(a.Refused):
|
||||
with self.assertRaises(Refused):
|
||||
a.plan(deployment(), dict(a.binding_for_case(CASE), value='synthetic'))
|
||||
|
||||
def test_build_pins_and_capabilities_required(self):
|
||||
s = Store(deployment())
|
||||
pins = self.admitted.pins
|
||||
for kw in (dict(allow_unimplemented=True),
|
||||
dict(case_source=self.case + b' '),
|
||||
dict(case_acceptance='case-acceptance:other'),
|
||||
dict(admitted=replace(self.admitted, pins=replace(pins, adapter=digest(b'x')))),
|
||||
dict(admitted=replace(self.admitted, step='register-client')),
|
||||
dict(admitted=replace(self.admitted, target='other')),
|
||||
dict(admitted=self.inv),
|
||||
dict(publish=None), dict(revoke_session='cmd'), dict(write=None)):
|
||||
with self.assertRaises(Exception, msg=kw):
|
||||
self.adapter(s, **kw)
|
||||
|
||||
def test_rebinding_and_source_drift_refused_before_write(self):
|
||||
s = Store(deployment())
|
||||
ad = self.adapter(s)
|
||||
with self.assertRaises(Refused):
|
||||
ad.invoke(replace(self.inv, principal='someone-else'))
|
||||
with patch.object(a, 'source_digest', return_value=digest(b'drift')):
|
||||
with self.assertRaises(Refused):
|
||||
ad.invoke(self.inv)
|
||||
self.assertEqual(s.writes, [])
|
||||
self.assertEqual(len(self.revocations), 2)
|
||||
|
||||
def test_tampered_or_unauthenticated_evidence_never_authorizes_undo(self):
|
||||
s = Store(deployment())
|
||||
ad = self.adapter(s)
|
||||
r = ad.invoke(self.inv)
|
||||
self.assertFalse(ad.verify(self.undo, r))
|
||||
self.assertFalse(ad.verify(self.inv, replace(r, evidence='owner-evidence:forged')))
|
||||
self.evidence[r.evidence] = b'tampered'
|
||||
self.assertFalse(ad.verify(self.inv, r))
|
||||
with self.assertRaises(Refused):
|
||||
ad.rollback(self.undo, (a.CREATED,))
|
||||
self.assertEqual(len(self.env(s)), 2)
|
||||
|
||||
def test_process_loss_does_not_reconstruct_created_authority(self):
|
||||
s = Store(deployment())
|
||||
self.adapter(s).invoke(self.inv)
|
||||
with self.assertRaises(Refused):
|
||||
self.adapter(s, expected_resource_version='2').rollback(self.undo, (a.CREATED,))
|
||||
self.assertEqual(len(self.env(s)), 2)
|
||||
|
||||
def test_revocation_or_publication_failure_is_never_success(self):
|
||||
for kw in (dict(revoke_session=lambda inv: None), dict(publish=lambda p: 'bad id!')):
|
||||
s = Store(deployment())
|
||||
with self.assertRaises(Refused):
|
||||
self.adapter(s, **kw).invoke(self.inv)
|
||||
|
||||
def test_rollback_refused_after_drift(self):
|
||||
s = Store(deployment())
|
||||
_, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
||||
run()
|
||||
ad = self.adapter(s)
|
||||
ad.invoke(self.inv)
|
||||
s.d['metadata']['resourceVersion'] = '9'
|
||||
with self.assertRaises(a.Refused):
|
||||
rollback()
|
||||
with self.assertRaises(Refused):
|
||||
ad.rollback(self.undo, (a.CREATED,))
|
||||
self.assertEqual(len(self.env(s)), 2)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue