feat: support gateway access keys

codex_retry_gateway_tui.py

@@ -28,6 +28,7 @@ DEFAULT_GATEWAY_URL = "http://127.0.0.1:4610/__codex_retry_gateway"
 DEFAULT_API_URL = DEFAULT_GATEWAY_URL
 DEFAULT_CONFIG_FILE = "~/.config/codex-retry-gateway-tui/api-url"
 DEFAULT_STATUS_CONFIG_FILE = "~/.config/codex-retry-gateway-tui/status-url"
+DEFAULT_ACCESS_KEY_FILE = "~/.config/codex-retry-gateway-tui/access-key"
 DEFAULT_GATEWAY_STATE_FILE = "~/.codex-retry-gateway/state.json"
 DEFAULT_GATEWAY_JSON_CONFIG_FILE = "~/.codex-retry-gateway/config/config.json"
 DEFAULT_VERSION_CHECK_URL = "https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui/raw/branch/main/pyproject.toml"
@@ -143,6 +144,10 @@ def config_file_path() -> Path:
     return Path(os.environ.get("CODEX_RETRY_GATEWAY_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser()
 
 
+def access_key_file_path() -> Path:
+    return Path(os.environ.get("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY_FILE", DEFAULT_ACCESS_KEY_FILE)).expanduser()
+
+
 def default_status_url() -> str:
     return configured_url(
         ("CODEX_RETRY_GATEWAY_TUI_STATUS_URL",),
@@ -151,6 +156,14 @@ def default_status_url() -> str:
     )
 
 
+def default_access_key() -> str:
+    return configured_url(
+        ("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY",),
+        os.environ.get("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY_FILE", DEFAULT_ACCESS_KEY_FILE),
+        "",
+    )
+
+
 def normalize_gateway_url(api_url: str) -> str:
     value = str(api_url or "").strip()
     if not value:
@@ -214,6 +227,21 @@ def write_api_url_config(api_url: str) -> Path:
     return path
 
 
+def write_access_key_config(access_key: str) -> Path:
+    path = access_key_file_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        path.parent.chmod(0o700)
+    except OSError:
+        pass
+    path.write_text(str(access_key or "").strip() + "\n", encoding="utf-8")
+    try:
+        path.chmod(0o600)
+    except OSError:
+        pass
+    return path
+
+
 def run_install_command() -> int:
     try:
         return subprocess.run(INSTALL_COMMAND_ARGS, check=False).returncode
@@ -468,8 +496,15 @@ def status_kind(value: Any) -> str:
     return "unknown"
 
 
-def fetch_payload(api_url: str, timeout: int) -> dict[str, Any]:
+def request_headers(access_key: str = "", *, accept: str = "application/json") -> dict[str, str]:
-    req = urllib.request.Request(api_url, headers={"Accept": "application/json"})
+    headers = {"Accept": accept}
+    if str(access_key or "").strip():
+        headers["x-codex-retry-gateway-key"] = str(access_key).strip()
+    return headers
+
+
+def fetch_payload(api_url: str, timeout: int, access_key: str = "") -> dict[str, Any]:
+    req = urllib.request.Request(api_url, headers=request_headers(access_key))
     with urllib.request.urlopen(req, timeout=timeout) as response:
         data = json.loads(response.read().decode("utf-8"))
     if not isinstance(data, dict):
@@ -477,11 +512,11 @@ def fetch_payload(api_url: str, timeout: int) -> dict[str, Any]:
     return data
 
 
-def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]:
+def fetch_optional_payload(url: str, timeout: int, access_key: str = "") -> tuple[dict[str, Any], str]:
     if not str(url or "").strip():
         return {}, ""
     try:
-        return fetch_payload(url, timeout), ""
+        return fetch_payload(url, timeout, access_key), ""
     except Exception as exc:
         return {}, str(exc)
 
@@ -1004,11 +1039,11 @@ def action_url(api_url: str, suffix: str) -> str:
     return f"{base}{suffix}"
 
 
-def post_json(url: str, timeout: int, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+def post_json(url: str, timeout: int, payload: dict[str, Any], access_key: str = "") -> tuple[int, dict[str, Any]]:
     req = urllib.request.Request(
         url,
         data=json.dumps(payload).encode("utf-8"),
-        headers={"Content-Type": "application/json", "Accept": "application/json"},
+        headers={**request_headers(access_key), "Content-Type": "application/json"},
         method="POST",
     )
     with urllib.request.urlopen(req, timeout=timeout) as response:
@@ -1018,8 +1053,8 @@ def post_json(url: str, timeout: int, payload: dict[str, Any]) -> tuple[int, dic
         return response.status, data
 
 
-def delete_json(url: str, timeout: int) -> tuple[int, dict[str, Any]]:
+def delete_json(url: str, timeout: int, access_key: str = "") -> tuple[int, dict[str, Any]]:
-    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="DELETE")
+    req = urllib.request.Request(url, headers=request_headers(access_key), method="DELETE")
     with urllib.request.urlopen(req, timeout=timeout) as response:
         data = json.loads(response.read().decode("utf-8"))
         if not isinstance(data, dict):
@@ -1042,6 +1077,7 @@ def fetch_dashboard_snapshot(
     status_url: str,
     timeout: int,
     *,
+    access_key: str = "",
     filter_text: str = "",
     current_view: str = "overview",
 ) -> dict[str, Any]:
@@ -1054,7 +1090,7 @@ def fetch_dashboard_snapshot(
     }
 
     def fetch_optional(url: str) -> tuple[dict[str, Any], str]:
-        return fetch_optional_payload(url, timeout)
+        return fetch_optional_payload(url, timeout, access_key)
 
     with ThreadPoolExecutor(max_workers=4) as pool:
         futures = {
@@ -1092,7 +1128,14 @@ def fetch_dashboard_snapshot(
     }
 
 
-def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: int, version_message: str = "") -> int:
+def run_textual(
+    api_url: str,
+    status_url: str,
+    refresh_seconds: int,
+    timeout: int,
+    version_message: str = "",
+    access_key: str = "",
+) -> int:
     try:
         from textual.app import App, ComposeResult
         from textual.coordinate import Coordinate
@@ -1350,6 +1393,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                     gateway_base_url,
                     status_api_url,
                     timeout,
+                    access_key=access_key,
                     filter_text=current_filter,
                     current_view=self.current_view,
                 )
@@ -1588,7 +1632,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                 if payload is None:
                     self._set_status(f"edit {row['name']} cancelled (no changes)")
                     return
-                status, result = post_json(build_api_url(gateway_base_url, "/api/profiles"), timeout, payload)
+                status, result = post_json(build_api_url(gateway_base_url, "/api/profiles"), timeout, payload, access_key)
                 self._set_status(f"edit {payload.get('name') or row['name']} -> {status} | {short_text(result, 96)}")
                 self.refresh_data(refresh=True)
             except Exception as exc:
@@ -1603,6 +1647,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                     build_api_url(gateway_base_url, "/api/profiles/probe"),
                     timeout,
                     {"profile": row["name"]},
+                    access_key,
                 )
                 self._set_status(f"probe {row['name']} -> {status} | {short_text(payload, 96)}")
             except Exception as exc:
@@ -1617,6 +1662,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                     build_api_url(gateway_base_url, "/api/profiles/switch"),
                     timeout,
                     {"profile": row["name"]},
+                    access_key,
                 )
                 self._set_status(f"switch {row['name']} -> {status} | {short_text(payload, 96)}")
                 self.refresh_data(refresh=True)
@@ -1629,7 +1675,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
                 return
             payload = profile_payload_from_row(row)
             try:
-                status, result = post_json(build_api_url(gateway_base_url, "/api/profiles"), timeout, payload)
+                status, result = post_json(build_api_url(gateway_base_url, "/api/profiles"), timeout, payload, access_key)
                 self._set_status(f"save profile -> {status} | {short_text(result, 96)}")
                 self.refresh_data(refresh=True)
             except Exception as exc:
@@ -1640,7 +1686,7 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
             if not row or row["active"]:
                 return
             try:
-                status, payload = delete_json(build_api_url(gateway_base_url, f"/api/profiles/{urllib.parse.quote(row['name'])}"), timeout)
+                status, payload = delete_json(build_api_url(gateway_base_url, f"/api/profiles/{urllib.parse.quote(row['name'])}"), timeout, access_key)
                 self._set_status(f"delete {row['name']} -> {status} | {short_text(payload, 96)}")
                 self.refresh_data(refresh=True)
             except Exception as exc:
@@ -1674,6 +1720,11 @@ def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: in
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Codex Retry Gateway TUI")
     parser.add_argument("--api-url", default=default_api_url())
+    parser.add_argument(
+        "--access-key",
+        default=default_access_key(),
+        help="optional management access key for protected gateway admin pages and APIs",
+    )
     parser.add_argument(
         "--status-url",
         default=default_status_url(),
@@ -1744,10 +1795,14 @@ def print_once(snapshot: dict[str, Any], filter_text: str = "") -> None:
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     args.api_url = gateway_admin_url(args.api_url)
+    access_key = str(args.access_key or "").strip()
     status_url = str(args.status_url or "").strip() or default_status_url()
     if args.save_config or args.install:
         config_path = write_api_url_config(args.api_url)
         print(f"saved api url to {config_path}")
+        if access_key:
+            key_path = write_access_key_config(access_key)
+            print(f"saved access key to {key_path}")
     if args.install:
         print("installing codex-retry-gateway-tui with uv tool...")
         return run_install_command()
@@ -1763,12 +1818,20 @@ def main(argv: list[str] | None = None) -> int:
             args.api_url,
             status_url,
             args.timeout,
+            access_key=access_key,
             filter_text=args.filter,
             current_view="overview",
         )
         print_once(snapshot, args.filter)
         return 0
-    return run_textual(args.api_url, status_url, max(1, args.refresh_seconds), max(1, args.timeout), version_message)
+    return run_textual(
+        args.api_url,
+        status_url,
+        max(1, args.refresh_seconds),
+        max(1, args.timeout),
+        version_message,
+        access_key,
+    )
 
 
 if __name__ == "__main__":