LEARN. PRACTICE. GROW.Define the required operational action
In a fictional funds service, APS needs to read closing status without changing settings or reading settlement credentials. Write a matrix before selecting a role: reporter identity, ccsp-team-a namespace, closing-status ConfigMap, get operation. Add negative checks for other-settings, ccsp-team-b, Secrets and PATCH. Availability of the status page does not establish that the matrix is correct. This exercise uses the Kubernetes API as a concrete management plane; the business API, user sessions and provider IAM policies are different layers requiring their own criteria.
Separate authentication and authorization in the exercise
The script creates the ServiceAccount with automountServiceAccountToken disabled and explicitly requests a ten-minute token through TokenRequest. TokenReview confirms the identity; the first ConfigMap read receives 403 because no grant exists. Neither --as nor a merely simulated identity is used: Python sends the token over HTTPS and validates the server certificate with the dedicated kubeconfig CA. The administrator prepares objects, but evaluated requests use the reporter identity. Record this distinction in the evidence. Disabling automatic mounting prevents that automatic distribution; it is not a mechanism for cancelling an explicitly issued token.
Exercise scope and request shape
The read-status Role and its RoleBinding allow reading closing-status by name in A. The lab obtains 200 for that object and 403 for another name, another namespace, a Secret and modification. An unfiltered list also receives 403; fieldSelector=metadata.name=closing-status returns 200 and only the expected object. The initial grant includes get and list with resourceNames. After narrowing it to get, listing is no longer authorized. An application using list-based discovery needs testing with its actual request shape rather than broader rights to conceal an incompatibility.
Locate the grant that preserves access
The injected fault is an additional ClusterRoleBinding allowing ConfigMap and Secret reads across namespaces. With the same token, the Secret and B’s ConfigMap now return 200. Narrowing the local Role to get does not remove that path. The exercise confirms that the Secret remains readable until the broad binding is withdrawn. After removal, the Secret and B return 403 while A’s closing-status remains accessible. For an access review, retain binding inventory and affected identities. Do not attribute the result solely to the name of the role you edited: observe effective permissions and legitimately required operations.
Hand the matrix to operations
Run the complete code below only in the dedicated cluster prepared using the lab README. Two runs passed 30 checks each; image and versions are recorded in the manifest. For APS handover, provide allowed and denied requests, change approval owners, containment procedure and evidence that legitimate reads recover. The token still authenticates after permission withdrawal: the team must distinguish credential state from authorization of new requests. This exercise neither closes existing watches nor erases secrets already read. Summarize acceptance by operation and scope, linking it with identity management, change and incident response.
#!/usr/bin/env python3
"""Original CCSP lab: real service-account HTTPS, RBAC and Metadata audit.
Requires an exclusive dr-ccsp-management-* kind cluster and the supplied policy.
Only synthetic objects are used. Tokens remain in process memory and stdin.
"""
import argparse, base64, datetime, hashlib, json, shutil, ssl, subprocess
import tempfile, time, urllib.error, urllib.request, uuid
from pathlib import Path
p = argparse.ArgumentParser(description=__doc__)
for name in ('kubectl','kubeconfig','cluster','output'):
p.add_argument('--'+name,required=True)
a = p.parse_args
assert a.cluster.startswith('dr-ccsp-management-')
assert Path(a.kubeconfig).is_absolute
out = Path(a.output); assert not out.exists
private = Path(tempfile.mkdtemp(prefix='dr-ccsp-management-private-'))
base = [a.kubectl,'--kubeconfig',a.kubeconfig,'--context','kind-'+a.cluster,
'--cache-dir',str(private/'cache'),'--request-timeout=15s']
ns_a,ns_b = 'ccsp-team-a','ccsp-team-b'
subject = 'system:serviceaccount:'+ns_a+':reporter'
trial = 'dr-ccsp-management/'+uuid.uuid4.hex
checks,requests,settlements,owned_ns = [],[],[],[]
cluster_role = 'dr-ccsp-broad-'+uuid.uuid4.hex[:8]
token = None; broad_created = False; broad_bound = False
def k(*args,obj=None,ok=True):
result = subprocess.run(base+list(args),text=True,capture_output=True,
input=json.dumps(obj) if obj is not None else None,timeout=40)
if ok and result.returncode:
# Never echo token-bearing input or raw kubeconfig/API output.
raise RuntimeError('Admin operation failed: '+str(args[:3]))
return result
def resource(kind,name,namespace=None,**fields):
api = 'rbac.authorization.k8s.io/v1' if kind in ('Role','RoleBinding','ClusterRole','ClusterRoleBinding') else 'v1'
metadata = dict(name=name)
if namespace: metadata['namespace']=namespace
return dict(apiVersion=api,kind=kind,metadata=metadata,**fields)
def create(obj):
return json.loads(k('create','-f','-','-o','json',obj=obj).stdout)
def binding:
return resource('RoleBinding','read-status',ns_a,
subjects=[dict(kind='ServiceAccount',name='reporter',namespace=ns_a)],
roleRef=dict(apiGroup='rbac.authorization.k8s.io',kind='Role',name='read-status'))
def record(name,inputs,actual,expected):
checks.append(dict(name=name,inputs=inputs,actual=actual,expected=expected))
if actual!= expected: raise AssertionError(name)
print(name+' PASS',flush=True)
def request(path,method='GET',body=None):
data=json.dumps(body).encode if body is not None else None
headers={'Authorization':'Bearer '+token,'User-Agent':trial}
if data is not None: headers['Content-Type']='application/merge-patch+json'
req=urllib.request.Request(server+path,data=data,headers=headers,method=method)
try:
with urllib.request.urlopen(req,context=ctx,timeout=15) as response:
result=(response.status,response.read,response.headers.get('Audit-Id'))
except urllib.error.HTTPError as error:
result=(error.code,error.read,error.headers.get('Audit-Id'))
requests.append(dict(method=method,path=path,status=result[0],auditID=result[2]))
return result
def check_request(name,path,expected,method='GET',body=None):
status,data,_=request(path,method,body)
record(name,dict(path=path,method=method),status,expected)
return data
def settle(path,expected):
start=time.monotonic
while time.monotonic-start<15:
if request(path)[0]==expected:
settlements.append(dict(path=path,expected=expected,elapsedSeconds=round(time.monotonic-start,3)))
return
time.sleep(0.2)
raise AssertionError('Authorization propagation exceeded bounded wait')
def authentication_review:
result=json.loads(k('create','-f','-','-o','json',obj=dict(
apiVersion='authentication.k8s.io/v1',kind='TokenReview',spec=dict(token=token))).stdout)
# TokenReview responses echo spec.token; retain only these status fields.
status=result.get('status',{})
return dict(authenticated=status.get('authenticated',False),username=status.get('user',{}).get('username'))
try:
cfg=json.loads(k('config','view','--minify','--raw','-o','json').stdout)
cluster=cfg['clusters'][0]['cluster'];server=cluster['server']
assert server.startswith('https://127.0.0.1:') and 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').stdout)
assert versions['serverVersion']['gitVersion']=='v1.37.0'
assert versions['clientVersion']['gitVersion']=='v1.37.1'
for ns in (ns_a,ns_b):
absent=k('get','namespace',ns,ok=False)
assert absent.returncode!=0 and 'NotFound' in absent.stderr
create(resource('Namespace',ns));owned_ns.append(ns)
create(resource('ServiceAccount','reporter',ns_a,automountServiceAccountToken=False))
for ns in (ns_a,ns_b):
create(resource('ConfigMap','closing-status',ns,data=dict(state='synthetic-ready')))
create(resource('ConfigMap','other-settings',ns_a,data=dict(mode='synthetic-private')))
marker='synthetic-secret-'+uuid.uuid4.hex
create(resource('Secret','settlement-credential',ns_a,type='Opaque',stringData=dict(value=marker)))
token=k('-n',ns_a,'create','token','reporter','--duration=10m').stdout.strip
record('explicit token authenticates with automount disabled',{'automount':False},authentication_review,dict(authenticated=True,username=subject))
cm_a='/api/v1/namespaces/'+ns_a+'/configmaps/closing-status'
cm_b='/api/v1/namespaces/'+ns_b+'/configmaps/closing-status'
secret_a='/api/v1/namespaces/'+ns_a+'/secrets/settlement-credential'
list_a='/api/v1/namespaces/'+ns_a+'/configmaps'
check_request('authenticated identity without grant denied',cm_a,403)
rules=[dict(apiGroups=[''],resources=['configmaps'],resourceNames=['closing-status'],verbs=['get','list'])]
create(resource('Role','read-status',ns_a,rules=rules));create(binding)
settle(cm_a,200)
check_request('named status read allowed',cm_a,200)
check_request('unfiltered collection list denied',list_a,403)
data=check_request('name-filtered collection list allowed',list_a+'?fieldSelector=metadata.name%3Dclosing-status',200)
record('filtered response contains only named object',{'selector':'metadata.name=closing-status'},[x['metadata']['name'] for x in json.loads(data)['items']],['closing-status'])
check_request('different object name denied','/api/v1/namespaces/'+ns_a+'/configmaps/other-settings',403)
check_request('other namespace denied',cm_b,403)
check_request('secret read denied before broad grant',secret_a,403)
check_request('status modification denied',cm_a,403,'PATCH',dict(data=dict(state='unapproved')))
create(resource('ClusterRole',cluster_role,rules=[dict(apiGroups=[''],resources=['configmaps','secrets'],verbs=['get','list'])]));broad_created=True
create(resource('ClusterRoleBinding',cluster_role,subjects=[dict(kind='ServiceAccount',name='reporter',namespace=ns_a)],roleRef=dict(apiGroup='rbac.authorization.k8s.io',kind='ClusterRole',name=cluster_role)));broad_bound=True
settle(secret_a,200)
exposed=check_request('broad cluster grant permits secret read',secret_a,200)
record('successful API read returned synthetic secret',{'resource':'settlement-credential'},json.loads(exposed)['data']['value']==base64.b64encode(marker.encode).decode,True);exposed=None
check_request('broad cluster grant crosses namespace',cm_b,200)
check_request('broad cluster grant permits collection list',list_a,200)
# Narrowing one Role cannot subtract a grant from another binding.
k('-n',ns_a,'patch','role','read-status','--type=merge','-p',json.dumps(dict(rules=[dict(apiGroups=[''],resources=['configmaps'],resourceNames=['closing-status'],verbs=['get'])])))
check_request('narrowing local role does not remove broad grant',secret_a,200)
k('delete','clusterrolebinding',cluster_role);broad_bound=False
settle(secret_a,403)
check_request('secret denied after broad binding removal',secret_a,403)
check_request('other namespace denied after broad binding removal',cm_b,403)
check_request('legitimate status read survives containment',cm_a,200)
check_request('collection list denied after local role narrowed',list_a,403)
record('token still authenticates after permission withdrawal',{'removed':'ClusterRoleBinding'},authentication_review,dict(authenticated=True,username=subject))
k('-n',ns_a,'delete','rolebinding','read-status');settle(cm_a,403)
check_request('removing last binding denies new status reads',cm_a,403)
create(binding);settle(cm_a,200)
check_request('restored least privilege recovers status reads',cm_a,200)
check_request('restored least privilege keeps secret denied',secret_a,403)
# Match actual client audit IDs, not a guess based on counts or timestamps.
wanted={r['auditID'] for r in requests}
assert None not in wanted
raw=''events=[]
for _ in range(50):
fetched=subprocess.run(['docker','exec',a.cluster+'-control-plane','cat','/var/log/dr-audit/audit.log'],text=True,capture_output=True,timeout=20)
assert fetched.returncode==0
raw=fetched.stdout
parsed=[json.loads(line) for line in raw.splitlines if line.strip]
events=[e for e in parsed if e.get('userAgent')==trial and e.get('stage')=='ResponseComplete']
if wanted<={e['auditID'] for e in events}:break
time.sleep(0.2)
by_id={e['auditID']:e for e in events}
record('every client request has a completed audit event',{'clientRequests':len(requests),'uniqueAuditIDs':len(wanted)},wanted<=set(by_id),True)
record('audit response codes match client observations',{'correlation':'Audit-Id'},all(by_id[r['auditID']]['responseStatus']['code']==r['status'] for r in requests),True)
record('audit identifies authenticated service account',{'expectedUser':subject},all(e['user']['username']==subject and not e.get('impersonatedUser') for e in events),True)
secret_events=[e for e in events if e.get('objectRef',{}).get('resource')=='secrets']
record('audit captures allowed and denied secret reads',{'resource':'secrets'},sorted(set(e['responseStatus']['code'] for e in secret_events)),[200,403])
record('Metadata events omit request and response bodies',{'level':'Metadata'},all(e['level']=='Metadata' and 'requestObject' not in e and 'responseObject' not in e for e in events),True)
record('synthetic secret absent from raw audit log',{'checkedForms':['plain','base64']},marker not in raw and base64.b64encode(marker.encode).decode not in raw,True)
record('token absent from raw audit log',{'tokenRecorded':False},token not in raw,True)
safe_events=[{key:e.get(key) for key in ['auditID','stage','level','verb','requestURI','requestReceivedTimestamp','stageTimestamp','objectRef','responseStatus']}|{'user':e['user']['username']} for e in events]
finally:
token=None
if broad_bound:k('delete','clusterrolebinding',cluster_role)
if broad_created:k('delete','clusterrole',cluster_role)
for ns in reversed(owned_ns):k('delete','namespace',ns,'--wait=true','--timeout=30s')
shutil.rmtree(private)
report=dict(checkedAt=datetime.datetime.now(datetime.timezone.utc).isoformat,scriptSha256=hashlib.sha256(Path(__file__).read_bytes).hexdigest,versions=versions,cluster=a.cluster,trial=trial,checks=checks,requests=requests,auditEvents=safe_events,propagationChecks=settlements,cleanup=dict(namespacesRemoved=True,clusterRoleRemoved=True,privateCacheRemoved=not private.exists,tokensRecorded=False,realCredentialsUsed=False),limitations=[
'Actual local Kubernetes service-account authentication, HTTPS authorization and API audit; no cloud-provider IAM, OIDC, managed control plane or production application.',
'New requests only; existing watches, cached secret copies and application sessions were not revoked or tested.',
'Metadata proves request attribution and response status, not response content, subsequent use, exfiltration destination or complete incident scope.',
'No network-policy isolation, Pod execution, remote SIEM delivery, audit retention attack, immutable logging or provider-wide incident containment.'
])
out.write_text(json.dumps(report,indent=2)+'\n')
print('PASS',len(checks),'checks; credentials not recorded',flush=True)
The same token reads a Secret before binding removal and receives 403 afterwards while still authenticating.
Common pitfalls
Testing only allowed paths; overlooking additional bindings; confusing 403 with a revoked token; inferring content or exfiltration from Metadata.
Related topics: Identity, tokens and authorization · Change, evidence and incident response
Execute authenticated requests and detect permissions that remain active through another binding.
Reference: Using RBAC Authorization · CCSP examination outline effective 2026-08-01; January2026 V2 PDF