39 lines
2.2 KiB
Python
39 lines
2.2 KiB
Python
|
import importlib.util,unittest,json
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import patch
|
||
|
|
from contextlib import contextmanager
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from urllib.error import HTTPError,URLError
|
||
|
|
spec=importlib.util.spec_from_file_location('native',Path(__file__).resolve().parents[1]/'scripts/t03-native-execution.py');m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
||
|
|
class TestProbe(unittest.TestCase):
|
||
|
|
def check(self,status):
|
||
|
|
scoped=SimpleNamespace(token='synthetic-revoked-token',token_capabilities=lambda *a,**k:['deny'])
|
||
|
|
session=SimpleNamespace(client=scoped,revocation_succeeded=False)
|
||
|
|
@contextmanager
|
||
|
|
def session_context(role):
|
||
|
|
yield session
|
||
|
|
session.revocation_succeeded=True;scoped.token=''
|
||
|
|
client=SimpleNamespace(addr='http://127.0.0.1:18200',approle_session=session_context)
|
||
|
|
class Opener:
|
||
|
|
def open(self,req,timeout):
|
||
|
|
self_test.assertEqual(req.get_header('X-vault-token'),'synthetic-revoked-token')
|
||
|
|
self_test.assertEqual(scoped.token,'')
|
||
|
|
if status:raise HTTPError(req.full_url,status,'synthetic',{},None)
|
||
|
|
return unittest.mock.MagicMock()
|
||
|
|
self_test=self
|
||
|
|
with patch('urllib.request.build_opener',return_value=Opener()):return m.verify_session_cleanup(client)
|
||
|
|
def test_revoked_explicit_token_refused(self):self.assertEqual(self.check(403)['revoked_token_lookup_status'],403)
|
||
|
|
def test_live_token_refused(self):
|
||
|
|
with self.assertRaisesRegex(ValueError,'revoked_token_still_usable'):self.check(0)
|
||
|
|
def test_server_failure_is_not_revocation_proof(self):
|
||
|
|
with self.assertRaisesRegex(ValueError,'revocation_probe_not_definitive'):self.check(500)
|
||
|
|
def test_reconciled_prior_phase(self):
|
||
|
|
prior={'phase':'verify_attempt_started','failure_code':'revoked_token_still_usable','actions':[{'action':'apply','approval_id':m.IDS['apply'],'exit_code':0,'limits':{'token_ttl':900,'token_max_ttl':1800,'secret_id_ttl':900,'secret_id_num_uses':1,'token_num_uses':8}}]}
|
||
|
|
with patch.object(Path,'read_text',return_value=json.dumps(prior)):
|
||
|
|
self.assertEqual(m.resume_receipt(),prior)
|
||
|
|
prior['actions'][0]['approval_id']='other'
|
||
|
|
with patch.object(Path,'read_text',return_value=json.dumps(prior)):
|
||
|
|
with self.assertRaisesRegex(ValueError,'prior_apply_not_verified'):m.resume_receipt()
|
||
|
|
|
||
|
|
if __name__=='__main__':unittest.main()
|