Prepare Playwright consumer credentials and identity activation handoff

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13690-09e0-7cd1-964b-ec83cdbcd348

scripts/playwright_consumer.py

@@ -0,0 +1,215 @@
+#!/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())