a570bc2f5b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a137ca-5c77-7a71-b9ae-23ad9b0b5684
201 lines
12 KiB
Python
201 lines
12 KiB
Python
"""Meaningful fail-closed tests; no credentials or network are used."""
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import hashlib
|
|
import yaml
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout, redirect_stderr
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('reuse_provision', Path(__file__).parents[1] / 'scripts/provision-reuse-surface-forgejo-source.py')
|
|
p = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(p)
|
|
|
|
class Journal:
|
|
attempt = 'mock-attempt'
|
|
def __init__(self): self.rows = []
|
|
def put(self, event, **kw): self.rows.append((event, kw))
|
|
|
|
class Fake:
|
|
def __init__(self):
|
|
self.journal = Journal()
|
|
self.user = None
|
|
self.team = None
|
|
self.members = []
|
|
self.repos = []
|
|
self.tokens = []
|
|
self.extra = []
|
|
self.mutations = []
|
|
def pages(self, path, sudo=False):
|
|
if path == '/orgs/coulomb/teams': return [] if self.team is None else [self.team]
|
|
if path.endswith('/members'): return self.members
|
|
if path.endswith('/repos') and path.startswith('/teams/'): return self.repos
|
|
if path == '/user/orgs': return [] if not self.members else [{'username':'coulomb'}]
|
|
if path == '/user/teams': return [] if not self.members else [{**self.team,'organization':{'username':'coulomb'}}]
|
|
if path == '/user/repos?order_by=id': return self.repos + self.extra
|
|
if path.endswith('/tokens'): return self.tokens
|
|
raise AssertionError(path)
|
|
def request(self, method, path, payload=None, expected=(200,), **kwargs):
|
|
if method != 'GET':
|
|
self.mutations.append((method,path))
|
|
raise AssertionError('mutation must not occur in preflight failure test')
|
|
if path == '/users/' + p.USER: return (404,None) if self.user is None else (200,self.user)
|
|
if path.startswith('/repos/'): return 200, {'private':True,'full_name':'coulomb/'+path.split('/')[3]}
|
|
raise AssertionError(path)
|
|
|
|
class Bao:
|
|
def __init__(self, data=None): self.data = data
|
|
def get(self): return self.data
|
|
|
|
class Tests(unittest.TestCase):
|
|
def complete(self):
|
|
c = Fake()
|
|
c.user = {'login':p.USER,'restricted':True,'is_admin':False,'visibility':'private'}
|
|
c.team = {'id':10,'name':p.TEAM,'permission':'read','includes_all_repositories':False,'can_create_org_repo':False,'units_map':p.UNITS}
|
|
c.members = [{'login':p.USER}]
|
|
c.repos = [{'name':x,'owner':{'login':p.ORG},'private':True,'full_name':p.ORG+'/'+x} for x in sorted(p.REPOS)]
|
|
return c
|
|
def test_server_capped_pages_continue_until_empty(self):
|
|
c=p.Client('mock-admin',Journal()); pages={1:[{'id':x} for x in range(1,51)],2:[{'id':51}],3:[]}
|
|
def get(method,path,**kwargs):
|
|
c.last_total_count=51
|
|
page=int(path.split('page=')[1]); return 200,pages[page]
|
|
with patch.object(c,'request',side_effect=get) as request:
|
|
self.assertEqual(len(c.pages('/user/repos?order_by=id',sudo=True)),51)
|
|
self.assertEqual(request.call_count,3)
|
|
def test_late_collaborator_private_grant_zero_mutations(self):
|
|
c=self.complete()
|
|
original_pages=c.pages
|
|
def pages(path,sudo=False):
|
|
if path.startswith('/user/repos'):
|
|
return p.Client.pages(c,path,sudo=sudo)
|
|
return original_pages(path,sudo=sudo)
|
|
original_request=c.request
|
|
def request(method,path,**kwargs):
|
|
if path.startswith('/user/repos?'):
|
|
page=int(path.split('page=')[1])
|
|
if page==1: return 200,[{'id':x,'private':False,'full_name':'someone/public-'+str(x),'owner':{'login':'someone'}} for x in range(1,51)]
|
|
if page==2: return 200,[{'id':51,'private':True,'full_name':'other-org/excluded-collaborator','owner':{'login':'other-org'}}]
|
|
return 200,[]
|
|
return original_request(method,path,**kwargs)
|
|
c.pages=pages; c.request=request
|
|
with self.assertRaisesRegex(p.Refused,'extra_private_repository_grant'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_duplicate_page_ids_refuse(self):
|
|
c=p.Client('mock-admin',Journal())
|
|
with patch.object(c,'request',return_value=(200,[{'id':1}])):
|
|
with self.assertRaisesRegex(p.Refused,'duplicate_page_id'): c.pages('/user/repos')
|
|
def test_total_count_mismatch_refuses(self):
|
|
c=p.Client('mock-admin',Journal())
|
|
def request(method,path,**kwargs):
|
|
c.last_total_count=2
|
|
return 200,[{'id':1}] if 'page=1' in path else []
|
|
with patch.object(c,'request',side_effect=request):
|
|
with self.assertRaisesRegex(p.Refused,'total_count_mismatch'): c.pages('/user/repos')
|
|
def test_unstable_total_count_refuses(self):
|
|
c=p.Client('mock-admin',Journal())
|
|
def request(method,path,**kwargs):
|
|
page=int(path.split('page=')[1]); c.last_total_count=2 if page==1 else 3
|
|
return 200,[{'id':page}]
|
|
with patch.object(c,'request',side_effect=request):
|
|
with self.assertRaisesRegex(p.Refused,'unstable_total_count'): c.pages('/user/repos')
|
|
def test_excess_pagination_refuses(self):
|
|
c=p.Client('mock-admin',Journal())
|
|
def request(method,path,**kwargs):
|
|
return 200,[{'id':int(path.split('page=')[1])}]
|
|
with patch.object(c,'request',side_effect=request):
|
|
with self.assertRaisesRegex(p.Refused,'pagination_bound_exceeded'): c.pages('/user/repos')
|
|
def test_existing_all_repository_team_refused_before_mutation(self):
|
|
c=self.complete(); c.team['includes_all_repositories']=True
|
|
with self.assertRaisesRegex(p.Refused,'team_conflict'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_existing_excluded_team_repo_refused_before_mutation(self):
|
|
c=self.complete(); c.repos.append({'name':'rapp-core-hub','owner':{'login':'coulomb'}})
|
|
with self.assertRaisesRegex(p.Refused,'unexpected_team_repository'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_existing_admin_user_refused_before_mutation(self):
|
|
c=self.complete(); c.user['is_admin']=True
|
|
with self.assertRaisesRegex(p.Refused,'user_conflict'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_existing_token_kv_binding_mismatch_refused(self):
|
|
c=self.complete(); c.tokens=[{'id':20,'name':p.PREFIX+'one','scopes':p.SCOPES,'token_last_eight':'different'}]
|
|
with self.assertRaisesRegex(p.Refused,'pat_kv_binding_unverifiable'): p.provision(c,Bao({p.FIELD:'mock-token'}),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_extra_private_grants_refused(self):
|
|
c=self.complete(); c.extra=[{'private':True,'full_name':'coulomb/secret-outside'}]
|
|
with self.assertRaisesRegex(p.Refused,'extra_private'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_orphan_pat_not_reminted(self):
|
|
c=self.complete(); c.tokens=[{'id':20,'name':p.PREFIX+'one','scopes':p.SCOPES,'token_last_eight':'ck-token'}]
|
|
with self.assertRaisesRegex(p.Refused,'orphan'): p.provision(c,Bao(),{})
|
|
self.assertEqual(c.mutations,[])
|
|
def test_verified_noop_rechecks_secret_and_scope(self):
|
|
c=self.complete(); c.tokens=[{'id':20,'name':p.PREFIX+'one','scopes':p.SCOPES,'token_last_eight':'ck-token'}]
|
|
with patch.object(p,'verify') as verify:
|
|
self.assertEqual(p.provision(c,Bao({p.FIELD:'mock-token'}),{}),'noop')
|
|
verify.assert_called_once_with(c,'mock-token',{})
|
|
self.assertEqual(c.mutations,[])
|
|
self.assertEqual(c.journal.rows[-1],('terminal',{'outcome':'verified_noop'}))
|
|
def test_existing_write_unit_refused(self):
|
|
c=self.complete(); c.team['units_map']={**p.UNITS,'repo.code':'write'}
|
|
with self.assertRaisesRegex(p.Refused,'team_conflict'): p.inspect(c,Bao())
|
|
def test_api_ambiguous_mutation_records_intent_without_raw_error(self):
|
|
j=Journal(); c=p.Client('fake-secret-admin',j)
|
|
with patch.object(c.opener,'open',side_effect=RuntimeError('fake-secret-admin sensitive provider body')):
|
|
with self.assertRaisesRegex(p.Refused,'request_observation_failed'):
|
|
c.request('POST','/admin/users',{'password':'fake-secret-password'},operation='user_create')
|
|
self.assertEqual([x[0] for x in j.rows],['intent','outcome'])
|
|
self.assertNotIn('secret',json.dumps(j.rows))
|
|
self.assertEqual(j.rows[-1][1]['status'],'ambiguous')
|
|
def test_network_error_never_reaches_stdout_stderr(self):
|
|
out,err=io.StringIO(),io.StringIO()
|
|
with patch.object(p,'approval',return_value={}), patch.object(p,'Journal',return_value=Journal()) as factory, patch.dict(p.os.environ,{'API_TOKEN':'fake-secret-admin'}), patch.object(p,'provision',side_effect=RuntimeError('fake-secret-admin body')):
|
|
factory.return_value.close=lambda:None
|
|
with redirect_stdout(out),redirect_stderr(err):
|
|
code=p.main(['--inner','--approval','mock','--receipt','mock'])
|
|
self.assertEqual(code,1); self.assertEqual(out.getvalue()+err.getvalue(),'')
|
|
self.assertNotIn('secret',json.dumps(factory.return_value.rows))
|
|
def test_unapproved_ccr_refuses_before_warden(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
f=Path(tmp)/'approval.json'; f.write_text(json.dumps({'ccr_id':p.CCR,'decision':'approved','supervisor':'mock','decision_ref':'mock'}))
|
|
with patch.object(p.subprocess,'run') as run:
|
|
self.assertEqual(p.main(['--approval',str(f),'--receipt',str(Path(tmp)/'receipt')]),1)
|
|
run.assert_not_called()
|
|
def test_approved_artifact_revision_mismatch_before_warden(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp); files={p.CCR_FILE,p.PLAN_FILE,p.SCRIPT_FILE,'openbao/policies/workload-kv-read-reuse-surface-runtime-eso.hcl'}
|
|
for rel in files:
|
|
f=root/rel; f.parent.mkdir(parents=True,exist_ok=True); f.write_text('mock')
|
|
(root/p.CCR_FILE).write_text(yaml.safe_dump({'status':'approved'}))
|
|
receipt={'ccr_id':p.CCR,'decision':'approved','supervisor':'founder','decision_ref':'mock-resolved-decision','approved_artifacts':{rel:hashlib.sha256((root/rel).read_bytes()).hexdigest() for rel in files}}
|
|
f=root/'approval.json'; f.write_text(json.dumps(receipt)); (root/p.SCRIPT_FILE).write_text('modified-after-approval')
|
|
with patch.object(p,'ROOT',root),patch.object(p.subprocess,'run') as run:
|
|
self.assertEqual(p.main(['--approval',str(f),'--receipt',str(root/'receipt')]),1)
|
|
run.assert_not_called()
|
|
def test_journal_symlink_refused(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
f=Path(tmp)/'real'; f.write_text(''); link=Path(tmp)/'link'; link.symlink_to(f)
|
|
with self.assertRaises(OSError): p.Journal(link)
|
|
def test_redirects_never_follow_auth_header(self):
|
|
with self.assertRaisesRegex(p.Refused,'redirect_refused'):
|
|
p.NoRedirect().redirect_request(None,None,302,'',{ },'https://other.example')
|
|
def test_bao_cas_write_uncertain_response_reconciled(self):
|
|
b=p.Bao(); j=Journal()
|
|
with patch.object(b,'run',return_value=type('R',(),{'returncode':1})()) as run, patch.object(b,'get',return_value={p.FIELD:'fake-source'}):
|
|
b.put('fake-source',j)
|
|
self.assertIn('-cas=0',run.call_args.args[0]); self.assertEqual(run.call_args.args[1],b'fake-source')
|
|
self.assertEqual(j.rows[-1],('verified',{'operation':'kv_create','status':'matched'}))
|
|
def test_bao_timeout_preserves_ambiguous_outcome(self):
|
|
b=p.Bao(); j=Journal()
|
|
with patch.object(b,'run',side_effect=p.Refused('bao_observation_failed')):
|
|
with self.assertRaises(p.Refused): b.put('fake-source',j)
|
|
self.assertEqual(j.rows[-1],('outcome',{'operation':'kv_create','status':'ambiguous'}))
|
|
def test_kv_denial_not_treated_as_absence(self):
|
|
b=p.Bao()
|
|
with patch.object(b,'run',return_value=type('R',(),{'returncode':1,'stderr':b'Code: 403 Error: fake-secret'})()):
|
|
with self.assertRaisesRegex(p.Refused,'bao_preflight_failed'): b.get()
|
|
|
|
if __name__=='__main__': unittest.main()
|