tegwick
b01a30bd49
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
252 lines
12 KiB
Python
252 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Private service Secret creation, exact-match verification and named rotation.
|
|
|
|
No credential is accepted on argv, evaluated as shell code, or printed. This
|
|
helper is not an authority grant or a multi-service rotation transaction.
|
|
"""
|
|
import base64
|
|
import contextlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
class Refused(Exception):
|
|
pass
|
|
|
|
|
|
TARGETS = {
|
|
'postgresql': ('databases', 'net-kingdom-pg-privacyidea-app', ('privacyidea',)),
|
|
'privacyidea': ('mfa', 'privacyidea-config', ('privacyidea',)),
|
|
'lldap': ('sso', 'lldap-secrets', ('lldap',)),
|
|
'authelia': ('sso', 'authelia-secrets', ('authelia', 'lldap')),
|
|
}
|
|
ROTATIONS = {
|
|
'postgresql': {'PI_DB_PASSWORD': 'password'},
|
|
'privacyidea': {'PI_SECRET_KEY': 'PI_SECRET_KEY', 'PI_DB_PASSWORD': 'PI_SQLALCHEMY_DATABASE_URI'},
|
|
'lldap': {'LLDAP_JWT_SECRET': 'LLDAP_JWT_SECRET'},
|
|
'authelia': {'AUTHELIA_SESSION_SECRET': 'session_secret'},
|
|
}
|
|
ANNOTATION = 'kubectl.kubernetes.io/last-applied-configuration'
|
|
|
|
|
|
def private(info, directory=False):
|
|
kind = stat.S_ISDIR if directory else stat.S_ISREG
|
|
if not kind(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
|
|
raise Refused()
|
|
if not directory and info.st_nlink != 1:
|
|
raise Refused()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def custody(root, components):
|
|
root = Path(os.path.abspath(root))
|
|
if any(p.is_symlink() for p in (root, *root.parents)): raise Refused()
|
|
descriptors = []
|
|
try:
|
|
flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
|
fd = os.open(root, flags); descriptors.append(fd); private(os.fstat(fd), True)
|
|
children = {}
|
|
for component in components:
|
|
child = os.open(component, flags, dir_fd=fd)
|
|
descriptors.append(child); private(os.fstat(child), True)
|
|
children[component] = child
|
|
yield children
|
|
finally:
|
|
for fd in reversed(descriptors): os.close(fd)
|
|
|
|
|
|
def read_private(parent, name):
|
|
fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
|
|
try:
|
|
private(os.fstat(fd))
|
|
raw = os.read(fd, 65537)
|
|
if len(raw) > 65536: raise Refused()
|
|
return raw
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def read_env(parent):
|
|
values = {}
|
|
for line in read_private(parent, 'secrets.env').decode().splitlines():
|
|
if not line.strip() or line.lstrip().startswith('#'): continue
|
|
name, separator, value = line.partition('=')
|
|
if not separator or not re.fullmatch(r'[A-Z][A-Z0-9_]*', name) or name in values:
|
|
raise Refused()
|
|
values[name] = value
|
|
return values
|
|
|
|
|
|
def literal(values, name):
|
|
value = values.get(name, '')
|
|
if not re.fullmatch(r'[A-Za-z0-9]{32,128}', value): raise Refused()
|
|
return value
|
|
|
|
|
|
def run(args, payload=None):
|
|
completed = subprocess.run(args, input=payload, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, timeout=60, check=False)
|
|
if completed.returncode or len(completed.stdout) > 2 * 1024 * 1024: raise Refused()
|
|
return completed.stdout
|
|
|
|
|
|
def decode_secret(raw, namespace, name):
|
|
obj = json.loads(raw)
|
|
if not isinstance(obj, dict) or obj.get('kind') != 'Secret' or obj.get('apiVersion') != 'v1':
|
|
raise Refused()
|
|
meta = obj.get('metadata')
|
|
if not isinstance(meta, dict) or meta.get('name') != name or meta.get('namespace') != namespace:
|
|
raise Refused()
|
|
if (not isinstance(meta.get('uid'), str) or not isinstance(meta.get('resourceVersion'), str)
|
|
or not re.fullmatch(r'[a-zA-Z0-9-]{1,128}', meta['uid'])
|
|
or not re.fullmatch(r'[0-9]{1,128}', meta['resourceVersion'])):
|
|
raise Refused()
|
|
if (meta.get('ownerReferences') or obj.get('immutable')
|
|
or obj.get('type') not in (('Opaque', 'kubernetes.io/basic-auth')
|
|
if name == 'net-kingdom-pg-privacyidea-app' else ('Opaque',))):
|
|
raise Refused()
|
|
annotations = meta.get('annotations', {})
|
|
labels = meta.get('labels', {})
|
|
if not isinstance(annotations, dict) or not isinstance(labels, dict): raise Refused()
|
|
if any(not isinstance(v, str) for v in (*annotations.values(), *labels.values())): raise Refused()
|
|
if any(k.startswith(('external-secrets.io/', 'reconcile.external-secrets.io/')) for k in labels): raise Refused()
|
|
if not isinstance(obj.get('data'), dict) or not obj['data']: raise Refused()
|
|
for key, value in obj['data'].items():
|
|
if not isinstance(key, str) or not isinstance(value, str): raise Refused()
|
|
base64.b64decode(value, validate=True)
|
|
return obj
|
|
|
|
|
|
def read_key(parent, configured, existing):
|
|
if configured:
|
|
path = Path(configured)
|
|
if not path.is_absolute() or any(p.is_symlink() for p in (path, *path.parents)):
|
|
raise Refused()
|
|
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
private(os.fstat(fd), True)
|
|
key = read_private(fd, path.name)
|
|
finally:
|
|
os.close(fd)
|
|
else:
|
|
try:
|
|
key = read_private(parent, 'oidc_private_key.pem')
|
|
except FileNotFoundError:
|
|
if existing is not None:
|
|
key = base64.b64decode(existing['data']['oidc_issuer_private_key'], validate=True)
|
|
else:
|
|
key = run(['openssl', 'genrsa', '2048'])
|
|
run(['openssl', 'rsa', '-check', '-noout'], key)
|
|
temp = '.issuer-' + secrets.token_hex(12)
|
|
fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
|
|
try:
|
|
with os.fdopen(fd, 'wb') as file:
|
|
file.write(key); file.flush(); os.fsync(file.fileno())
|
|
os.link(temp, 'oidc_private_key.pem', src_dir_fd=parent, dst_dir_fd=parent, follow_symlinks=False)
|
|
os.fsync(parent)
|
|
finally:
|
|
os.unlink(temp, dir_fd=parent)
|
|
run(['openssl', 'rsa', '-check', '-noout'], key)
|
|
return key.decode()
|
|
|
|
|
|
def service_values(component, values):
|
|
# Validate all required literals before any cluster call or local key creation.
|
|
if component == 'postgresql':
|
|
return {'username': 'privacyidea', 'password': literal(values['privacyidea'], 'PI_DB_PASSWORD')}
|
|
if component == 'privacyidea':
|
|
env = values['privacyidea']; password = literal(env, 'PI_DB_PASSWORD')
|
|
return {'PI_SECRET_KEY': literal(env, 'PI_SECRET_KEY'), 'PI_PEPPER': literal(env, 'PI_PEPPER'),
|
|
'PI_SQLALCHEMY_DATABASE_URI': 'postgresql://privacyidea:' + password +
|
|
'@net-kingdom-pg-rw.databases.svc.cluster.local:5432/privacyidea_db'}
|
|
if component == 'lldap':
|
|
return {key: literal(values['lldap'], key) for key in ('LLDAP_JWT_SECRET', 'LLDAP_LDAP_USER_PASS')}
|
|
env = values['authelia']
|
|
return {'jwt_secret': literal(env, 'AUTHELIA_JWT_SECRET'),
|
|
'session_secret': literal(env, 'AUTHELIA_SESSION_SECRET'),
|
|
'storage_encryption_key': literal(env, 'AUTHELIA_STORAGE_ENCRYPTION_KEY'),
|
|
'oidc_hmac_secret': literal(env, 'AUTHELIA_OIDC_HMAC_SECRET'),
|
|
'ldap_password': literal(values['lldap'], 'LLDAP_LDAP_USER_PASS')}
|
|
|
|
|
|
def publish(namespace, name, desired, old, allowed):
|
|
encoded = {k: base64.b64encode(v.encode()).decode() for k, v in desired.items()}
|
|
if old is None:
|
|
if allowed: raise Refused() # Rotation may not silently create missing state.
|
|
obj = {'apiVersion': 'v1', 'kind': 'Secret', 'type': 'Opaque',
|
|
'metadata': {'namespace': namespace, 'name': name}, 'data': encoded}
|
|
result = run(['kubectl', 'create', '-o', 'json', '-f', '-'], json.dumps(obj).encode())
|
|
expected = encoded; outcome = 'created'
|
|
else:
|
|
if not set(encoded).issubset(old['data']): raise Refused()
|
|
different = {k for k, value in encoded.items() if old['data'][k] != value}
|
|
if different - allowed: raise Refused()
|
|
annotations = old['metadata'].get('annotations', {})
|
|
if not different and ANNOTATION not in annotations: return 'verified'
|
|
patch = [{'op': 'test', 'path': '/metadata/uid', 'value': old['metadata']['uid']},
|
|
{'op': 'test', 'path': '/metadata/resourceVersion', 'value': old['metadata']['resourceVersion']},
|
|
{'op': 'test', 'path': '/data', 'value': old['data']}]
|
|
patch.extend({'op': 'replace', 'path': '/data/' + key, 'value': encoded[key]} for key in sorted(different))
|
|
if ANNOTATION in annotations:
|
|
patch.append({'op': 'remove', 'path': '/metadata/annotations/kubectl.kubernetes.io~1last-applied-configuration'})
|
|
result = run(['kubectl', 'patch', 'secret', name, '-n', namespace, '--type=json',
|
|
'--patch-file=/dev/stdin', '-o', 'json'], json.dumps(patch).encode())
|
|
expected = {**old['data'], **{k: encoded[k] for k in different}}
|
|
outcome = 'updated'
|
|
readback = decode_secret(result, namespace, name)
|
|
if readback['data'] != expected or ANNOTATION in readback['metadata'].get('annotations', {}): raise Refused()
|
|
if old is not None:
|
|
if readback['metadata']['uid'] != old['metadata']['uid'] or readback['type'] != old['type']: raise Refused()
|
|
expected_annotations = {k: v for k, v in old['metadata'].get('annotations', {}).items() if k != ANNOTATION}
|
|
if readback['metadata'].get('annotations', {}) != expected_annotations: raise Refused()
|
|
if readback['metadata'].get('labels', {}) != old['metadata'].get('labels', {}): raise Refused()
|
|
return outcome
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) not in (2, 3, 5): raise Refused()
|
|
component = sys.argv[1]
|
|
if component not in TARGETS: raise Refused()
|
|
namespace, name, inputs = TARGETS[component]
|
|
root = sys.argv[2] if len(sys.argv) >= 3 else Path(__file__).resolve().parent / 'secrets'
|
|
rotation = None
|
|
if len(sys.argv) == 5:
|
|
if sys.argv[3] != '--rotate-key' or sys.argv[4] not in ROTATIONS[component]: raise Refused()
|
|
rotation = ROTATIONS[component][sys.argv[4]]
|
|
with custody(root, inputs) as directories:
|
|
values = {c: read_env(directories[c]) for c in inputs}
|
|
desired = service_values(component, values)
|
|
if component == 'authelia':
|
|
import bcrypt
|
|
client = literal(values['authelia'], 'AUTHELIA_KEYCAPE_CLIENT_SECRET').encode()
|
|
if len(client) > 72: raise Refused()
|
|
raw = run(['kubectl', 'get', 'secret', name, '-n', namespace, '--ignore-not-found', '-o', 'json'])
|
|
old = decode_secret(raw, namespace, name) if raw.strip() else None
|
|
if rotation and old is None: raise Refused()
|
|
if component == 'authelia':
|
|
if old is None:
|
|
hashed = bcrypt.hashpw(client, bcrypt.gensalt(rounds=12))
|
|
else:
|
|
hashed = base64.b64decode(old['data']['keycape_client_secret_hash'], validate=True)
|
|
if not bcrypt.checkpw(client, hashed): raise Refused()
|
|
desired['keycape_client_secret_hash'] = hashed.decode()
|
|
desired['oidc_issuer_private_key'] = read_key(directories['authelia'],
|
|
values['authelia'].get('AUTHELIA_OIDC_PRIVATE_KEY_FILE', ''), old)
|
|
outcome = publish(namespace, name, desired, old, {rotation} if rotation else set())
|
|
print('Service Secret ' + outcome + '; native consumer acceptance remains required.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except (Refused, OSError, ValueError, KeyError, TypeError, ImportError, subprocess.SubprocessError):
|
|
print('ERROR: service Secret operation refused or failed. Check private inputs, current ownership '
|
|
'and the named rotation through approved metadata-only diagnostics. A failed write has an '
|
|
'unverified remote outcome; reconcile before retry. Local recovery keys are retained.', file=sys.stderr)
|
|
sys.exit(1)
|