66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
|
"""Unattended OpenBao credential fetch through an AppRole.
|
||
|
|
|
||
|
|
Each run logs in with the delivered role_id/secret_id, reads one KV v2 entry
|
||
|
|
and revokes the token. Values stay in memory for the run. They are never logged,
|
||
|
|
printed or written to disk.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
class CredentialUnavailable(Exception):
|
||
|
|
"""The credential could not be obtained. The message never contains a secret."""
|
||
|
|
|
||
|
|
|
||
|
|
def _call(addr, method, path, token=None, body=None, timeout=15):
|
||
|
|
request = urllib.request.Request(
|
||
|
|
addr.rstrip('/') + '/v1/' + path.lstrip('/'), method=method,
|
||
|
|
data=None if body is None else json.dumps(body).encode(),
|
||
|
|
headers={'Content-Type': 'application/json', **({'X-Vault-Token': token} if token else {})})
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||
|
|
raw = response.read()
|
||
|
|
return json.loads(raw) if raw else {}
|
||
|
|
except urllib.error.HTTPError as exc:
|
||
|
|
# Status only. The response body could echo request data.
|
||
|
|
raise CredentialUnavailable(f'OpenBao {method} {path.split("/")[0]}/… returned {exc.code}') from None
|
||
|
|
except (urllib.error.URLError, OSError) as exc:
|
||
|
|
raise CredentialUnavailable(f'OpenBao unreachable at {addr}: {type(exc).__name__}') from None
|
||
|
|
|
||
|
|
|
||
|
|
def kv_v2_data_path(kv_path):
|
||
|
|
mount, _, rest = kv_path.partition('/')
|
||
|
|
if not rest:
|
||
|
|
raise ValueError('kv_path must be <mount>/<path>')
|
||
|
|
return f'{mount}/data/{rest}'
|
||
|
|
|
||
|
|
|
||
|
|
def fetch(spec, call=_call):
|
||
|
|
"""Return the KV fields as a dict, read through AppRole login."""
|
||
|
|
addr = os.environ.get('BAO_ADDR') or spec.get('addr', 'http://127.0.0.1:18200')
|
||
|
|
approle = Path(os.path.expanduser(spec.get('approle_dir', '~/.local/inbound-library/approle')))
|
||
|
|
try:
|
||
|
|
role_id = (approle / 'role_id').read_text().strip()
|
||
|
|
secret_id = (approle / 'secret_id').read_text().strip()
|
||
|
|
except OSError:
|
||
|
|
raise CredentialUnavailable(f'AppRole credentials missing in {approle}') from None
|
||
|
|
login = call(addr, 'POST', f"auth/{spec.get('approle_mount', 'approle')}/login",
|
||
|
|
body={'role_id': role_id, 'secret_id': secret_id})
|
||
|
|
token = (login.get('auth') or {}).get('client_token')
|
||
|
|
if not token:
|
||
|
|
raise CredentialUnavailable('AppRole login returned no token')
|
||
|
|
try:
|
||
|
|
read = call(addr, 'GET', kv_v2_data_path(spec['kv_path']), token=token)
|
||
|
|
data = ((read.get('data') or {}).get('data')) or {}
|
||
|
|
if not data:
|
||
|
|
raise CredentialUnavailable(f"no data at {spec['kv_path']}")
|
||
|
|
return data
|
||
|
|
finally:
|
||
|
|
try:
|
||
|
|
call(addr, 'POST', 'auth/token/revoke-self', token=token)
|
||
|
|
except CredentialUnavailable:
|
||
|
|
pass # The token expires on its own (bounded TTL and number of uses).
|