key-cape/scripts/render-upstream-issuer-probe.py

#!/usr/bin/env python3
"""Render a bounded, metadata-only issuer probe for NetKingdom review.

No cluster mutation or secret lookup. The deployment owner creates the Job
first, then re-renders children with its real UID before creating them. This
makes every temporary route/policy/service follow the Job's cleanup lifetime.
"""
import argparse
import base64
import hashlib
import json
import re
import secrets
from pathlib import Path

IMAGE_RE = re.compile(r'forgejo\.coulomb\.social/coulomb/key-cape@sha256:[0-9a-f]{64}')
ISSUERS = ['https://auth.coulomb.social', 'http://auth.coulomb.social',
           'http://authelia.sso.svc.cluster.local:9091']


def render(image, state, job_uid=None):
    if not IMAGE_RE.fullmatch(image):
        raise ValueError('an immutable KeyCape image digest is required')
    if not re.fullmatch(r'[A-Za-z0-9_-]{43}', state):
        raise ValueError('invalid state')
    decoded = base64.urlsafe_b64decode(state + '=')
    if len(decoded) != 32 or base64.urlsafe_b64encode(decoded).decode().rstrip('=') != state:
        raise ValueError('noncanonical state')
    if job_uid is not None and not re.fullmatch(r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}', job_uid):
        raise ValueError('invalid Job UID')
    name = 'keycape-issuer-proof-' + hashlib.sha256(state.encode()).hexdigest()[:12]
    labels = {'keycape.coulomb.social/issuer-proof': name}
    peer = {'podSelector': {'matchLabels': labels}}
    metadata = {'name': name, 'namespace': 'sso', 'labels': labels}
    args = ['probe-upstream-issuer', '--config=/etc/keycape/config.yaml',
            '--listen=:8081', '--lifetime=10m', '--allowed-issuers=' + ','.join(ISSUERS)]
    job = {
        'apiVersion': 'batch/v1', 'kind': 'Job', 'metadata': dict(metadata),
        'spec': {'backoffLimit': 0, 'activeDeadlineSeconds': 600, 'ttlSecondsAfterFinished': 300,
                 'template': {'metadata': {'labels': labels}, 'spec': {
                     'restartPolicy': 'Never', 'automountServiceAccountToken': False,
                     'securityContext': {'runAsNonRoot': True, 'runAsUser': 65534, 'runAsGroup': 65534,
                                         'fsGroup': 65534, 'seccompProfile': {'type': 'RuntimeDefault'}},
                     'containers': [{'name': 'probe', 'image': image, 'args': args,
                                     'env': [{'name': 'KEYCAPE_ISSUER_PROBE_STATE', 'value': state}],
                                     'ports': [{'name': 'probe', 'containerPort': 8081}],
                                     'securityContext': {'allowPrivilegeEscalation': False,
                                                         'readOnlyRootFilesystem': True,
                                                         'capabilities': {'drop': ['ALL']}},
                                     'resources': {'requests': {'cpu': '25m', 'memory': '32Mi'},
                                                   'limits': {'cpu': '200m', 'memory': '128Mi'}},
                                     'readinessProbe': {'httpGet': {'path': '/healthz', 'port': 8081},
                                                        'periodSeconds': 2},
                                     'volumeMounts': [{'name': 'config', 'mountPath': '/etc/keycape', 'readOnly': True}]}],
                     'volumes': [{'name': 'config', 'secret': {'secretName': 'keycape-config',
                                 'items': [{'key': 'config.yaml', 'path': 'config.yaml', 'mode': 288}]}}]
                 }}}}
    children_metadata = dict(metadata)
    if job_uid:
        children_metadata['ownerReferences'] = [{'apiVersion': 'batch/v1', 'kind': 'Job',
             'name': name, 'uid': job_uid, 'blockOwnerDeletion': False}]
    # Backticks belong to Traefik's rule language, not shell execution.
    rule = ('Host(`kc.coulomb.social`) && (Path(`/upstream-issuer-proof/' + state +
            '`) || (Path(`/authorize/callback`) && Query(`state`, `' + state + '`)))')
    children = [
        {'apiVersion': 'v1', 'kind': 'Service', 'metadata': dict(children_metadata),
         'spec': {'selector': labels, 'ports': [{'port': 8081, 'targetPort': 8081}]}},
        {'apiVersion': 'traefik.io/v1alpha1', 'kind': 'IngressRoute', 'metadata': dict(children_metadata),
         'spec': {'entryPoints': ['websecure'], 'tls': {'secretName': 'kc-tls'},
                  'routes': [{'kind': 'Rule', 'match': rule, 'priority': 10000,
                              'middlewares': [{'name': 'keycape-rate-limit'}, {'name': 'keycape-hsts'}],
                              'observability': {'accessLogs': False, 'metrics': False, 'tracing': False},
                              'services': [{'name': name, 'port': 8081}]}]}},
        {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy', 'metadata': dict(children_metadata),
         'spec': {'podSelector': {'matchLabels': labels}, 'policyTypes': ['Ingress', 'Egress'],
                  'ingress': [{'from': [{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': 'kube-system'}},
                                        'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'traefik'}}}],
                               'ports': [{'port': 8081, 'protocol': 'TCP'}]}],
                  'egress': [{'to': [{'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'authelia'}}}],
                              'ports': [{'port': 9091, 'protocol': 'TCP'}]},
                             {'to': [{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': 'kube-system'}},
                                      'podSelector': {'matchLabels': {'k8s-app': 'kube-dns'}}}],
                              'ports': [{'port': 53, 'protocol': protocol} for protocol in ['TCP', 'UDP']]}]}},
        {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy',
         'metadata': dict(children_metadata, name=name + '-authelia'),
         'spec': {'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'authelia'}},
                  'policyTypes': ['Ingress'], 'ingress': [{'from': [peer], 'ports': [{'port': 9091, 'protocol': 'TCP'}]}]}}
    ]
    return {'name': name, 'start_url': 'https://kc.coulomb.social/upstream-issuer-proof/' + state,
            'job': job, 'children': {'apiVersion': 'v1', 'kind': 'List', 'items': children}}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--image', required=True)
    parser.add_argument('--state', help='reuse the exact random state from the first rendering')
    parser.add_argument('--job-uid', help='actual UID returned by creating the reviewed Job')
    parser.add_argument('--output', type=Path, required=True, help='new private packet directory')
    args = parser.parse_args()
    state = args.state or secrets.token_urlsafe(32)
    packet = render(args.image, state, args.job_uid)
    args.output.mkdir(mode=0o700, parents=False, exist_ok=False)
    for name, data in [('job.json', packet['job']), ('children.json', packet['children']),
                       ('packet.json', {'state': state, 'image': args.image, 'job_uid': args.job_uid,
                                        'name': packet['name'], 'start_url': packet['start_url'],
                                        'children_bound_to_job': args.job_uid is not None})]:
        dest = args.output / name
        with dest.open('x') as stream:
            dest.chmod(0o600)
            json.dump(data, stream, indent=2)
            stream.write('\n')
    print(json.dumps({'packet_directory': str(args.output), 'job_name': packet['name'],
                      'children_bound_to_job': args.job_uid is not None, 'cluster_changed': False}))


if __name__ == '__main__':
    main()