216 lines
10 KiB
Python
216 lines
10 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Build an isolated consumer kubeconfig and deliver KeyCape tokens to kubectl.
|
||
|
|
|
||
|
|
Token claim checks here prevent accidental credential misuse. Signature verification
|
||
|
|
and authoritative group mapping belong to the Kubernetes JWT authenticator.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import base64
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import stat
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
SERVER = "https://127.0.0.1:16444"
|
||
|
|
CA_SHA256 = "794d3badd7c3743fe66c3671e7f152277f2ac9adddbc5cebe534802432861bd7"
|
||
|
|
ISSUER = "https://kc.coulomb.social"
|
||
|
|
AUDIENCE = "playwright-kubernetes"
|
||
|
|
SCOPE = "playwright:connect"
|
||
|
|
API = "client.authentication.k8s.io/v1"
|
||
|
|
|
||
|
|
|
||
|
|
def require(condition: bool, message: str) -> None:
|
||
|
|
if not condition:
|
||
|
|
raise ValueError(message)
|
||
|
|
|
||
|
|
|
||
|
|
def private_path(value: str) -> Path:
|
||
|
|
path = Path(os.path.abspath(os.path.expanduser(value)))
|
||
|
|
require(path.resolve() == path, "symlink paths are refused")
|
||
|
|
require(not any((p / ".git").exists() for p in path.parents),
|
||
|
|
"consumer files must be outside Git worktrees")
|
||
|
|
parent = path.parent.stat()
|
||
|
|
require(parent.st_uid == os.getuid() and stat.S_IMODE(parent.st_mode) == 0o700,
|
||
|
|
"parent directory must be owned by the caller with mode 0700")
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def read_token(value: str) -> tuple[str, int]:
|
||
|
|
path = private_path(value)
|
||
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
||
|
|
with os.fdopen(fd) as stream:
|
||
|
|
info = os.fstat(stream.fileno())
|
||
|
|
require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid()
|
||
|
|
and stat.S_IMODE(info.st_mode) == 0o600 and info.st_nlink == 1,
|
||
|
|
"token must be a caller-owned mode-0600 regular file with one link")
|
||
|
|
require(info.st_size <= 65536, "token response is too large")
|
||
|
|
response = json.loads(stream.read(65537))
|
||
|
|
require(isinstance(response, dict), "invalid token response")
|
||
|
|
token = response.get("access_token")
|
||
|
|
require(response.get("token_type", "").lower() == "bearer"
|
||
|
|
and isinstance(token, str) and len(token.split(".")) == 3,
|
||
|
|
"expected a KeyCape bearer access token")
|
||
|
|
part = token.split(".")[1]
|
||
|
|
claims = json.loads(base64.urlsafe_b64decode(part + "=" * (-len(part) % 4)))
|
||
|
|
require(isinstance(claims, dict), "invalid token claims")
|
||
|
|
require(claims.get("iss") == ISSUER, "token issuer mismatch")
|
||
|
|
aud = claims.get("aud")
|
||
|
|
require(aud == AUDIENCE or aud == [AUDIENCE], "token audience mismatch")
|
||
|
|
require(isinstance(claims.get("scope"), str)
|
||
|
|
and SCOPE in claims["scope"].split(), "token scope mismatch")
|
||
|
|
principal = claims.get("principal_type")
|
||
|
|
roles = claims.get("roles", [])
|
||
|
|
require(principal == "human" or (principal == "service"
|
||
|
|
and isinstance(roles, list) and "coding-agent" in roles),
|
||
|
|
"token is not a human or registered coding agent")
|
||
|
|
require(isinstance(claims.get("sub"), str) and bool(claims["sub"]),
|
||
|
|
"token subject is missing")
|
||
|
|
expires = claims.get("exp")
|
||
|
|
issued = claims.get("iat")
|
||
|
|
now = time.time()
|
||
|
|
require(type(expires) is int and type(issued) is int
|
||
|
|
and issued <= now + 30 and 0 < expires - issued <= 3600 and expires > now + 30,
|
||
|
|
"token expired or has invalid validity times; authenticate again")
|
||
|
|
return token, expires
|
||
|
|
|
||
|
|
|
||
|
|
def credential(token_file: str) -> dict:
|
||
|
|
# Check kubectl's destination BEFORE opening the credential file.
|
||
|
|
info = json.loads(os.environ.get("KUBERNETES_EXEC_INFO", "{}"))
|
||
|
|
require(info.get("apiVersion") == API and info.get("kind") == "ExecCredential",
|
||
|
|
"invoke through the generated kubectl configuration")
|
||
|
|
cluster = info.get("spec", {}).get("cluster", {})
|
||
|
|
require(cluster.get("server") == SERVER and not cluster.get("proxy-url")
|
||
|
|
and not cluster.get("tls-server-name")
|
||
|
|
and not cluster.get("insecure-skip-tls-verify", False),
|
||
|
|
"kubectl destination mismatch")
|
||
|
|
ca = base64.b64decode(cluster.get("certificate-authority-data", ""), validate=True)
|
||
|
|
require(hashlib.sha256(ca).hexdigest() == CA_SHA256, "kubectl CA mismatch")
|
||
|
|
require(not sys.stdout.isatty(), "credential output must be consumed by kubectl")
|
||
|
|
token, expires = read_token(token_file)
|
||
|
|
return {"apiVersion": API, "kind": "ExecCredential", "status": {
|
||
|
|
"token": token,
|
||
|
|
"expirationTimestamp": datetime.fromtimestamp(expires, timezone.utc)
|
||
|
|
.isoformat().replace("+00:00", "Z"),
|
||
|
|
}}
|
||
|
|
|
||
|
|
|
||
|
|
def configuration(token_file: str) -> dict:
|
||
|
|
token_path = private_path(token_file)
|
||
|
|
ca = (ROOT / "config/playwright/railiance01-ca.pem").read_bytes()
|
||
|
|
require(hashlib.sha256(ca).hexdigest() == CA_SHA256, "public CA source changed")
|
||
|
|
return {
|
||
|
|
"apiVersion": "v1", "kind": "Config", "current-context": "playwright",
|
||
|
|
"clusters": [{"name": "railiance01-playwright", "cluster": {
|
||
|
|
"server": SERVER, "certificate-authority-data": base64.b64encode(ca).decode(),
|
||
|
|
}}],
|
||
|
|
"contexts": [{"name": "playwright", "context": {
|
||
|
|
"cluster": "railiance01-playwright", "user": "playwright-consumer",
|
||
|
|
"namespace": "playwright",
|
||
|
|
}}],
|
||
|
|
"users": [{"name": "playwright-consumer", "user": {"exec": {
|
||
|
|
"apiVersion": API, "command": sys.executable,
|
||
|
|
"args": [str(Path(__file__).resolve()), "credential", "--token-file", str(token_path)],
|
||
|
|
"interactiveMode": "Never", "provideClusterInfo": True,
|
||
|
|
}}}],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def setup(token_file: str, output: str) -> None:
|
||
|
|
config = configuration(token_file)
|
||
|
|
target = private_path(output)
|
||
|
|
require(private_path(token_file) != target, "token and kubeconfig paths must differ")
|
||
|
|
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
||
|
|
with os.fdopen(fd, "w") as stream:
|
||
|
|
json.dump(config, stream, indent=2)
|
||
|
|
stream.write("\n")
|
||
|
|
print("Created isolated consumer kubeconfig; authentication and API transport are required.")
|
||
|
|
|
||
|
|
|
||
|
|
def verify(kubeconfig: str) -> dict:
|
||
|
|
path = private_path(kubeconfig)
|
||
|
|
config = json.loads(path.read_text())
|
||
|
|
token_file = config["users"][0]["user"]["exec"]["args"][-1]
|
||
|
|
require(config == configuration(token_file),
|
||
|
|
"verification requires the unmodified generated consumer configuration")
|
||
|
|
|
||
|
|
def post(endpoint: str, payload: dict) -> dict:
|
||
|
|
result = subprocess.run(
|
||
|
|
["kubectl", "--kubeconfig", str(path), "--context", "playwright",
|
||
|
|
"--request-timeout=15s", "create", "--raw", endpoint, "-f", "-"],
|
||
|
|
input=json.dumps(payload), text=True, capture_output=True, timeout=25,
|
||
|
|
)
|
||
|
|
require(result.returncode == 0, "consumer API request failed; check login and API transport")
|
||
|
|
return json.loads(result.stdout)
|
||
|
|
|
||
|
|
identity = post("/apis/authentication.k8s.io/v1/selfsubjectreviews", {
|
||
|
|
"apiVersion": "authentication.k8s.io/v1", "kind": "SelfSubjectReview",
|
||
|
|
})["status"]["userInfo"]
|
||
|
|
username = identity["username"]
|
||
|
|
require(username.startswith(("keycape:human:", "keycape:service:"))
|
||
|
|
and set(identity.get("groups", [])) == {"playwright-consumers", "system:authenticated"},
|
||
|
|
"API returned an unexpected consumer identity or groups")
|
||
|
|
checks = [
|
||
|
|
("pods-list", "playwright", "", "pods", "", "list", "", True),
|
||
|
|
("deployment-get", "playwright", "apps", "deployments", "", "get", "playwright", True),
|
||
|
|
("forward", "playwright", "", "pods", "portforward", "create", "", True),
|
||
|
|
("secrets-denied", "playwright", "", "secrets", "", "list", "", False),
|
||
|
|
("exec-denied", "playwright", "", "pods", "exec", "create", "", False),
|
||
|
|
("write-denied", "playwright", "apps", "deployments", "", "patch", "playwright", False),
|
||
|
|
("other-namespace-denied", "default", "", "pods", "", "list", "", False),
|
||
|
|
("other-deployment-denied", "playwright", "apps", "deployments", "", "get", "other", False),
|
||
|
|
("impersonation-denied", "", "", "users", "", "impersonate", "", False),
|
||
|
|
]
|
||
|
|
for name, namespace, group, resource, subresource, verb, resource_name, expected in checks:
|
||
|
|
result = post("/apis/authorization.k8s.io/v1/selfsubjectaccessreviews", {
|
||
|
|
"apiVersion": "authorization.k8s.io/v1", "kind": "SelfSubjectAccessReview",
|
||
|
|
"spec": {"resourceAttributes": {
|
||
|
|
"namespace": namespace, "group": group, "resource": resource,
|
||
|
|
"subresource": subresource, "verb": verb, "name": resource_name,
|
||
|
|
}},
|
||
|
|
})
|
||
|
|
status = result["status"]
|
||
|
|
require(not status.get("evaluationError") and status.get("allowed") is expected,
|
||
|
|
"consumer permission check failed: " + name)
|
||
|
|
return {"identity": username, "groups": sorted(identity["groups"]),
|
||
|
|
"impersonated": False, "checks_passed": [c[0] for c in checks]}
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
subs = parser.add_subparsers(dest="action", required=True)
|
||
|
|
config = subs.add_parser("setup", help="create a new isolated kubeconfig")
|
||
|
|
config.add_argument("--token-file", required=True)
|
||
|
|
config.add_argument("--out", required=True)
|
||
|
|
auth = subs.add_parser("credential", help="kubectl exec plugin; never run directly")
|
||
|
|
auth.add_argument("--token-file", required=True)
|
||
|
|
check = subs.add_parser("verify", help="prove current identity and namespace permissions")
|
||
|
|
check.add_argument("--kubeconfig", required=True)
|
||
|
|
args = parser.parse_args()
|
||
|
|
try:
|
||
|
|
if args.action == "setup":
|
||
|
|
setup(args.token_file, args.out)
|
||
|
|
elif args.action == "credential":
|
||
|
|
print(json.dumps(credential(args.token_file)))
|
||
|
|
else:
|
||
|
|
print(json.dumps(verify(args.kubeconfig), indent=2))
|
||
|
|
except (OSError, ValueError, KeyError, IndexError, TypeError, AttributeError, OverflowError,
|
||
|
|
subprocess.TimeoutExpired):
|
||
|
|
# Parse/provider errors may contain secret input; never echo the exception.
|
||
|
|
print("Playwright consumer command failed. Check private file permissions, token validity, "
|
||
|
|
"the pinned API route and owner activation status.", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|