74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160 | def check_cluster(client, *, namespace: str, service_account: str, timeout: float = 10) -> DoctorReport:
"""Read service-account metadata and perform non-persistent self-access reviews."""
validate_target(namespace, service_account)
if timeout is None or timeout <= 0 or timeout > 30:
raise ValueError("timeout must be between 0 (exclusive) and 30 seconds")
deadline = WaitDeadline(timeout)
report = DoctorReport(namespace, service_account)
api, authorization = k8s.CoreV1Api(client), k8s.AuthorizationV1Api(client)
try:
account = api.read_namespaced_service_account(
name=service_account,
namespace=namespace,
**deadline.request_options(),
)
report.findings.append(
DiagnosticFinding(
"service_account_found",
"info",
service_account,
"Driver service account exists",
f"ServiceAccount {namespace}/{account.metadata.name}",
"Existence does not establish its permissions; ask an administrator to review the "
"scoped driver role.",
)
)
if account.automount_service_account_token is False:
report.findings.append(
DiagnosticFinding(
"service_account_token_disabled",
"warning",
service_account,
"Automatic API credentials are disabled",
"ServiceAccount automountServiceAccountToken is false.",
"Confirm that the driver receives an explicit projected token or another "
"supported Kubernetes credential.",
)
)
except (k8s.ApiException, TimeoutError, HTTPTimeoutError) as error:
code = getattr(error, "status", "timeout")
report.findings.append(
DiagnosticFinding(
"service_account_unavailable",
"warning",
service_account,
"Driver service account could not be verified",
f"Service-account lookup returned {code}.",
"Check the namespace and service-account name, and the caller's permission to "
"read serviceaccounts.",
)
)
for resource, verb in CALLER_CHECKS:
name, _, subresource = resource.partition("/")
check = {"resource": resource, "verb": verb, "allowed": None}
try:
review = authorization.create_self_subject_access_review(
body=k8s.V1SelfSubjectAccessReview(
spec=k8s.V1SelfSubjectAccessReviewSpec(
resource_attributes=k8s.V1ResourceAttributes(
namespace=namespace,
group="",
resource=name,
subresource=subresource or None,
verb=verb,
),
)
),
**deadline.request_options(),
)
check["allowed"] = bool(review.status.allowed) if review.status else None
except (k8s.ApiException, TimeoutError, HTTPTimeoutError):
pass
report.caller_permissions.append(check)
if check["allowed"] is not True:
report.findings.append(
DiagnosticFinding(
"caller_permission_denied" if check["allowed"] is False else "caller_permission_unknown",
"warning",
resource,
f"Caller permission: {verb} {resource}",
"Denied by Kubernetes authorization."
if check["allowed"] is False
else "Access review unavailable.",
"Review the caller's namespace-scoped role. These checks do not test or grant "
"driver service-account permissions.",
)
)
return report
|