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
64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Renew the ESO parent tokens that project OpenBao dynamic database leases.
|
|
|
|
ESO never renews a tokenSecretRef token, and OpenBao revokes every lease a
|
|
token created when that token expires (RPF-WP-0046). Each token Secret is
|
|
mounted read-only at <ESO_TOKEN_ROOT>/<lane>/token. For each lane this calls
|
|
auth/token/renew-self and prints one JSON line with the lane, the resulting
|
|
TTL and an error class. It never prints a token.
|
|
|
|
Exit 1 if any lane fails or ends below MIN_TTL_SECONDS, so the Job is marked
|
|
failed. A non-periodic token cannot be renewed past its max TTL, so it trips
|
|
the threshold as it nears expiry instead of failing silently.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
ADDR = os.environ.get('BAO_ADDR', 'http://openbao.openbao.svc:8200').rstrip('/')
|
|
ROOT = Path(os.environ.get('ESO_TOKEN_ROOT', '/var/run/eso-tokens'))
|
|
MIN_TTL = int(os.environ.get('MIN_TTL_SECONDS', str(72 * 3600)))
|
|
|
|
|
|
def renew(token, opener=urllib.request.urlopen):
|
|
request = urllib.request.Request(
|
|
ADDR + '/v1/auth/token/renew-self', data=b'{}', method='POST',
|
|
headers={'X-Vault-Token': token, 'Content-Type': 'application/json'})
|
|
with opener(request, timeout=15) as response:
|
|
auth = json.load(response).get('auth') or {}
|
|
return {'ttl': auth.get('lease_duration'), 'renewable': auth.get('renewable')}
|
|
|
|
|
|
def check_lane(lane_dir, opener=urllib.request.urlopen):
|
|
result = {'lane': lane_dir.name, 'ok': False}
|
|
try:
|
|
token = (lane_dir / 'token').read_text(encoding='utf-8').strip()
|
|
if not token:
|
|
return dict(result, error='empty_token')
|
|
result.update(renew(token, opener))
|
|
except urllib.error.HTTPError as error:
|
|
return dict(result, error='http_%s' % error.code)
|
|
except (urllib.error.URLError, OSError, ValueError) as error:
|
|
return dict(result, error=type(error).__name__)
|
|
ttl = result.get('ttl')
|
|
if not isinstance(ttl, int) or ttl < MIN_TTL:
|
|
return dict(result, error='ttl_below_threshold')
|
|
return dict(result, ok=True)
|
|
|
|
|
|
def main(opener=urllib.request.urlopen):
|
|
lanes = sorted(p for p in ROOT.iterdir() if p.is_dir() and not p.name.startswith('.'))
|
|
if not lanes:
|
|
print(json.dumps({'ok': False, 'error': 'no_lanes_mounted'}))
|
|
return 1
|
|
results = [check_lane(lane, opener) for lane in lanes]
|
|
for item in results:
|
|
print(json.dumps(item, sort_keys=True))
|
|
return 0 if all(item['ok'] for item in results) else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|