diff --git a/codex_retry_gateway_tui.py b/codex_retry_gateway_tui.py index 7dceebf..7bd23e5 100644 --- a/codex_retry_gateway_tui.py +++ b/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]: - req = urllib.request.Request(api_url, headers={"Accept": "application/json"}) +def request_headers(access_key: str = "", *, accept: str = "application/json") -> dict[str, str]: + 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]]: - req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="DELETE") +def delete_json(url: str, timeout: int, access_key: str = "") -> tuple[int, dict[str, Any]]: + 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__":