Kubernetes Secret 需要明确的访问边界

import json
import subprocess
import sys
import time
from uuid import uuid4

assert len(sys.argv) == 3, 'usage: python3 example.py DISPOSABLE_KUBECONFIG CONTEXT'
base = ['kubectl', '--kubeconfig', sys.argv[1], '--context', sys.argv[2], '--request-timeout=30s']
image = 'python:3.11-slim-bookworm'


def k(*args, payload=None, expected=0):
    result = subprocess.run(base + list(args), input=json.dumps(payload) if payload is not None else None,
                            capture_output=True, text=True, timeout=150)
    if expected is not None:
        assert result.returncode == expected, result.stderr
    return result


def obj(namespace, kind, name=None, selector=None):
    args = ['get', kind, '-n', namespace, '-o', 'json']
    if name:
        args.append(name)
    if selector:
        args += ['-l', selector]
    return json.loads(k(*args).stdout)


def apply(namespace, data):
    return k('apply', '-n', namespace, '-f', '-', payload=data)


def wait_for(read, ready, seconds=120):
    until = time.monotonic() + seconds
    while time.monotonic() < until:
        value = read()
        if ready(value):
            return value
        time.sleep(0.5)
    raise AssertionError('expected Kubernetes condition did not arrive')


def pod_ready(pod):
    return any(c['type'] == 'Ready' and c['status'] == 'True'
               for c in pod.get('status', {}).get('conditions', []))


def spec(script='import time; time.sleep(600)'):
    return {'automountServiceAccountToken': False, 'terminationGracePeriodSeconds': 1,
            'containers': [{'name': 'app', 'image': image, 'imagePullPolicy': 'IfNotPresent',
                            'command': ['python3', '-c', script],
                            'resources': {'requests': {'cpu': '10m', 'memory': '24Mi'},
                                          'limits': {'cpu': '200m', 'memory': '96Mi'}}}]}

import base64
namespace = 'notes-secrets-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    demo = 'DEMO_ONLY_NOT_A_CREDENTIAL'
    k('create', '-n', namespace, '-f', '-', payload={'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'name': 'example'},
                      'type': 'Opaque', 'stringData': {'needed': demo, 'unused': 'UNUSED_DEMO'}})
    stored = obj(namespace, 'secret', 'example')
    assert base64.b64decode(stored['data']['needed']).decode() == demo
    apply(namespace, {'apiVersion': 'v1', 'kind': 'ServiceAccount', 'metadata': {'name': 'creator'},
                      'automountServiceAccountToken': False})
    apply(namespace, {'apiVersion': 'rbac.authorization.k8s.io/v1', 'kind': 'Role',
        'metadata': {'name': 'pod-creator'},
        'rules': [{'apiGroups': [''], 'resources': ['pods'], 'verbs': ['create', 'get']},
                  {'apiGroups': [''], 'resources': ['pods/log'], 'verbs': ['get']}]})
    apply(namespace, {'apiVersion': 'rbac.authorization.k8s.io/v1', 'kind': 'RoleBinding',
        'metadata': {'name': 'pod-creator'},
        'subjects': [{'kind': 'ServiceAccount', 'name': 'creator', 'namespace': namespace}],
        'roleRef': {'apiGroup': 'rbac.authorization.k8s.io', 'kind': 'Role', 'name': 'pod-creator'}})
    identity = 'system:serviceaccount:' + namespace + ':creator'
    denied = k('--as', identity, 'get', 'secret', 'example', '-n', namespace, expected=None)
    assert denied.returncode != 0 and 'Forbidden' in denied.stderr
    probe = "from pathlib import Path\nimport json,time\nroot=Path('/example')\nprint(json.dumps({'synthetic_value_matches':(root/'one').read_text()=='DEMO_ONLY_NOT_A_CREDENTIAL',\n                  'unused_key_absent':not (root/'unused').exists(),\n                  'service_account_token_absent':not Path('/var/run/secrets/kubernetes.io/serviceaccount/token').exists()}),flush=True)\ntime.sleep(600)"
    podspec = spec(probe)
    podspec['serviceAccountName'] = 'creator'
    podspec['volumes'] = [{'name': 'example', 'secret': {'secretName': 'example',
        'items': [{'key': 'needed', 'path': 'one'}], 'defaultMode': 0o400}}]
    podspec['containers'][0]['volumeMounts'] = [{'name': 'example', 'mountPath': '/example', 'readOnly': True}]
    definition = {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'reader', 'namespace': namespace},
                  'spec': podspec}
    k('--as', identity, 'create', '-f', '-', payload=definition)
    wait_for(lambda: obj(namespace, 'pod', 'reader'), pod_ready)
    output = wait_for(lambda: k('--as', identity, 'logs', '-n', namespace, 'reader').stdout.strip(), bool)
    observed = json.loads(output)
    assert all(observed.values())
    print(json.dumps({'api_base64_decodes_to_synthetic_value': True,
                      'direct_secret_get_forbidden': True, 'same_identity_can_create_consuming_pod': True,
                      'mount_observation': observed, 'etcd_encryption_tested': False}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')