fix: secure login flow with copy-paste API key exchange#337
Conversation
|
Promptless prepared a documentation update related to this change. Triggered by runpod/flash#337 Updated the |
There was a problem hiding this comment.
Pull request overview
This PR updates the flash CLI login flow to avoid fetching newly-issued API keys over an unauthenticated polling endpoint by switching to a user copy/paste exchange, and removes the no-longer-needed GraphQL polling method.
Changes:
- Replace the CLI’s polling-based login with a copy/paste API key prompt; remove
--timeoutand related polling/deadline logic. - Remove
get_flash_auth_request_statusfrom the GraphQL client and update unit tests accordingly. - Introduce new SSE log parsing/streaming helpers in
request_logs.pyand broaden the supported Python version range in packaging metadata.
Reviewed changes
Copilot reviewed 6 out of 7 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/unit/test_login.py | Updates unit tests to match the copy/paste login flow and removes polling-related coverage. |
| tests/unit/test_login_extended.py | Updates extended login/GraphQL tests to remove polling status checks and align with paste-based login. |
| src/runpod_flash/core/resources/request_logs.py | Adds SSE event/log parsing and a pod log streaming generator. |
| src/runpod_flash/core/api/runpod.py | Removes the polling query method get_flash_auth_request_status. |
| src/runpod_flash/cli/commands/login.py | Implements copy/paste API key login flow and removes timeout/polling logic. |
| pyproject.toml | Relaxes requires-python from >=3.10,<3.13 to >=3.10. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| if status in {"DENIED", "EXPIRED", "CONSUMED"}: | ||
| raise RuntimeError(f"login failed: {status.lower()}") | ||
| api_key = console.input("Paste the API key shown after authorization: ").strip() |
| async with get_authenticated_httpx_client() as client: | ||
| async with client.get(url) as response: | ||
| async for line in response.aiter_lines(): | ||
| event = parse_sse_event(line) | ||
| if event: | ||
| log_event = parse_log_event(event.data) | ||
| if log_event: | ||
| yield log_event |
| def parse_sse_event(data: str) -> Optional[SSEEvent]: | ||
| """ | ||
| Parses an SSE line into a dictionary | ||
| """ | ||
| if not data: | ||
| return None | ||
|
|
||
| try: | ||
| event_id_line, data_line = filter(bool, data.split("\n")) | ||
| event_id = event_id_line.split(":", 1)[1].strip() | ||
| data_json = data_line.split(":", 1)[1].strip() | ||
| data = json.loads(data_json) | ||
| return SSEEvent(id=event_id, data=data) | ||
| except Exception as e: | ||
| log.error("Failed to parse SSE event: %s", e) | ||
| return None |
| Streams logs from pod using SSE | ||
| """ | ||
| if tail < 0: | ||
| raise ValueError("tail must be greater than 0") |
| @dataclass | ||
| class SSEEvent: | ||
| id: str | ||
| data: dict[str, Any] | ||
|
|
||
|
|
||
| @dataclass | ||
| class LogEvent: | ||
| source: str | ||
| line: str | ||
| ts: str | ||
|
|
||
|
|
||
| def parse_sse_event(data: str) -> Optional[SSEEvent]: | ||
| """ | ||
| Parses an SSE line into a dictionary | ||
| """ | ||
| if not data: | ||
| return None | ||
|
|
||
| try: | ||
| event_id_line, data_line = filter(bool, data.split("\n")) | ||
| event_id = event_id_line.split(":", 1)[1].strip() | ||
| data_json = data_line.split(":", 1)[1].strip() | ||
| data = json.loads(data_json) | ||
| return SSEEvent(id=event_id, data=data) | ||
| except Exception as e: | ||
| log.error("Failed to parse SSE event: %s", e) | ||
| return None | ||
|
|
||
|
|
||
| def parse_log_event(data: dict[str, Any]) -> Optional[LogEvent]: | ||
| """ | ||
| Parses a log event from a dictionary | ||
| """ | ||
| try: | ||
| return LogEvent(source=data["source"], line=data["line"], ts=data["ts"]) | ||
| except Exception as e: | ||
| log.error("Failed to parse log event: %s", e) | ||
| return None | ||
|
|
||
|
|
||
| async def stream_pod_logs(pod_id: str, tail: int = 0): | ||
| """ | ||
| Streams logs from pod using SSE | ||
| """ | ||
| if tail < 0: | ||
| raise ValueError("tail must be greater than 0") | ||
|
|
||
| url = f"{RUNPOD_HAPI_URL}/v1/pod/{pod_id}/logs?stream=true&tail={tail}" | ||
|
|
||
| async with get_authenticated_httpx_client() as client: | ||
| async with client.get(url) as response: | ||
| async for line in response.aiter_lines(): | ||
| event = parse_sse_event(line) | ||
| if event: | ||
| log_event = parse_log_event(event.data) | ||
| if log_event: | ||
| yield log_event | ||
|
|
| @dataclass | ||
| class SSEEvent: | ||
| id: str | ||
| data: dict[str, Any] | ||
|
|
||
|
|
||
| @dataclass | ||
| class LogEvent: | ||
| source: str | ||
| line: str | ||
| ts: str | ||
|
|
||
|
|
||
| def parse_sse_event(data: str) -> Optional[SSEEvent]: | ||
| """ | ||
| Parses an SSE line into a dictionary | ||
| """ | ||
| if not data: | ||
| return None | ||
|
|
||
| try: | ||
| event_id_line, data_line = filter(bool, data.split("\n")) | ||
| event_id = event_id_line.split(":", 1)[1].strip() | ||
| data_json = data_line.split(":", 1)[1].strip() | ||
| data = json.loads(data_json) | ||
| return SSEEvent(id=event_id, data=data) | ||
| except Exception as e: | ||
| log.error("Failed to parse SSE event: %s", e) | ||
| return None | ||
|
|
||
|
|
||
| def parse_log_event(data: dict[str, Any]) -> Optional[LogEvent]: | ||
| """ | ||
| Parses a log event from a dictionary | ||
| """ | ||
| try: | ||
| return LogEvent(source=data["source"], line=data["line"], ts=data["ts"]) | ||
| except Exception as e: | ||
| log.error("Failed to parse log event: %s", e) | ||
| return None | ||
|
|
||
|
|
||
| async def stream_pod_logs(pod_id: str, tail: int = 0): | ||
| """ | ||
| Streams logs from pod using SSE | ||
| """ |
deanq
left a comment
deanq
left a comment
There was a problem hiding this comment.
AE-3128 review — login flow security. A few suggestions on the CLI paste UX.
|
|
||
| if status in {"DENIED", "EXPIRED", "CONSUMED"}: | ||
| raise RuntimeError(f"login failed: {status.lower()}") | ||
| api_key = console.input("Paste the API key shown after authorization: ").strip() |
There was a problem hiding this comment.
This echoes the key to the terminal — it ends up in scrollback, screen recordings, and shoulder-surfing range. For a security-focused PR, prefer rich.prompt.Prompt.ask("Paste the API key", password=True) or getpass.getpass. The user already has the key visible in the browser; they don't need it echoed again here.
| if dt.datetime.now(dt.timezone.utc) >= deadline: | ||
| raise RuntimeError("login timed out") | ||
| if not api_key: | ||
| raise RuntimeError("no api key provided") |
There was a problem hiding this comment.
Right now any non-empty string gets written to the credentials file. Worth at minimum a prefix check (e.g. rpa_) so a fat-fingered paste fails loudly. Even better: fire a cheap authenticated call (e.g. myself) to confirm the key works before persisting, so users don't discover the bad paste on the next command.
|
|
||
| if status in {"DENIED", "EXPIRED", "CONSUMED"}: | ||
| raise RuntimeError(f"login failed: {status.lower()}") | ||
| api_key = console.input("Paste the API key shown after authorization: ").strip() |
There was a problem hiding this comment.
A wrong/empty paste currently forces the user to rerun flash login, which means re-approving in the browser. A small retry loop (e.g. 3 attempts) would be friendlier without weakening security.
the old login flow had the CLI poll for the API key after browser approval. any process that knew the request ID could intercept the key via the same unauthenticated query.
replaces polling with a copy-paste flow: the browser displays the generated API key after approval, and the user pastes it into the CLI prompt. the CLI never fetches the key over the network.
also removes the
get_flash_auth_request_statuspolling method from the GraphQL client since it is no longer needed.AE-3128