tegwick
d5ae2d4924
Lock exact data-path capabilities and the absence of parent and sibling paths in the unit test, and mark T02 done after those tests passed. Assistant: grok Assistant-Session: 01a10946-16b2-7771-88ac-c6c2d3e14d0b
82 lines
4.7 KiB
Python
82 lines
4.7 KiB
Python
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)
|
|
prefix='platform/data/workloads/coulomb/fluid-telegram/hall-of-helix/'
|
|
paths=[line.split('"')[1] for line in h.splitlines() if line.startswith('path ')]
|
|
assert paths==[prefix+'operator-app',prefix+'operator-session',prefix+'bot-token',prefix+'redaction-salt','auth/token/revoke-self']
|
|
assert 'metadata/' not in h and '*' not in h and '/destroy/' not in h
|
|
assert 'platform/data/workloads/coulomb/fluid-telegram"' not in h
|
|
assert 'platform/data/workloads/coulomb/another-workload' not in h
|
|
def caps(entry):
|
|
return h.split(prefix+entry+'"')[1].split('}')[0]
|
|
assert caps('operator-app').strip()=='{\n capabilities = ["read"]'
|
|
assert caps('operator-session').strip()==caps('bot-token').strip()=='{\n capabilities = ["create", "read", "update"]'
|
|
salt=caps('redaction-salt')
|
|
assert '"create", "read"' in salt and all(x not in salt for x in ('"update"','"patch"','"delete"','"list"'))
|
|
|
|
|
|
@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()
|