Plano de gestão: identidade, âmbito e privilégio mínimo

#!/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)

Tópicos relacionados: Identidade, tokens e autorização · Mudança, evidência e resposta a incidentes

Referência: Using RBAC Authorization · CCSP examination outline effective 2026-08-01; January2026 V2 PDF