diff --git a/CLAUDE.md b/CLAUDE.md index 2134da43..ab660fed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -333,7 +333,7 @@ kbagent auth register-projects [--stack URL|alias] [--all] [--project-id ID ...] # instead of silently skipping the second project). See docs/programmatic-auth-login-plan.md # section 4.5 for the full design. -kbagent auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE] [--read-only] [--ttl-days N] +kbagent auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE | --webauthn] [--read-only] [--ttl-days N] [--project-id ID ...] kbagent auth pat-revoke PAT_ID [--stack URL|alias] [--yes] # `auth pat-*` (0.81.0+): mints/revokes a Personal Access Token (`kbc_pat_...`) for one-time CI/CD # setup -- the sanctioned alternative to a raw Storage token for pipelines, and the only session-auth diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 9f50a83f..2319fa76 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -25,7 +25,7 @@ unattended agent task.** Issues a USER-scoped "programmatic session" - `auth logout [--stack URL|alias] [--remove-projects] [--yes]` -- revoke the refresh token server-side and delete the local session from `auth.json`. `--remove-projects` also removes `config.json` aliases pointing at this session (sentinel-token projects only; a static-token project on the same stack is never touched). - `auth register-projects [--stack URL|alias] [--all] [--project-id ID ...] [--alias ID=ALIAS ...] [--yes]` -- register an EXISTING session's accessible projects as `config.json` aliases, without re-running `login`. Fixes two usability gaps in plain `login`: nothing was registered unless `--register-projects` was passed, and the suggested alias was always slugified from the project NAME, so a project id like `9840` from the login table never resolved as `--project 9840`. `--all` selects every accessible project; `--project-id ID` (repeatable) selects specific ones (an inaccessible id raises a `ConfigError`); passing neither starts an interactive arrow-key + spacebar checkbox picker -- every not-yet-registered project preselected, up/down or `j`/`k` move, `space` toggles, `a` selects/deselects all, `enter` accepts, `q`/`esc`/`ctrl-c` cancels -- followed by a single `Edit aliases?` confirm (default no) that opens the old per-project alias prompt only if you opt in (each row already shows its suggested alias), then a final `typer.confirm`. On a piped stdin or a terminal without real interactive capabilities, the picker falls back to the original typed prompt (numbers / ranges `1-3` / `all` / `none`). In a non-TTY or `--json` context with neither `--all` nor `--project-id`, the command fails fast telling the caller to pass `--all` or `--project-id` instead of hanging on a prompt. `--alias ID=ALIAS` (repeatable) overrides the suggested alias for a given project id in every mode, including as the picker's prefilled default. `--yes` skips only the picker's final confirmation. Two collision rules, in both modes: a project already registered under an alias for this project+stack reports `status: "exists"` (no-op -- rename via `project edit --new-alias` instead of re-registering); an alias already claimed by a different project (or a static-token project) reports `status: "skipped"` with a rename-hint note -- an existing `config.json` entry is never overwritten. `auth login` (without `--register-projects`) now also offers this same picker interactively right after a successful login, when stdout is a TTY and `--json` was not used; otherwise it just prints the hint to run this command later, and a failure in that optional follow-up never changes `login`'s own (already-successful) exit code. -- `auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE] [--read-only] [--ttl-days N]` -- mint a Personal Access Token (`kbc_pat_...`) from an EXISTING `auth login` session (does not log in itself). Does a TOTP step-up (`POST /v1/auth/sudo`) then mints the token (`POST /v1/auth/pat`); `--totp-code` is prompted interactively when omitted and is REQUIRED under `--json` or a non-TTY stdin -- no unattended path to mint a PAT, same boundary as `auth login`. Prints the token exactly once (`access_token`); kbagent never stores it. Intended use: one-time CI/CD setup -- store the result as `KBC_TOKEN` (`KBAGENT_PROJECT_FROM_ENV=1`) or via `project add --token`. `make_client_factory` (`services/base.py`) recognizes the `kbc_pat_` prefix on a plain static token and routes it to `Authorization: Bearer` instead of `X-StorageApi-Token` (distinct auth schemes on the Storage API), so a PAT works everywhere a session project already does (`sync`, `storage`, `config`, ...) -- it just doesn't rotate, so replace it (`pat-create` again) instead of expecting a refresh. New error codes: `AUTH_SUDO_REQUIRED`, `AUTH_MFA_INVALID`. +- `auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE | --webauthn] [--read-only] [--ttl-days N] [--project-id ID ...]` -- mint a Personal Access Token (`kbc_pat_...`) from an EXISTING `auth login` session (does not log in itself). Does a step-up then mints the token (`POST /v1/auth/pat`): `--totp-code` is prompted interactively when omitted and is REQUIRED under `--json` or a non-TTY stdin -- no unattended path to mint a PAT, same boundary as `auth login`; `--webauthn` opens a browser for a passkey ceremony instead (mutually exclusive with `--totp-code`; see `auth/webauthn_browser.py` for the current placeholder ceremony-page contract this assumes). `--project-id` (repeatable) narrows the PAT's scope to an explicit allow-list instead of every project the signed-in user can access -- use it for a one-project-per-CI-secret setup. Prints the token exactly once (`access_token`); kbagent never stores it. Intended use: one-time CI/CD setup -- store the result as `KBC_TOKEN` (`KBAGENT_PROJECT_FROM_ENV=1`) or via `project add --token`. `make_client_factory` (`services/base.py`) recognizes the `kbc_pat_` prefix on a plain static token and routes it to `Authorization: Bearer` instead of `X-StorageApi-Token` (distinct auth schemes on the Storage API), so a PAT works everywhere a session project already does (`sync`, `storage`, `config`, ...) -- it just doesn't rotate, so replace it (`pat-create` again) instead of expecting a refresh. New error codes: `AUTH_SUDO_REQUIRED`, `AUTH_MFA_INVALID`. - `auth pat-revoke PAT_ID [--stack URL|alias] [--yes]` -- revoke a Personal Access Token. No sudo step-up needed. Idempotent: revoking an already-revoked id is not an error. v1 scope: the Storage + Manage paths. `serve` reaches them too (it delegates to diff --git a/src/keboola_agent_cli/auth/auth_client.py b/src/keboola_agent_cli/auth/auth_client.py index ec618dc5..9163ea65 100644 --- a/src/keboola_agent_cli/auth/auth_client.py +++ b/src/keboola_agent_cli/auth/auth_client.py @@ -42,6 +42,7 @@ AUTH_REFRESH_CONTENTION_STRING_CODE, AUTH_REFRESH_TIMEOUT, AUTH_SESSIONS_PATH, + AUTH_SUDO_CHALLENGE_PATH, AUTH_SUDO_PATH, AUTH_TOKEN_INTROSPECT_PATH, AUTH_TOKEN_REFRESH_PATH, @@ -58,6 +59,7 @@ IntrospectResponse, PatCreateResult, RevokeResult, + SudoChallengeResult, SudoResult, ) @@ -659,10 +661,10 @@ def sudo_totp(self, access_token: str, totp_code: str) -> SudoResult: """Activate the sudo window on the current session via TOTP (``POST /v1/auth/sudo``). Required before `create_pat` -- a PAT cannot be minted outside an - active sudo window. Only the TOTP factor is wired here: WebAuthn - step-up needs a live browser ceremony this CLI has nowhere to host, - and password step-up is rejected outright once MFA is configured - (the API's own rule, not a restriction added here). + active sudo window. Password step-up is rejected outright once MFA + is configured (the API's own rule, not a restriction added here). + See `sudo_challenge`/`sudo_webauthn` for the WebAuthn/passkey factor, + which needs a live browser ceremony instead of a typed code. """ response = self._do_request( "POST", @@ -677,6 +679,49 @@ def sudo_totp(self, access_token: str, totp_code: str) -> SudoResult: timeout_seconds=int(data.get("sudoTimeoutSeconds", 0)), ) + def sudo_challenge(self, access_token: str) -> SudoChallengeResult: + """Start a WebAuthn sudo challenge (``POST /v1/auth/sudo/challenge``). + + Returns the `challengeToken` + `PublicKeyCredentialRequestOptions` + the caller hands to a browser ceremony page; complete it with + `sudo_webauthn` once the page redirects back with an assertion. + """ + response = self._do_request( + "POST", + AUTH_SUDO_CHALLENGE_PATH, + json={}, + headers={"Authorization": f"Bearer {access_token}"}, + ) + data = response.json() + return SudoChallengeResult( + challenge_token=str(data.get("challengeToken", "")), + options=data.get("options") or {}, + expires_in=int(data.get("expiresIn", 0)), + ) + + def sudo_webauthn(self, access_token: str, challenge_token: str, assertion: str) -> SudoResult: + """Complete a WebAuthn sudo step-up (``POST /v1/auth/sudo``). + + `challenge_token` and `assertion` are the values `sudo_challenge` + issued and the browser ceremony redirected back with, respectively. + """ + response = self._do_request( + "POST", + AUTH_SUDO_PATH, + json={ + "type": "webauthn", + "challengeToken": challenge_token, + "webauthnAssertion": assertion, + }, + headers={"Authorization": f"Bearer {access_token}"}, + ) + data = response.json() + return SudoResult( + verified=bool(data.get("sudoVerified")), + expires_at=str(data.get("sudoExpiresAt", "")), + timeout_seconds=int(data.get("sudoTimeoutSeconds", 0)), + ) + def create_pat( self, access_token: str, @@ -684,6 +729,7 @@ def create_pat( name: str, read_only: bool = False, expires_in: int | None = None, + project_ids: list[str] | None = None, ) -> PatCreateResult: """Mint a Personal Access Token (``POST /v1/auth/pat``). @@ -691,12 +737,24 @@ def create_pat( the server answers 403 otherwise. The returned `PatCreateResult` carries the bearer value exactly once -- the caller must print it and never persist it. + + `project_ids` narrows the scope to an explicit allow-list (least + privilege for a CI secret meant to touch exactly one project) + instead of the server's default of every project the signed-in user + can access. """ body: dict[str, Any] = {"name": name} if expires_in is not None: body["expiresIn"] = expires_in + scope: dict[str, Any] = {} + if project_ids: + scope["projects"] = project_ids + else: + scope["all"] = True if read_only: - body["scope"] = {"all": True, "readOnly": True} + scope["readOnly"] = True + if scope != {"all": True}: + body["scope"] = scope response = self._do_request( "POST", AUTH_PAT_PATH, diff --git a/src/keboola_agent_cli/auth/models.py b/src/keboola_agent_cli/auth/models.py index 5c58fae6..fc7507bb 100644 --- a/src/keboola_agent_cli/auth/models.py +++ b/src/keboola_agent_cli/auth/models.py @@ -204,6 +204,20 @@ class SudoResult: timeout_seconds: int = 0 +@dataclass(frozen=True) +class SudoChallengeResult: + """Outcome of POST /v1/auth/sudo/challenge -- a WebAuthn ceremony to complete in a browser. + + ``options`` is the serialized `PublicKeyCredentialRequestOptions` the + ceremony page feeds to `navigator.credentials.get({publicKey: options})` + -- passed through opaquely, this CLI never inspects its contents. + """ + + challenge_token: str + options: dict + expires_in: int = 0 + + class PatItem(BaseModel): """A Personal Access Token's metadata. Never carries the secret value.""" diff --git a/src/keboola_agent_cli/auth/webauthn_browser.py b/src/keboola_agent_cli/auth/webauthn_browser.py new file mode 100644 index 00000000..c40b2508 --- /dev/null +++ b/src/keboola_agent_cli/auth/webauthn_browser.py @@ -0,0 +1,269 @@ +"""Browser-based WebAuthn sudo step-up: loopback callback half. + +Mirrors `pkce.py`'s pattern deliberately: a WebAuthn ceremony +(`navigator.credentials.get()`) can only run on a page whose origin matches +(or is a registrable suffix of) the credential's relying-party id -- a page +kbagent hosts itself on `127.0.0.1` cannot complete it, the same way this +CLI cannot complete an OAuth authorization step itself. The ceremony page +has to be served by the stack, exactly like `/admin/auth/pkce/authorize` is +for login; this module is only the CLI-side half that opens the browser at +that page and waits on a loopback listener for the result. + +PLACEHOLDER CONTRACT -- confirm before pointing this at a live stack: +`AUTH_SUDO_WEBAUTHN_CEREMONY_PATH` (constants.py) and the query parameter +names below (`options`, `challengeToken`, `redirect_uri`, `state` out; +`assertion`, `state` back) are this module's best guess at a page that +mirrors the existing PKCE authorize/callback contract -- run its own +`navigator.credentials.get()` using the options it's handed, then redirect +the browser to `redirect_uri` with the resulting assertion. Everything else +here (the loopback listener, the state check, opening the browser, wiring +the result into `POST /v1/auth/sudo`) is real and does not change once the +actual page/contract is confirmed -- only the one path constant and the +query parameter names would need adjusting. +""" + +from __future__ import annotations + +import http.server +import json +import secrets +import socket +import threading +from dataclasses import dataclass +from http import HTTPStatus +from urllib.parse import parse_qs, urlencode, urlsplit + +from ..constants import AUTH_CALLBACK_TIMEOUT, AUTH_PKCE_STATE_BYTES + +_LOOPBACK_CANDIDATES: tuple[tuple[str, int], ...] = ( + ("127.0.0.1", socket.AF_INET), + ("::1", socket.AF_INET6), +) + +_SUCCESS_BODY = ( + "Keboola CLI sudo step-up" + "

Step-up complete. You can close this tab and return to your terminal.

" + "" +) +_FAILURE_BODY = ( + "Keboola CLI sudo step-up" + "

Step-up failed. You can close this tab and return to your terminal.

" + "" +) + + +def generate_webauthn_state() -> str: + """A fresh CSRF-style nonce for one ceremony, sized like the PKCE `state`.""" + return secrets.token_urlsafe(AUTH_PKCE_STATE_BYTES) + + +@dataclass(frozen=True) +class WebAuthnCallback: + """The WebAuthn assertion and state received on the loopback callback. + + ``assertion`` is the raw string the ceremony page redirected back with -- + passed through verbatim to `AuthClient.sudo_webauthn`'s `webauthnAssertion` + body field, never parsed or re-encoded here. + """ + + assertion: str + state: str + + +class WebAuthnCeremonySetupError(Exception): + """Pre-ceremony failure (loopback bind, no usable browser). No credential spent yet.""" + + +class WebAuthnCeremonyTimeout(WebAuthnCeremonySetupError): + """No callback arrived within the wait timeout.""" + + +class WebAuthnStateMismatch(Exception): + """The callback's ``state`` did not match the one this CLI generated. + + Terminal, same reasoning as `PkceStateMismatch`: the redirect did not + originate from the ceremony this process started. + """ + + +class WebAuthnCeremonyDenied(Exception): + """The ceremony page redirected back with ``error=`` (cancelled, no authenticator, ...).""" + + def __init__(self, error: str, description: str = "") -> None: + message = f"WebAuthn step-up failed: {error}" + if description: + message = f"{message} ({description})" + super().__init__(message) + self.error = error + self.description = description + + +def _first_query_value(params: dict[str, list[str]], key: str) -> str | None: + values = params.get(key) + return values[0] if values else None + + +class _CallbackHTTPServer(http.server.HTTPServer): + def __init__( + self, server_address: tuple[str, int], address_family: int, expected_state: str + ) -> None: + self.address_family = address_family + super().__init__(server_address, _CallbackHandler) + self.expected_state = expected_state + self.result: WebAuthnCallback | None = None + self.error: Exception | None = None + self.event = threading.Event() + + +class _CallbackHandler(http.server.BaseHTTPRequestHandler): + """Handles exactly one meaningful `GET /callback` request per server.""" + + def log_message(self, format: str, *args: object) -> None: + """Silence the stdlib access log -- the URL carries the assertion.""" + return + + def do_GET(self) -> None: + server = self.server + if not isinstance(server, _CallbackHTTPServer): # pragma: no cover - defensive + self.send_response(HTTPStatus.INTERNAL_SERVER_ERROR) + self.end_headers() + return + + params = parse_qs(urlsplit(self.path).query) + assertion = _first_query_value(params, "assertion") + error = _first_query_value(params, "error") + + if assertion is None and error is None: + self.send_response(HTTPStatus.NOT_FOUND) + self.end_headers() + return + + state = _first_query_value(params, "state") + + if not _constant_time_eq(state or "", server.expected_state): + server.error = WebAuthnStateMismatch( + "The redirect's state parameter did not match the one this CLI " + "sent; refusing to treat this callback as a valid step-up." + ) + self._respond(ok=False) + server.event.set() + return + + if error is not None: + description = _first_query_value(params, "error_description") or "" + server.error = WebAuthnCeremonyDenied(error, description) + self._respond(ok=False) + server.event.set() + return + + server.result = WebAuthnCallback(assertion=assertion or "", state=state or "") + self._respond(ok=True) + server.event.set() + + def _respond(self, *, ok: bool) -> None: + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.end_headers() + self.wfile.write((_SUCCESS_BODY if ok else _FAILURE_BODY).encode("utf-8")) + + +def _constant_time_eq(a: str, b: str) -> bool: + import hmac + + return hmac.compare_digest(a, b) + + +class WebAuthnCallbackServer: + """Loopback HTTP listener that receives the WebAuthn-ceremony redirect. + + Binds an ephemeral port on `127.0.0.1`, falling back to `[::1]`. Use as + a context manager. See the module docstring for the placeholder ceremony + page / query contract this listener expects to be redirected back with. + """ + + def __init__(self, expected_state: str) -> None: + self._closed = True + self._expected_state = expected_state + self._httpd = self._bind(expected_state) + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + try: + self._thread.start() + except BaseException: + self._httpd.server_close() + raise + self._closed = False + + @staticmethod + def _bind(expected_state: str) -> _CallbackHTTPServer: + last_error: OSError | None = None + for host, family in _LOOPBACK_CANDIDATES: + try: + return _CallbackHTTPServer((host, 0), family, expected_state) + except OSError as exc: + last_error = exc + continue + raise WebAuthnCeremonySetupError( + f"Could not bind a loopback callback listener on 127.0.0.1 or [::1]: {last_error}" + ) from last_error + + def __enter__(self) -> WebAuthnCallbackServer: + return self + + def __exit__(self, *args: object) -> None: + self.close() + + @property + def redirect_uri(self) -> str: + _host, port = self._httpd.server_address[:2] + if self._httpd.address_family == socket.AF_INET6: + return f"http://[::1]:{port}/callback" + return f"http://127.0.0.1:{port}/callback" + + def wait(self, timeout: float = AUTH_CALLBACK_TIMEOUT) -> WebAuthnCallback: + """Block until a valid callback resolves the step-up, or ``timeout`` elapses.""" + if not self._httpd.event.wait(timeout): + raise WebAuthnCeremonyTimeout( + f"Timed out after {timeout:.0f}s waiting for the browser to " + "complete the WebAuthn step-up and redirect back to the CLI." + ) + if self._httpd.error is not None: + raise self._httpd.error + if self._httpd.result is None: # pragma: no cover - defensive + raise WebAuthnCeremonySetupError( + "Callback server signalled completion without a result." + ) + return self._httpd.result + + def close(self) -> None: + if self._closed: + return + self._closed = True + self._httpd.shutdown() + self._httpd.server_close() + + +def build_ceremony_url( + *, + stack_url: str, + ceremony_path: str, + options: dict, + challenge_token: str, + redirect_uri: str, + state: str, +) -> str: + """Build the browser-facing WebAuthn ceremony URL. + + ``options`` (the `PublicKeyCredentialRequestOptions` from + `POST /v1/auth/sudo/challenge`) travels as a JSON query parameter so the + ceremony page does not need a second authenticated call to fetch it -- + it only has the one-time `challenge_token`, not this session's bearer. + """ + query = urlencode( + { + "challengeToken": challenge_token, + "options": json.dumps(options, separators=(",", ":")), + "redirectUri": redirect_uri, + "state": state, + } + ) + return f"{stack_url.rstrip('/')}{ceremony_path}?{query}" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 2b65938c..8f880b98 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -25,13 +25,18 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { "0.81.0": [ - "New: `kbagent auth pat-create --name NAME [--totp-code CODE] [--read-only] " - "[--ttl-days N]` -- mint a Personal Access Token (`kbc_pat_...`) from an EXISTING " - "`auth login` session, for one-time CI/CD setup. Does a TOTP step-up " - "(`POST /v1/auth/sudo`) then mints the token (`POST /v1/auth/pat`); `--totp-code` is " - "prompted interactively when omitted and is REQUIRED under `--json` or a non-TTY " - "stdin -- there is no unattended path to mint a PAT, the same human-required boundary " - "as `auth login` itself. The token is printed exactly once and never stored by kbagent.", + "New: `kbagent auth pat-create --name NAME [--totp-code CODE | --webauthn] " + "[--read-only] [--ttl-days N] [--project-id ID ...]` -- mint a Personal Access " + "Token (`kbc_pat_...`) from an EXISTING `auth login` session, for one-time CI/CD " + "setup. Two mutually exclusive step-up factors: `--totp-code` (prompted " + "interactively when omitted, and REQUIRED under `--json` or a non-TTY stdin -- " + "there is no unattended path to mint a PAT, the same human-required boundary as " + "`auth login` itself) or `--webauthn` (opens a browser for a passkey ceremony via " + "a new loopback listener, mirroring the PKCE login flow -- the exact ceremony-page " + "contract is a documented placeholder pending platform confirmation). `--project-id` " + "(repeatable) narrows the scope to an explicit allow-list instead of every accessible " + "project, for a one-project-per-CI-secret setup. The token is printed exactly once " + "and never stored by kbagent.", "New: `kbagent auth pat-revoke PAT_ID [--yes]` -- revoke a Personal Access Token. No " "sudo step-up needed; idempotent (revoking an already-revoked id is not an error).", "New: a `kbc_pat_...` token now works as a drop-in for `KBC_TOKEN` " diff --git a/src/keboola_agent_cli/commands/auth.py b/src/keboola_agent_cli/commands/auth.py index 4f62a4ee..1b9feff7 100644 --- a/src/keboola_agent_cli/commands/auth.py +++ b/src/keboola_agent_cli/commands/auth.py @@ -572,9 +572,15 @@ def auth_pat_create( totp_code: str | None = typer.Option( None, "--totp-code", - help="Current 6-digit TOTP code. Omitted: prompted interactively " - "(never accept this as a hardcoded value in a script -- it is a live, " - "30-second code you type from your authenticator app each time).", + help="Current 6-digit TOTP code. Omitted (and --webauthn not passed): prompted " + "interactively (never accept this as a hardcoded value in a script -- it is a " + "live, 30-second code you type from your authenticator app each time).", + ), + webauthn: bool = typer.Option( + False, + "--webauthn", + help="Step up via a WebAuthn/passkey browser ceremony instead of a typed TOTP " + "code. Opens a browser; mutually exclusive with --totp-code.", ), read_only: bool = typer.Option( False, "--read-only", help="Issue a read-only PAT (denies writes on Storage routes)" @@ -582,13 +588,22 @@ def auth_pat_create( ttl_days: int | None = typer.Option( None, "--ttl-days", help="PAT lifetime in days (default: org policy maximum)" ), + project_id: list[str] | None = typer.Option( + None, + "--project-id", + help="Restrict the PAT to this project id (repeatable). Omitted: every project " + "the signed-in user can access -- pass this for a one-project-per-CI-secret setup.", + ), ) -> None: """Mint a Personal Access Token from the current session, for one-time CI/CD setup. Requires `kbagent auth login` to already be signed in on this stack -- this command spends that session's access token, it does not start a new - login. Also requires a live TOTP code (step-up authentication), because - the auth service will not mint a PAT without an active sudo window. + login. Also requires a live step-up (sudo), because the auth service will + not mint a PAT without an active sudo window: either a typed TOTP code + (the default -- no browser needed) or `--webauthn` (opens a browser for a + passkey ceremony; see `auth/webauthn_browser.py` for the current + placeholder ceremony-page contract this assumes). The token is printed exactly once, in `access_token`, and never stored by kbagent -- copy it into a CI secret immediately. Store it as `KBC_TOKEN` @@ -599,13 +614,24 @@ def auth_pat_create( (`sync`, `storage`, `config`, ...) works the same way with a PAT -- with the difference that a PAT does not rotate, so replace it (this command again) instead of expecting an automatic refresh. + + `--project-id` (repeatable) scopes the token to an explicit allow-list + instead of every accessible project -- use it once per project when + minting a separate CI secret per project (e.g. `KBC_TOKEN_`). """ formatter = get_formatter(ctx) check_cli_operation(ctx, "auth.pat-create") - if totp_code is None: + if webauthn and totp_code is not None: + formatter.error( + message="--webauthn and --totp-code are mutually exclusive.", + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) + if not webauthn and totp_code is None: if formatter.json_mode or not _is_stdout_tty(): formatter.error( - message="--totp-code is required in --json mode or when stdin is not a TTY.", + message="--totp-code (or --webauthn) is required in --json mode or when " + "stdin is not a TTY.", error_code=ErrorCode.INVALID_ARGUMENT, ) raise typer.Exit(code=2) @@ -615,9 +641,11 @@ def auth_pat_create( result = service.create_pat( stack=stack, totp_code=totp_code, + webauthn=webauthn, name=name, read_only=read_only, expires_in=ttl_days * 86400 if ttl_days else None, + project_ids=project_id, ) except (ConfigError, KeboolaApiError) as exc: _handle_errors(formatter, exc) diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 3d4e67f1..19a54524 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -140,18 +140,27 @@ hint to run this command later. A failure in this optional follow-up never changes login's own (already-successful) exit code. - kbagent auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE] [--read-only] [--ttl-days N] + kbagent auth pat-create --name NAME [--stack URL|alias] [--totp-code CODE | --webauthn] [--read-only] [--ttl-days N] [--project-id ID ...] kbagent auth pat-revoke PAT_ID [--stack URL|alias] [--yes] Mint or revoke a Personal Access Token (kbc_pat_...) -- the sanctioned way to give CI/CD a long-lived credential instead of a raw Storage token. pat-create spends an EXISTING `auth login` session's access - token (it does NOT log in itself) to do a TOTP step-up - (POST /v1/auth/sudo) then mint the token (POST /v1/auth/pat). - --totp-code is prompted interactively when omitted, and is REQUIRED - under --json or a non-TTY stdin -- there is no unattended path to mint - a PAT, same "needs a human" boundary as `auth login` itself. AN AI + token (it does NOT log in itself) to do a step-up then mint the token + (POST /v1/auth/pat). Two mutually exclusive step-up factors: + --totp-code (prompted interactively when omitted, and REQUIRED under + --json or a non-TTY stdin -- there is no unattended path to mint a + PAT, same "needs a human" boundary as `auth login` itself), or + --webauthn (opens a browser for a passkey ceremony -- see + auth/webauthn_browser.py for the current placeholder ceremony-page + contract this assumes; works fine under --json/non-TTY since it never + prompts for a typed code). + --project-id (repeatable) narrows the PAT's scope to an explicit + allow-list instead of every project the signed-in user can access -- + use it once per project when minting a separate CI secret per project + (e.g. KBC_TOKEN_). AN AI AGENT MUST NOT invent or guess a TOTP code; either the human supplies - --totp-code, or the agent must not run this command at all. + --totp-code (or completes --webauthn themselves), or the agent must + not run this command at all. The token is printed exactly once (in `access_token`) and never stored by kbagent -- copy it into a CI secret immediately. Store it as KBC_TOKEN under KBAGENT_PROJECT_FROM_ENV=1, or via `project add diff --git a/src/keboola_agent_cli/constants.py b/src/keboola_agent_cli/constants.py index 4fc89f3b..617a06e3 100644 --- a/src/keboola_agent_cli/constants.py +++ b/src/keboola_agent_cli/constants.py @@ -665,8 +665,16 @@ def _resolve_app_name() -> str: # window must be active on the current session before `AUTH_PAT_PATH` (POST) # will mint a token -- see `AuthService.create_pat`. AUTH_SUDO_PATH: str = "/v1/auth/sudo" +AUTH_SUDO_CHALLENGE_PATH: str = "/v1/auth/sudo/challenge" AUTH_PAT_PATH: str = "/v1/auth/pat" +# PLACEHOLDER -- the browser-facing page that runs the WebAuthn ceremony and +# redirects back to the CLI's loopback listener, mirroring +# AUTH_PKCE_AUTHORIZE_PATH for login. Not yet confirmed against a live stack +# (auth/webauthn_browser.py's module docstring has the full contract this +# assumes); adjust this one constant once the real page exists. +AUTH_SUDO_WEBAUTHN_CEREMONY_PATH: str = "/admin/auth/sudo/webauthn" + AUTH_DEVICE_DEFAULT_INTERVAL: int = 5 # RFC 8628 default poll interval (s) AUTH_DEVICE_MAX_INTERVAL: int = 60 # cap after repeated slow_down AUTH_DEVICE_SLOW_DOWN_INCREMENT: int = 5 # bump when the server sends no interval diff --git a/src/keboola_agent_cli/services/auth_service.py b/src/keboola_agent_cli/services/auth_service.py index e57d2f2b..585d1054 100644 --- a/src/keboola_agent_cli/services/auth_service.py +++ b/src/keboola_agent_cli/services/auth_service.py @@ -21,7 +21,7 @@ from ..auth.auth_client import AuthClient from ..auth.device import run_device_flow from ..auth.environment import BrowserEnvironment, detect_browser_environment, open_browser -from ..auth.models import CliTokenResponse, DeviceAuthorization, StackSession +from ..auth.models import CliTokenResponse, DeviceAuthorization, StackSession, SudoResult from ..auth.pkce import ( PkceAuthorizationError, PkceCallbackServer, @@ -32,7 +32,13 @@ from ..auth.sentinel import is_session_token from ..auth.state_store import AuthStateStore from ..auth.token_provider import SessionTokenProvider, reset_provider_registry +from ..auth.webauthn_browser import ( + WebAuthnCallbackServer, + build_ceremony_url, + generate_webauthn_state, +) from ..config_store import ConfigStore +from ..constants import AUTH_SUDO_WEBAUTHN_CEREMONY_PATH from ..errors import ConfigError, ErrorCode, KeboolaApiError from ..models import normalize_stack_url from ._auth_registration import ( @@ -440,6 +446,32 @@ def _perform_pkce(self, client: AuthClient) -> CliTokenResponse: code_verifier=challenge.code_verifier, ) + def _perform_webauthn_sudo( + self, client: AuthClient, stack_url: str, access_token: str + ) -> SudoResult: + """Run one WebAuthn sudo ceremony: challenge -> browser -> loopback -> verify. + + Mirrors `_perform_pkce`'s shape. See `auth/webauthn_browser.py`'s + module docstring for the placeholder ceremony-page contract this + assumes -- everything below it (the loopback listener, opening the + browser, wiring the result into `sudo_webauthn`) is real regardless + of what the confirmed page path/params turn out to be. + """ + challenge = client.sudo_challenge(access_token) + state = generate_webauthn_state() + with WebAuthnCallbackServer(expected_state=state) as server: + ceremony_url = build_ceremony_url( + stack_url=stack_url, + ceremony_path=AUTH_SUDO_WEBAUTHN_CEREMONY_PATH, + options=challenge.options, + challenge_token=challenge.challenge_token, + redirect_uri=server.redirect_uri, + state=state, + ) + self._browser_opener(ceremony_url) + callback = server.wait() + return client.sudo_webauthn(access_token, challenge.challenge_token, callback.assertion) + # ------------------------------------------------------------------ # project candidates / registration # ------------------------------------------------------------------ @@ -787,19 +819,25 @@ def create_pat( self, *, stack: str | None, - totp_code: str, + totp_code: str | None = None, + webauthn: bool = False, name: str, read_only: bool = False, expires_in: int | None = None, + project_ids: list[str] | None = None, ) -> PatCreateCliResult: """Mint a PAT from the already-logged-in session on this stack. - This does NOT log in or open a browser -- it spends an *existing* - session's access token to complete the sudo-then-create-PAT sequence, - which are ordinary bearer-authenticated API calls once a live session - exists (unlike `auth login` itself, neither needs a browser). Run + This does NOT log in itself -- it spends an *existing* session's + access token to complete the sudo-then-create-PAT sequence. Run `kbagent auth login` first if there is no stored session yet. + Exactly one step-up factor: `totp_code` (a typed 6-digit code -- no + browser needed, the ordinary case) or `webauthn=True` (opens a + browser for a WebAuthn/passkey ceremony -- see + `auth/webauthn_browser.py`'s module docstring for the ceremony-page + contract this assumes and its current placeholder status). + The intended use is one-time CI/CD setup: mint a long-lived, scoped PAT here interactively, then store `access_token` as a CI secret and use it in place of a Storage token (`KBC_TOKEN` under @@ -807,7 +845,14 @@ def create_pat( recognizes the `kbc_pat_...` prefix and sends it as `Authorization: Bearer` instead of `X-StorageApi-Token` (`services/base.py`'s `make_client_factory`). + + `project_ids`, when given, narrows the PAT to exactly those projects + (least privilege for a one-project-per-secret CI setup) instead of + every project the signed-in user can access. """ + if not webauthn and not totp_code: + raise ConfigError("Either totp_code or webauthn=True is required to step up.") + stack_url = self._resolve_stack_url(stack) session = self._state_store.get_session(stack_url) if session is None: @@ -822,7 +867,11 @@ def create_pat( ) access_token = provider.get_access_token() with self._auth_client_factory(stack_url) as client: - sudo = client.sudo_totp(access_token, totp_code) + if webauthn: + sudo = self._perform_webauthn_sudo(client, stack_url, access_token) + else: + assert totp_code is not None # guarded above + sudo = client.sudo_totp(access_token, totp_code) if not sudo.verified: raise KeboolaApiError( message="Sudo step-up was not verified. Check the TOTP code and try again.", @@ -831,7 +880,11 @@ def create_pat( retryable=False, ) result = client.create_pat( - access_token, name=name, read_only=read_only, expires_in=expires_in + access_token, + name=name, + read_only=read_only, + expires_in=expires_in, + project_ids=project_ids, ) return PatCreateCliResult( diff --git a/tests/test_auth_client.py b/tests/test_auth_client.py index 50835730..9b2b30da 100644 --- a/tests/test_auth_client.py +++ b/tests/test_auth_client.py @@ -30,6 +30,7 @@ IntrospectResponse, PatCreateResult, RevokeResult, + SudoChallengeResult, SudoResult, ) from keboola_agent_cli.commands._helpers import map_error_to_exit_code @@ -1337,6 +1338,74 @@ def test_404_maps_to_auth_not_supported(self, httpx_mock) -> None: assert excinfo.value.error_code == ErrorCode.AUTH_NOT_SUPPORTED_ON_STACK +class TestSudoChallenge: + def test_returns_challenge_token_and_options(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{STACK_URL}/v1/auth/sudo/challenge", + method="POST", + status_code=200, + json={ + "challengeToken": "kbc_mfa_xyz", + "options": {"challenge": "abc", "rpId": "keboola.com"}, + "expiresIn": 120, + }, + ) + client = _make_client() + try: + result = client.sudo_challenge("kbc_at_live") + finally: + client.close() + + assert isinstance(result, SudoChallengeResult) + assert result.challenge_token == "kbc_mfa_xyz" + assert result.options == {"challenge": "abc", "rpId": "keboola.com"} + assert result.expires_in == 120 + + request = httpx_mock.get_requests()[0] + assert request.headers["Authorization"] == "Bearer kbc_at_live" + + def test_404_maps_to_auth_not_supported(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{STACK_URL}/v1/auth/sudo/challenge", method="POST", status_code=404, json={} + ) + client = _make_client() + try: + with pytest.raises(KeboolaApiError) as excinfo: + client.sudo_challenge("kbc_at_live") + finally: + client.close() + assert excinfo.value.error_code == ErrorCode.AUTH_NOT_SUPPORTED_ON_STACK + + +class TestSudoWebauthn: + def test_verified_sends_type_and_assertion(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{STACK_URL}/v1/auth/sudo", + method="POST", + status_code=200, + json={ + "sudoVerified": True, + "sudoExpiresAt": "2026-01-01T00:05:00Z", + "sudoTimeoutSeconds": 300, + }, + ) + client = _make_client() + try: + result = client.sudo_webauthn("kbc_at_live", "kbc_mfa_xyz", "fake-assertion-json") + finally: + client.close() + + assert result.verified is True + + request = httpx_mock.get_requests()[0] + assert request.headers["Authorization"] == "Bearer kbc_at_live" + assert json.loads(request.read().decode()) == { + "type": "webauthn", + "challengeToken": "kbc_mfa_xyz", + "webauthnAssertion": "fake-assertion-json", + } + + class TestCreatePat: def test_minimal_request(self, httpx_mock) -> None: httpx_mock.add_response( @@ -1405,6 +1474,37 @@ def test_read_only_and_ttl_in_body(self, httpx_mock) -> None: "scope": {"all": True, "readOnly": True}, } + def test_project_ids_scope_narrows_to_allow_list(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{STACK_URL}/v1/auth/pat", + method="POST", + status_code=201, + json={ + "accessToken": "kbc_pat_scoped", + "expiresIn": 86400, + "pat": { + "id": "pat-3", + "name": "n", + "scope": {"projects": ["9840"]}, + "projects": [], + "readOnly": False, + "expiresAt": "2026-01-02T00:00:00Z", + "createdAt": "2026-01-01T00:00:00Z", + }, + }, + ) + client = _make_client() + try: + client.create_pat("kbc_at_live", name="n", project_ids=["9840"]) + finally: + client.close() + + request = httpx_mock.get_requests()[0] + assert json.loads(request.read().decode()) == { + "name": "n", + "scope": {"projects": ["9840"]}, + } + def test_sudo_not_active_raises(self, httpx_mock) -> None: httpx_mock.add_response( url=f"{STACK_URL}/v1/auth/pat", diff --git a/tests/test_auth_service.py b/tests/test_auth_service.py index 03fb6d21..c84c0844 100644 --- a/tests/test_auth_service.py +++ b/tests/test_auth_service.py @@ -25,6 +25,7 @@ PatCreateResult, PatItem, RevokeResult, + SudoChallengeResult, SudoResult, ) from keboola_agent_cli.auth.pkce import ( @@ -35,6 +36,7 @@ PkceStateMismatch, ) from keboola_agent_cli.auth.state_store import AuthStateStore +from keboola_agent_cli.auth.webauthn_browser import WebAuthnCallback, WebAuthnStateMismatch from keboola_agent_cli.config_store import ConfigStore from keboola_agent_cli.errors import ConfigError, ErrorCode, KeboolaApiError from keboola_agent_cli.models import ProjectConfig @@ -81,6 +83,9 @@ def __init__(self) -> None: pat=PatItem(id="pat-1", name="ci-token", readOnly=False), ) self.revoke_pat_result = RevokeResult(confirmed=True) + self.sudo_challenge_result = SudoChallengeResult( + challenge_token="kbc_mfa_xyz", options={"challenge": "abc"}, expires_in=120 + ) def __enter__(self) -> _FakeAuthClient: return self @@ -128,6 +133,14 @@ def sudo_totp(self, access_token: str, totp_code: str): self.calls.append(("sudo_totp", (access_token, totp_code))) return self.sudo_result + def sudo_challenge(self, access_token: str): + self.calls.append(("sudo_challenge", (access_token,))) + return self.sudo_challenge_result + + def sudo_webauthn(self, access_token: str, challenge_token: str, assertion: str): + self.calls.append(("sudo_webauthn", (access_token, challenge_token, assertion))) + return self.sudo_result + def create_pat(self, access_token: str, **kwargs: Any): self.calls.append(("create_pat", (access_token, kwargs))) return self.pat_create_response @@ -171,6 +184,44 @@ def _reset_fake_callback_server() -> None: _FakeCallbackServer.wait_state_override = None +class _FakeWebAuthnCallbackServer: + """Stand-in for `WebAuthnCallbackServer`: succeeds with a fixed callback.""" + + wait_error: Exception | None = None + wait_state_override: str | None = None + + def __init__(self, *, expected_state: str) -> None: + self._expected_state = expected_state + self.redirect_uri = "http://127.0.0.1:1/callback" + + def __enter__(self) -> _FakeWebAuthnCallbackServer: + return self + + def __exit__(self, *args: object) -> bool: + return False + + def wait(self, timeout: float | None = None) -> WebAuthnCallback: + wait_error = type(self).wait_error + if wait_error is not None: + raise wait_error + state = type(self).wait_state_override or self._expected_state + return WebAuthnCallback(assertion="fake-assertion-json", state=state) + + +def _reset_fake_webauthn_server() -> None: + _FakeWebAuthnCallbackServer.wait_error = None + _FakeWebAuthnCallbackServer.wait_state_override = None + + +@pytest.fixture(autouse=True) +def _patch_webauthn_server(monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: + """Replace the real WebAuthn loopback listener with the in-memory fake for every test.""" + _reset_fake_webauthn_server() + monkeypatch.setattr(svc_mod, "WebAuthnCallbackServer", _FakeWebAuthnCallbackServer) + yield + _reset_fake_webauthn_server() + + @pytest.fixture(autouse=True) def _patch_pkce_server(monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: """Replace the real loopback listener with the in-memory fake for every test.""" @@ -1300,6 +1351,7 @@ def test_success_does_sudo_then_create_with_live_access_token(self, store, state "name": "ci-salesforce", "read_only": False, "expires_in": None, + "project_ids": None, } def test_sudo_not_verified_raises_sudo_required(self, store, state_store) -> None: @@ -1324,6 +1376,72 @@ def test_ttl_days_converted_to_seconds(self, store, state_store) -> None: create_call = next(c for c in client.calls if c[0] == "create_pat") assert create_call[1][1]["expires_in"] == 90 * 86400 + def test_project_ids_forwarded(self, store, state_store) -> None: + state_store.put_session(_existing_session(session_id="sess-1", refresh_token="rt-1")) + client = _FakeAuthClient() + service = _make_service(store, state_store, client) + + service.create_pat( + stack=STACK_URL, totp_code="123456", name="ci", project_ids=["9840", "9841"] + ) + + create_call = next(c for c in client.calls if c[0] == "create_pat") + assert create_call[1][1]["project_ids"] == ["9840", "9841"] + + def test_neither_totp_nor_webauthn_raises_config_error(self, store, state_store) -> None: + state_store.put_session(_existing_session(session_id="sess-1", refresh_token="rt-1")) + client = _FakeAuthClient() + service = _make_service(store, state_store, client) + + with pytest.raises(ConfigError): + service.create_pat(stack=STACK_URL, name="ci") + + def test_webauthn_does_challenge_then_browser_ceremony_then_verify( + self, store, state_store + ) -> None: + state_store.put_session(_existing_session(session_id="sess-1", refresh_token="rt-1")) + client = _FakeAuthClient() + opened: list[str] = [] + + def _record_open(url: str) -> bool: + opened.append(url) + return True + + service = _make_service(store, state_store, client, browser_opener=_record_open) + + result = service.create_pat(stack=STACK_URL, webauthn=True, name="ci-salesforce") + + assert result.access_token == "kbc_pat_abc123" + assert client.calls[0] == ("sudo_challenge", ("old-at",)) + assert client.calls[1][0] == "sudo_webauthn" + assert client.calls[1][1] == ("old-at", "kbc_mfa_xyz", "fake-assertion-json") + assert client.calls[2][0] == "create_pat" + # The browser was opened at a URL carrying the challenge token. + assert len(opened) == 1 + assert "kbc_mfa_xyz" in opened[0] + assert "connection.keboola.com" in opened[0] + + def test_webauthn_state_mismatch_propagates(self, store, state_store) -> None: + state_store.put_session(_existing_session(session_id="sess-1", refresh_token="rt-1")) + client = _FakeAuthClient() + service = _make_service(store, state_store, client) + _FakeWebAuthnCallbackServer.wait_error = WebAuthnStateMismatch("state did not match") + + with pytest.raises(WebAuthnStateMismatch): + service.create_pat(stack=STACK_URL, webauthn=True, name="ci") + assert not any(c[0] == "create_pat" for c in client.calls) + + def test_webauthn_sudo_not_verified_raises_sudo_required(self, store, state_store) -> None: + state_store.put_session(_existing_session(session_id="sess-1", refresh_token="rt-1")) + client = _FakeAuthClient() + client.sudo_result = SudoResult(verified=False) + service = _make_service(store, state_store, client) + + with pytest.raises(KeboolaApiError) as exc_info: + service.create_pat(stack=STACK_URL, webauthn=True, name="ci") + assert exc_info.value.error_code == ErrorCode.AUTH_SUDO_REQUIRED + assert not any(c[0] == "create_pat" for c in client.calls) + class TestRevokePat: def test_no_session_raises_config_error(self, store, state_store) -> None: diff --git a/tests/test_auth_webauthn_browser.py b/tests/test_auth_webauthn_browser.py new file mode 100644 index 00000000..c6fe1369 --- /dev/null +++ b/tests/test_auth_webauthn_browser.py @@ -0,0 +1,133 @@ +"""Tests for auth/webauthn_browser.py: the WebAuthn sudo-ceremony loopback server. + +Mirrors test_auth_pkce.py's approach: a real loopback HTTP server driven by +real HTTP requests, not a mock of http.server. +""" + +from __future__ import annotations + +import contextlib +import json +import threading +import urllib.error +import urllib.request +from urllib.parse import parse_qs, urlencode, urlsplit + +import pytest + +from keboola_agent_cli.auth.webauthn_browser import ( + WebAuthnCallback, + WebAuthnCallbackServer, + WebAuthnCeremonyDenied, + WebAuthnCeremonyTimeout, + WebAuthnStateMismatch, + build_ceremony_url, + generate_webauthn_state, +) + + +def _get(base_url: str, params: dict[str, str] | None = None) -> None: + query = urlencode(params) if params else "" + url = f"{base_url}?{query}" if query else base_url + with contextlib.suppress(urllib.error.URLError, OSError): + urllib.request.urlopen(url, timeout=5) + + +def _get_after(delay: float, base_url: str, params: dict[str, str] | None = None) -> None: + timer = threading.Timer(delay, _get, args=(base_url, params)) + timer.daemon = True + timer.start() + + +class TestGenerateWebAuthnState: + def test_produces_distinct_nonces(self) -> None: + first = generate_webauthn_state() + second = generate_webauthn_state() + assert first + assert first != second + + +class TestWebAuthnCallbackServerSuccess: + def test_real_loopback_request_resolves_wait(self) -> None: + with WebAuthnCallbackServer(expected_state="expected-state") as server: + assert server.redirect_uri.startswith("http://127.0.0.1:") + assert server.redirect_uri.endswith("/callback") + + _get_after( + 0.05, + server.redirect_uri, + {"assertion": "fake-assertion-json", "state": "expected-state"}, + ) + + result = server.wait(timeout=5.0) + + assert result == WebAuthnCallback(assertion="fake-assertion-json", state="expected-state") + + def test_favicon_style_request_does_not_resolve_wait(self) -> None: + with WebAuthnCallbackServer(expected_state="expected-state") as server: + _get_after(0.02, f"{server.redirect_uri.rsplit('/callback', 1)[0]}/favicon.ico") + _get_after( + 0.08, + server.redirect_uri, + {"assertion": "real-assertion", "state": "expected-state"}, + ) + + result = server.wait(timeout=5.0) + + assert result == WebAuthnCallback(assertion="real-assertion", state="expected-state") + + +class TestWebAuthnCallbackServerStateMismatch: + def test_wrong_state_raises_and_never_resolves_a_result(self) -> None: + with WebAuthnCallbackServer(expected_state="expected-state") as server: + _get_after(0.02, server.redirect_uri, {"assertion": "a", "state": "wrong-state"}) + with pytest.raises(WebAuthnStateMismatch): + server.wait(timeout=5.0) + + +class TestWebAuthnCallbackServerDenied: + def test_error_param_raises_ceremony_denied(self) -> None: + with WebAuthnCallbackServer(expected_state="expected-state") as server: + _get_after( + 0.02, + server.redirect_uri, + { + "error": "cancelled", + "error_description": "user dismissed", + "state": "expected-state", + }, + ) + with pytest.raises(WebAuthnCeremonyDenied) as excinfo: + server.wait(timeout=5.0) + assert excinfo.value.error == "cancelled" + + +class TestWebAuthnCallbackServerTimeout: + def test_no_callback_raises_timeout(self) -> None: + with ( + WebAuthnCallbackServer(expected_state="expected-state") as server, + pytest.raises(WebAuthnCeremonyTimeout), + ): + server.wait(timeout=0.2) + + +class TestBuildCeremonyUrl: + def test_query_carries_all_fields(self) -> None: + url = build_ceremony_url( + stack_url="https://connection.keboola.com", + ceremony_path="/admin/auth/sudo/webauthn", + options={"challenge": "abc", "rpId": "keboola.com"}, + challenge_token="kbc_mfa_xyz", + redirect_uri="http://127.0.0.1:12345/callback", + state="state123", + ) + parsed = urlsplit(url) + assert parsed.scheme == "https" + assert parsed.netloc == "connection.keboola.com" + assert parsed.path == "/admin/auth/sudo/webauthn" + + params = parse_qs(parsed.query) + assert params["challengeToken"] == ["kbc_mfa_xyz"] + assert params["redirectUri"] == ["http://127.0.0.1:12345/callback"] + assert params["state"] == ["state123"] + assert json.loads(params["options"][0]) == {"challenge": "abc", "rpId": "keboola.com"} diff --git a/tests/test_cli_auth.py b/tests/test_cli_auth.py index 9df4f3fa..e4e91ead 100644 --- a/tests/test_cli_auth.py +++ b/tests/test_cli_auth.py @@ -1019,7 +1019,13 @@ def test_success_with_explicit_totp_code(self, tmp_path: Path) -> None: assert "kbc_pat_shownonceonly00000000" in result.output assert "ONLY ONCE" in result.output svc.create_pat.assert_called_once_with( - stack=None, totp_code="123456", name="ci-salesforce", read_only=False, expires_in=None + stack=None, + totp_code="123456", + webauthn=False, + name="ci-salesforce", + read_only=False, + expires_in=None, + project_ids=None, ) def test_read_only_and_ttl_forwarded(self, tmp_path: Path) -> None: @@ -1044,7 +1050,45 @@ def test_read_only_and_ttl_forwarded(self, tmp_path: Path) -> None: ) assert result.exit_code == 0, result.output svc.create_pat.assert_called_once_with( - stack=None, totp_code="123456", name="ci", read_only=True, expires_in=30 * 86400 + stack=None, + totp_code="123456", + webauthn=False, + name="ci", + read_only=True, + expires_in=30 * 86400, + project_ids=None, + ) + + def test_project_id_repeatable_forwarded(self, tmp_path: Path) -> None: + config_dir = tmp_path / "c" + config_dir.mkdir() + svc = MagicMock() + svc.create_pat.return_value = _pat_create_result() + result = _invoke( + config_dir, + svc, + [ + "auth", + "pat-create", + "--name", + "ci-salesforce", + "--totp-code", + "123456", + "--project-id", + "9840", + "--project-id", + "9841", + ], + ) + assert result.exit_code == 0, result.output + svc.create_pat.assert_called_once_with( + stack=None, + totp_code="123456", + webauthn=False, + name="ci-salesforce", + read_only=False, + expires_in=None, + project_ids=["9840", "9841"], ) def test_json_mode_without_totp_code_fails_fast(self, tmp_path: Path) -> None: @@ -1071,6 +1115,47 @@ def test_sudo_required_error_surfaces(self, tmp_path: Path) -> None: data = json.loads(result.stdout) assert data["error"]["code"] == "AUTH_SUDO_REQUIRED" + def test_webauthn_flag_forwarded_without_totp_code(self, tmp_path: Path) -> None: + config_dir = tmp_path / "c" + config_dir.mkdir() + svc = MagicMock() + svc.create_pat.return_value = _pat_create_result() + result = _invoke(config_dir, svc, ["auth", "pat-create", "--name", "ci", "--webauthn"]) + assert result.exit_code == 0, result.output + svc.create_pat.assert_called_once_with( + stack=None, + totp_code=None, + webauthn=True, + name="ci", + read_only=False, + expires_in=None, + project_ids=None, + ) + + def test_webauthn_and_totp_code_are_mutually_exclusive(self, tmp_path: Path) -> None: + config_dir = tmp_path / "c" + config_dir.mkdir() + svc = MagicMock() + result = _invoke( + config_dir, + svc, + ["auth", "pat-create", "--name", "ci", "--webauthn", "--totp-code", "123456"], + ) + assert result.exit_code == 2, result.output + svc.create_pat.assert_not_called() + + def test_webauthn_works_under_json_mode_without_totp_code(self, tmp_path: Path) -> None: + """--webauthn needs no typed code, so it must not hit the --json/no-TTY fail-fast.""" + config_dir = tmp_path / "c" + config_dir.mkdir() + svc = MagicMock() + svc.create_pat.return_value = _pat_create_result() + result = _invoke( + config_dir, svc, ["--json", "auth", "pat-create", "--name", "ci", "--webauthn"] + ) + assert result.exit_code == 0, result.output + svc.create_pat.assert_called_once() + class TestPatRevoke: def test_confirmed_revoke(self, tmp_path: Path) -> None: