#!/usr/bin/env python3 """Request and resolve TapAuth static-secret bundles without caching plaintext.""" from __future__ import annotations import argparse import hashlib import json import os import re import shutil import stat import subprocess import sys import tempfile import time from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener KEY_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") ITEM_KEY_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$") SECRET_ID_RE = re.compile(r"^sec_[A-Za-z0-9_-]{32,}$") RESOLVE_TOKEN_RE = re.compile(r"^sr_[A-Za-z0-9_-]{32,}$") AGE_RECIPIENT_RE = re.compile(r"^age1[ac-hj-np-z02-9]{58}$") MAX_RESPONSE_BYTES = 128 * 1024 MAX_SCHEMA_BYTES = 64 * 1024 class ApiError(Exception): def __init__(self, status: int, message: str): super().__init__(message) self.status = status class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): del req, fp, code, msg, headers, newurl return None def die(message: str) -> None: print(f"tapauth-static: {message}", file=sys.stderr) raise SystemExit(1) def private_dir(path: Path) -> Path: if path.is_symlink(): die(f"directory must not be a symlink: {path}") existed = path.exists() path.mkdir(parents=True, exist_ok=True, mode=0o700) info = path.stat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): die(f"directory must be owned by the current user: {path}") if existed and info.st_mode & (stat.S_IRWXG | stat.S_IRWXO): die(f"directory must be private (mode 700): {path}") path.chmod(0o700) return path def private_file(path: Path) -> None: try: info = path.lstat() except OSError: die(f"required private file is missing: {path}") if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode): die(f"private file must be a regular file: {path}") if info.st_uid != os.getuid() or info.st_mode & (stat.S_IRWXG | stat.S_IRWXO): die(f"private file must be user-owned with mode 600: {path}") def cache_root() -> Path: configured = os.environ.get("TAPAUTH_HOME") if configured: root = Path(configured).expanduser() elif os.environ.get("CLAUDE_PLUGIN_DATA"): root = Path(os.environ["CLAUDE_PLUGIN_DATA"]).expanduser() else: root = Path.home() / ".tapauth" return private_dir(root) def static_dir(root: Path) -> Path: return private_dir(root / "static") def state_path(root: Path, key_id: str) -> Path: return static_dir(root) / f"{key_id}.json" def identity_path(root: Path, key_id: str) -> Path: return static_dir(root) / f"{key_id}.agekey" def atomic_json(path: Path, value: dict) -> None: directory = private_dir(path.parent) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=directory) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(value, handle, separators=(",", ":"), ensure_ascii=False) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) path.chmod(0o600) finally: try: os.unlink(temporary) except FileNotFoundError: pass def load_state(root: Path, key_id: str) -> dict: path = state_path(root, key_id) private_file(path) if path.stat().st_size > MAX_SCHEMA_BYTES: die("cached static-secret state is too large") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): die("cached static-secret state is invalid") if not isinstance(value, dict) or value.get("key_id") != key_id: die("cached static-secret state has the wrong key id") if not SECRET_ID_RE.fullmatch(str(value.get("id", ""))): die("cached static-secret id is invalid") if not RESOLVE_TOKEN_RE.fullmatch(str(value.get("resolve_token", ""))): die("cached static-secret resolve token is invalid") if value.get("identity_file") != str(identity_path(root, key_id)): die("cached static-secret identity path is invalid") items = value.get("items") if not isinstance(items, list) or not items or not all(isinstance(item, str) and ITEM_KEY_RE.fullmatch(item) for item in items): die("cached static-secret item list is invalid") return value def base_url() -> str: raw = os.environ.get("TAPAUTH_BASE_URL", "https://tapauth.ai") if raw != raw.strip() or len(raw) > 2048: die("TAPAUTH_BASE_URL is invalid") parsed = urlsplit(raw) loopback = parsed.hostname in {"localhost", "127.0.0.1", "::1"} if ( not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or (parsed.scheme != "https" and not (parsed.scheme == "http" and loopback)) ): die("TAPAUTH_BASE_URL must use HTTPS, except for loopback testing") return raw.rstrip("/") def read_limited(response) -> bytes: length = response.headers.get("Content-Length") if length: try: if int(length) > MAX_RESPONSE_BYTES: raise ApiError(response.status, "response is too large") except ValueError as error: raise ApiError(response.status, "response has invalid length") from error data = response.read(MAX_RESPONSE_BYTES + 1) if len(data) > MAX_RESPONSE_BYTES: raise ApiError(response.status, "response is too large") return data def api_json(method: str, path: str, body: dict | None = None, token: str = "") -> dict: payload = None headers = {"Accept": "application/json"} if body is not None: payload = json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode() headers["Content-Type"] = "application/json" if token: headers["Authorization"] = f"Bearer {token}" request = Request(f"{base_url()}{path}", data=payload, headers=headers, method=method) try: with build_opener(NoRedirect).open(request, timeout=15) as response: raw = read_limited(response) status = response.status except HTTPError as error: raw = error.read(MAX_RESPONSE_BYTES + 1) if len(raw) > MAX_RESPONSE_BYTES: raise ApiError(error.code, "response is too large") from error try: detail = json.loads(raw).get("error", "request failed") except (json.JSONDecodeError, AttributeError): detail = "request failed" detail = str(detail) if len(detail) > 500 or any(ord(char) < 0x20 or ord(char) == 0x7F for char in detail): detail = "request failed" raise ApiError(error.code, detail) from error except (URLError, TimeoutError, OSError) as error: raise ApiError(0, "failed to contact TapAuth") from error if status == 202: raise ApiError(status, "pending") try: value = json.loads(raw) except json.JSONDecodeError as error: raise ApiError(status, "TapAuth returned invalid JSON") from error if not isinstance(value, dict): raise ApiError(status, "TapAuth returned an invalid response") return value def trusted_binary(name: str) -> str: found = shutil.which(name) if not found: die(f"{name} is required for static-secret support") path = Path(found).resolve(strict=True) info = path.stat() if not path.is_file() or not os.access(path, os.X_OK): die(f"{name} is not executable") if info.st_uid not in {0, os.getuid()} or info.st_mode & (stat.S_IWGRP | stat.S_IWOTH): die(f"{name} must not be writable by another user") return str(path) def child_env() -> dict[str, str]: keep = ("PATH", "HOME", "USERPROFILE", "SYSTEMROOT", "TMPDIR", "TEMP", "LANG", "LC_ALL") result = {key: os.environ[key] for key in keep if os.environ.get(key)} result["NO_COLOR"] = "1" return result def ensure_identity(root: Path, key_id: str) -> tuple[Path, str]: path = identity_path(root, key_id) keygen = trusted_binary("age-keygen") if not path.exists(): temporary = path.with_name(f".{path.name}.{os.getpid()}") try: subprocess.run( [keygen, "-o", str(temporary)], env=child_env(), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15, check=True, ) temporary.chmod(0o600) try: os.link(temporary, path) except FileExistsError: pass except (OSError, subprocess.SubprocessError): die("failed to generate age identity") finally: try: temporary.unlink() except FileNotFoundError: pass private_file(path) try: process = subprocess.run( [keygen, "-y", str(path)], env=child_env(), stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=15, check=True, ) except (OSError, subprocess.SubprocessError): die("failed to derive age recipient") recipient = process.stdout.strip() if not AGE_RECIPIENT_RE.fullmatch(recipient): die("age-keygen returned an invalid recipient") return path, recipient def trimmed(value, field: str, maximum: int, required: bool = False): if value is None: if required: die(f"{field} is required") return None if not isinstance(value, str): die(f"{field} must be a string") result = value.strip() if required and not result: die(f"{field} is required") if len(result) > maximum: die(f"{field} is too long") return result or None def load_schema(path: Path) -> tuple[dict, str]: try: if path.stat().st_size > MAX_SCHEMA_BYTES: die("schema file is too large") source = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): die("schema file must contain valid JSON") if not isinstance(source, dict) or not isinstance(source.get("items"), list): die("schema must contain an items array") if not 1 <= len(source["items"]) <= 50: die("schema must contain between 1 and 50 items") normalized = [] seen = set() for index, raw in enumerate(source["items"]): if not isinstance(raw, dict): die(f"items[{index}] must be an object") key = trimmed(raw.get("key"), f"items[{index}].key", 128, True) if not ITEM_KEY_RE.fullmatch(key) or key in seen: die(f"items[{index}].key is invalid or duplicated") seen.add(key) if "required" in raw and not isinstance(raw["required"], bool): die(f"items[{index}].required must be a boolean") normalized.append( { "key": key, "description": trimmed(raw.get("description"), f"items[{index}].description", 500), "validation_regex": trimmed(raw.get("validation_regex"), f"items[{index}].validation_regex", 500), "validation_hint": trimmed(raw.get("validation_hint"), f"items[{index}].validation_hint", 500), "required": raw.get("required") is not False, } ) expires = source.get("expires_in", 86400) if ( isinstance(expires, bool) or not isinstance(expires, int) or not 60 <= expires <= 30 * 24 * 60 * 60 ): die("expires_in must be between 60 seconds and 30 days") schema_json = json.dumps({"items": normalized}, separators=(",", ":"), ensure_ascii=False) return { "description": trimmed(source.get("description"), "description", 1000), "expires_in": expires, "items": normalized, }, hashlib.sha256(schema_json.encode()).hexdigest() def valid_created(value: dict, key_id: str) -> None: if value.get("key_id") != key_id: die("TapAuth returned the wrong key id") if not SECRET_ID_RE.fullmatch(str(value.get("id", ""))): die("TapAuth returned an invalid secret id") if not RESOLVE_TOKEN_RE.fullmatch(str(value.get("resolve_token", ""))): die("TapAuth returned an invalid resolve token") if value.get("status") not in {"pending", "active"}: die("TapAuth returned an invalid static-secret status") approval = str(value.get("approval_url", "")) approval_url = urlsplit(approval) service_url = urlsplit(base_url()) try: approval_port = approval_url.port or (443 if approval_url.scheme == "https" else 80) service_port = service_url.port or (443 if service_url.scheme == "https" else 80) except ValueError: die("TapAuth returned an invalid approval URL") if ( approval_url.scheme != service_url.scheme or approval_url.hostname != service_url.hostname or approval_port != service_port or not approval_url.fragment or approval_url.username is not None or approval_url.password is not None or len(approval) > 4096 or any(ord(char) < 0x20 or ord(char) == 0x7F for char in approval) ): die("TapAuth returned an invalid approval URL") def print_approval(state: dict) -> None: print(f"Approve static secret: {state['approval_url']}") print() print("Show this URL to the user, then start --token immediately; it waits until approval completes.") def create_request(args, root: Path) -> None: schema, schema_hash = load_schema(Path(args.schema)) old_state = None path = state_path(root, args.key_id) if path.exists(): old_state = load_state(root, args.key_id) if not args.fresh: if old_state.get("schema_hash") != schema_hash: die("cached schema differs; use --fresh to request replacement input") try: status = api_json("GET", f"/api/v1/secrets/{old_state['id']}").get("status") except ApiError as error: if error.status not in {404, 410}: die(str(error)) status = "missing" if status == "active": print(f"Already authorized for static key {args.key_id}. Use --token to retrieve an item.") return if status == "pending": print_approval(old_state) return identity, recipient = ensure_identity(root, args.key_id) body = { "key_id": args.key_id, "name": args.name, "description": schema["description"], "age_recipient": recipient, "expires_in": schema["expires_in"], "items": schema["items"], } try: created = api_json("POST", "/api/v1/secrets", body) except ApiError as error: die(str(error)) valid_created(created, args.key_id) state = { "id": created["id"], "key_id": args.key_id, "resolve_token": created["resolve_token"], "approval_url": created["approval_url"], "identity_file": str(identity), "schema_hash": schema_hash, "schema_version": 1, "items": [item["key"] for item in schema["items"]], } atomic_json(path, state) if old_state and old_state.get("id") != state["id"]: try: api_json("DELETE", f"/api/v1/secrets/{old_state['id']}", token=old_state["resolve_token"]) except ApiError: pass print_approval(state) def decrypt(ciphertext: str, identity: Path) -> dict: private_file(identity) age = trusted_binary("age") try: process = subprocess.run( [age, "--decrypt", "-i", str(identity)], input=ciphertext, env=child_env(), capture_output=True, text=True, timeout=30, check=True, ) except (OSError, subprocess.SubprocessError): die("failed to decrypt static secret") if len(process.stdout.encode()) > MAX_RESPONSE_BYTES: die("decrypted static secret is too large") try: value = json.loads(process.stdout) except json.JSONDecodeError: die("decrypted static secret is not valid JSON") if not isinstance(value, dict) or not all(isinstance(key, str) and isinstance(item, str) for key, item in value.items()): die("decrypted static secret must be a string map") return value def resolve_item(args, root: Path) -> None: state = load_state(root, args.key_id) if args.item not in state.get("items", []): die("requested item is not in the cached schema") try: timeout = int(os.environ.get("TAPAUTH_POLL_TIMEOUT_SECONDS", "600")) except ValueError: die("TAPAUTH_POLL_TIMEOUT_SECONDS must be an integer") if timeout < 0: die("TAPAUTH_POLL_TIMEOUT_SECONDS must not be negative") started = time.monotonic() while True: try: response = api_json( "GET", f"/api/v1/secrets/{state['id']}/resolve", token=state["resolve_token"], ) break except ApiError as error: if error.status != 202: die(str(error)) if time.monotonic() - started >= timeout: die("timed out") time.sleep(2) if response.get("id") != state["id"] or response.get("key_id") != args.key_id: die("TapAuth returned the wrong static-secret identity") if response.get("status") != "active": die("TapAuth returned an invalid static-secret status") if response.get("schema_hash") != state.get("schema_hash"): die("TapAuth returned an unexpected static-secret schema") if ( type(response.get("schema_version")) is not int or response.get("schema_version") != state.get("schema_version") ): die("TapAuth returned an unexpected static-secret schema version") if ( type(response.get("ciphertext_version")) is not int or response["ciphertext_version"] < 1 ): die("TapAuth returned an invalid static-secret ciphertext version") ciphertext = response.get("ciphertext") if not isinstance(ciphertext, str) or not ciphertext or len(ciphertext.encode()) > MAX_RESPONSE_BYTES: die("TapAuth returned invalid static-secret ciphertext") values = decrypt(ciphertext, Path(state["identity_file"])) value = values.get(args.item) if not isinstance(value, str) or not value: die("resolved static-secret item is missing or empty") print(value) def revoke(args, root: Path) -> None: state = load_state(root, args.key_id) try: api_json("DELETE", f"/api/v1/secrets/{state['id']}", token=state["resolve_token"]) except ApiError as error: if error.status not in {404, 410}: die(str(error)) state_path(root, args.key_id).unlink(missing_ok=True) print(f"Revoked static key: {args.key_id}") def parse_args(): parser = argparse.ArgumentParser(prog="tapauth-static") action = parser.add_mutually_exclusive_group() action.add_argument("--token", action="store_true", help="resolve one approved item") action.add_argument("--revoke", action="store_true", help="revoke the cached request") parser.add_argument("--fresh", "--no-cache", dest="fresh", action="store_true") parser.add_argument("--key-id", required=True) parser.add_argument("--name") parser.add_argument("--schema") parser.add_argument("--item") args = parser.parse_args() if not KEY_ID_RE.fullmatch(args.key_id): die("--key-id must be a stable machine-readable id") if args.token: if args.fresh or args.name or args.schema or not args.item or not ITEM_KEY_RE.fullmatch(args.item): die("--token requires --item and does not accept --fresh") elif args.revoke: if args.fresh or args.item or args.name or args.schema: die("--revoke accepts only --key-id") elif not args.name or not args.schema or args.item: die("request mode requires --name and --schema only") return args def main() -> None: args = parse_args() root = cache_root() if args.token: resolve_item(args, root) elif args.revoke: revoke(args, root) else: create_request(args, root) if __name__ == "__main__": main()