LEARN. PRACTICE. GROW.1. Define what was contained
At 17:20, the security team finds a dashboard credential in a diagnostic artifact. Fund closing is at 17:30. APS removes a RoleBinding and receives 403 on a fresh request. That evidence is useful, but “the account is blocked” remains imprecise. Identify the tested credential, operation, namespace, object, endpoint and time. This fictional case does not describe a bank’s internal procedures. Build a matrix with four questions: does the server recognize the identity; does it allow the new operation; does it keep delivering data on an earlier connection; are copies held outside the API? Each row needs its own evidence. In this module, positive TokenReview answers the first question, a fresh GET or watch the second, and a marker created after the change the third. The fourth requires investigating clients and export destinations outside this lab. Before making changes, record the minimum state needed to explain the incident without placing the token in the ticket. Pair an allowed-operation test with one that must fail. Repeat both after the change using the correct identity. An administrator successfully reading the object demonstrates availability for that administrator, not absence of attacker access. End the update with a bounded conclusion: “fresh GET requests with this credential to this object were denied at 17:23; the earlier stream still requires containment”. Assign an owner to each unresolved row.
2. Withdraw grants without accidentally restoring them
An RBAC change must account for whoever manages the object. In the dashboard example, the removed RoleBinding returns and the audit log identifies a GitOps controller. Repeated deletion leaves a race between the operator and reconciler. Correct the declarative source, coordinate reconciliation and inspect the effective object. Document the temporary exception so the next synchronization does not restore a compromised grant. Kubernetes default roles have a different mechanism: the API server can restore missing rules and subjects at startup. Before changing objects managed by the control plane, identify their purpose and dependencies. An annotation disabling auto-reconciliation is not a universal hardening recipe. A role used by essential components may need rights that appear broad when read without context. Test intended behavior and component continuity within an approved change. During fund closing, recovery does not require returning the original role to the exposed account. Define the specific status read, assign it to a clean identity and limit its scope. Record an owner, deadline and evidence of later removal. If access must temporarily expand, state the need and identify who accepts the risk. A grant “only during the incident” without a verifiable deadline tends to become permanent. Also check whether another binding or another authorizer continues to allow the operation you intended to block. Preserve the evidence before declaring the grant fully withdrawn.
3. Choose the unit of revocation
The exercise creates two tokens for the same ServiceAccount: one bound to a control Secret and one without an object binding. The Secret contains only synthetic text; it is not where we store the token. Deleting that object causes the bound token to be rejected, while the other remains usable once permissions return. An identical account name does not make every credential’s dependencies identical. Next, we recreate the Secret with the same name. The old token remains rejected because the new object has a different UID. Issuing a token bound to the new instance allows legitimate access to be tested again. Finally, deleting the ServiceAccount itself affects both token types. Choosing between invalidating one binding and withdrawing the whole identity depends on inventory and impact. The team should know which processes share the account before applying a broad measure. Disabling automount reduces automatic mounting into Pods; it neither removes an already issued credential nor prevents an authorized issuer requesting another. Pod mounting has separate rules and was not exercised in this lab. We also did not measure the time allowance for objects pending deletion with finalizers. The documentation addresses that condition separately. Tests wait for propagation and check each credential individually within a deadline instead of announcing universal instantaneous revocation. A passing probe for one token is not substituted for a probe of the other token.
4. Prepare and run the isolated exercise
Save this lesson’s code as run.py. You need Python 3, kind, Docker and a compatible kubectl. Use an exclusive disposable cluster whose name starts with dr-cks-identity-response- and a separate kubeconfig with mode 0600. The reference execution used kind 0.33.0, node image kindest/node:v1.37.0@sha256:a1ed56cfb0e7b93589bdf97c8cd566405a265939e3620fc4f5de89adff580ae5 and kubectl 1.37.1. The official exam states runtime 1.35; this difference is explicit and the script rejects versions other than those checked. Do not point it at your workstation’s usual context. Create the cluster using kind create cluster --name dr-cks-identity-response-study --image IMAGE --kubeconfig /absolute/path/kubeconfig, replacing IMAGE with the identifier above. Then pass python3 run.py the arguments --kubectl /absolute/path/kubectl, --kubeconfig /absolute/path/kubeconfig, --cluster dr-cks-identity-response-study and --output /new/path/result.json. The result file must not already exist. Confirm versions before execution and do not change assertions merely to obtain a green result. The script creates a random namespace, account, Secret, ConfigMap and limited RBAC. Tokens remain in memory and stdin, with no credential-bearing command arguments. Complete TokenReview responses are not published. The finally block removes the namespace and private cache. When finished, delete only the dedicated cluster with kind delete cluster --name dr-cks-identity-response-study and remove its kubeconfig. If execution fails, inspect the cause and cleanup before repeating with a new result filename.
5. Interpret the stream and result sequence
Predict the results on paper before execution. First, a token authenticates but cannot read. After the binding is created, reading succeeds. A watch opens from the observed resourceVersion and an initial marker is published. After RBAC withdrawal, fresh GET and watch requests fail; a second marker created after those denials arrives over the already open stream. Deleting the bound Secret makes new requests return 401, but a third marker still arrives over that same stream. This sequence distinguishes a new event from older data already held in a buffer. The code does more than check whether a client process exists. It observes delivery of the expected synthetic value after each change. The watch has a bounded timeoutSeconds; its reader ends before the response object is closed. We did not execute forced server-side termination or demonstrate a universal procedure for proxies, load balancers or different versions. The report contains 23 observations, versions, the script hash and propagation-probe timings. Compare the hash against your executed code before associating evidence with the lesson. If you obtain a different result, record configuration, version and exact ordering rather than hiding it. In a real environment, response may require terminating connections through supported mechanisms or applying broader authorized containment. Acceptance needs to demonstrate that delivery ended and that the legitimate recovery flow remains available. In both reference runs, the probe waiting for 401 after Secret deletion took about 9.8 seconds. This is a measurement of this environment, not an SLA or a universal cache or revocation rule. The script’s 15-second deadline is an exercise limit.
6. Retain evidence without exposing the credential again
TokenReview accepts a credential and returns a verdict; its response can repeat spec.token. Publishing the complete JSON in a ticket, log or report recreates the problem you are resolving. The lab retains only the authentication boolean. An operational procedure should add approved non-secret identifiers, time, tested audience and conclusion, with appropriate access and retention. Do not use public JWT decoders to investigate real tokens. Analyzing a JWT raises two different questions: are the signed bytes valid for this recipient, and is the object binding still valid now? Offline verification can answer the first but does not automatically query current cluster state. TokenReview lets you consult the API server. If a service depends on that state when accepting calls, define audience, unavailability handling and cache lifetime. An older cached result is not equivalent to a check performed at this instant. This exercise does not implement an offline cryptographic verifier. Reading or decoding claims alone does not validate a signature. It also does not allow old reviews to be retrieved with kubectl get: TokenReview is not stored as a normal queryable resource. For the exercise, submit a result table without tokens, identify negative checks and explain what each one proves. A screenshot omitting identity, operation or timing can look convincing while remaining insufficient to close the incident. The evidence should let another operator understand the boundary of the claim.
7. Handle external authentication and API exposure
The exercised mechanism uses cluster-local ServiceAccounts. Do not generalize its results to a human user’s login. In an OIDC model with locally validated JWTs, signing out of the IdP does not necessarily notify the API server on each request. An already issued token may still contain a group removed from the directory. Before claiming revocation, confirm the integration model, remaining validity and requests made with the targeted credential. This lab runs no IdP and proves no particular provider behavior. Also distinguish a health route from a data route. In the disposable cluster, /livez without credentials returns 200 while an anonymous ConfigMap read receives 403. We do not infer that every cluster should expose the same routes. In v1.35, AuthenticationConfiguration can restrict which paths accept anonymous authentication. That design must be combined with authorization, load-balancer needs and negative tests of other routes. Imagine a change that indiscriminately closes anonymous access and removes every API server from the load balancer. Security intent does not remove operational impact. First define the probe, its source, identity and path. Test the change within a controlled scope and prepare approved recovery. The acceptance report should cover both resource protection and management-path health, because losing that path can also prevent incident response. Identify who owns the probe configuration and how they confirm successful recovery.
8. Upgrade, recover and hand over to RUN
Responding to vulnerabilities requires confirming the affected component, version, exploitation conditions and fix identified by the official source. A version list alone demonstrates neither exposure nor change effectiveness. Connect each action to the concrete risk and avoid inventing a CVE to justify the exercise. This lesson’s scenarios use hypothetical transitions; we did not execute a Kubernetes upgrade. Plan supported ordering, including every API server each consumer can reach. In an HA state with 1.34 and 1.35, a 1.35 kubelet would be newer than the still-reachable 1.34 server. Tolerance for older components does not authorize reversing that relationship. Do not skip API server minors, and check additional tool or provider restrictions. Webhooks and clients belong in acceptance alongside the version displayed by kubectl version. Hand RUN a matrix containing previous state, change, expected result, observed result, owner and limitation. Include old credentials, a clean recovery credential, fresh GET, fresh watch and the earlier stream. Add the approved recovery destination, exception deadline and business confirmation. The final assessment should require justified decisions: what evidence is missing, which action belongs to the team, and what has not been established? Summarize the principle: containing an identity requires observing credentials, rights, connections and already obtained data. These four related controls are not interchangeable, and each needs closure criteria. Retain those criteria in the runbook so the next shift can continue without reconstructing assumptions.
"""Original CKS lab: permission withdrawal, bound-token invalidation and watches.
Use only a dedicated kind cluster. Tokens stay in process memory and stdin;
the report contains status codes and synthetic object observations, not tokens.
"""
import argparse
import base64
import datetime
import hashlib
import json
import queue
import shutil
import ssl
import subprocess
import tempfile
import threading
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path
p = argparse.ArgumentParser
for name in ('kubectl', 'kubeconfig', 'cluster', 'output'):
p.add_argument('--' + name, required=True)
a = p.parse_args
assert a.cluster.startswith('dr-cks-identity-response-')
assert Path(a.kubeconfig).is_absolute
out = Path(a.output)
assert not out.exists
private = Path(tempfile.mkdtemp(prefix='dr-cks-identity-response-private-'))
ns = 'dr-response-' + uuid.uuid4.hex[:8]
base = [a.kubectl, '--kubeconfig', a.kubeconfig, '--context', 'kind-' + a.cluster,
'--cache-dir', str(private / 'cache'), '--request-timeout=15s']
tokens, records, streams, settlements = {}, [], [], []
created = False
def k(*args, obj=None):
result = subprocess.run(base + list(args), text=True, capture_output=True,
input=json.dumps(obj) if obj is not None else None, timeout=30)
if result.returncode:
# Do not echo input or full output from a token-bearing API operation.
raise RuntimeError('Admin operation failed: ' + str(args[:3]))
return result.stdout
def resource(kind, name, **fields):
group = 'rbac.authorization.k8s.io/v1' if kind in ('Role', 'RoleBinding') else 'v1'
return dict(apiVersion=group, kind=kind, metadata=dict(name=name), **fields)
def create(obj):
return json.loads(k('-n', ns, 'create', '-f', '-', '-o', 'json', obj=obj))
def binding:
return resource('RoleBinding', 'read-status',
roleRef=dict(apiGroup='rbac.authorization.k8s.io', kind='Role', name='read-status'),
subjects=[dict(kind='ServiceAccount', name='reporter', namespace=ns)])
def token(name, secret=None):
args = ['-n', ns, 'create', 'token', 'reporter', '--duration=10m']
if secret:
args += ['--bound-object-kind=Secret', '--bound-object-name=token-anchor',
'--bound-object-uid=' + secret]
tokens[name] = k(*args).strip
def request(who, path):
headers = {'Authorization': 'Bearer ' + tokens[who]} if who else {}
req = urllib.request.Request(server + path, headers=headers)
try:
with urllib.request.urlopen(req, context=ctx, timeout=15) as response:
return response.status, response.read
except urllib.error.HTTPError as error:
return error.code, error.read
def record(name, observed, expected):
passed = observed == expected
records.append(dict(name=name, observed=observed, expected=expected, passed=passed))
print(name, 'PASS' if passed else 'FAIL', flush=True)
assert passed, name
def check(name, who, path, expected):
status, _ = request(who, path)
record(name, status, expected)
def settle(who, path, expected):
started = time.monotonic
deadline = started + 15
while time.monotonic < deadline:
if request(who, path)[0] == expected:
settlements.append(dict(credentialLabel=who, expected=expected, elapsedSeconds=round(time.monotonic-started, 3)))
return
time.sleep(0.2)
raise AssertionError('Propagation deadline exceeded')
def review(name, who, expected):
result = json.loads(k('create', '-f', '-', '-o', 'json', obj={
'apiVersion': 'authentication.k8s.io/v1', 'kind': 'TokenReview',
'spec': {'token': tokens[who]}}))
# The response echoes spec.token. Retain only the authentication verdict.
record(name, result.get('status', {}).get('authenticated', False), expected)
def start_watch(who, resource_version):
events = queue.Queue
path = watch_path + '&resourceVersion=' + resource_version
req = urllib.request.Request(server + path, headers={'Authorization': 'Bearer ' + tokens[who]})
stream = urllib.request.urlopen(req, context=ctx, timeout=30)
streams.append(stream)
def reader:
try:
for line in stream:
event = json.loads(line)
obj = event.get('object', {})
if event.get('type') == 'MODIFIED':
events.put(obj.get('data', {}).get('state'))
except (OSError, ValueError):
pass
thread = threading.Thread(target=reader, daemon=True)
thread.start
return events, thread
def update_state(state):
k('-n', ns, 'patch', 'configmap', 'closing-status', '--type=merge',
'-p', json.dumps({'data': {'state': state}}))
try:
cfg = json.loads(k('config', 'view', '--minify', '--raw', '-o', 'json'))
cluster = cfg['clusters'][0]['cluster']
server = cluster['server']
assert server.startswith('https://127.0.0.1:')
assert not cluster.get('insecure-skip-tls-verify')
ctx = ssl.create_default_context(cadata=base64.b64decode(cluster['certificate-authority-data']).decode)
cfg = None
versions = json.loads(k('version', '-o', 'json'))
assert versions['serverVersion']['gitVersion'] == 'v1.37.0'
assert versions['clientVersion']['gitVersion'] == 'v1.37.1'
k('create', 'namespace', ns)
created = True
create(resource('ServiceAccount', 'reporter', automountServiceAccountToken=False))
anchor = create(resource('Secret', 'token-anchor', type='Opaque', stringData={'purpose': 'synthetic-revocation-anchor'}))
first_uid = anchor['metadata']['uid']
token('bound', first_uid)
token('unbound')
create(resource('ConfigMap', 'closing-status', data={'state': 'initial'}))
core = '/api/v1/namespaces/' + ns
target = core + '/configmaps/closing-status'
watch_path = core + '/configmaps?watch=true&timeoutSeconds=25&fieldSelector=metadata.name%3Dclosing-status'
review('explicit-token-valid-with-automount-disabled', 'bound', True)
check('valid-identity-without-permission', 'bound', target, 403)
create(resource('Role', 'read-status', rules=[dict(apiGroups=[''], resources=['configmaps'],
resourceNames=['closing-status'], verbs=['get', 'watch'])]))
create(binding)
settle('bound', target, 200)
check('bound-token-permitted', 'bound', target, 200)
check('unbound-token-same-permission', 'unbound', target, 200)
rv = json.loads(request('bound', target)[1])['metadata']['resourceVersion']
events, thread = start_watch('bound', rv)
update_state('before-revocation')
record('watch-receives-before-revocation', events.get(timeout=10), 'before-revocation')
k('-n', ns, 'delete', 'rolebinding', 'read-status')
settle('bound', target, 403)
check('new-get-denied-after-rbac-withdrawal', 'bound', target, 403)
check('new-watch-denied-after-rbac-withdrawal', 'bound', watch_path, 403)
review('identity-still-valid-after-rbac-withdrawal', 'bound', True)
update_state('after-rbac-withdrawal')
record('existing-watch-receives-after-rbac-withdrawal', events.get(timeout=10), 'after-rbac-withdrawal')
k('-n', ns, 'delete', 'secret', 'token-anchor')
settle('bound', target, 401)
check('new-request-rejects-deleted-anchor', 'bound', target, 401)
review('review-rejects-deleted-anchor', 'bound', False)
update_state('after-anchor-deletion')
record('existing-watch-receives-after-anchor-deletion', events.get(timeout=10), 'after-anchor-deletion')
# Let the bounded watch finish before closing its response object.
# This does not exercise an emergency server-side forced disconnect.
thread.join(timeout=30)
for stream in streams:
stream.close
record('watch-reader-ended', thread.is_alive, False)
anchor = create(resource('Secret', 'token-anchor', type='Opaque', stringData={'purpose': 'new-synthetic-anchor'}))
record('recreated-anchor-has-new-uid', anchor['metadata']['uid']!= first_uid, True)
create(binding)
settle('unbound', target, 200)
check('unbound-token-survives-unrelated-anchor-deletion', 'unbound', target, 200)
check('same-name-does-not-restore-old-bound-token', 'bound', target, 401)
review('review-rejects-old-uid-after-recreation', 'bound', False)
token('replacement', anchor['metadata']['uid'])
review('replacement-token-authenticated', 'replacement', True)
check('replacement-token-permitted', 'replacement', target, 200)
check('anonymous-health-path', None, '/livez', 200)
check('anonymous-workload-read-denied', None, target, 403)
k('-n', ns, 'delete', 'serviceaccount', 'reporter')
settle('unbound', target, 401)
check('account-deletion-rejects-unbound-token', 'unbound', target, 401)
settle('replacement', target, 401)
check('account-deletion-rejects-replacement-token', 'replacement', target, 401)
assert len(records) == 23
finally:
for stream in streams:
stream.close
tokens.clear
if created:
k('delete', 'namespace', ns, '--wait=true', '--timeout=60s')
shutil.rmtree(private)
report = dict(executedAt=datetime.datetime.now(datetime.timezone.utc).isoformat,
scriptSha256=hashlib.sha256(Path(__file__).read_bytes).hexdigest,
versions=versions, examVersion='1.35', passed=all(r['passed'] for r in records),
observations=records, propagationChecks=settlements,
cleanup=dict(namespaceRemoved=True, privateCacheRemoved=not private.exists,
tokensRecorded=False, realCredentialsUsed=False, watchReaderEnded=not thread.is_alive),
scope='Actual HTTPS requests, TokenReview, Secret-bound TokenRequest, RBAC withdrawal and an existing watch on disposable kind server1.37.0/client1.37.1. Exam runtime1.35 differs. No OIDC provider, offline cryptographic JWT validation, finalizer grace timing, workload execution, server-side forced disconnect or upgrade was exercised.')
out.write_text(json.dumps(report, indent=2) + '\n')
print('PASS', len(records))
An old watch receives a fresh marker after RoleBinding withdrawal; a new watch receives 403.
Common pitfalls
Confusing 403 with an invalid token, automount with revocation, a name with a UID and fresh denial with stream termination.
Related topics: RBAC and reconciliation · TokenRequest and TokenReview · Incident management and continuity · Upgrades and compatibility
Measure identity, permission, the in-flight request and exposure of already obtained data separately.
Reference: CKS certification and domains · Kubernetes v1.35; current six-domain CKS outline