Advance dependency roots with guarded source implementations

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091

tests/test_oidc_matrix.py

@@ -0,0 +1,77 @@
+from dataclasses import replace
+from unittest.mock import MagicMock, patch
+import json
+import pytest
+from ops_mason.executor import OIDCKVSpec, BuildRefused, BuildError, oidc_policy_hcl, oidc_role_document, build_oidc_kv_lane, build_spec_document
+from test_executor import _plan
+
+
+def spec(tmp_path):
+    return OIDCKVSpec(policy_name='workload-kv-fluid-telegram-hall-of-helix-operator',
+        role_name='fluid-telegram-hall-of-helix-workload-kv',
+        campaign_path='platform/workloads/coulomb/fluid-telegram/hall-of-helix',
+        bound_group='reviewed-test-operators', allowed_redirect_uris=('http://localhost:8250/oidc/callback',),
+        identity_confirmation_ref='test-only:iam-owner-receipt', platform_matrix_review_ref='test-only:platform-owner-receipt',
+        audit_log_path=tmp_path/'audit.jsonl')
+
+
+def test_matrix_has_exact_data_scope_and_write_once_salt(tmp_path):
+    s=spec(tmp_path);h=oidc_policy_hcl(s)
+    assert h.count('platform/data/workloads/coulomb/fluid-telegram/hall-of-helix/')==4
+    assert 'metadata/' not in h and '*' not in h and '/destroy/' not in h
+    salt=h.split('/redaction-salt"')[1].split('}')[0]
+    assert '"create", "read"' in salt and '"update"' not in salt
+    app=h.split('/operator-app"')[1].split('}')[0]
+    assert '"read"' in app and '"create"' not in app
+    assert 'auth/token/revoke-self' in h
+
+
+@pytest.mark.parametrize('field,value',[
+ ('campaign_path','platform/workloads/coulomb/fluid-telegram/*'),
+ ('campaign_path','platform/workloads/coulomb/fluid-telegram'),
+ ('campaign_path','platform/workloads/coulomb/fluid-telegram/../sibling'),
+ ('campaign_path','platform/workloads/coulomb/another-workload/campaign'),
+ ('bound_group','*'),('bound_group','group,other'),
+ ('identity_confirmation_ref','unconfirmed'),('platform_matrix_review_ref','pending'),
+ ('allowed_redirect_uris',('https://public.example/ui/callback',)),
+ ('allowed_redirect_uris',('http://user@localhost:8250/oidc/callback',)),
+ ('allowed_redirect_uris',('http://localhost:8250/oidc/callback?x=1',)),
+ ('allowed_redirect_uris',()),('allowed_redirect_uris',('http://localhost:bad/oidc/callback',)),
+ ('token_ttl','16m'),('token_max_ttl','31m'),('token_max_ttl','1m')])
+def test_refuses_unsafe_matrix_before_bao(tmp_path,field,value):
+    s=replace(spec(tmp_path),**{field:value})
+    with patch('ops_mason.executor.subprocess.run') as run:
+        with pytest.raises(BuildRefused):build_oidc_kv_lane(_plan(tmp_path,spec=spec(tmp_path)),s)
+    run.assert_not_called()
+
+
+def test_draft_and_changed_approval_never_execute(tmp_path):
+    s=spec(tmp_path)
+    with patch('ops_mason.executor.subprocess.run') as run:
+        with pytest.raises(BuildRefused):build_oidc_kv_lane(_plan(tmp_path,status='draft',spec=s),s)
+        p=_plan(tmp_path,spec=s)
+        with pytest.raises(BuildRefused):build_oidc_kv_lane(p,replace(s,bound_group='other-reviewed-group'))
+    run.assert_not_called()
+
+
+def test_exact_role_binding_and_value_free_build(tmp_path):
+    s=spec(tmp_path);p=_plan(tmp_path,spec=s)
+    with patch('ops_mason.executor.subprocess.run',return_value=MagicMock(returncode=0,stdout='suppressed-response')) as run:
+        result=build_oidc_kv_lane(p,s)
+    assert len(run.call_args_list)==2
+    assert run.call_args_list[0].args[0]==['bao','policy','write',s.policy_name,'-']
+    assert run.call_args_list[1].args[0]==['bao','write','auth/netkingdom/role/'+s.role_name,'-']
+    role=json.loads(run.call_args_list[1].kwargs['input'])
+    assert role['bound_claims']=={'groups':s.bound_group}
+    assert role['token_no_default_policy'] is True and role['token_explicit_max_ttl']=='30m'
+    assert role['token_policies']==[s.policy_name]
+    assert build_spec_document(s)['engine']=='openbao-oidc-kv-read-write'
+    assert 'suppressed-response' not in s.audit_log_path.read_text()
+    assert result['approved_spec_sha256']==p.approved_spec_sha256
+
+
+def test_failure_suppresses_output_and_never_claims_success(tmp_path):
+    s=spec(tmp_path)
+    with patch('ops_mason.executor.subprocess.run',return_value=MagicMock(returncode=1,stdout='sensitive-marker',stderr='sensitive-marker')):
+        with pytest.raises(BuildError) as e:build_oidc_kv_lane(_plan(tmp_path,spec=s),s)
+    assert 'sensitive-marker' not in str(e.value) and not s.audit_log_path.exists()