#!/usr/bin/env python3 """ claude-code manager — 制御プレーン 責務: * Web UI(モバイル対応・ビルド不要の素の HTML/JS) * セッションのライフサイクル API(作成 / 一覧 / 破棄 / touch) * gateway(nginx) からの auth_request 検証(token モード) * アイドル / 寿命超過セッションの回収ループ 依存: Python 標準ライブラリのみ(追加 pip 不要)。 Kubernetes API へは in-cluster ServiceAccount トークンで直接アクセスする。 セッション NS 内の Pod / Service のみを操作する(RBAC で制限済み)。 """ import base64 import hmac import html import json import logging import os import re import secrets import ssl import threading import time import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # -------------------------------------------------------------------------- # 設定(すべて環境変数から) # -------------------------------------------------------------------------- def env(key, default=""): return os.environ.get(key, default) def env_json(key, default): raw = os.environ.get(key, "").strip() if not raw: return default try: return json.loads(raw) except ValueError: logging.error("invalid JSON in env %s", key) return default LISTEN_PORT = int(env("LISTEN_PORT", "8080")) SESSIONS_NS = env("SESSIONS_NAMESPACE", "default") FULLNAME = env("FULLNAME", "claude-code") SESSION_SA = env("SESSION_SERVICE_ACCOUNT", "claude-code-session") SESSION_IMAGE = env("SESSION_IMAGE", "docker.io/library/node:22-alpine") SESSION_IMAGE_PULL_POLICY = env("SESSION_IMAGE_PULL_POLICY", "IfNotPresent") SESSION_IMAGE_PULL_SECRETS = env_json("SESSION_IMAGE_PULL_SECRETS_JSON", []) BOOTSTRAP_CM = env("SESSION_BOOTSTRAP_CONFIGMAP", FULLNAME + "-session") ANTHROPIC_SECRET = env("ANTHROPIC_SECRET", FULLNAME + "-anthropic") ANTHROPIC_AUTH_MODE = env("ANTHROPIC_AUTH_MODE", "apiKey") ANTHROPIC_CREDENTIALS_KEY = env("ANTHROPIC_CREDENTIALS_KEY", ".credentials.json") GIT_IDENTITIES = env_json("GIT_IDENTITIES_JSON", {}) # Helm 管理(読み取り専用) GUI_IDENTITIES_ENABLED = env("GUI_IDENTITIES_ENABLED", "false").lower() == "true" GIT_FULLNAME_PREFIX = env("GIT_FULLNAME_PREFIX", FULLNAME) GIT_ALLOWED_HOSTS = [h.strip().lower() for h in env("GIT_ALLOWED_HOSTS", "").split(",") if h.strip()] GIT_ALLOW_ARBITRARY = env("GIT_ALLOW_ARBITRARY", "false").lower() == "true" GIT_AUTHOR_NAME = env("GIT_DEFAULT_AUTHOR_NAME", "Claude") GIT_AUTHOR_EMAIL = env("GIT_DEFAULT_AUTHOR_EMAIL", "claude@cafepieters.com") IDLE_TIMEOUT = int(env("IDLE_TIMEOUT_SECONDS", "1800")) MAX_LIFETIME = int(env("MAX_LIFETIME_SECONDS", "43200")) REAP_INTERVAL = int(env("REAP_INTERVAL_SECONDS", "60")) # 終了済み(Succeeded / Failed)セッション Pod を残す秒数。0 なら即回収。 # 死因を kubectl logs / UI「ログ」で調べられるよう既定で猶予を持たせる。 TERMINATED_RETENTION = int(env("TERMINATED_RETENTION_SECONDS", "900")) MAX_SESSIONS = int(env("MAX_SESSIONS", "10")) TTYD_PORT = int(env("TTYD_PORT", "7681")) SESSION_RESOURCES = env_json("SESSION_RESOURCES_JSON", {}) WORKSPACE_DIR = env("WORKSPACE_DIR", "/workspace") AUTOSTART_CLAUDE = env("AUTOSTART_CLAUDE", "true") VERIFY_COMMAND = env("VERIFY_COMMAND", "") AUTOPUSH = env_json("AUTOPUSH_JSON", {"enabled": False}) EXTRA_APK = env("EXTRA_APK_PACKAGES", "") CLAUDE_CODE_VERSION = env("CLAUDE_CODE_VERSION", "") ANTHROPIC_EXTRA_ENV = env_json("ANTHROPIC_EXTRA_ENV_JSON", {}) POD_LABELS = env_json("SESSION_POD_LABELS_JSON", {}) POD_ANNOTATIONS = env_json("SESSION_POD_ANNOTATIONS_JSON", {}) NODE_SELECTOR = env_json("SESSION_NODE_SELECTOR_JSON", {}) TOLERATIONS = env_json("SESSION_TOLERATIONS_JSON", []) PRIORITY_CLASS = env("SESSION_PRIORITY_CLASS", "") SESSION_RUN_UID = int(env("SESSION_RUN_UID", "10001")) AUTH_MODE = env("AUTH_MODE", "token") _auth_token_file = env("AUTH_TOKEN_FILE", "") if _auth_token_file and os.path.exists(_auth_token_file): with open(_auth_token_file, "r", encoding="utf-8") as fh: AUTH_TOKEN = fh.read().strip() else: AUTH_TOKEN = env("AUTH_TOKEN", "") logging.basicConfig( level=getattr(logging, env("LOG_LEVEL", "info").upper(), logging.INFO), format="%(asctime)s %(levelname)s %(message)s", ) log = logging.getLogger("manager") LABEL_APP = "app.kubernetes.io/name" LABEL_COMPONENT = "app.kubernetes.io/component" LABEL_SID = "claude-code.cafepieters.com/session-id" ANN_PREFIX = "claude-code.cafepieters.com/" SESSION_SELECTOR = LABEL_COMPONENT + "=session," + LABEL_APP + "=claude-code" LABEL_GIT_IDENT = "claude-code.cafepieters.com/git-identity" LABEL_IDENT_SOURCE = "claude-code.cafepieters.com/identity-source" LABEL_IDENT_TYPE = "claude-code.cafepieters.com/identity-type" SID_RE = re.compile(r"^[a-f0-9]{10}$") # identity 名は Secret 名の一部になるため DNS ラベル相当に制限 IDENT_NAME_RE = re.compile(r"^[a-z0-9]([a-z0-9-]{0,38}[a-z0-9])?$") # 受け入れる repo URL: # https://host[:port]/path # ssh://[user@]host[:port]/path # [user@]host:path (SCP 形式。Gitea/GitHub の SSH ユーザーは常に "git") _USER = r"(?:[A-Za-z0-9_.\-]+@)?" REPO_RE = re.compile( r"^(?:https://(?P

[A-Za-z0-9.\-]+)(?::\d+)?/[\w./\-~%]+" r"|ssh://" + _USER + r"(?P

[A-Za-z0-9.\-]+)(?::\d+)?/[\w./\-~%]+" r"|" + _USER + r"(?P

[A-Za-z0-9.\-]+):(?!//)[\w./\-~%]+)$" ) REPO_HELP = ("URL の形式: https:////.git / " "ssh://git@//.git / git@:/.git " "(Gitea・GitHub の SSH ユーザーは常に 'git'。あなたのアカウント名ではありません)") REF_RE = re.compile(r"^[\w./\-]{1,200}$") # -------------------------------------------------------------------------- # Kubernetes API クライアント(標準ライブラリのみ) # -------------------------------------------------------------------------- K8S_HOST = "https://kubernetes.default.svc" _sa_dir = "/var/run/secrets/kubernetes.io/serviceaccount" with open(_sa_dir + "/token", "r", encoding="utf-8") as fh: K8S_TOKEN = fh.read().strip() _ssl_ctx = ssl.create_default_context(cafile=_sa_dir + "/ca.crt") def k8s(method, path, body=None, ctype="application/json"): url = K8S_HOST + path data = None if body is not None: data = body if isinstance(body, bytes) else json.dumps(body).encode("utf-8") req = urllib.request.Request(url=url, method=method, data=data) req.add_header("Authorization", "Bearer " + K8S_TOKEN) req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", ctype) try: with urllib.request.urlopen(req, context=_ssl_ctx, timeout=30) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as exc: raw = exc.read() try: payload = json.loads(raw) except ValueError: payload = {"message": raw.decode("utf-8", "replace")} return exc.code, payload def core(path): return "/api/v1/namespaces/" + urllib.parse.quote(SESSIONS_NS) + path def k8s_text(path): """text/plain を返すエンドポイント(Pod ログ)用。(status, text) を返す。""" req = urllib.request.Request(url=K8S_HOST + path, method="GET") req.add_header("Authorization", "Bearer " + K8S_TOKEN) try: with urllib.request.urlopen(req, context=_ssl_ctx, timeout=30) as resp: return resp.status, resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: return exc.code, exc.read().decode("utf-8", "replace") def session_log(sid, tail=800): if not SID_RE.match(sid): return 400, "bad session id" return k8s_text(core("/pods/claude-sess-" + sid + "/log?container=claude&tailLines=" + str(int(tail)))) # -------------------------------------------------------------------------- # git identity(Helm 管理 + GUI 登録) # -------------------------------------------------------------------------- def gui_identity_secret(name): return GIT_FULLNAME_PREFIX + "-git-" + name def list_gui_identities(): """GUI から登録された identity(source=gui ラベルの Secret)の一覧。""" if not GUI_IDENTITIES_ENABLED: return [] status, payload = k8s("GET", core("/secrets?labelSelector=" + urllib.parse.quote(LABEL_IDENT_SOURCE + "=gui"))) if status != 200: log.error("list identity secrets failed: %s", status) return [] out = [] for sec in payload.get("items", []): lbl = sec.get("metadata", {}).get("labels", {}) nm = lbl.get(LABEL_GIT_IDENT) if nm: out.append({"name": nm, "type": lbl.get(LABEL_IDENT_TYPE, "ssh"), "source": "gui"}) return out def list_identities(): """UI 用: Helm 管理 + GUI 登録の identity 一覧。""" out = [{"name": n, "type": (v or {}).get("type", "ssh"), "source": "helm"} for n, v in GIT_IDENTITIES.items()] names = {i["name"] for i in out} for i in list_gui_identities(): if i["name"] not in names: out.append(i) return sorted(out, key=lambda x: x["name"]) def resolve_identity(name): """build_pod 用にマウント情報を返す。無ければ None。""" if not name: return None if name in GIT_IDENTITIES: return GIT_IDENTITIES[name] if not GUI_IDENTITIES_ENABLED: return None status, sec = k8s("GET", core("/secrets/" + gui_identity_secret(name))) if status != 200: return None lbl = sec.get("metadata", {}).get("labels", {}) if lbl.get(LABEL_IDENT_SOURCE) != "gui": return None typ = lbl.get(LABEL_IDENT_TYPE, "ssh") data = sec.get("data", {}) or {} if typ == "https-token": user = "x-access-token" if data.get("username"): try: user = base64.b64decode(data["username"]).decode("utf-8").strip() or user except Exception: # noqa: BLE001 pass return {"type": "https-token", "secret": gui_identity_secret(name), "keyToken": "token", "username": user} return {"type": "ssh", "secret": gui_identity_secret(name), "keyPrivateKey": "id", "keyKnownHosts": "known_hosts"} def create_gui_identity(body): if not GUI_IDENTITIES_ENABLED: return 403, {"error": "GUI からの identity 登録は無効です(git.guiIdentities.enabled)"} name = str(body.get("name", "")).strip() typ = str(body.get("type", "")).strip() if not IDENT_NAME_RE.match(name): return 400, {"error": "identity 名は英小文字・数字・ハイフン(先頭末尾は英数字、40 文字以内)"} if name in GIT_IDENTITIES: return 409, {"error": "Helm 管理の identity と同名です: " + name} if typ == "ssh": key = str(body.get("privateKey", "")).strip() kh = str(body.get("knownHosts", "")).strip() if "PRIVATE KEY" not in key: return 400, {"error": "privateKey が OpenSSH/PEM 秘密鍵ではありません"} if not kh: return 400, {"error": "knownHosts は必須です(`ssh-keyscan ` の出力)"} string_data = {"id": key + ("\n" if not key.endswith("\n") else ""), "known_hosts": kh + "\n"} elif typ == "https-token": token = str(body.get("token", "")).strip() if not token: return 400, {"error": "token は必須です"} string_data = {"username": str(body.get("username", "") or "x-access-token").strip(), "token": token} else: return 400, {"error": "type は ssh か https-token"} manifest = { "apiVersion": "v1", "kind": "Secret", "type": "Opaque", "metadata": { "name": gui_identity_secret(name), "namespace": SESSIONS_NS, "labels": { LABEL_APP: "claude-code", LABEL_GIT_IDENT: name, LABEL_IDENT_SOURCE: "gui", LABEL_IDENT_TYPE: typ, }, }, "stringData": string_data, } # 既存なら置き換え(PUT)、無ければ作成(POST) status, _ = k8s("GET", core("/secrets/" + gui_identity_secret(name))) if status == 200: st, payload = k8s("PUT", core("/secrets/" + gui_identity_secret(name)), manifest) else: st, payload = k8s("POST", core("/secrets"), manifest) if st not in (200, 201): return st, {"error": "Secret 保存に失敗: %s" % payload.get("message", st)} log.info("gui identity %s (%s) saved", name, typ) return 200, {"ok": True, "name": name, "type": typ} def delete_gui_identity(name): if not GUI_IDENTITIES_ENABLED: return 403, {"error": "無効です"} if not IDENT_NAME_RE.match(name): return 400, {"error": "bad name"} status, sec = k8s("GET", core("/secrets/" + gui_identity_secret(name))) if status != 200: return 404, {"error": "見つかりません"} if sec.get("metadata", {}).get("labels", {}).get(LABEL_IDENT_SOURCE) != "gui": return 403, {"error": "Helm 管理の identity は GUI から削除できません"} k8s("DELETE", core("/secrets/" + gui_identity_secret(name))) log.info("gui identity %s deleted", name) return 200, {"ok": True} # -------------------------------------------------------------------------- # セッション操作 # -------------------------------------------------------------------------- def repo_host(url): m = REPO_RE.match(url) if not m: return None return (m.group("h1") or m.group("h2") or m.group("h3") or "").lower() def list_sessions(): status, payload = k8s("GET", core("/pods?labelSelector=" + urllib.parse.quote(SESSION_SELECTOR))) if status != 200: log.error("list pods failed: %s %s", status, payload) return [] out = [] for pod in payload.get("items", []): meta = pod.get("metadata", {}) ann = meta.get("annotations", {}) st = pod.get("status", {}) info = { "id": meta.get("labels", {}).get(LABEL_SID, ""), "phase": st.get("phase", "Unknown"), "repo": ann.get(ANN_PREFIX + "repo", ""), "ref": ann.get(ANN_PREFIX + "ref", ""), "identity": ann.get(ANN_PREFIX + "git-identity", ""), "owner": ann.get(ANN_PREFIX + "owner", ""), "createdAt": ann.get(ANN_PREFIX + "created-at", ""), "lastActivity": ann.get(ANN_PREFIX + "last-activity", ""), "reason": st.get("reason", ""), # DeadlineExceeded 等 "message": st.get("message", ""), "finishedAt": "", "exitCode": None, } for cs in st.get("containerStatuses", []) or []: term = (cs.get("state") or {}).get("terminated") if term: info["finishedAt"] = term.get("finishedAt", "") info["exitCode"] = term.get("exitCode") if not info["reason"]: info["reason"] = term.get("reason", "") out.append(info) return out def _rfc3339_epoch(value): """RFC3339(...Z)→ epoch 秒。失敗時は 0。""" if not value: return 0 try: import datetime return int(datetime.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()) except Exception: # noqa: BLE001 return 0 def active_session_count(): return sum(1 for s in list_sessions() if s["phase"] in ("Pending", "Running")) def build_pod(sid, repo, ref, identity_name, owner): now = str(int(time.time())) env_list = [ {"name": "SESSION_ID", "value": sid}, {"name": "REPO_URL", "value": repo}, {"name": "REPO_REF", "value": ref}, {"name": "HOME", "value": "/home/claude"}, {"name": "WORKSPACE_DIR", "value": WORKSPACE_DIR}, {"name": "TTYD_PORT", "value": str(TTYD_PORT)}, {"name": "IDLE_TIMEOUT_SECONDS", "value": str(IDLE_TIMEOUT)}, {"name": "AUTOSTART_CLAUDE", "value": AUTOSTART_CLAUDE}, {"name": "VERIFY_COMMAND", "value": VERIFY_COMMAND}, {"name": "AUTOPUSH_ENABLED", "value": str(bool(AUTOPUSH.get("enabled"))).lower()}, {"name": "AUTOPUSH_BRANCH_PREFIX", "value": AUTOPUSH.get("branchPrefix", "claude/")}, {"name": "AUTOPUSH_PROTECTED", "value": ",".join(AUTOPUSH.get("protectedBranches", []))}, {"name": "EXTRA_APK_PACKAGES", "value": EXTRA_APK}, {"name": "CLAUDE_CODE_VERSION", "value": CLAUDE_CODE_VERSION}, {"name": "GIT_AUTHOR_NAME", "value": GIT_AUTHOR_NAME}, {"name": "GIT_AUTHOR_EMAIL", "value": GIT_AUTHOR_EMAIL}, {"name": "GIT_COMMITTER_NAME", "value": GIT_AUTHOR_NAME}, {"name": "GIT_COMMITTER_EMAIL", "value": GIT_AUTHOR_EMAIL}, ] for key, val in ANTHROPIC_EXTRA_ENV.items(): env_list.append({"name": str(key), "value": str(val)}) volumes = [ {"name": "workspace", "emptyDir": {"sizeLimit": "4Gi"}}, {"name": "home", "emptyDir": {"sizeLimit": "1Gi"}}, {"name": "tmp", "emptyDir": {"medium": "Memory", "sizeLimit": "64Mi"}}, {"name": "npm-cache", "emptyDir": {"sizeLimit": "1Gi"}}, {"name": "bootstrap", "configMap": {"name": BOOTSTRAP_CM, "defaultMode": 0o555}}, ] mounts = [ {"name": "workspace", "mountPath": WORKSPACE_DIR}, {"name": "home", "mountPath": "/home/claude"}, {"name": "tmp", "mountPath": "/tmp"}, {"name": "npm-cache", "mountPath": "/home/claude/.npm"}, {"name": "bootstrap", "mountPath": "/bootstrap", "readOnly": True}, ] ident = resolve_identity(identity_name) if ident: env_list.append({"name": "GIT_IDENTITY_TYPE", "value": ident.get("type", "")}) if ident.get("type") == "ssh": volumes.append({ "name": "git-identity", "secret": { "secretName": ident["secret"], "defaultMode": 0o400, "items": [ {"key": ident.get("keyPrivateKey", "id"), "path": "id"}, {"key": ident.get("keyKnownHosts", "known_hosts"), "path": "known_hosts"}, ], }, }) mounts.append({"name": "git-identity", "mountPath": "/git-identity", "readOnly": True}) elif ident.get("type") == "https-token": volumes.append({ "name": "git-identity", "secret": { "secretName": ident["secret"], "defaultMode": 0o400, "items": [{"key": ident.get("keyToken", "token"), "path": "token"}], }, }) mounts.append({"name": "git-identity", "mountPath": "/git-identity", "readOnly": True}) env_list.append({"name": "GIT_HTTPS_USERNAME", "value": ident.get("username", "x-access-token")}) # ---- Claude 認証情報の渡し方 ---- # credentials モード: Secret をファイルとしてマウントし、エントリポイントが # $HOME/.claude/.credentials.json へコピーする(書き込み可能にして claude 自身の # リフレッシュを許す)。env var 名に "." を含められないため envFrom は使わない。 # apiKey / oauth モード: 有効な env var 名なので envFrom でそのまま注入する。 env_from = [] if ANTHROPIC_AUTH_MODE == "credentials": volumes.append({ "name": "anthropic-creds", "secret": { "secretName": ANTHROPIC_SECRET, "defaultMode": 0o400, "items": [{"key": ANTHROPIC_CREDENTIALS_KEY, "path": "credentials.json"}], }, }) mounts.append({"name": "anthropic-creds", "mountPath": "/anthropic", "readOnly": True}) env_list.append({"name": "SEED_CREDENTIALS_FILE", "value": "/anthropic/credentials.json"}) else: env_from = [{"secretRef": {"name": ANTHROPIC_SECRET}}] container = { "name": "claude", "image": SESSION_IMAGE, "imagePullPolicy": SESSION_IMAGE_PULL_POLICY, "command": ["/bin/sh", "/bootstrap/entrypoint.sh"], "env": env_list, "envFrom": env_from, "ports": [{"name": "ttyd", "containerPort": TTYD_PORT}], "volumeMounts": mounts, "resources": SESSION_RESOURCES, "securityContext": { "allowPrivilegeEscalation": False, "readOnlyRootFilesystem": True, "runAsNonRoot": True, "runAsUser": SESSION_RUN_UID, "capabilities": {"drop": ["ALL"]}, }, "readinessProbe": { "tcpSocket": {"port": TTYD_PORT}, "initialDelaySeconds": 5, "periodSeconds": 5, "failureThreshold": 60, }, } labels = { LABEL_APP: "claude-code", LABEL_COMPONENT: "session", "app.kubernetes.io/managed-by": "claude-code-manager", LABEL_SID: sid, } labels.update({str(k): str(v) for k, v in POD_LABELS.items()}) annotations = { ANN_PREFIX + "repo": repo, ANN_PREFIX + "ref": ref, ANN_PREFIX + "git-identity": identity_name or "", ANN_PREFIX + "owner": owner or "anon", ANN_PREFIX + "created-at": now, ANN_PREFIX + "last-activity": now, } annotations.update({str(k): str(v) for k, v in POD_ANNOTATIONS.items()}) spec = { "serviceAccountName": SESSION_SA, "automountServiceAccountToken": False, "restartPolicy": "Never", "activeDeadlineSeconds": MAX_LIFETIME, "terminationGracePeriodSeconds": 30, "enableServiceLinks": False, "securityContext": { "runAsNonRoot": True, "runAsUser": SESSION_RUN_UID, "runAsGroup": SESSION_RUN_UID, "fsGroup": SESSION_RUN_UID, "seccompProfile": {"type": "RuntimeDefault"}, }, "containers": [container], "volumes": volumes, } if SESSION_IMAGE_PULL_SECRETS: spec["imagePullSecrets"] = SESSION_IMAGE_PULL_SECRETS if NODE_SELECTOR: spec["nodeSelector"] = NODE_SELECTOR if TOLERATIONS: spec["tolerations"] = TOLERATIONS if PRIORITY_CLASS: spec["priorityClassName"] = PRIORITY_CLASS return { "apiVersion": "v1", "kind": "Pod", "metadata": {"name": "claude-sess-" + sid, "namespace": SESSIONS_NS, "labels": labels, "annotations": annotations}, "spec": spec, } def build_service(sid): return { "apiVersion": "v1", "kind": "Service", "metadata": { "name": "claude-sess-" + sid, "namespace": SESSIONS_NS, "labels": {LABEL_APP: "claude-code", LABEL_COMPONENT: "session", LABEL_SID: sid}, }, "spec": { "selector": {LABEL_SID: sid}, "ports": [{"name": "ttyd", "port": TTYD_PORT, "targetPort": TTYD_PORT}], }, } def create_session(repo, ref, identity_name, owner): repo = (repo or "").strip() ref = (ref or "main").strip() if not REPO_RE.match(repo): return None, "リポジトリ URL の形式が不正です。" + REPO_HELP if not REF_RE.match(ref): return None, "ref(ブランチ / タグ)の形式が不正です" host = repo_host(repo) if not GIT_ALLOW_ARBITRARY and host not in GIT_ALLOWED_HOSTS: return None, ("許可されていない Git ホストです: %s(許可済み: %s)" % (host, ", ".join(GIT_ALLOWED_HOSTS) or "なし")) if identity_name and resolve_identity(identity_name) is None: return None, "未知の git identity: " + identity_name if active_session_count() >= MAX_SESSIONS: return None, "同時セッション数の上限(%d)に達しています" % MAX_SESSIONS sid = secrets.token_hex(5) status, payload = k8s("POST", core("/pods"), build_pod(sid, repo, ref, identity_name, owner)) if status not in (200, 201): return None, "Pod 作成に失敗: %s" % payload.get("message", status) status, payload = k8s("POST", core("/services"), build_service(sid)) if status not in (200, 201): k8s("DELETE", core("/pods/claude-sess-" + sid)) return None, "Service 作成に失敗: %s" % payload.get("message", status) log.info("session %s created repo=%s ref=%s owner=%s", sid, repo, ref, owner) return sid, None def delete_session(sid): if not SID_RE.match(sid): return False k8s("DELETE", core("/pods/claude-sess-" + sid), {"gracePeriodSeconds": 10}) k8s("DELETE", core("/services/claude-sess-" + sid)) log.info("session %s deleted", sid) return True _touch_cache = {} def touch_session(sid, min_interval=15): if not SID_RE.match(sid): return now = time.time() if now - _touch_cache.get(sid, 0) < min_interval: return _touch_cache[sid] = now patch = {"metadata": {"annotations": {ANN_PREFIX + "last-activity": str(int(now))}}} k8s("PATCH", core("/pods/claude-sess-" + sid), patch, ctype="application/merge-patch+json") # -------------------------------------------------------------------------- # 回収ループ # -------------------------------------------------------------------------- def reaper_loop(): while True: time.sleep(REAP_INTERVAL) try: reap_once() except Exception as exc: # noqa: BLE001 - ループを止めない log.exception("reaper error: %s", exc) def reap_once(): now = int(time.time()) live_ids = set() for s in list_sessions(): sid = s["id"] live_ids.add(sid) reason = None if s["phase"] in ("Succeeded", "Failed"): # 終了済み Pod は TERMINATED_RETENTION の猶予後に回収(死因調査のため残す) fin = _rfc3339_epoch(s.get("finishedAt")) or (int(s["createdAt"] or 0) + 120) if now - fin >= TERMINATED_RETENTION: reason = "phase=" + s["phase"] else: try: created = int(s["createdAt"] or 0) last = int(s["lastActivity"] or 0) except ValueError: created = last = now if created and now - created > MAX_LIFETIME: reason = "max-lifetime" elif last and now - last > IDLE_TIMEOUT: reason = "idle" if reason: log.info("reaping session %s (%s)", sid, reason) delete_session(sid) live_ids.discard(sid) # 孤児 Service の掃除 status, payload = k8s("GET", core("/services?labelSelector=" + urllib.parse.quote(SESSION_SELECTOR))) if status == 200: for svc in payload.get("items", []): sid = svc.get("metadata", {}).get("labels", {}).get(LABEL_SID, "") if sid and sid not in live_ids: log.info("reaping orphan service %s", sid) k8s("DELETE", core("/services/claude-sess-" + sid)) # -------------------------------------------------------------------------- # 認証(token モード) # -------------------------------------------------------------------------- def extract_token(handler): auth = handler.headers.get("Authorization", "") if auth.startswith("Bearer "): return auth[7:].strip() cookie = handler.headers.get("Cookie", "") for part in cookie.split(";"): part = part.strip() if part.startswith("cc_token="): return urllib.parse.unquote(part[9:]) qs = urllib.parse.urlparse(handler.path).query params = urllib.parse.parse_qs(qs) if "token" in params: return params["token"][0] return "" def token_ok(handler): if AUTH_MODE != "token": return True if not AUTH_TOKEN: return False return hmac.compare_digest(extract_token(handler), AUTH_TOKEN) # -------------------------------------------------------------------------- # HTTP ハンドラ # -------------------------------------------------------------------------- INDEX_HTML = """ Claude Code — Remote Sessions

Claude Code — Remote Sessions

""" class Handler(BaseHTTPRequestHandler): server_version = "claude-code-manager" protocol_version = "HTTP/1.1" def _send(self, code, body=b"", ctype="application/json; charset=utf-8", extra=None): if isinstance(body, str): body = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") for key, val in (extra or {}).items(): self.send_header(key, val) self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _json(self, code, obj, extra=None): self._send(code, json.dumps(obj), "application/json; charset=utf-8", extra) def _body(self): length = int(self.headers.get("Content-Length", "0") or "0") if length <= 0: return {} try: return json.loads(self.rfile.read(length) or b"{}") except ValueError: return {} def log_message(self, fmt, *args): log.debug("%s - %s", self.address_string(), fmt % args) # ---- ルーティング ---- def do_GET(self): path = urllib.parse.urlparse(self.path).path if path == "/healthz": return self._send(200, b'{"ok":true}') if path == "/auth": # nginx auth_request 用(token モードのみ意味を持つ) if token_ok(self): orig = self.headers.get("X-Original-URI", "") m = re.search(r"/s/([a-f0-9]{10})/", orig) if m: try: touch_session(m.group(1)) except Exception: # noqa: BLE001 pass return self._send(200, b"ok", "text/plain") return self._send(401, b"unauthorized", "text/plain") if path == "/" or path == "/index.html": return self._send(200, INDEX_HTML, "text/html; charset=utf-8") if path == "/api/config": if not token_ok(self): return self._json(401, {"error": "unauthorized"}) return self._json(200, { "identities": list_identities(), "guiIdentities": GUI_IDENTITIES_ENABLED, "idleTimeoutSeconds": IDLE_TIMEOUT, "maxLifetimeSeconds": MAX_LIFETIME, "maxSessions": MAX_SESSIONS, "authMode": AUTH_MODE, }) if path == "/api/identities": if not token_ok(self): return self._json(401, {"error": "unauthorized"}) return self._json(200, {"identities": list_identities()}) if path == "/api/sessions": if not token_ok(self): return self._json(401, {"error": "unauthorized"}) return self._json(200, {"sessions": list_sessions()}) m = re.match(r"^/api/sessions/([a-f0-9]{10})/log$", path) if m: if not token_ok(self): return self._json(401, {"error": "unauthorized"}) status, text = session_log(m.group(1)) return self._send(status if status == 200 else 200, text or "(ログはまだありません)", "text/plain; charset=utf-8") return self._send(404, b'{"error":"not found"}') def do_POST(self): path = urllib.parse.urlparse(self.path).path if path == "/login": body = self._body() token = str(body.get("token", "")) if AUTH_MODE == "token" and AUTH_TOKEN and hmac.compare_digest(token, AUTH_TOKEN): cookie = ("cc_token=" + urllib.parse.quote(token) + "; HttpOnly; SameSite=Lax; Path=/; Max-Age=2592000") return self._json(200, {"ok": True}, {"Set-Cookie": cookie}) return self._json(401, {"error": "invalid token"}) if not token_ok(self): return self._json(401, {"error": "unauthorized"}) if path == "/api/sessions": body = self._body() owner = self.headers.get("X-Forwarded-User", "") or "anon" sid, err = create_session(body.get("repo", ""), body.get("ref", "main"), body.get("identity", ""), owner) if err: return self._json(400, {"error": err}) return self._json(201, {"id": sid, "url": "/s/" + sid + "/"}) if path == "/api/identities": code, payload = create_gui_identity(self._body()) return self._json(code, payload) m = re.match(r"^/api/sessions/([a-f0-9]{10})/touch$", path) if m: touch_session(m.group(1), min_interval=0) return self._json(200, {"ok": True}) return self._send(404, b'{"error":"not found"}') def do_DELETE(self): if not token_ok(self): return self._json(401, {"error": "unauthorized"}) path = urllib.parse.urlparse(self.path).path m = re.match(r"^/api/sessions/([a-f0-9]{10})$", path) if m: delete_session(m.group(1)) return self._json(200, {"ok": True}) m = re.match(r"^/api/identities/([a-z0-9][a-z0-9-]{0,39})$", path) if m: code, payload = delete_gui_identity(m.group(1)) return self._json(code, payload) return self._send(404, b'{"error":"not found"}') def main(): log.info("claude-code manager starting: ns=%s image=%s idle=%ds maxLife=%ds maxSessions=%d authMode=%s", SESSIONS_NS, SESSION_IMAGE, IDLE_TIMEOUT, MAX_LIFETIME, MAX_SESSIONS, AUTH_MODE) if AUTH_MODE == "token" and not AUTH_TOKEN: log.warning("AUTH_MODE=token だがトークンが空。全リクエストを拒否します。") threading.Thread(target=reaper_loop, name="reaper", daemon=True).start() httpd = ThreadingHTTPServer(("0.0.0.0", LISTEN_PORT), Handler) httpd.serve_forever() if __name__ == "__main__": import sys if "--reap-once" in sys.argv[1:]: # 予備の CronJob から呼ばれる一回きりの回収。manager 障害時の保険。 log.info("one-shot reap starting (ns=%s)", SESSIONS_NS) reap_once() log.info("one-shot reap done") else: main()