b2ebe10849
The five dynamic-database ClusterSecretStores use 768h static tokens that nothing renews; expiry revoked their DB leases on 2026-09-23 and recurs around 2026-10-25. Kubernetes auth is not a drop-in fix: ESO v0.16.1 revokes its login token after each reconcile, which revokes the leases it created. - eso-token-renewer CronJob (ArgoCD draft, no RBAC, mounted Secrets). - Attended periodic mint script for all five lanes. - CronJob added to the platform-addons AppProject in git (not yet applied). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 150322@bnt-lap001 Assistant-Session: 16a7b788-374e-4915-a1df-fc87ffd9a5e4
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import urllib.error
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADDON = ROOT / 'argocd/platform-addons/eso-token-renewer'
|
|
MINT = ROOT / 'scripts/openbao-eso-db-token-periodic-attended.sh'
|
|
spec = importlib.util.spec_from_file_location('renew', ADDON / 'renew.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
def opener_for(ttls):
|
|
seen = []
|
|
|
|
def opener(request, timeout):
|
|
token = request.headers['X-vault-token']
|
|
seen.append((request.full_url, request.get_method(), token))
|
|
ttl = ttls[token]
|
|
if isinstance(ttl, Exception):
|
|
raise ttl
|
|
return Response(json.dumps({'auth': {'lease_duration': ttl, 'renewable': True}}).encode())
|
|
return opener, seen
|
|
|
|
|
|
def lanes(tmp_path, tokens):
|
|
for lane, token in tokens.items():
|
|
(tmp_path / lane).mkdir()
|
|
(tmp_path / lane / 'token').write_text(token + '\n')
|
|
m.ROOT = tmp_path
|
|
|
|
|
|
def test_all_lanes_renewed_and_no_token_printed(tmp_path, capsys):
|
|
lanes(tmp_path, {'core-hub': 'tok-a', 'tenant-engine': 'tok-b'})
|
|
opener, seen = opener_for({'tok-a': 604800, 'tok-b': 604800})
|
|
assert m.main(opener) == 0
|
|
assert [s[1] for s in seen] == ['POST', 'POST']
|
|
assert all(s[0].endswith('/v1/auth/token/renew-self') for s in seen)
|
|
out = capsys.readouterr().out
|
|
assert 'tok-a' not in out and 'tok-b' not in out
|
|
assert [json.loads(line)['lane'] for line in out.splitlines()] == ['core-hub', 'tenant-engine']
|
|
|
|
|
|
def test_capped_ttl_and_http_errors_fail_the_job(tmp_path, capsys):
|
|
lanes(tmp_path, {'a': 't1', 'b': 't2', 'c': 't3'})
|
|
denied = urllib.error.HTTPError('u', 403, 'denied', {}, None)
|
|
opener, _ = opener_for({'t1': 604800, 't2': 3600, 't3': denied})
|
|
assert m.main(opener) == 1
|
|
rows = {r['lane']: r for r in map(json.loads, capsys.readouterr().out.splitlines())}
|
|
assert rows['a']['ok'] and rows['b']['error'] == 'ttl_below_threshold'
|
|
assert rows['c']['error'] == 'http_403'
|
|
|
|
|
|
def test_no_lanes_is_a_failure(tmp_path):
|
|
m.ROOT = tmp_path
|
|
assert m.main() == 1
|
|
|
|
|
|
def test_cronjob_mounts_match_mint_script_lanes():
|
|
docs = list(yaml.safe_load_all((ADDON / 'cronjob.yaml').read_text()))
|
|
cron = next(d for d in docs if d['kind'] == 'CronJob')
|
|
pod = cron['spec']['jobTemplate']['spec']['template']['spec']
|
|
mounted = {v['name']: v['secret']['secretName'] for v in pod['volumes'] if 'secret' in v}
|
|
script = MINT.read_text()
|
|
minted = dict(re.findall(r'^\s+([a-z-]+)\) echo "(openbao-[a-z-]+-eso-token) ', script, re.M))
|
|
assert mounted == minted
|
|
assert pod['automountServiceAccountToken'] is False
|
|
assert all(m_['readOnly'] for m_ in pod['containers'][0]['volumeMounts'])
|
|
|
|
|
|
def test_mint_script_uses_period_and_is_silent():
|
|
script = MINT.read_text()
|
|
assert 'exec >/dev/null 2>&1' in script
|
|
assert '-period="$TOKEN_PERIOD"' in script and '-ttl=' not in script
|
|
assert subprocess.run(['sh', '-n', str(MINT)]).returncode == 0
|
|
assert subprocess.run(['sh', str(MINT), '--confirm', 'wrong', 'core-hub']).returncode == 2
|
|
|
|
|
|
def test_mint_policies_exist_or_are_known_external():
|
|
script = MINT.read_text()
|
|
policies = set()
|
|
for group in re.findall(r'-eso-token ([a-z0-9,-]+)"', script):
|
|
policies.update(group.split(','))
|
|
declared = {p.stem for p in (ROOT / 'openbao/policies').glob('*.hcl')}
|
|
# Configured by rapp-postgres scripts (configure-*-openbao.sh), not as .hcl here.
|
|
external = {'external-secrets-canned-prompts-database', 'external-secrets-sbom-nexus-database',
|
|
'credential-broker-tenant-engine-runtime', 'credential-broker-tenant-engine-migration'}
|
|
assert policies - declared <= external
|