117 lines
6.8 KiB
Python
117 lines
6.8 KiB
Python
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
PATH = Path(__file__).with_name('attended_private_registration.py')
|
|
spec = importlib.util.spec_from_file_location('attended_private', PATH)
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
mason = m.module(m.MASON, 'mason_contract_test', m.MASON_SHA)
|
|
# Synthetic shape only; never issued by a credential service.
|
|
TOKEN = 's.' + 'synthetic_test_value_' * 3
|
|
|
|
class Process:
|
|
def __init__(self, raw=TOKEN, error=''):
|
|
self.stdout = io.StringIO(raw)
|
|
self.stderr = io.StringIO(error)
|
|
self.returncode = 0
|
|
def wait(self, **kwargs): return self.returncode
|
|
def poll(self): return self.returncode
|
|
def kill(self): self.returncode = -9
|
|
|
|
class Checks(unittest.TestCase):
|
|
def caller(self):
|
|
return {'data': {'policies': [m.POLICY], 'identity_policies': [],
|
|
'entity_id': m.ENTITY, 'ttl': 1800}}
|
|
def test_exact_caller(self): m.verify_caller(json.dumps(self.caller()))
|
|
def test_wrong_caller_boundaries(self):
|
|
cases = [('ttl', True), ('ttl', 1801), ('ttl', 0),
|
|
('policies', ['default', m.POLICY]),
|
|
('policies', [m.POLICY, m.POLICY]),
|
|
('identity_policies', ['platform-admin']), ('identity_policies', 0), ('entity_id', 'admin')]
|
|
for key, value in cases:
|
|
data = self.caller(); data['data'][key] = value
|
|
with self.subTest(key=key, value=value), self.assertRaises(ValueError):
|
|
m.verify_caller(json.dumps(data))
|
|
def test_ambiguous_token_not_selected(self):
|
|
self.assertEqual(m.extract_token(TOKEN, mason.TOKEN_RE), TOKEN)
|
|
self.assertEqual(m.extract_token(TOKEN+' '+TOKEN, mason.TOKEN_RE), '')
|
|
self.assertEqual(m.extract_token('BAD '+TOKEN, mason.TOKEN_RE), TOKEN)
|
|
def test_missing_preapproval_stops_before_login(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
args = SimpleNamespace(approval=Path(tmp)/'missing', rollout_proof=Path(tmp)/'rollout', native_proof=Path(tmp)/'native', kubeconfig=Path(tmp)/'kube')
|
|
with patch.object(m.subprocess, 'Popen') as login, self.assertRaises(FileNotFoundError):
|
|
m.prepare(args)
|
|
login.assert_not_called()
|
|
def test_journal_refuses_symlink_and_wrong_mode(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
target = Path(tmp)/'target'; target.write_text(''); target.chmod(0o600)
|
|
link=Path(tmp)/'link'; link.symlink_to(target)
|
|
with self.assertRaises(OSError): m.append(link, {'safe': True})
|
|
target.chmod(0o644)
|
|
with self.assertRaises(ValueError): m.append(target, {'safe': True})
|
|
self.assertEqual(target.read_text(), '')
|
|
def exercise(self, error='', raw=TOKEN, bad_caller=False, child_exit=0, revoke_exit=0):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
approval=Path(tmp)/'approval'; approval.write_text('{}')
|
|
args=SimpleNamespace(approval=approval, rollout_proof=Path(tmp)/'rollout', native_proof=Path(tmp)/'native', kubeconfig=Path(tmp)/'kube')
|
|
helper=SimpleNamespace(PLAN=Path(tmp)/'plan', PUBLIC=Path(tmp)/'public', DISCLOSURE=Path(tmp)/'disclosure', CONSENT=Path(tmp)/'consent', IMAGE_PUBLICATION=Path(tmp)/'publication', load=lambda p:{}, gates=lambda *a:None)
|
|
events=[]; calls=[]; process=Process(raw, error)
|
|
def run(argv, **kwargs):
|
|
calls.append((argv, kwargs))
|
|
if argv[:3]==['bao','token','lookup']:
|
|
data=self.caller()
|
|
if bad_caller: data['data']['policies']=['platform-admin']
|
|
return subprocess.CompletedProcess(argv,0,json.dumps(data),'SECRET_LOOKUP_ERROR')
|
|
if argv[:3]==['bao','token','revoke']:
|
|
return subprocess.CompletedProcess(argv,revoke_exit,b'SECRET_REVOKE_OUTPUT',b'SECRET_REVOKE_ERROR')
|
|
return subprocess.CompletedProcess(argv,child_exit,b'SECRET_CHILD_OUTPUT',b'SECRET_CHILD_ERROR')
|
|
browser=SimpleNamespace(parse_args=lambda *a: 'https://example/authorize',launch=lambda value:None)
|
|
stdout=io.StringIO()
|
|
with patch.object(m.yaml,'safe_load',return_value={'state_dir':'unused'}),patch.object(m.Path,'read_text',return_value='{}'),patch.object(m.subprocess,'Popen',return_value=process) as popen,patch.object(m.subprocess,'run',side_effect=run),redirect_stdout(stdout):
|
|
result=m.execute(args,helper,mason,browser,events.append)
|
|
text=stdout.getvalue()+json.dumps(events)
|
|
self.assertNotIn('SECRET', text)
|
|
argv=popen.call_args.args[0]
|
|
self.assertIn('-token-only',argv); self.assertIn('-no-store',argv)
|
|
self.assertIn('skip_browser=true',argv)
|
|
env=popen.call_args.kwargs['env']
|
|
self.assertNotIn('BAO_TOKEN',env); self.assertNotIn('VAULT_TOKEN',env)
|
|
self.assertNotIn('REUSE_SURFACE_TOKEN',env)
|
|
for argv,kw in calls:
|
|
self.assertNotIn(TOKEN, ' '.join(argv))
|
|
self.assertTrue(kw.get('capture_output'))
|
|
return result, events, calls
|
|
def test_success_always_revokes_and_suppresses_child_bytes(self):
|
|
result, events, calls=self.exercise()
|
|
self.assertEqual(result,0); self.assertTrue(events[-1]['revoked'])
|
|
self.assertEqual(calls[-1][0],['bao','token','revoke','-self'])
|
|
self.assertTrue(any(a[0]=='warden' for a,k in calls))
|
|
def test_wrong_caller_revokes_without_writer(self):
|
|
result,events,calls=self.exercise(bad_caller=True)
|
|
self.assertEqual(result,1); self.assertTrue(events[-1]['revoked'])
|
|
self.assertFalse(any(a[0]=='warden' for a,k in calls))
|
|
def test_login_error_never_exposed_revokes_if_token_present(self):
|
|
result,events,calls=self.exercise(error='SECRET_ERROR_DETAIL\n')
|
|
self.assertEqual(result,1); self.assertTrue(events[-1]['revoked'])
|
|
self.assertFalse(any(a[0]=='warden' for a,k in calls))
|
|
def test_malformed_login_with_token_revokes_without_writer(self):
|
|
result,events,calls=self.exercise(raw='BAD '+TOKEN)
|
|
self.assertEqual(result,1); self.assertTrue(events[-1]['revoked'])
|
|
self.assertFalse(any(a[0]=='warden' for a,k in calls))
|
|
def test_child_failure_revokes(self):
|
|
result,events,calls=self.exercise(child_exit=11)
|
|
self.assertEqual(result,11); self.assertTrue(events[-1]['revoked'])
|
|
def test_failed_revoke_overrides_success(self):
|
|
result,events,calls=self.exercise(revoke_exit=1)
|
|
self.assertEqual(result,1); self.assertFalse(events[-1]['revoked'])
|
|
|
|
if __name__=='__main__': unittest.main()
|