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