安全上下文需要运行时证据来验证

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'}}}]}

namespace = 'notes-security-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    probe = '''from pathlib import Path
import errno,json,os,subprocess,time
status=dict(line.split(':',1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line)
try:
 Path('/tmp/forbidden-write').write_text('x')
 raise AssertionError('root filesystem unexpectedly writable')
except OSError as error:
 assert error.errno==errno.EROFS
try:
 os.setuid(0)
 raise AssertionError('unexpected UID escalation')
except OSError as error:
 assert error.errno==errno.EPERM
work=Path('/work'); (work/'data').write_text('writable')
(work/'example.py').write_text("print('interpreted from the writable volume')")
ran=subprocess.check_output(['python3',str(work/'example.py')],text=True).strip()
result={'uid':os.getuid(),'gid':os.getgid(),'cap_eff':int(status['CapEff'].strip(),16),
        'no_new_privs':int(status['NoNewPrivs']),'seccomp':int(status['Seccomp']),
        'root_write_errno':errno.EROFS,'uid_zero_denied':True,
        'volume_value':(work/'data').read_text(),'writable_volume_code':ran}
assert result['uid']==result['gid']==10001
assert result['cap_eff']==0 and result['no_new_privs']==1 and result['seccomp']==2
print(json.dumps(result),flush=True); time.sleep(600)
'''
    podspec = spec(probe)
    podspec['securityContext'] = {'runAsNonRoot': True, 'runAsUser': 10001,
        'runAsGroup': 10001, 'fsGroup': 10001, 'seccompProfile': {'type': 'RuntimeDefault'}}
    podspec['volumes'] = [{'name': 'work', 'emptyDir': {}}]
    c = podspec['containers'][0]
    c['securityContext'] = {'allowPrivilegeEscalation': False, 'readOnlyRootFilesystem': True,
                             'capabilities': {'drop': ['ALL']}}
    c['volumeMounts'] = [{'name': 'work', 'mountPath': '/work'}]
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'restricted'}, 'spec': podspec})
    wait_for(lambda: obj(namespace, 'pod', 'restricted'), pod_ready)
    observed = json.loads(wait_for(lambda: k('logs', '-n', namespace, 'restricted').stdout.strip(), bool))
    conflicting = spec()
    conflicting['securityContext'] = {'runAsNonRoot': True, 'runAsUser': 0}
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'conflict'}, 'spec': conflicting})
    refused = wait_for(lambda: obj(namespace, 'pod', 'conflict'), lambda p:
        any(c.get('state', {}).get('waiting', {}).get('reason') == 'CreateContainerConfigError'
            for c in p.get('status', {}).get('containerStatuses', [])))
    waiting = refused['status']['containerStatuses'][0]
    assert 'runAsUser breaks non-root policy' in waiting['state']['waiting']['message']
    assert waiting['restartCount'] == 0
    print(json.dumps({'runtime_observation': observed,
                      'conflicting_uid_accepted_by_api_but_container_start_refused': True,
                      'container_escape_resistance_tested': False}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')