60101b0932
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13690-09e0-7cd1-964b-ec83cdbcd348
94 lines
4.9 KiB
Python
94 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only native RBAC checks; impersonation is not consumer login proof."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import json
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
|
|
|
|
CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
|
|
GROUP = 'playwright-consumers'
|
|
ROLE = 'playwright-consumer'
|
|
|
|
|
|
def verify(target):
|
|
def kubectl(args, body=None):
|
|
result = subprocess.run(
|
|
['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', target,
|
|
shlex.join(['kubectl', '--request-timeout=10s', *args])],
|
|
input=json.dumps(body) if body is not None else None,
|
|
capture_output=True, text=True, timeout=20, check=True)
|
|
return json.loads(result.stdout)
|
|
|
|
uid = kubectl(['get', 'ns', 'kube-system', '-o', 'json'])['metadata']['uid']
|
|
if uid != CLUSTER_UID:
|
|
raise RuntimeError('wrong cluster')
|
|
binding = kubectl(['get', 'rolebinding', GROUP, '-n', 'playwright', '-o', 'json'])
|
|
role = kubectl(['get', 'role', ROLE, '-n', 'playwright', '-o', 'json'])
|
|
if binding['subjects'] != [{'apiGroup': 'rbac.authorization.k8s.io',
|
|
'kind': 'Group', 'name': GROUP}]:
|
|
raise RuntimeError('consumer binding subjects differ')
|
|
if binding['roleRef'] != {'apiGroup': 'rbac.authorization.k8s.io',
|
|
'kind': 'Role', 'name': ROLE}:
|
|
raise RuntimeError('consumer binding role differs')
|
|
expected = [
|
|
{'apiGroups': [''], 'resources': ['pods'], 'verbs': ['get', 'list']},
|
|
{'apiGroups': [''], 'resources': ['pods/portforward'], 'verbs': ['create']},
|
|
{'apiGroups': ['apps'], 'resources': ['deployments'],
|
|
'resourceNames': ['playwright'], 'verbs': ['get']},
|
|
]
|
|
if role['rules'] != expected:
|
|
raise RuntimeError('consumer role rules differ')
|
|
cases = [
|
|
(True, 'get', 'pods', '', '', 'playwright', '', True),
|
|
(True, 'list', 'pods', '', '', 'playwright', '', True),
|
|
(True, 'create', 'pods', 'portforward', '', 'playwright', '', True),
|
|
(True, 'get', 'deployments', '', 'apps', 'playwright', 'playwright', True),
|
|
(True, 'get', 'deployments', '', 'apps', 'playwright', 'other', False),
|
|
(True, 'get', 'secrets', '', '', 'playwright', '', False),
|
|
(True, 'create', 'pods', 'exec', '', 'playwright', '', False),
|
|
(True, 'create', 'pods', '', '', 'playwright', '', False),
|
|
(True, 'patch', 'deployments', '', 'apps', 'playwright', 'playwright', False),
|
|
(True, 'create', 'pods', 'portforward', '', 'default', '', False),
|
|
(True, 'create', 'rolebindings', '', 'rbac.authorization.k8s.io', 'playwright', '', False),
|
|
(False, 'list', 'pods', '', '', 'playwright', '', False),
|
|
(False, 'create', 'pods', 'portforward', '', 'playwright', '', False),
|
|
]
|
|
checks = []
|
|
for member, verb, resource, subresource, group, namespace, name, expected_allowed in cases:
|
|
attributes = dict(verb=verb, resource=resource, subresource=subresource,
|
|
group=group, namespace=namespace, name=name)
|
|
args = ['--as=rpf-playwright-rbac-probe', '--as-group=system:authenticated']
|
|
if member:
|
|
args.append('--as-group=' + GROUP)
|
|
# Raw review endpoint avoids kubectl's unrelated CRD-discovery permission.
|
|
review = kubectl([*args, 'create', '--raw',
|
|
'/apis/authorization.k8s.io/v1/selfsubjectaccessreviews',
|
|
'-f', '-'], {
|
|
'apiVersion': 'authorization.k8s.io/v1', 'kind': 'SelfSubjectAccessReview',
|
|
'spec': {'resourceAttributes': attributes}})
|
|
status = review['status']
|
|
if status.get('evaluationError') or type(status.get('allowed')) is not bool:
|
|
raise RuntimeError('authorization review incomplete')
|
|
checks.append(dict(member=member, attributes=attributes,
|
|
expected=expected_allowed, allowed=status['allowed'],
|
|
passed=status['allowed'] == expected_allowed))
|
|
return dict(schema='platform-playwright-rbac/v1',
|
|
observed_at=datetime.now(timezone.utc).isoformat(), cluster_uid=uid,
|
|
role_uid=role['metadata']['uid'], binding_uid=binding['metadata']['uid'],
|
|
group=GROUP, checks=checks, passed=all(c['passed'] for c in checks),
|
|
actual_consumer_authentication_proven=False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--target', default='railiance01', choices=['railiance01'])
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
args = parser.parse_args()
|
|
result = verify(args.target)
|
|
args.output.write_text(json.dumps(result, indent=2) + '\n')
|
|
print(json.dumps({'passed': result['passed'], 'checks': len(result['checks']),
|
|
'evidence': str(args.output)}))
|
|
raise SystemExit(0 if result['passed'] else 1)
|