LEARN. PRACTICE. GROW.From accepted configuration to available service
On a fictional funds-processing platform, the security team requests Restricted before a rollout. The project team must preserve closing operations and hand RUN a recoverable configuration. API-server acceptance of a Deployment confirms an operation on that object. Pod creation, container startup, readiness and the business transaction still follow. Draw this sequence and associate evidence with each stage: admission result, ReplicaSet conditions, container state and integration test. FailedCreate mentioning PodSecurity belongs to creation, rather than diagnosis of an application that has already started. Begin with the message and the actual applied template. Compare the named field with the approved requirement, including init containers. If correction changes an image or privileges, involve application and security owners. Define the investigation deadline and contingency configuration before rollout. Healthy old replicas may temporarily sustain business processing, but they do not establish that the system can replace them after failure. Include recreation capability in acceptance criteria. In a meeting, distinguish the service currently running from the configuration the team has proved it can deploy again.
Apply and version Pod Security
Enforce, warn and audit serve different purposes. In a namespace using enforce=baseline and warn=restricted, a Pod can be admitted while producing a useful migration warning. That warning is not a pending approval. Raising enforce to Restricted does not automatically evict existing Pods. A test should create a new Pod from the template instead of relying only on old Pods remaining Running. Policy applied to controller requests also explains why accepting a Deployment does not guarantee admission of the resulting Pods. Pin policy versions when controlled comparison is needed. The lab uses enforce-version=v1.35 and warn-version=v1.35 on a 1.37.0 server. A label selects policy controls; it does not change the server binary. Review both Pod-level and individual container fields: a valid inherited setting can be replaced by an incompatible override. On Linux, assess runAsNonRoot, allowPrivilegeEscalation, capabilities and seccomp alongside inherited Baseline controls. An auxiliary container also participates in evaluation. Document exemptions by identity, namespace or RuntimeClass and who can change them. A broad controller exemption can let many authors bypass policy beyond the release under investigation.
Separate tenants and select runtimes
Begin with the trust model. Two internal teams sharing change processes may have different needs from customers running untrusted code. Namespaces help organize objects and policies; an isolation conclusion requires assessment of API access, networking, resources, storage and execution. For every boundary, write an allowed action and one that should be denied. Quotas address consumption and do not replace authorization. A NetworkPolicy stored in the API needs a networking implementation that enforces it. This lab does not test that layer. RuntimeClass selects a handler already configured in the CRI. Creating an object named sandboxed does not install a sandbox. In a heterogeneous pool, connect selection to prepared nodes and confirm execution under the expected runtime. Pod and RuntimeClass nodeSelectors combine requirements; contradictory values are not alternatives. Tolerations address eligibility against taints without fixing conflicting selectors. Consider overhead and representative measurements before promising capacity or latency. Sound acceptance identifies the running technology, mitigated risk, update ownership and recovery behavior. Passing Restricted reduces certain dangerous configurations but does not prove absence of kernel vulnerabilities or a complete tenant boundary.
Deliver, consume and rotate Secrets
Divide rotation into four moments: the object is updated, the value reaches the consumer, the process starts using it and the previous version can be retired. A successful GET covers only part of this path. A regular volume receives eventually consistent updates. A subPath mount does not receive those automatic updates. An existing environment variable retains its startup value. Even after a file changes, an application that read it once may keep the earlier value in memory. Define reload or recreation and test actual authentication without printing credentials in logs. Deliver data only to the container that needs it. An observer without the mount or variable has less direct exposure, but this does not establish strong isolation between untrusted code inside one Pod. Review indirect permissions, such as creating workloads that consume Secrets. For immutable objects, use a replacement strategy with controlled references and retirement planning. When a required key is missing, investigate name, namespace and the release contract before making the reference optional. That change may simply move a clear startup failure into a less visible business-processing error.
Protect storage and traffic
Encrypting stored objects does not automatically solve how applications use credentials. A configuration encrypting new writes does not establish migration of older objects. Before removing a key or the ability to read old records, validate rewriting and readability, API-server consistency and backup recovery. Retain recovery material according to the approved plan. This lesson’s lab does not inspect etcd or rotate encryption keys; these decisions are documentation-based preparation. For Pod-to-Pod traffic, distinguish encrypted connection, authenticated identity and authorized operation. In an Istio sidecar scenario, PERMISSIVE admits plaintext clients during migration. Inventory those clients before imposing STRICT and define positive and negative tests against the correct destination. A portLevelMtls exception applies to the selected workload port, which can differ from the Service’s exposed port. A valid mTLS request does not automatically grant business authorization. Hand RUN the allowed identities, effective policy, failure signals and rotation or recovery procedure. Do not use HTTP 200 as the sole encryption evidence or confuse a DNS-caused failure with security enforcement. The exercise here does not install or validate a service mesh.
Prepare practice and predict outcomes
Save the code as run.py. Use a disposable kind cluster whose name starts with dr-cks-workload-, a dedicated kubeconfig, kubectl client 1.37.1 and server 1.37.0. The script refuses a different context and requires a local HTTPS API with active verification. The Python image must already be imported and referenced by digest on the node; imagePullPolicy=Never prevents the exercise from fetching a different image. The laboratory README describes ARM64 import and the local reference used in the trials. Do not reuse production credentials or namespaces. Run python3 run.py --kubectl /path/kubectl --kubeconfig /path/kubeconfig --cluster NAME --image DIGEST_REFERENCE --output evidence.json. Choose a new report. The script creates a random namespace, configures only its policy and uses explicitly synthetic values. Before execution, predict five results: an incompatible Pod admitted with a warning, direct creation denied after enforcement, an accepted Deployment with ReplicaSet denial, an updated volume and consumers retaining initial values. Write what evidence distinguishes each from an image, scheduler or application failure. Policy is pinned to v1.35; the server remains 1.37.0. Results do not prove complete equivalence to the official exam.
Interpret the 17 observations
The first group compares admission and execution. The old Pod remains Running after the policy change; a new incompatible request is denied. The Deployment passes the API, while its controller records FailedCreate. A corrected Pod starts with UID 1000, NoNewPrivs set and active seccomp; its ServiceAccount token is not mounted. These results demonstrate the observed requests and processes. They do not establish acceptance of every image, init container or additional configuration. The second group compares delivery and consumption. The worker begins with the same synthetic value in its volume, subPath, environment and cache. The observer receives neither the mount nor the variable. After patching, the script waits for volume change within a bounded deadline and confirms that the other three forms retain initial values. Measured delay is an observation of this trial, not an SLA. Finally, the API rejects immutable data changes and reversal of immutable; a missing required key prevents another container starting. The report retains 17 results and the code hash. Namespace and temporary files are removed. Remove the disposable cluster and its dedicated kubeconfig when closing the exercise.
Decide completion and hand over to RUN
In the first case, a closing release has 20 minutes to decide contingency. The main signal is admission denial despite traffic on old replicas. In the second, the earlier credential expires in 30 minutes and consumers use different delivery mechanisms. Both decisions need observable criteria, owners and a deadline. If evidence remains insufficient within the window, apply approved contingency and preserve investigation material. A broad exception can restore functionality while invalidating the change’s security requirement. Hand RUN a matrix of consumers, versions and controls: who creates Pods, which policy applies, how each process receives credentials, how it reloads, what should be denied and who resolves each failure. Include recreation capability in continuity testing. Separate this laboratory’s results from practice still needed: CNI, sandboxing, service mesh, etcd encryption and complete recovery. Scenarios are original and fictional and do not represent internal BNP Paribas procedures. The final summary should let a colleague explain the difference between an accepted object, a functioning process and a proven security requirement using course evidence. Keep the evidence scope visible when presenting the change to technical and business stakeholders.
"""Original CKS admission and synthetic Secret-delivery lab on a disposable kind cluster."""
import argparse,datetime,hashlib,json,shutil,subprocess,tempfile,time,uuid
from pathlib import Path
p=argparse.ArgumentParser;p.add_argument('--kubectl',required=True);p.add_argument('--kubeconfig',required=True);p.add_argument('--cluster',required=True);p.add_argument('--image',required=True);p.add_argument('--output',required=True);a=p.parse_args
assert '@sha256:' in a.image;assert a.cluster.startswith('dr-cks-workload-');assert Path(a.kubeconfig).is_absolute;out=Path(a.output);assert not out.exists
ns='dr-workload-'+uuid.uuid4.hex[:8];tmp=Path(tempfile.mkdtemp(prefix='dr-workload-private-'));records=[];created=False
base=[a.kubectl,'--kubeconfig',a.kubeconfig,'--context','kind-'+a.cluster,'--cache-dir',str(tmp/'cache'),'--request-timeout=15s']
def k(*args,obj=None,ok=True):
r=subprocess.run(base+list(args),input=json.dumps(obj)if obj is not None else None,capture_output=True,text=True,timeout=35)
if ok and r.returncode:raise RuntimeError(str(args[:3])+': '+r.stderr[:500])
return r
def check(name,condition,observed):
assert condition,(name,observed);records.append(dict(name=name,passed=True,observed=observed));print(name,flush=True)
def get(kind,name=None):return json.loads(k('-n',ns,'get',kind,*([name]if name else []),'-o','json').stdout)
def poll(fn,seconds=90):
end=time.monotonic+seconds
while time.monotonic<end:
value=fn
if value:return value
time.sleep(1)
raise AssertionError('Condition deadline exceeded')
def create(obj,ok=True):return k('-n',ns,'create','-f','-',obj=obj,ok=ok)
def pod(name,restricted=True):
return dict(apiVersion='v1',kind='Pod',metadata=dict(name=name,labels={'app':name}),spec=dict(automountServiceAccountToken=False,terminationGracePeriodSeconds=1,securityContext=dict(runAsNonRoot=True,runAsUser=1000,runAsGroup=1000,seccompProfile={'type':'RuntimeDefault'}),containers=[dict(name='worker',image=a.image,imagePullPolicy='Never',command=['python','-c','import time; time.sleep(600)'],resources=dict(requests={'cpu':'10m','memory':'24Mi'},limits={'cpu':'200m','memory':'64Mi'}),securityContext=dict(allowPrivilegeEscalation=not restricted,readOnlyRootFilesystem=True,capabilities={'drop':['ALL']}))]))
def execpy(name,code,container='worker'):return json.loads(k('-n',ns,'exec',name,'-c',container,'--','python','-c',code).stdout)
try:
cfg=json.loads(k('config','view','--minify','--raw','-o','json').stdout)['clusters'][0]['cluster'];assert cfg['server'].startswith('https://127.0.0.1:')and not cfg.get('insecure-skip-tls-verify');cfg=None
version=json.loads(k('version','-o','json').stdout);assert version['serverVersion']['gitVersion']=='v1.37.0'and version['clientVersion']['gitVersion']=='v1.37.1'
k('create','namespace',ns);created=True
k('label','namespace',ns,'pod-security.kubernetes.io/enforce=baseline','pod-security.kubernetes.io/enforce-version=v1.35','pod-security.kubernetes.io/warn=restricted','pod-security.kubernetes.io/warn-version=v1.35')
legacy=pod('legacy',False);r=create(legacy);check('baseline-admits-with-restricted-warning','would violate PodSecurity' in r.stderr and 'allowPrivilegeEscalation' in r.stderr,dict(admitted=True,restrictedWarning=True))
poll(lambda:get('pod','legacy').get('status',{}).get('phase')=='Running');uid=get('pod','legacy')['metadata']['uid']
r=k('label','namespace',ns,'pod-security.kubernetes.io/enforce=restricted','--overwrite');check('enforce-update-warns-existing-pod','legacy' in r.stderr and 'allowPrivilegeEscalation'in r.stderr,dict(existingViolationReported=True))
current=get('pod','legacy');check('existing-pod-not-evicted',current['metadata']['uid']==uid and current['status']['phase']=='Running',dict(sameUID=True,phase=current['status']['phase']))
r=create(pod('denied',False),ok=False);check('new-violating-pod-rejected',r.returncode!=0 and 'Forbidden'in r.stderr and 'allowPrivilegeEscalation'in r.stderr,dict(rejected=True,control='allowPrivilegeEscalation'))
bad=pod('blocked-release',False);deployment=dict(apiVersion='apps/v1',kind='Deployment',metadata=dict(name='blocked-release'),spec=dict(replicas=1,selector={'matchLabels':{'app':'blocked-release'}},template={'metadata':{'labels':{'app':'blocked-release'}},'spec':bad['spec']}))
r=create(deployment);check('deployment-object-admitted','would violate PodSecurity'in r.stderr,dict(admitted=True,warning=True))
def controller_denial:
for rs in get('replicasets')['items']:
if rs['metadata'].get('labels',{}).get('app')=='blocked-release':
for c in rs.get('status',{}).get('conditions',[]):
if c.get('reason')=='FailedCreate' and 'violates PodSecurity'in c.get('message',''):return c
condition=poll(controller_denial);check('controller-pod-create-denied',condition['status']=='True',dict(type=condition['type'],reason=condition['reason'],policyDenial=True))
k('-n',ns,'delete','deployment','blocked-release','--wait=true')
corrected=pod('corrected');create(corrected);poll(lambda:get('pod','corrected').get('status',{}).get('phase')=='Running')
runtime=execpy('corrected',"import os,json;from pathlib import Path;s=dict(x.split(':',1)for x in Path('/proc/self/status').read_text.splitlines);print(json.dumps({'uid':os.getuid,'nnp':s['NoNewPrivs'].strip,'seccomp':s['Seccomp'].strip,'tokenMounted':Path('/var/run/secrets/kubernetes.io/serviceaccount/token').exists}))")
check('corrected-pod-runs-restricted',runtime['uid']==1000 and runtime['nnp']=='1' and runtime['seccomp']=='2',runtime)
check('serviceaccount-token-not-mounted',not runtime['tokenMounted'],dict(tokenMounted=False))
create(dict(apiVersion='v1',kind='Secret',metadata=dict(name='rotation'),stringData={'value':'synthetic-v1'}))
consumer=pod('consumer');c=consumer['spec']['containers'][0];c['env']=[dict(name='DR_VALUE',valueFrom={'secretKeyRef':{'name':'rotation','key':'value'}})];c['volumeMounts']=[dict(name='secret',mountPath='/rotating',readOnly=True),dict(name='secret',mountPath='/frozen',subPath='value',readOnly=True),dict(name='scratch',mountPath='/tmp')]
c['command']=['python','-c',"import json,os,time;from pathlib import Path;Path('/tmp/startup.json').write_text(json.dumps({'env':os.environ['DR_VALUE'],'cached':Path('/rotating/value').read_text}));time.sleep(600)"]
consumer['spec']['volumes']=[dict(name='secret',secret={'secretName':'rotation'}),dict(name='scratch',emptyDir={})]
observer=pod('x')['spec']['containers'][0];observer['name']='observer'consumer['spec']['containers'].append(observer);create(consumer)
poll(lambda:all(x.get('ready')for x in get('pod','consumer').get('status',{}).get('containerStatuses',[]))and len(get('pod','consumer').get('status',{}).get('containerStatuses',[]))==2)
inspect="import json,os;from pathlib import Path;d=json.loads(Path('/tmp/startup.json').read_text);d.update(volume=Path('/rotating/value').read_text,subPath=Path('/frozen').read_text);print(json.dumps(d))"
before=execpy('consumer',inspect);check('three-delivery-paths-initialized',all(v=='synthetic-v1'for v in before.values),dict(paths=list(before),allInitial=True))
other=execpy('consumer',"import json,os;from pathlib import Path;print(json.dumps({'filePresent':Path('/rotating/value').exists,'envPresent':'DR_VALUE'in os.environ}))",'observer');check('mount-scoped-to-consumer',not other['filePresent']and not other['envPresent'],other)
k('-n',ns,'patch','secret','rotation','--type=merge','-p',json.dumps({'stringData':{'value':'synthetic-v2'}}))
start=time.monotonic;after=poll(lambda:(lambda x:x if x['volume']=='synthetic-v2'else None)(execpy('consumer',inspect)),180)
check('regular-volume-eventually-updated',after['volume']=='synthetic-v2',dict(updated=True,observedDelaySeconds=round(time.monotonic-start,2)))
check('environment-retains-initial-value',after['env']=='synthetic-v1',dict(retainedInitial=True))
check('subpath-retains-initial-value',after['subPath']=='synthetic-v1',dict(retainedInitial=True))
check('application-cache-retains-initial-value',after['cached']=='synthetic-v1',dict(retainedInitial=True))
create(dict(apiVersion='v1',kind='Secret',metadata=dict(name='fixed'),immutable=True,stringData={'value':'synthetic-fixed'}))
r=k('-n',ns,'patch','secret','fixed','--type=merge','-p',json.dumps({'stringData':{'value':'synthetic-new'}}),ok=False);check('immutable-data-change-rejected',r.returncode!=0 and 'immutable'in r.stderr,dict(rejected=True))
r=k('-n',ns,'patch','secret','fixed','--type=merge','-p','{"immutable":false}',ok=False);check('immutable-reversal-rejected',r.returncode!=0 and 'immutable'in r.stderr,dict(rejected=True))
missing=pod('missing-key');missing['spec']['containers'][0]['env']=[dict(name='DR_VALUE',valueFrom={'secretKeyRef':{'name':'rotation','key':'absent'}})];create(missing)
def missing_reason:
statuses=get('pod','missing-key').get('status',{}).get('containerStatuses',[])
return next((x.get('state',{}).get('waiting')for x in statuses if x.get('state',{}).get('waiting',{}).get('reason')=='CreateContainerConfigError'),None)
error=poll(missing_reason);check('missing-required-key-blocks-start',error['reason']=='CreateContainerConfigError',dict(reason=error['reason'],missingKey='absent'))
images={x['name']:x['imageID']for x in get('pod','consumer')['status']['containerStatuses']};assert len(records)==17,len(records)
finally:
if created:k('delete','namespace',ns,'--wait=true','--timeout=90s')
shutil.rmtree(tmp)
report=dict(executedAt=datetime.datetime.now(datetime.timezone.utc).isoformat,scriptSha256=hashlib.sha256(Path(__file__).read_bytes).hexdigest,version=version,examVersion='1.35',policyVersion='v1.35',versionDifferenceExplicit=True,images=images,observations=records,cleanup=dict(namespaceRemoved=True,temporaryMaterialRemoved=not tmp.exists,realSecretsUsed=False),scope='Actual Pod Security admission, controller rejection, running Linux Pods and synthetic Secret delivery on Kubernetes1.37.0 with policy pinned v1.35. No CNI isolation, service mesh, sandboxed runtime, encryption-at-rest, tenant security boundary or full practical mock demonstrated.')
out.write_text(json.dumps(report,indent=2)+'\n');print('PASS',len(records))
A Deployment is accepted but its ReplicaSet cannot create Pods; a Secret changes in the volume while the process retains the old version.
Common pitfalls
Confusing apply with availability, projection with reload or namespaces with complete isolation; using global exemptions without assessing scope.
Related topics: Identities and delegation · Linux hardening · Supply-chain security
Useful acceptance follows each boundary: request, process, consumption and business function, with allowed and denied tests.
Reference: CKS certification and domains · Kubernetes v1.35; current six-domain CKS outline