NK-WP-0058-T03 step 1: name the causes (window-command identity phase, approval_entry field)
executor-lanes.py: the identity phase distinguishes openbao_unreachable, openbao_sealed and platform_admin_login_required, with a one-line hint; LaneError keeps its fixed code. executor.py: approval_entry:<approver_not_listed|not_human|not_aal2|self_approval>, never the subject. Note: executor.py changed, so the next admission pins a new image. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 943911@bnt-lap001 Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0
This commit is contained in:
parent
f85a22d245
commit
7350ba8003
7 changed files with 103 additions and 16 deletions
|
|
@ -372,7 +372,7 @@
|
|||
| task | NK-WP-0057-T04 | todo | — | workplans/NK-WP-0057-decision-routing-by-responsibility.md |
|
||||
| task | NK-WP-0058-T01 | done | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T02 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T03 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T03 | progress | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T04 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T05 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
| task | NK-WP-0058-T06 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
|
||||
|
|
|
|||
|
|
@ -122,6 +122,19 @@ def entry_is_aal2(assurance) -> bool:
|
|||
and assurance.get("mfa") is True)
|
||||
|
||||
|
||||
def entry_failure(entry: dict, allowed: set, executor_identity: str) -> str | None:
|
||||
"""The first check an approval entry fails, as a fixed name, or None. Never the subject itself."""
|
||||
if entry.get("subject_id") not in allowed:
|
||||
return "approver_not_listed"
|
||||
if str(entry.get("principal_type", "")).lower() != "human":
|
||||
return "not_human"
|
||||
if not entry_is_aal2(entry.get("assurance")):
|
||||
return "not_aal2"
|
||||
if entry.get("subject_id") == executor_identity:
|
||||
return "self_approval"
|
||||
return None
|
||||
|
||||
|
||||
def iso(t: dt.datetime) -> str:
|
||||
return t.astimezone(dt.UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
|
||||
|
||||
|
|
@ -192,10 +205,9 @@ class Run:
|
|||
if not entries:
|
||||
raise Refused("approval_no_entries")
|
||||
for entry in entries: # aal2 is recorded, not enforced, by approval-engine today: check every entry
|
||||
if (entry.get("subject_id") not in allowed or str(entry.get("principal_type", "")).lower() != "human"
|
||||
or not entry_is_aal2(entry.get("assurance"))
|
||||
or entry.get("subject_id") == self.cfg.executor_identity):
|
||||
raise Refused("approval_entry")
|
||||
failed = entry_failure(entry, allowed, self.cfg.executor_identity)
|
||||
if failed: # names the first failed field, value-free (NK-WP-0058-T03; 2026-11-04 needed a diagnostic pod)
|
||||
raise Refused("approval_entry:" + failed)
|
||||
self.approver_subject = entries[0]["subject_id"]
|
||||
window_end = bea.parse_time(spec["window"]["not_after"])
|
||||
expires = (approval.get("validity") or {}).get("expires_at")
|
||||
|
|
|
|||
|
|
@ -322,7 +322,7 @@ def exercise(args) -> dict:
|
|||
engine.add_entry(aid, APPROVER_SUB, assurance=keycape_assurance("aal1"), principal_type="human",
|
||||
evidence_ref="synthetic-store-fixture", tenant_source="directory")
|
||||
assert found(aid)["status"] == "approved"
|
||||
assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_entry"
|
||||
assert str(expect(lambda: X.Run(doc, aid, run_cfg, deps_for(doc)).preflight(), X.Refused)) == "approval_entry:not_aal2"
|
||||
assert found(aid)["status"] == "approved"
|
||||
checks["aal1_human_entry_refused_by_executor"] = True
|
||||
|
||||
|
|
|
|||
|
|
@ -228,7 +228,7 @@ class RefusalTests(Base):
|
|||
"trust_not_satisfied:runtime_secret_trust": lambda: setattr(self.w, "trust_ok", False),
|
||||
"approval_binding": lambda: self.w.carrier.approval["binding"]["target"].update(admission_digest="sha256:" + "0" * 64),
|
||||
"approval_status": lambda: self.w.carrier.approval.update(status="requested"),
|
||||
"approval_entry": lambda: self.w.carrier.approval["entries"][0].update(assurance=AAL1),
|
||||
"approval_entry:not_aal2": lambda: self.w.carrier.approval["entries"][0].update(assurance=AAL1),
|
||||
"approval_expires_before_window_end": lambda: self.w.carrier.approval["validity"].update(expires_at="2026-11-03T10:00:00Z"),
|
||||
}
|
||||
for reason, mutate in cases.items():
|
||||
|
|
@ -238,12 +238,13 @@ class RefusalTests(Base):
|
|||
self.assert_no_effect(self.run_it(), reason)
|
||||
|
||||
def test_approval_by_someone_not_listed_or_not_human(self):
|
||||
for entry in ({"subject_id": "kc-sub-other", "assurance": AAL2, "principal_type": "human"},
|
||||
{"subject_id": ADMIN_SUB, "assurance": AAL2, "principal_type": "service"}):
|
||||
for entry, field in (({"subject_id": "kc-sub-other", "assurance": AAL2, "principal_type": "human"},
|
||||
"approver_not_listed"),
|
||||
({"subject_id": ADMIN_SUB, "assurance": AAL2, "principal_type": "service"}, "not_human")):
|
||||
with self.subTest(entry=entry):
|
||||
self.setUp()
|
||||
self.w.carrier.approval["entries"] = [entry]
|
||||
self.assert_no_effect(self.run_it(), "approval_entry")
|
||||
self.assert_no_effect(self.run_it(), "approval_entry:" + field)
|
||||
|
||||
def test_assurance_must_be_keycapes_verified_aal2(self):
|
||||
"""T19 finding: approval-engine stores the token's assurance object as JSON, never a bare level."""
|
||||
|
|
@ -252,7 +253,7 @@ class RefusalTests(Base):
|
|||
with self.subTest(assurance=assurance):
|
||||
self.setUp()
|
||||
self.w.carrier.approval["entries"][0]["assurance"] = assurance
|
||||
self.assert_no_effect(self.run_it(), "approval_entry")
|
||||
self.assert_no_effect(self.run_it(), "approval_entry:not_aal2")
|
||||
for assurance in (AAL2, json.loads(AAL2), json.dumps({"level": "aal3", "mfa": True})):
|
||||
self.assertTrue(X.entry_is_aal2(assurance))
|
||||
|
||||
|
|
@ -449,3 +450,16 @@ class FailureTests(Base):
|
|||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class EntryFailureNames(unittest.TestCase):
|
||||
"""NK-WP-0058-T03: the refusal names the failed field, never a subject."""
|
||||
|
||||
def test_each_field(self):
|
||||
ok = {"subject_id": "sub-a", "principal_type": "human", "assurance": AAL2}
|
||||
allowed = {"sub-a", "service:x"}
|
||||
self.assertIsNone(X.entry_failure(ok, allowed, "service:x"))
|
||||
self.assertEqual(X.entry_failure({**ok, "subject_id": "sub-b"}, allowed, "service:x"), "approver_not_listed")
|
||||
self.assertEqual(X.entry_failure({**ok, "principal_type": "service"}, allowed, "service:x"), "not_human")
|
||||
self.assertEqual(X.entry_failure({**ok, "assurance": AAL1}, allowed, "service:x"), "not_aal2")
|
||||
self.assertEqual(X.entry_failure({**ok, "subject_id": "service:x"}, allowed, "service:x"), "self_approval")
|
||||
|
|
|
|||
|
|
@ -59,11 +59,15 @@ class FakeBao:
|
|||
self.login_ok = {('nk-bind-executor', 'openbao', m.LANES['journey']['role'])}
|
||||
self.wide = False
|
||||
self.revoked = []
|
||||
self.status_code = 0 # bao status: 0 unsealed, 2 sealed, 1 unreachable
|
||||
self.lookup_code = 0 # token lookup: non-zero means no usable login in this shell
|
||||
|
||||
def __call__(self, args, *, payload=None, token=None, allow_failure=False):
|
||||
out = lambda d, code=0, err=b'': SimpleNamespace(returncode=code, stdout=json.dumps(d).encode(), stderr=err)
|
||||
missing = out({}, 2, b'No value found')
|
||||
if args[:2] == ['token', 'lookup']: return out({'data': self.identity})
|
||||
if args[0] == 'status': return out({'sealed': self.status_code == 2}, self.status_code)
|
||||
if args[:2] == ['token', 'lookup']:
|
||||
return out({}, self.lookup_code, b'permission denied') if self.lookup_code else out({'data': self.identity})
|
||||
if args[:2] == ['token', 'create']: return out({'auth': {'client_token': 'child'}})
|
||||
if args[0] == 'read' and args[-1] == m_key(self.m):
|
||||
return out({'data': {'cas_required': self.cas, **self.meta}})
|
||||
|
|
@ -225,3 +229,36 @@ def test_window_open_refusals(tmp_path, monkeypatch, mutate, reason):
|
|||
with pytest.raises(m.Refused, match=reason):
|
||||
m.window_open(d, path, {}, now=mutate(m, fake, now))
|
||||
assert 'auth/kubernetes/role/' + m.LANES['write']['role'] not in fake.store
|
||||
|
||||
|
||||
@pytest.mark.parametrize('status, lookup, reason', [
|
||||
(1, 0, 'openbao_unreachable'), (2, 0, 'openbao_sealed'), (0, 2, 'platform_admin_login_required')])
|
||||
def test_window_close_names_the_identity_cause(tmp_path, monkeypatch, status, lookup, reason):
|
||||
"""2026-11-05 a close failed as bare `LaneError` in a shell without a login (NK-WP-0058-T03)."""
|
||||
m, fake, d = build(tmp_path, monkeypatch)
|
||||
path, _ = admission_at(tmp_path, m)
|
||||
fake.status_code, fake.lookup_code = status, lookup
|
||||
with pytest.raises(m.Refused, match=reason):
|
||||
m.window_close(path, {})
|
||||
|
||||
|
||||
def test_main_receipt_and_line_carry_reason_and_hint(tmp_path, monkeypatch, capsys):
|
||||
m, fake, d = build(tmp_path, monkeypatch)
|
||||
path, _ = admission_at(tmp_path, m)
|
||||
fake.lookup_code = 2
|
||||
receipt = tmp_path / 'close.json'
|
||||
monkeypatch.setattr(sys, 'argv', ['executor-lanes.py', 'window-close', '--admission', str(path),
|
||||
'--receipt', str(receipt)])
|
||||
assert m.main() == 1
|
||||
line = json.loads(capsys.readouterr().out)
|
||||
assert line == {'status': 'failed', 'phase': 'identity', 'reason': 'platform_admin_login_required',
|
||||
'hint': m.HINTS['platform_admin_login_required']}
|
||||
assert json.loads(receipt.read_text())['reason'] == 'platform_admin_login_required'
|
||||
|
||||
|
||||
def test_lane_error_keeps_its_fixed_code(tmp_path, monkeypatch):
|
||||
m, *_ = build(tmp_path, monkeypatch)
|
||||
lane_error = sys.modules['state_hub_preflight_lane'].LaneError
|
||||
assert m.reason_of(lane_error('command_failed')) == 'LaneError:command_failed'
|
||||
assert m.reason_of(lane_error('not a fixed code')) == 'LaneError'
|
||||
assert m.reason_of(ValueError('anything')) == 'ValueError'
|
||||
|
|
|
|||
|
|
@ -102,7 +102,14 @@ def contract(name, expected):
|
|||
|
||||
|
||||
def attended_platform_admin():
|
||||
ident = data(bao(['token', 'lookup', '-format=json']))['data']
|
||||
"""Name the cause instead of a bare LaneError (NK-WP-0058-T03): OpenBao unreachable or sealed, no login in this
|
||||
shell, or a login that is not an attended platform-admin one."""
|
||||
status = bao(['status', '-format=json'], allow_failure=True).returncode
|
||||
if status == 2: raise Refused('openbao_sealed')
|
||||
if status: raise Refused('openbao_unreachable')
|
||||
lookup = bao(['token', 'lookup', '-format=json'], allow_failure=True)
|
||||
if lookup.returncode: raise Refused('platform_admin_login_required')
|
||||
ident = data(lookup)['data']
|
||||
if ('platform-admin' not in ident.get('policies', []) or 'root' in ident.get('policies', [])
|
||||
or not ident.get('path', '').startswith('auth/netkingdom/')):
|
||||
raise Refused('attended_platform_admin_required')
|
||||
|
|
@ -287,6 +294,22 @@ def window_close(admission_path, receipt):
|
|||
receipt.update(phase='done', status='window_closed', write_role_was_present=was_open, write_role_absent=True)
|
||||
|
||||
|
||||
HINTS = {
|
||||
'openbao_unreachable': 'check BAO_ADDR and the openbao-ui-railiance01 tunnel (bridge status)',
|
||||
'openbao_sealed': 'OpenBao is sealed; unseal before any window command',
|
||||
'platform_admin_login_required': 'bao login -method=oidc -path=netkingdom role=platform-admin in this shell',
|
||||
'attended_platform_admin_required': 'the token is not an attended platform-admin login (root or another path)',
|
||||
}
|
||||
|
||||
|
||||
def reason_of(error):
|
||||
"""A value-free reason: Refused carries a fixed code; the platform's LaneError carries a fixed code by its
|
||||
contract ('Only fixed, non-secret diagnostics may leave the envelope'); anything else is named by type only."""
|
||||
if isinstance(error, Refused): return str(error)
|
||||
if type(error).__name__ == 'LaneError' and str(error).isidentifier(): return 'LaneError:' + str(error)
|
||||
return type(error).__name__
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
p.add_argument('action', choices=['setup', 'verify', 'window-open', 'window-close'])
|
||||
|
|
@ -310,10 +333,11 @@ def main():
|
|||
elif a.action == 'window-open': window_open(digests, a.admission, r)
|
||||
else: window_close(a.admission, r)
|
||||
except Exception as error:
|
||||
r.update(status='failed', reason=str(error) if isinstance(error, Refused) else type(error).__name__)
|
||||
r.update(status='failed', reason=reason_of(error))
|
||||
if r['reason'] in HINTS: r['hint'] = HINTS[r['reason']]
|
||||
finally:
|
||||
with os.fdopen(fd, 'w') as out: json.dump(r, out, indent=2, sort_keys=True); out.write('\n')
|
||||
print(json.dumps({k: r[k] for k in ('status', 'phase', 'reason') if k in r}))
|
||||
print(json.dumps({k: r[k] for k in ('status', 'phase', 'reason', 'hint') if k in r}))
|
||||
return int(r['status'] == 'failed')
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -279,7 +279,7 @@ duplicates in other repos, a removal request is sent to their owners and recorde
|
|||
|
||||
```task
|
||||
id: NK-WP-0058-T03
|
||||
status: todo
|
||||
status: progress
|
||||
priority: high
|
||||
assignee: claude-code
|
||||
depends_on: [NK-WP-0058-T01, NK-WP-0058-T09]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue