#!/usr/bin/env python3 """Bounded apps-pg demo consumer provisioning; never emits credential values.""" import argparse import base64 import json from pathlib import Path import secrets import subprocess from urllib.parse import quote import yaml ROOT = Path(__file__).resolve().parents[1] CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786' NAME = 'vergabe_demo_company' NS = 'vergabe-demo-company' SECRET = 'vergabe-demo-company-credentials' RUNTIME = 'vergabe-demo-env' OWNER = {'railiance.io/provisioning-workplan': 'RPF-WP-0039'} KUBE = ['kubectl', '--kubeconfig', '/home/worsch/.kube/config-hosteurope', '--server', 'https://127.0.0.1:16444', '--request-timeout=20s'] class Refused(Exception): pass def require(condition, reason): if not condition: raise Refused(reason) def kube(args, value=None, *, raw=False): # kubectl error bodies may contain Secret material; neither stream escapes. result = subprocess.run(KUBE + args, input=None if value is None else json.dumps(value), capture_output=True, text=True, timeout=30) require(result.returncode == 0, 'kubernetes_operation_failed') if raw: return result.stdout return json.loads(result.stdout) if result.stdout.strip() else None def get(kind, name, namespace): return kube(['-n', namespace, 'get', kind, name, '--ignore-not-found', '-o', 'json']) def metadata(obj): return {k: obj['metadata'][k] for k in ['name', 'namespace', 'uid', 'resourceVersion']} def declarations(): role = yaml.safe_load((ROOT / 'helm/vergabe-demo-company-role.yaml').read_text()) database = yaml.safe_load((ROOT / 'helm/vergabe-demo-company-database.yaml').read_text()) require(role == {'name': NAME, 'ensure': 'present', 'login': True, 'inherit': False, 'superuser': False, 'createdb': False, 'createrole': False, 'connectionLimit': 20, 'passwordSecret': {'name': SECRET}}, 'role_declaration_changed') require(database['metadata'] == {'name': 'vergabe-demo-company-db', 'namespace': 'databases'} and database['spec'] == {'cluster': {'name': 'apps-pg'}, 'name': NAME, 'owner': NAME}, 'database_declaration_changed') return role, database def valid_owned_secret(obj, keys): require(obj['metadata'].get('labels', {}).get(next(iter(OWNER))) == next(iter(OWNER.values())) and not obj['metadata'].get('ownerReferences') and set(obj['data']) == set(keys), 'existing_secret_is_not_exact_owned_bundle') def secret_object(name, namespace, fields): return {'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'name': name, 'namespace': namespace, 'labels': dict(OWNER, **{'cnpg.io/reload': 'true'})}, 'type': 'kubernetes.io/basic-auth' if name == SECRET else 'Opaque', 'stringData': fields} def role_patch(cluster, role): roles = cluster['spec']['managed']['roles'] names = [x['name'] for x in roles] require(len(names) == len(set(names)) and len(roles) <= 3, 'consumer_ceiling_or_duplicate') require(all(x.get('connectionLimit') == 20 for x in roles), 'consumer_limits_changed') if NAME in names: require(roles[names.index(NAME)] == role, 'existing_role_differs') return None require(set(names) == {'vergabe', 'coulomb_social'} and len(roles) == 2, 'unexpected_consumers') return [{'op': 'test', 'path': '/metadata/uid', 'value': cluster['metadata']['uid']}, {'op': 'test', 'path': '/metadata/resourceVersion', 'value': cluster['metadata']['resourceVersion']}, {'op': 'test', 'path': '/spec/managed/roles', 'value': roles}, {'op': 'add', 'path': '/spec/managed/roles/-', 'value': role}] def run(mode, expected_revision): require(kube(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER_UID, 'wrong_cluster') namespace = get('namespace', NS, NS) require(namespace and namespace['metadata']['labels'].get('railiance.io/postgres-client') == 'apps-pg', 'namespace_not_bound') role, database = declarations() cluster = get('cluster', 'apps-pg', 'databases') patch = role_patch(cluster, role) db = get('database', 'vergabe-demo-company-db', 'databases') require(not db or all(db['spec'].get(k) == v for k, v in database['spec'].items()), 'database_resource_conflict') primary = cluster['status']['currentPrimary'] require(primary.startswith('apps-pg-'), 'unexpected_primary') sql = ("SELECT 'role' FROM pg_roles WHERE rolname='vergabe_demo_company'; " "SELECT 'database' FROM pg_database WHERE datname='vergabe_demo_company';") existing = kube(['-n', 'databases', 'exec', primary, '-c', 'postgres', '--', 'psql', '-U', 'postgres', '-d', 'postgres', '-At', '-c', sql], raw=True).splitlines() require('role' not in existing or patch is None, 'unmanaged_existing_role') require('database' not in existing or db is not None, 'unmanaged_existing_database') # These are the two newly named bundles for this lane, never bootstrap or historical secrets. provider = get('secret', SECRET, 'databases') consumer = get('secret', RUNTIME, NS) if provider: valid_owned_secret(provider, ['username', 'password']) require(base64.b64decode(provider['data']['username']).decode() == NAME, 'username_conflict') if consumer: valid_owned_secret(consumer, ['SECRET_KEY', 'DATABASE_URL']) require(provider is not None, 'runtime_without_provider') receipt = {'mode': mode, 'cluster': metadata(cluster), 'managed_consumers': len(cluster['spec']['managed']['roles']), 'role_change_needed': patch is not None, 'database_exists': db is not None, 'provider_exists': provider is not None, 'runtime_exists': consumer is not None, 'values_emitted': False} if mode == 'inspect': return receipt require(expected_revision == cluster['metadata']['resourceVersion'], 'cluster_revision_changed') password = base64.b64decode(provider['data']['password']).decode() if provider else secrets.token_urlsafe(48) uri = f'postgresql://{NAME}:{quote(password, safe="")}@apps-pg-rw.databases.svc.cluster.local:5432/{NAME}' if consumer: require(base64.b64decode(consumer['data']['DATABASE_URL']).decode() == uri, 'runtime_database_binding_conflict') objects = [] if not provider: objects.append(secret_object(SECRET, 'databases', {'username': NAME, 'password': password})) if not consumer: objects.append(secret_object(RUNTIME, NS, {'SECRET_KEY': secrets.token_urlsafe(64), 'DATABASE_URL': uri})) dry = ['--dry-run=server'] if mode == 'dry-run' else [] for obj in objects: result = kube(['create', '-f', '-', '-o', 'json', *dry], obj) expected = {k: base64.b64encode(v.encode()).decode() for k, v in obj['stringData'].items()} require(result['data'] == expected, 'secret_delivery_mismatch') if patch: result = kube(['-n', 'databases', 'patch', 'cluster', 'apps-pg', '--type=json', '--patch-file=/dev/stdin', '-o', 'json', *dry], patch) require(result['spec'] == dict(cluster['spec'], managed=dict(cluster['spec']['managed'], roles=cluster['spec']['managed']['roles'] + [role])), 'unrelated_cluster_spec_changed') if not db: kube(['create', '-f', '-', '-o', 'json', *dry], database) receipt.update(result='dry-run-passed' if mode == 'dry-run' else 'provisioned', provider_secret='databases/' + SECRET, runtime_secret=NS + '/' + RUNTIME, next='Wait for Database applied, then apply only the demo controls SQL and verify consumer login.') return receipt def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('mode', choices=['inspect', 'dry-run', 'apply']) parser.add_argument('--expected-cluster-resource-version') args = parser.parse_args() try: print(json.dumps(run(args.mode, args.expected_cluster_resource_version), sort_keys=True)) return 0 except Refused as error: print(json.dumps({'result': 'refused', 'reason': str(error)})) except Exception: print(json.dumps({'result': 'failed', 'reason': 'contained_operation_failed'})) return 1 if __name__ == '__main__': raise SystemExit(main())