tegwick
b46355d6b0
ILIB-WP-0003-T04: stdlib AppRole login, single KV v2 read and token revoke-self; no env-file cache; credential_unavailable keeps matching and notices running. Five fake-OpenBao tests; 19 total pass. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 1170398@bnt-lap001 Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
91 lines
3.8 KiB
Python
91 lines
3.8 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from inbound_library import intake, vault
|
|
from inbound_library.__main__ import run_due
|
|
|
|
SECRET = 's3cret-imap-password'
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self, deny_read=False, down=False):
|
|
self.calls, self.deny_read, self.down = [], deny_read, down
|
|
|
|
def __call__(self, addr, method, path, token=None, body=None):
|
|
self.calls.append((method, path, token))
|
|
if self.down:
|
|
raise vault.CredentialUnavailable(f'OpenBao unreachable at {addr}: URLError')
|
|
if path == 'auth/approle/login':
|
|
assert body == {'role_id': 'role-1', 'secret_id': 'secret-1'}
|
|
return {'auth': {'client_token': 'tok-1'}}
|
|
if path == 'platform/data/workloads/inbound-library/imap':
|
|
if self.deny_read:
|
|
raise vault.CredentialUnavailable('OpenBao GET platform/… returned 403')
|
|
return {'data': {'data': {'IMAP_USERNAME': 'library@coulomb.social',
|
|
'IMAP_PASSWORD': SECRET, 'IMAP_HOST': 'imap.ionos.de',
|
|
'IMAP_PORT': '993', 'IMAP_FOLDER': 'INBOX'}}}
|
|
if path == 'auth/token/revoke-self':
|
|
return {}
|
|
raise AssertionError(path)
|
|
|
|
|
|
class Vault(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.approle = Path(self.tmp.name) / 'approle'
|
|
self.approle.mkdir()
|
|
(self.approle / 'role_id').write_text('role-1\n')
|
|
(self.approle / 'secret_id').write_text('secret-1\n')
|
|
self.spec = {'addr': 'http://bao', 'approle_dir': str(self.approle),
|
|
'kv_path': 'platform/workloads/inbound-library/imap'}
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_login_read_and_revoke(self):
|
|
bao = FakeBao()
|
|
fields = vault.fetch(self.spec, call=bao)
|
|
self.assertEqual(fields['IMAP_PASSWORD'], SECRET)
|
|
self.assertEqual([c[1] for c in bao.calls], [
|
|
'auth/approle/login', 'platform/data/workloads/inbound-library/imap',
|
|
'auth/token/revoke-self'])
|
|
|
|
def test_denied_read_still_revokes_and_hides_secret(self):
|
|
bao = FakeBao(deny_read=True)
|
|
with self.assertRaises(vault.CredentialUnavailable) as caught:
|
|
vault.fetch(self.spec, call=bao)
|
|
self.assertEqual(bao.calls[-1][1], 'auth/token/revoke-self')
|
|
self.assertNotIn('secret-1', str(caught.exception))
|
|
|
|
def test_missing_approle_files(self):
|
|
(self.approle / 'secret_id').unlink()
|
|
with self.assertRaises(vault.CredentialUnavailable):
|
|
vault.fetch(self.spec, call=FakeBao())
|
|
|
|
def test_mailbox_built_from_vault_fields(self):
|
|
box = intake.mailbox_from_config(
|
|
{'credential': self.spec}, environ={},
|
|
fetch=lambda spec: vault.fetch(spec, call=FakeBao()))
|
|
self.assertEqual((box.user, box.host, box.port, box.folder),
|
|
('library@coulomb.social', 'imap.ionos.de', 993, 'INBOX'))
|
|
|
|
def test_run_due_reports_unreachable_bao_without_secret_and_still_matches(self):
|
|
root = Path(self.tmp.name) / 'lib'
|
|
(root / 'registry').mkdir(parents=True)
|
|
(root / 'registry' / 'topics.json').write_text(json.dumps({'version': 1, 'topics': {}}))
|
|
original = vault.fetch
|
|
vault.fetch = lambda spec: original(spec, call=FakeBao(down=True))
|
|
try:
|
|
report = run_due(root, {'credential': self.spec}, post=lambda *a: None)
|
|
finally:
|
|
vault.fetch = original
|
|
self.assertFalse(report['ok'])
|
|
self.assertEqual(report['intake']['status'], 'credential_unavailable')
|
|
self.assertIn('notices', report)
|
|
self.assertNotIn(SECRET, json.dumps(report))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|