From c0a2bfe5a5190a3c2ce33af61af8b2ba7f0eaed1 Mon Sep 17 00:00:00 2001 From: JS Date: Tue, 20 Oct 2026 18:54:19 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(cli):=20campus=20audit=20command=20gro?= =?UTF-8?q?up=20=E2=80=94=20API-key=20auth,=20keys,=20traces=20(#54)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit campus.audit (/audit/v1) authenticates with its own API keys (audit_v1_... Bearer), not OAuth tokens, so this adds the CLI's second credential lane alongside the device-flow machinery: - campus audit login/logout: store/clear an audit key from --api-key, CAMPUS_AUDIT_API_KEY, or a prompt, in the shared credential store, bound to the audit URL it was stored for (endpoint-mismatch fails fast, like OAuth tokens). Login verifies the key against the service first (401 rejects; 403 counts as verified for trace-only keys; unreachable stores unverified) unless --skip-verify. - campus audit keys list/new/get/update/revoke/regenerate: the operator workflow for minting and managing per-student keys (--owner, repeatable --scope, --rate-limit, --expires-at). Plaintext key material prints exactly once, at creation/regeneration (mirroring the server, which keeps only the hash). - campus audit traces list/get/search: read-only trace queries with cursor pagination (--cursor continuation hint), gated on traces:read. URL resolution mirrors campus_python exactly: CAMPUS_AUDIT_URL, then the config file, then ENV/CAMPUS_ENV (development Railway default, staging/production domains). All API-backed commands support --dry-run with the audit preamble (AuditRoot + API-key bearer) and only show kwargs actually passed. Beyond the issue's proposed surface (login, keys list/new/get/revoke, traces list/get/search): logout for symmetry with auth logout, and keys update/regenerate to complete the lifecycle the issue summary describes (PATCH fields; rotation with immediate old-key death). The server's login-journey search filter is not exposed yet — campus-api-python's traces.search() does not carry it. --- README.md | 91 ++++++- campus_cli/audit/__init__.py | 25 ++ campus_cli/audit/client.py | 159 ++++++++++++ campus_cli/audit/keys.py | 393 ++++++++++++++++++++++++++++++ campus_cli/audit/login.py | 170 +++++++++++++ campus_cli/audit/traces.py | 235 ++++++++++++++++++ campus_cli/cli.py | 2 + campus_cli/config.py | 53 ++++ campus_cli/credentials.py | 42 ++++ campus_cli/utils/output.py | 26 +- tests/integration/test_dry_run.py | 138 ++++++++++- tests/smoke/test_imports.py | 14 ++ tests/unit/test_audit.py | 343 ++++++++++++++++++++++++++ 13 files changed, 1684 insertions(+), 7 deletions(-) create mode 100644 campus_cli/audit/__init__.py create mode 100644 campus_cli/audit/client.py create mode 100644 campus_cli/audit/keys.py create mode 100644 campus_cli/audit/login.py create mode 100644 campus_cli/audit/traces.py create mode 100644 tests/unit/test_audit.py diff --git a/README.md b/README.md index da7bec4..bf394f1 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,17 @@ campus grant list campus grant grant --grantee-type user --grantee-id someone@school.edu \ --resource-type users --level write +# campus.audit: mint a student's trace-reading key (operator, #54) +campus audit login --api-key +campus audit keys new --name "Alice" --owner alice@school.edu \ + --scope traces:read --rate-limit 60 +campus audit keys list --owner alice@school.edu + +# …and the queries that key unlocks (from any machine the student logs in on) +campus audit traces list --limit 10 +campus audit traces search --path /auth/v1/users/ --status 500 +campus audit traces get --trace-id <32-char hex id> + # List vault entries campus vault list --vault myvault @@ -128,11 +139,70 @@ section of the campus login-flow doc, and the campus (A2 no silent widening, A5 supersession-is-deletion, A8 management scopes never self-confer). +### campus.audit: API-key auth (#54) + +The audit service (`/audit/v1`) is a separate, user-blind deployment +that does **not** accept OAuth tokens: it authenticates with its own +API keys (`audit_v1_…`) sent as Bearer credentials. `campus audit` is +the CLI's only command group using this second credential, stored in +the same credential store as your OAuth tokens and bound to the audit +URL it was stored for (per `ENV`/`CAMPUS_ENV` or `CAMPUS_AUDIT_URL`, +like every other endpoint): + +```bash +# Store a key: --api-key, the CAMPUS_AUDIT_API_KEY env var, or an +# interactive prompt. The key is verified against the service first +# (--skip-verify stores it without the round trip) +campus audit login --api-key audit_v1_… + +# …and the login state is cleared locally by: +campus audit logout +``` + +Two roles use the group: + +- **The operator** holds the key with `apikeys:*` scopes (the one + seeded from the server's `AUDIT_OPERATOR_API_KEY`) and administers + everyone else's keys — mint, inspect, update, rotate, revoke: + + ```bash + campus audit keys new --name "Alice" --owner alice@school.edu \ + --scope traces:read --rate-limit 60 --expires-at 2026-12-31T23:59:59Z + campus audit keys list --owner alice@school.edu + campus audit keys get --key-id + campus audit keys update --key-id --scope traces:read --rate-limit 120 + campus audit keys regenerate --key-id # rotation; old value dies + campus audit keys revoke --key-id # record kept for audit + ``` + + Plaintext key material is shown exactly once, at creation and at + regeneration (the server keeps only the hash) — pass it to the + student, who stores it with `campus audit login`. `owner_id` is an + email-format label used for organization and filtering only; + campus.audit is user-blind and self-contained by design, so it + never resolves to a campus user account. + +- **Students** log in with their minted key and query traces + (gated on `traces:read`): + + ```bash + campus audit traces list --limit 10 # recent, newest first + campus audit traces search --path /auth/v1/users/ --status 500 + campus audit traces search --user alice@school.edu --since 2026-10-20T00:00:00Z + campus audit traces get --trace-id <32-char hex> + ``` + + List and search paginate: when a page has more, the command prints + the `--cursor` token to continue with. `logout` only clears the + local copy of a key — a leaked key is disabled with + `campus audit keys revoke` (operator) or by letting `--expires-at` + pass. + ### Dry-run mode -All `client`, `user` and `vault` commands accept `--dry-run`. Instead of -authenticating and calling the API, the command prints the equivalent -Python snippet using the +All `client`, `user`, `vault` and `audit` commands accept `--dry-run`. +Instead of authenticating and calling the API, the command prints the +equivalent Python snippet using the [campus-api-python](https://pypi.org/project/campus-api-python/) library — useful for learning how to call the Campus API from Python: @@ -149,10 +219,23 @@ client = campus.auth.clients.new( print(client.id) ``` +`campus audit` snippets open with the audit client instead — the +service takes its own API key, not an OAuth token: + +```python +from campus_python import AuditRoot, CampusRequest + +campus_audit = AuditRoot(json_client=CampusRequest( + base_url="", timeout=30, mode="device", +)) +campus_audit.client.set_bearer_authorization("") +``` + ### Non-interactive use (scripts, CI, agents) Destructive commands — `auth logout`, `client delete`, `client revoke`, -`user delete`, `vault delete` — confirm before acting. When no terminal is attached +`user delete`, `vault delete`, `audit logout`, `audit keys revoke`, +`audit keys regenerate` — confirm before acting. When no terminal is attached the CLI never blocks waiting for input: it refuses immediately with an error naming the remedy. Waive the confirmation per command with `--confirm`/`-y`, or for a whole session with the `CAMPUS_ASSUME_YES` diff --git a/campus_cli/audit/__init__.py b/campus_cli/audit/__init__.py new file mode 100644 index 0000000..d1ce2d8 --- /dev/null +++ b/campus_cli/audit/__init__.py @@ -0,0 +1,25 @@ +"""campus.audit commands (#54). + +The audit service (/audit/v1) is a separate, user-blind deployment +authenticating with its own API keys (`audit_v1_...`, Bearer) rather +than OAuth user tokens. The command group mirrors its two +capabilities: the operator's key lifecycle (campus audit keys) and +read-only trace queries (campus audit traces), plus login/logout over +the stored key — the CLI's only non-OAuth credential. +""" + +import typer + +from campus_cli.audit.keys import keys_app +from campus_cli.audit.login import login_cmd, logout_cmd +from campus_cli.audit.traces import traces_app + +audit_app = typer.Typer(help="Campus Audit service commands (API-key auth)") + +# login/logout attach directly; keys and traces are sub-groups. +audit_app.command("login")(login_cmd) +audit_app.command("logout")(logout_cmd) +audit_app.add_typer(keys_app, name="keys", help="Audit API key lifecycle (operator)") +audit_app.add_typer(traces_app, name="traces", help="Trace queries (read-only)") + +__all__ = ["audit_app"] diff --git a/campus_cli/audit/client.py b/campus_cli/audit/client.py new file mode 100644 index 0000000..e4209b0 --- /dev/null +++ b/campus_cli/audit/client.py @@ -0,0 +1,159 @@ +"""Shared plumbing for campus.audit commands (#54). + +The audit service (/audit/v1) authenticates with one of its own API +keys (`audit_v1_...`) sent as a Bearer token — OAuth user tokens are +not accepted — so it gets its own credential slot and client here, +parallel to the OAuth machinery in campus_cli.auth. Like the OAuth +tokens, the stored key is bound to the service URL it was stored +against and rejected on a target mismatch. +""" + +import typer + +from campus_cli.config import config +from campus_cli.credentials import credentials +from campus_cli.utils.output import print_error + +# Key material of the campus.audit service is minted with this prefix; +# anything else is almost certainly a token or a paste error. +AUDIT_KEY_PREFIX = "audit_v1_" + + +def resolve_audit_url() -> str: + """Resolve the audit service base URL the CLI currently targets. + + Exits with a clean, actionable error when ENV/CAMPUS_ENV is invalid, + instead of letting config.audit_url's ValueError escape as a + traceback from whichever command touched it first. + + Returns: + The targeted audit service base URL (no /audit/v1 prefix; the + resource root carries it). + """ + try: + return config.audit_url + except ValueError as e: + print_error(str(e)) + raise typer.Exit(1) from e + + +def audit_endpoint_mismatch() -> tuple[str, str] | None: + """ + Check whether the stored audit key was stored for the targeted service URL. + + Returns: + (stored_url, current_url) on mismatch, None otherwise. Keys + stored before URL binding (no stored URL) are treated as + matching; they bind at the next 'campus audit login'. + """ + stored = credentials.get_audit_key_url() + if not stored: + return None + current = resolve_audit_url() + if stored.rstrip("/") != current.rstrip("/"): + return (stored, current) + return None + + +def ensure_audit_endpoint_match() -> None: + """ + Fail fast if the stored audit key targets a different service URL. + + Raises: + typer.Exit: If the stored key's audit URL differs from the one + the CLI currently targets. + """ + mismatch = audit_endpoint_mismatch() + if mismatch is None: + return + stored, current = mismatch + print_error( + f"Stored audit API key was stored for {stored} but the CLI is " + f"targeting {current}. Re-run 'campus audit login' against " + f"{current}." + ) + raise typer.Exit(1) + + +def get_audit_client(): + """ + Get an authenticated campus.audit client (#54). + + Returns: + A campus_python AuditRoot authenticating with the stored key. + + Raises: + typer.Exit: If no key is stored, or it targets another URL. + """ + api_key = credentials.get_audit_api_key() + if not api_key: + print_error( + "Not authenticated with campus.audit. Run 'campus audit login'" + " (--api-key, or set CAMPUS_AUDIT_API_KEY) first." + ) + raise typer.Exit(1) + + ensure_audit_endpoint_match() + + try: + from campus_python import AuditRoot, CampusRequest + + client = CampusRequest( + base_url=resolve_audit_url(), timeout=30, mode="device" + ) + client.set_bearer_authorization(api_key) + return AuditRoot(json_client=client) + except ImportError: + print_error("campus-api-python library not available.") + raise typer.Exit(1) from None + + +def api_error_details(e: Exception) -> tuple[int | None, str]: + """Best-effort (HTTP status, human message) from a failed audit call. + + campus_python maps non-2xx responses to errors.APIError carrying + the status code and the server's error_description; raw transport + failures (requests.HTTPError or connection errors) keep their + exception text, with any structured Campus error body unpacked. + """ + status = getattr(e, "status_code", None) + description = getattr(e, "error_description", None) + if status is None: + response = getattr(e, "response", None) + status = getattr(response, "status_code", None) + try: + error_obj = response.json().get("error", "") + except (ValueError, AttributeError): + error_obj = "" + if isinstance(error_obj, dict): + description = error_obj.get("message") or error_obj.get("code") + elif error_obj: + description = error_obj + if description: + return status, str(description) + return status, str(e) + + +def api_error_message(e: Exception) -> str: + """Extract a human-readable message from a failed audit API call.""" + return api_error_details(e)[1] + + +def fail_api(action: str, e: Exception) -> None: + """Report a failed audit API call and exit 1. + + Appends the standing remedy for the two rejections the operator can + act on without reading the docs: a rejected key (re-login) and a + scope shortfall (mint or widen the key). + """ + status, message = api_error_details(e) + hint = "" + if status == 401: + hint = " Re-run 'campus audit login' with a valid key." + elif status == 403: + hint = ( + " The stored key lacks the required scope;" + " mint or update it with 'campus audit keys'." + ) + print_error(f"Failed to {action}: {message}.{hint}") + raise typer.Exit(1) from e diff --git a/campus_cli/audit/keys.py b/campus_cli/audit/keys.py new file mode 100644 index 0000000..29cc548 --- /dev/null +++ b/campus_cli/audit/keys.py @@ -0,0 +1,393 @@ +"""campus.audit API key management (#54). + +The operator workflow for minting and managing audit API keys, e.g. +one per student for dev-environment access: `owner_id` is an +email-format label (organizational only — campus.audit is user-blind +and self-contained by design), `scopes` gate what the key may call +(e.g. ["traces:read"]), and key management itself requires apikeys:* +scopes, held only by the operator key. Plaintext key material is +shown exactly once, at creation and at regeneration (mirroring the +server). +""" + +from typing import Optional + +import typer +from rich.console import Console + +from campus_cli.audit.client import fail_api, get_audit_client +from campus_cli.auth.common import ( + confirm_destructive, + dry_run_option, +) +from campus_cli.utils.output import ( + AUDIT_PYTHON_API_PREAMBLE, + print_error, + print_json, + print_python_api, + print_success, +) + +keys_app = typer.Typer(help="Audit API key lifecycle (operator, apikeys:*)") +console = Console() + +ApiKeyOption = typer.Option(..., "--key-id", help="API key ID (from keys list)") +OwnerOption = typer.Option( + ..., + "--owner", + help=( + "Owner label, conventionally the student email (organizational" + " only — campus.audit is user-blind)" + ), +) +ScopeOption = typer.Option( + ..., + "--scope", + help=( + "Scope the key grants, e.g. traces:read; repeat for multiple" + " (the server requires at least one)" + ), +) + + +def _print_reveal_key(key_value: str) -> None: + """Print plaintext key material exactly once, unwrapped. + + Plain stdout write: Rich's word wrap can split long tokens + mid-string, which would corrupt a copy-pasted key. + """ + console.print("[dim]-------8<------- (copy now)[/dim]") + typer.echo(key_value) + console.print("[dim]-------8<-------[/dim]") + + +def _print_key_record(record: dict) -> None: + """Print one API key record in human-readable form.""" + status = "[green]active[/green]" + if record.get("revoked_at"): + status = "[red]revoked[/red]" + elif record.get("expires_at"): + status = "[yellow]expires[/yellow]" + console.print(f"[cyan]{record.get('name')}[/cyan] [dim]{record.get('id')}[/dim]") + console.print( + f" Owner: {record.get('owner_id')} Status: {status}" + f" Scopes: {' '.join(record.get('scopes') or []) or '(none)'}" + ) + extras = [] + if record.get("rate_limit") is not None: + extras.append(f"rate limit {record['rate_limit']}/min") + if record.get("expires_at"): + extras.append(f"expires {record['expires_at']}") + if record.get("revoked_at"): + extras.append(f"revoked {record['revoked_at']}") + if extras: + console.print(f" [dim]{'; '.join(extras)}[/dim]") + console.print() + + +@keys_app.command("list") +def keys_list( + owner: Optional[str] = typer.Option(None, "--owner", help="Filter by owner"), + include_inactive: bool = typer.Option( + False, + "--all", + help="Include revoked and expired keys (default: active only)", + ), + limit: Optional[int] = typer.Option(None, "--limit", help="Page size"), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """ + List audit API keys, optionally filtered by owner. + + Requires apikeys:read on the stored key (the operator key). + """ + if dry_run: + print_python_api( + "campus audit keys list", + [ + "result = campus_audit.apikeys.list(", + f" owner_id={owner!r},", + f" active_only={not include_inactive!r},", + f" limit={limit!r},", + ")", + "for key in result['api_keys']:", + " print(key['id'], key['name'], key['scopes'])", + ], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + try: + api = get_audit_client() + result = api.apikeys.list( + owner_id=owner, active_only=not include_inactive, limit=limit + ) + keys = result.get("api_keys", []) + + if output_json: + print_json(keys) + else: + if not keys: + console.print("[dim]No API keys found.[/dim]") + return + console.print(f"[bold]Found {len(keys)} API key(s):[/bold]\n") + for record in keys: + _print_key_record(record) + + except typer.Exit: + raise + except Exception as e: + fail_api("list API keys", e) + + +@keys_app.command("new") +def keys_new( + name: str = typer.Option(..., "--name", help="Key name, e.g. the student's name"), + owner: str = OwnerOption, + scope: list[str] = ScopeOption, # noqa: B008 + rate_limit: Optional[int] = typer.Option( + None, "--rate-limit", help="Requests per minute (optional)" + ), + expires_at: Optional[str] = typer.Option( + None, "--expires-at", help="ISO 8601 expiry (optional)" + ), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Create an audit API key (mint a student key). + + The plaintext key is shown exactly once, at creation — the server + keeps only its hash. Pass it to the student; they store it with + 'campus audit login --api-key'. + """ + if dry_run: + new_kwargs = [ + f" name={name!r},", + f" owner_id={owner!r},", + f" scopes={list(scope)!r},", + ] + if rate_limit is not None: + new_kwargs.append(f" rate_limit={rate_limit!r},") + if expires_at is not None: + new_kwargs.append(f" expires_at={expires_at!r},") + print_python_api( + "campus audit keys new", + [ + "record = campus_audit.apikeys.new(", + *new_kwargs, + ")", + "print(record['api_key']) # shown only here", + ], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + try: + api = get_audit_client() + record = api.apikeys.new( + name=name, + owner_id=owner, + scopes=list(scope), + rate_limit=rate_limit, + expires_at=expires_at, + ) + + if output_json: + print_json(record) + else: + print_success( + "API key created. The plaintext value is shown only here —" + " copy it now:" + ) + _print_reveal_key(record.get("api_key", "")) + console.print("[dim]Key record:[/dim]") + _print_key_record(record) + + except typer.Exit: + raise + except Exception as e: + fail_api("create API key", e) + + +@keys_app.command("get") +def keys_get( + key_id: str = ApiKeyOption, + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """Show one audit API key's record (never its plaintext).""" + if dry_run: + print_python_api( + "campus audit keys get", + [f"record = campus_audit.apikeys[{key_id!r}].get()", "print(record)"], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + try: + api = get_audit_client() + record = api.apikeys[key_id].get() + + if output_json: + print_json(record) + else: + _print_key_record(record) + + except typer.Exit: + raise + except Exception as e: + fail_api(f"get API key '{key_id}'", e) + + +@keys_app.command("update") +def keys_update( + key_id: str = ApiKeyOption, + name: Optional[str] = typer.Option(None, "--name", help="New key name"), + scope: Optional[list[str]] = typer.Option( # noqa: B008 + None, + "--scope", + help=( + "Replacement scope set; repeat for multiple. Omit to keep" + " the current scopes" + ), + ), + rate_limit: Optional[int] = typer.Option( + None, "--rate-limit", help="Requests per minute" + ), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Update an audit API key's name, scopes, or rate limit. + + Only the fields passed are changed. Expiry and revocation are + immutable — mint a fresh key instead. + """ + if name is None and scope is None and rate_limit is None: + print_error( + "Nothing to update: pass --name, --scope, or --rate-limit." + ) + raise typer.Exit(1) + + if dry_run: + update_kwargs = [] + if name is not None: + update_kwargs.append(f" name={name!r},") + if scope is not None: + update_kwargs.append(f" scopes={list(scope)!r},") + if rate_limit is not None: + update_kwargs.append(f" rate_limit={rate_limit!r},") + print_python_api( + "campus audit keys update", + [ + f"record = campus_audit.apikeys[{key_id!r}].update(", + *update_kwargs, + ")", + "print(record)", + ], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + try: + api = get_audit_client() + record = api.apikeys[key_id].update( + name=name, + scopes=list(scope) if scope is not None else None, + rate_limit=rate_limit, + ) + + if output_json: + print_json(record) + else: + print_success("API key updated.") + _print_key_record(record) + + except typer.Exit: + raise + except Exception as e: + fail_api(f"update API key '{key_id}'", e) + + +@keys_app.command("revoke") +def keys_revoke( + key_id: str = ApiKeyOption, + confirm: bool = typer.Option(False, "--confirm", "-y", help="Skip confirmation"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Revoke an audit API key (the record is kept for audit). + + The key stops authenticating immediately; its record stays listed + under --all for the audit trail. + """ + if dry_run: + print_python_api( + "campus audit keys revoke", + [f"campus_audit.apikeys[{key_id!r}].revoke()"], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + confirm_destructive(f"revoke audit API key '{key_id}'", confirm) + + try: + api = get_audit_client() + api.apikeys[key_id].revoke() + print_success(f"API key '{key_id}' revoked (record kept for audit).") + + except typer.Exit: + raise + except Exception as e: + fail_api(f"revoke API key '{key_id}'", e) + + +@keys_app.command("regenerate") +def keys_regenerate( + key_id: str = ApiKeyOption, + confirm: bool = typer.Option(False, "--confirm", "-y", help="Skip confirmation"), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Replace an audit API key's plaintext value (rotation). + + The old value stops working immediately; the new one is shown + exactly once here. Scopes, rate limit, and expiry carry over. + """ + if dry_run: + print_python_api( + "campus audit keys regenerate", + [ + f"key = campus_audit.apikeys[{key_id!r}].regenerate()", + "print(key) # shown only here", + ], + preamble=AUDIT_PYTHON_API_PREAMBLE, + ) + return + + confirm_destructive( + f"regenerate audit API key '{key_id}' (the old value stops" + " working immediately)", + confirm, + ) + + try: + api = get_audit_client() + key_value = api.apikeys[key_id].regenerate() + + if output_json: + print_json({"id": key_id, "api_key": key_value}) + else: + print_success( + "API key regenerated. The new plaintext value is shown only" + " here — copy it now:" + ) + _print_reveal_key(key_value) + + except typer.Exit: + raise + except Exception as e: + fail_api(f"regenerate API key '{key_id}'", e) diff --git a/campus_cli/audit/login.py b/campus_cli/audit/login.py new file mode 100644 index 0000000..f970d67 --- /dev/null +++ b/campus_cli/audit/login.py @@ -0,0 +1,170 @@ +"""campus.audit login/logout (#54). + +The audit service only accepts its own API keys, so 'campus audit +login' stores a key from --api-key, CAMPUS_AUDIT_API_KEY, or an +interactive prompt instead of running the OAuth device flow. The key +is verified against the service (best-effort, never blocking on +network errors) and bound to the audit URL it was stored for, like +OAuth tokens are bound to their auth endpoint. +""" + +import os +import sys + +import requests +import typer +from rich.console import Console + +from campus_cli.audit.client import ( + AUDIT_KEY_PREFIX, + api_error_details, + api_error_message, + resolve_audit_url, +) +from campus_cli.auth.common import confirm_destructive +from campus_cli.credentials import CredentialError, credentials +from campus_cli.utils.output import print_error, print_success, print_warning + +console = Console() + +# Environment fallback for the key material, for scripts and CI. +AUDIT_API_KEY_ENV = "CAMPUS_AUDIT_API_KEY" + + +def _resolve_key_input(api_key_option: str | None) -> str: + """Get the API key from the option, the environment, or a prompt. + + Raises: + typer.Exit: When no source yields a key (non-interactive). + """ + api_key = api_key_option or os.environ.get(AUDIT_API_KEY_ENV) or "" + if not api_key and sys.stdin.isatty(): + api_key = typer.prompt("Audit API key", hide_input=True) + api_key = api_key.strip() + if not api_key: + print_error( + f"No API key given. Pass --api-key or set {AUDIT_API_KEY_ENV}." + ) + raise typer.Exit(1) + if not api_key.startswith(AUDIT_KEY_PREFIX): + print_warning( + f"The key does not look like an audit API key (expected the" + f" {AUDIT_KEY_PREFIX}… prefix); storing it anyway." + ) + return api_key + + +def verify_api_key(audit_url: str, api_key: str) -> str: + """Check the key against the audit service before storing it. + + Any authenticated response proves the key: 200 needs apikeys:read, + while 403 means the key is valid but scoped elsewhere (fine for + trace-only student keys). + + Args: + audit_url: The audit service base URL to verify against. + api_key: The candidate plaintext key. + + Returns: + "verified" when the server accepted the key, "unverified" when + the service could not be reached (the key is stored anyway). + + Raises: + typer.Exit: When the server rejected the key (nothing stored). + """ + try: + from campus_python import AuditRoot, CampusRequest, errors + except ImportError: + print_error("campus-api-python library not available.") + raise typer.Exit(1) from None + + try: + client = CampusRequest(base_url=audit_url, timeout=10, mode="device") + client.set_bearer_authorization(api_key) + AuditRoot(json_client=client).apikeys.list(limit=1) + except (errors.APIError, requests.RequestException) as e: + status, _ = api_error_details(e) + if status == 401: + print_error( + "The audit service rejected this key (401);" + " nothing was stored." + ) + raise typer.Exit(1) from e + if status == 403: + # An authenticated rejection: the key works but lacks + # apikeys:read (fine for trace-only student keys). + console.print( + "[dim]Key verified (lacks apikeys:read, so it cannot" + " list keys — fine for trace-only keys).[/dim]" + ) + return "verified" + # Anything else (network error, 404 on older deployments, 5xx) + # does not prove the key; store it unverified and say so. + print_warning( + f"Could not verify the key against the audit service" + f" ({api_error_message(e)}); storing it unverified." + ) + return "unverified" + return "verified" + + +def login_cmd( + api_key: str | None = typer.Option( + None, + "--api-key", + help=( + "The audit API key (audit_v1_...). Falls back to the" + f" {AUDIT_API_KEY_ENV} environment variable, then an" + " interactive prompt." + ), + ), + skip_verify: bool = typer.Option( + False, + "--skip-verify", + help="Store the key without checking it against the service", + ), +) -> None: + """ + Store an API key for the campus.audit service (#54). + + Unlike OAuth tokens, audit keys never expire server-side into a + refresh flow: a revoked or deleted key is cleared by re-running + this command with a new one. The key is stored in the shared + credential store, bound to the audit URL the CLI currently targets. + """ + audit_url = resolve_audit_url() + api_key = _resolve_key_input(api_key) + + if not skip_verify: + verify_api_key(audit_url, api_key) + + credentials.set_audit_api_key(api_key, audit_url) + print_success("Audit API key stored.") + console.print(f"[dim]Audit service: {audit_url}[/dim]") + + +def logout_cmd( + confirm: bool = typer.Option( + False, "--confirm", "-y", help="Skip confirmation prompt" + ), +) -> None: + """ + Clear the stored audit API key (local only). + + This does not disable the key server-side — audit keys cannot + revoke themselves. To invalidate a key, revoke it first: + 'campus audit keys revoke --key-id '. + """ + try: + if not credentials.get_audit_api_key(): + print_success("Not logged in to campus.audit.") + return + confirm_destructive( + "clear the stored campus.audit API key", confirm + ) + credentials.delete_audit_api_key() + except CredentialError as e: + print_error(f"Failed to log out: {e}") + raise typer.Exit(1) from e + + print_success("Logged out of campus.audit (local key cleared).") diff --git a/campus_cli/audit/traces.py b/campus_cli/audit/traces.py new file mode 100644 index 0000000..5bc444c --- /dev/null +++ b/campus_cli/audit/traces.py @@ -0,0 +1,235 @@ +"""campus.audit trace queries (#54). + +Read-only access to the traces the audit service records: recent +traces, full span trees, and filtered search, gated on traces:read. +Trace IDs are 32-char hex, span IDs 16-char hex (OpenTelemetry +convention). +""" + +from typing import Optional + +import typer +from rich.console import Console + +from campus_cli.audit.client import fail_api, get_audit_client +from campus_cli.auth.common import dry_run_option +from campus_cli.utils.output import ( + AUDIT_PYTHON_API_PREAMBLE, + print_json, + print_python_api, +) + +traces_app = typer.Typer(help="Trace queries (read-only, traces:read)") +console = Console() + + +def _status_markup(status: Optional[int]) -> str: + """Color a status code: green 2xx, yellow 3xx/4xx, red 5xx/none.""" + if status is None: + return "[dim]—[/dim]" + if 200 <= status < 300: + return f"[green]{status}[/green]" + if 200 <= status < 500: + return f"[yellow]{status}[/yellow]" + return f"[red]{status}[/red]" + + +def _format_trace_summary(trace: dict) -> str: + """Render a trace summary line: method, path, status, timing.""" + root = trace.get("root_span") or {} + method = root.get("method", "?") + path = root.get("path", "?") + user = root.get("user_id") or root.get("client_id") or "anonymous" + return ( + f"{method} {path} → {_status_markup(root.get('status_code'))}" + f" {trace.get('duration_ms', '?')} ms," + f" {trace.get('span_count', '?')} span(s), user: {user}" + ) + + +def _print_trace_page(result: dict) -> None: + """Print a list/search result page with its continuation cursor.""" + traces = result.get("traces", []) + if not traces: + console.print("[dim]No traces found.[/dim]") + return + console.print(f"[bold]Showing {len(traces)} trace(s):[/bold]\n") + for trace in traces: + console.print(f"[cyan]{trace.get('trace_id', '?')}[/cyan]") + console.print(f" {_format_trace_summary(trace)}") + console.print(f" [dim]{trace.get('started_at', '')}[/dim]") + console.print() + cursor = result.get("cursor") or {} + if cursor.get("has_more") and cursor.get("next"): + console.print( + "[dim]More traces available; continue with " + f"--cursor {cursor['next']}[/dim]" + ) + + +def _dry_run(command: str, operation: list[str]) -> None: + """Print a dry-run snippet carrying the audit client preamble.""" + print_python_api(command, operation, preamble=AUDIT_PYTHON_API_PREAMBLE) + + +def _query_operation(call: str, kwargs: dict) -> list[str]: + """Operation lines for list/search: only the given kwargs appear.""" + given = [f" {k}={v!r}," for k, v in kwargs.items() if v is not None] + if given: + head = [f"result = campus_audit.traces.{call}(", *given, ")"] + else: + head = [f"result = campus_audit.traces.{call}()"] + return [ + *head, + "for trace in result['traces']:", + " print(trace['trace_id'], trace['root_span']['path'])", + ] + + +@traces_app.command("list") +def traces_list( + since: Optional[str] = typer.Option( + None, "--since", help="ISO 8601 lower bound on trace time" + ), + until: Optional[str] = typer.Option( + None, "--until", help="ISO 8601 upper bound on trace time" + ), + limit: Optional[int] = typer.Option(None, "--limit", help="Page size"), + cursor: Optional[str] = typer.Option( + None, "--cursor", help="Opaque token from a previous page" + ), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """List recent traces, newest first.""" + if dry_run: + _dry_run( + "campus audit traces list", + _query_operation("list", { + "since": since, "until": until, + "limit": limit, "cursor": cursor, + }), + ) + return + + try: + api = get_audit_client() + result = api.traces.list( + since=since, until=until, limit=limit, cursor=cursor + ) + + if output_json: + print_json(result) + else: + _print_trace_page(result) + + except typer.Exit: + raise + except Exception as e: + fail_api("list traces", e) + + +@traces_app.command("get") +def traces_get( + trace_id: str = typer.Option(..., "--trace-id", help="32-char hex trace ID"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Show a trace's full span tree (as JSON). + + The tree is the trace: the root span with its children nested, + each span carrying the recorded request/response detail. + """ + if dry_run: + _dry_run( + "campus audit traces get", + [ + f"tree = campus_audit.traces[{trace_id!r}].get()", + "print(tree)", + ], + ) + return + + try: + api = get_audit_client() + tree = api.traces[trace_id].get() + print_json(tree) + + except typer.Exit: + raise + except Exception as e: + fail_api(f"get trace '{trace_id}'", e) + + +@traces_app.command("search") +def traces_search( + path: Optional[str] = typer.Option( + None, "--path", help="Filter by endpoint path" + ), + status: Optional[int] = typer.Option( + None, "--status", help="Filter by HTTP status code" + ), + user: Optional[str] = typer.Option( + None, "--user", help="Filter by user ID" + ), + client: Optional[str] = typer.Option( + None, "--client", help="Filter by OAuth client ID" + ), + api_key_id: Optional[str] = typer.Option( + None, "--api-key-id", help="Filter by audit API key ID" + ), + since: Optional[str] = typer.Option( + None, "--since", help="ISO 8601 lower bound on trace time" + ), + until: Optional[str] = typer.Option( + None, "--until", help="ISO 8601 upper bound on trace time" + ), + limit: Optional[int] = typer.Option(None, "--limit", help="Page size"), + cursor: Optional[str] = typer.Option( + None, "--cursor", help="Opaque token from a previous page" + ), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), + dry_run: bool = dry_run_option(), +) -> None: + """ + Search traces by endpoint, outcome, principal, or time. + + Filters combine (AND). The server's login-journey filter is not + exposed here yet — campus-api-python's search() does not carry it. + """ + if dry_run: + _dry_run( + "campus audit traces search", + _query_operation("search", { + "path": path, "status": status, + "user_id": user, "client_id": client, + "api_key_id": api_key_id, + "since": since, "until": until, + "limit": limit, "cursor": cursor, + }), + ) + return + + try: + api = get_audit_client() + result = api.traces.search( + path=path, + status=status, + user_id=user, + client_id=client, + api_key_id=api_key_id, + since=since, + until=until, + limit=limit, + cursor=cursor, + ) + + if output_json: + print_json(result) + else: + _print_trace_page(result) + + except typer.Exit: + raise + except Exception as e: + fail_api("search traces", e) diff --git a/campus_cli/cli.py b/campus_cli/cli.py index 17c8dc9..faa70b5 100644 --- a/campus_cli/cli.py +++ b/campus_cli/cli.py @@ -37,6 +37,7 @@ def version() -> None: # Import auth commands # noqa: E402 - Imported here to avoid circular dependencies +from campus_cli.audit import audit_app # noqa: E402 from campus_cli.auth import ( # noqa: E402 client_app, grants_app, @@ -51,6 +52,7 @@ def version() -> None: app.add_typer(user_app, name="user", help="User management commands") app.add_typer(grants_app, name="grant", help="Access-grant store commands") app.add_typer(vault_app, name="vault", help="Vault management commands") +app.add_typer(audit_app, name="audit", help="Campus Audit service commands") def main() -> None: diff --git a/campus_cli/config.py b/campus_cli/config.py index 8a24757..9117085 100644 --- a/campus_cli/config.py +++ b/campus_cli/config.py @@ -31,6 +31,12 @@ class Config: # /auth/v1 prefix because the CLI appends /oauth/... paths itself STAGING_AUTH_URL = "https://auth.campus.nyjc.dev/auth/v1" PRODUCTION_AUTH_URL = "https://auth.campus.nyjc.app/auth/v1" + # Audit service base URLs (#54). No /audit/v1 prefix: the resource + # root (campus_python.AuditRoot) carries the url_prefix itself, so + # these match campus_python's URL resolution exactly. + DEFAULT_AUDIT_URL = "https://campusaudit-development.up.railway.app" + STAGING_AUDIT_URL = "https://audit.campus.nyjc.dev" + PRODUCTION_AUDIT_URL = "https://audit.campus.nyjc.app" DEFAULT_AUTO_REFRESH = True DEFAULT_REFRESH_THRESHOLD = 300 # 5 minutes @@ -156,6 +162,53 @@ def auth_url(self, value: str) -> None: """Set the auth server URL.""" self.set("auth_url", value) + @property + def audit_url(self) -> str: + """ + Get the audit service base URL (#54). + + Resolution order mirrors campus_python's audit URL resolution + (CAMPUS_AUDIT_URL env var, then ENV/CAMPUS_ENV), with the config + file's audit_url key taking precedence over the ENV mapping — + the same precedence auth_url uses: + + 1. CAMPUS_AUDIT_URL environment variable + 2. audit_url key in the config file + 3. ENV/CAMPUS_ENV environment variable (ENV wins) mapped to the + deployment environment's audit service URL + 4. Built-in development default + """ + env_audit_url = os.getenv("CAMPUS_AUDIT_URL") + if env_audit_url: + return env_audit_url + config_audit_url = self.get("audit_url") + if config_audit_url: + return config_audit_url + return self._env_audit_url() + + @staticmethod + def _env_audit_url() -> str: + """Resolve the audit URL from the ENV/CAMPUS_ENV deployment environment.""" + campus_env = os.environ.get("ENV", os.environ.get("CAMPUS_ENV", "development")) + match campus_env: + case "development" | "testing": + return Config.DEFAULT_AUDIT_URL + case "staging": + return Config.STAGING_AUDIT_URL + case "production": + return Config.PRODUCTION_AUDIT_URL + case _: + raise ValueError( + f"Invalid deployment environment {campus_env!r} in " + "ENV/CAMPUS_ENV; expected development, staging, or " + "production (or set CAMPUS_AUDIT_URL explicitly)" + ) + + @audit_url.setter + def audit_url(self, value: str) -> None: + """Set the audit service base URL.""" + self.set("audit_url", value) + @property def auto_refresh(self) -> bool: """Get whether to automatically refresh expired tokens.""" diff --git a/campus_cli/credentials.py b/campus_cli/credentials.py index 7c06887..5325487 100644 --- a/campus_cli/credentials.py +++ b/campus_cli/credentials.py @@ -267,6 +267,48 @@ def set_token_auth_url(self, auth_url: str) -> None: """ self.set_password("token_auth_url", auth_url) + def get_audit_api_key(self) -> str | None: + """ + Get the stored campus.audit API key (#54). + + Returns: + The API key (audit_v1_...), or None if not found. + """ + return self.get_password("audit_api_key") + + def set_audit_api_key(self, api_key: str, audit_url: str) -> None: + """ + Store the campus.audit API key, bound to the service URL (#54). + + Args: + api_key: The plaintext API key. + audit_url: The audit service base URL the key was stored + (and verified) against. + """ + self.set_password("audit_api_key", api_key) + self.set_password("audit_key_url", audit_url) + + def get_audit_key_url(self) -> str | None: + """ + Get the audit service URL the stored API key was stored against. + + Returns: + The audit URL, or None if not found. + """ + return self.get_password("audit_key_url") + + def delete_audit_api_key(self) -> None: + """ + Delete the stored campus.audit API key and its URL binding (#54). + + Raises: + CredentialError: If no audit key is stored. + """ + self.delete_password("audit_api_key") + with contextlib.suppress(CredentialError): + # May not exist (keys stored before URL binding, #54) + self.delete_password("audit_key_url") + def get_login_session_id(self) -> str | None: """ Get the stored login-session id (#837). diff --git a/campus_cli/utils/output.py b/campus_cli/utils/output.py index e4a9a17..991fb94 100644 --- a/campus_cli/utils/output.py +++ b/campus_cli/utils/output.py @@ -24,6 +24,21 @@ "", ] +# Preamble for campus.audit --dry-run snippets (#54): the audit +# service authenticates with one of its own API keys, not an OAuth +# token, so its resource root is built directly. +AUDIT_PYTHON_API_PREAMBLE = [ + "from campus_python import AuditRoot, CampusRequest", + "", + "campus_audit = AuditRoot(json_client=CampusRequest(", + ' base_url="", timeout=30, mode="device",', + "))", + "", + "# Set your audit API key (campus audit login)", + 'campus_audit.client.set_bearer_authorization("")', + "", +] + def print_success(message: str) -> None: """ @@ -93,7 +108,11 @@ def print_json(data: Any, pretty: bool = True) -> None: console.print(json.dumps(data)) -def print_python_api(command: str, operation: list[str]) -> None: +def print_python_api( + command: str, + operation: list[str], + preamble: Optional[list[str]] = None, +) -> None: """ Print the equivalent Python API code for a --dry-run invocation. @@ -103,8 +122,11 @@ def print_python_api(command: str, operation: list[str]) -> None: Args: command: The CLI command the snippet corresponds to. operation: The API operation lines (after client setup). + preamble: Client-setup lines opening the snippet; defaults to + the OAuth-token device client (audit commands pass + the audit API-key setup instead, #54). """ - code = "\n".join([*_PYTHON_API_PREAMBLE, *operation]) + code = "\n".join([*(preamble or _PYTHON_API_PREAMBLE), *operation]) console.print("[bold]Dry run:[/bold] no API call made.") console.print(f"Equivalent Python code for [cyan]{command}[/cyan]:\n") console.print(Syntax(code, "python", word_wrap=True, background_color="default")) diff --git a/tests/integration/test_dry_run.py b/tests/integration/test_dry_run.py index 7fd9cdc..ec0a407 100644 --- a/tests/integration/test_dry_run.py +++ b/tests/integration/test_dry_run.py @@ -33,6 +33,22 @@ def _invoke_dry_run(*args): return result +def _invoke_audit_dry_run(*args): + """Invoke an audit dry-run command without touching the audit client. + + Audit snippets carry the API-key preamble (AuditRoot + key bearer) + instead of the OAuth one (#54). + """ + with patch("campus_cli.audit.client.get_audit_client") as mock_get: + result = runner.invoke(app, [*args, "--dry-run"]) + mock_get.assert_not_called() + assert result.exit_code == 0 + assert "Dry run" in result.stdout + assert "import AuditRoot, CampusRequest" in result.stdout + assert "set_bearer_authorization" in result.stdout + return result + + def test_client_new_dry_run(): """client new --dry-run shows the clients.new() call with arguments.""" result = _invoke_dry_run( @@ -215,9 +231,111 @@ def test_vault_list_dry_run(): assert "campus.auth.vaults['myvault'].keys()" in result.stdout +def test_audit_keys_new_dry_run(): + """audit keys new --dry-run shows the apikeys.new() call (#54).""" + result = _invoke_audit_dry_run( + "audit", "keys", "new", + "--name", "Alice", "--owner", "alice@school.edu", + "--scope", "traces:read", + ) + + assert "campus_audit.apikeys.new(" in result.stdout + assert "name='Alice'" in result.stdout + assert "owner_id='alice@school.edu'" in result.stdout + assert "scopes=['traces:read']" in result.stdout + + +def test_audit_keys_get_dry_run(): + """audit keys get --dry-run shows indexed resource access.""" + result = _invoke_audit_dry_run( + "audit", "keys", "get", "--key-id", "apikey-abc123" + ) + + assert "campus_audit.apikeys['apikey-abc123'].get()" in result.stdout + + +def test_audit_keys_revoke_dry_run_skips_confirmation(): + """audit keys revoke --dry-run shows the revoke call without prompting.""" + result = _invoke_audit_dry_run( + "audit", "keys", "revoke", "--key-id", "apikey-abc123" + ) + + assert "campus_audit.apikeys['apikey-abc123'].revoke()" in result.stdout + assert "Are you sure" not in result.stdout + + +def test_audit_keys_update_dry_run_only_passed_fields(): + """audit keys update --dry-run shows only the fields being updated.""" + result = _invoke_audit_dry_run( + "audit", "keys", "update", + "--key-id", "apikey-abc123", "--scope", "traces:read", + ) + + assert "campus_audit.apikeys['apikey-abc123'].update(" in result.stdout + assert "scopes=['traces:read']" in result.stdout + assert "name=" not in result.stdout + + plain = _invoke_audit_dry_run( + "audit", "keys", "update", + "--key-id", "apikey-abc123", "--name", "N", + ) + assert "name='N'" in plain.stdout + assert "scopes=" not in plain.stdout + + +def test_audit_keys_update_requires_a_field(): + """audit keys update with nothing to change exits before any call.""" + result = runner.invoke( + app, ["audit", "keys", "update", "--key-id", "k"] + ) + assert result.exit_code != 0 + + +def test_audit_keys_regenerate_dry_run_skips_confirmation(): + """audit keys regenerate --dry-run previews the rotation.""" + result = _invoke_audit_dry_run( + "audit", "keys", "regenerate", "--key-id", "apikey-abc123" + ) + + assert "campus_audit.apikeys['apikey-abc123'].regenerate()" in result.stdout + assert "Are you sure" not in result.stdout + + +def test_audit_traces_list_dry_run(): + """audit traces list --dry-run shows the traces.list() call.""" + result = _invoke_audit_dry_run("audit", "traces", "list", "--limit", "10") + + assert "campus_audit.traces.list(" in result.stdout + assert "limit=10" in result.stdout + + +def test_audit_traces_get_dry_run(): + """audit traces get --dry-run shows indexed trace access.""" + result = _invoke_audit_dry_run( + "audit", "traces", "get", "--trace-id", "a" * 32 + ) + + assert f"campus_audit.traces[{'a' * 32!r}].get()" in result.stdout + + +def test_audit_traces_search_dry_run(): + """audit traces search --dry-run passes only the given filters.""" + result = _invoke_audit_dry_run( + "audit", "traces", "search", + "--path", "/auth/v1/users/", "--status", "500", "--user", "alice@school.edu", + ) + + assert "campus_audit.traces.search(" in result.stdout + assert "path='/auth/v1/users/'" in result.stdout + assert "status=500" in result.stdout + assert "user_id='alice@school.edu'" in result.stdout + + def test_dry_run_flag_documented_in_help(): """--dry-run appears in help for API-backed command groups.""" - for group_args in (["client"], ["vault"]): + for group_args in ( + ["client"], ["vault"], ["audit", "keys"], ["audit", "traces"], + ): result = runner.invoke(app, [*group_args, "list", "--help"]) assert result.exit_code == 0 assert "--dry-run" in result.stdout @@ -267,6 +385,24 @@ def test_dry_run_snippets_are_valid_python(): compile(code, f"", "exec") +def test_audit_dry_run_snippets_are_valid_python(): + """Generated audit snippets compile (#54).""" + cases = [ + ("audit", "keys", "new", "--name", "A", + "--owner", "o@school.edu", "--scope", "traces:read"), + ("audit", "keys", "get", "--key-id", "apikey-abc123"), + ("audit", "keys", "update", "--key-id", "apikey-abc123", + "--scope", "traces:read"), + ("audit", "traces", "list"), + ("audit", "traces", "list", "--limit", "10", "--since", "2026-10-01"), + ("audit", "traces", "search", "--path", "/x", "--status", "500"), + ] + for case in cases: + result = _invoke_audit_dry_run(*case) + code = _extract_snippet(result.stdout) + compile(code, f"", "exec") + + def test_dry_run_quotes_tricky_values(): """Values containing quotes are escaped in the snippet.""" result = _invoke_dry_run( diff --git a/tests/smoke/test_imports.py b/tests/smoke/test_imports.py index f0254d0..4535b23 100644 --- a/tests/smoke/test_imports.py +++ b/tests/smoke/test_imports.py @@ -33,6 +33,20 @@ def test_import_auth_modules(): assert spec is not None, f"{module} module not found" +def test_import_audit_modules(): + """Test that audit submodules can be imported without errors (#54).""" + audit_modules = [ + "campus_cli.audit.client", + "campus_cli.audit.keys", + "campus_cli.audit.login", + "campus_cli.audit.traces", + ] + + for module in audit_modules: + spec = importlib.util.find_spec(module) + assert spec is not None, f"{module} module not found" + + def test_import_utils_modules(): """Test that utility modules can be imported without errors.""" utils_modules = [ diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py new file mode 100644 index 0000000..e177bcd --- /dev/null +++ b/tests/unit/test_audit.py @@ -0,0 +1,343 @@ +"""Unit tests for the campus.audit command group (#54).""" + +from unittest import mock + +import pytest +import requests +import typer +from campus_python import errors +from typer.testing import CliRunner + +from campus_cli.audit.client import ( + api_error_details, + audit_endpoint_mismatch, + resolve_audit_url, +) +from campus_cli.audit.login import _resolve_key_input, verify_api_key +from campus_cli.audit.traces import _format_trace_summary, _status_markup +from campus_cli.cli import app +from campus_cli.config import Config +from campus_cli.credentials import CredentialError + +AUDIT_URL = "https://audit.example.test" + + +def _http_error(status, error_body=None): + """Build an HTTPError whose response carries the given status/body.""" + response = mock.Mock(spec=requests.Response, status_code=status) + if error_body is None: + response.json.side_effect = ValueError("no json body") + else: + response.json.return_value = error_body + return requests.HTTPError("error", response=response) + + +class TestApiErrorMessage: + """api_error_details extracts status and message from both error shapes.""" + + def test_client_library_api_error(self): + # campus_python raises errors.APIError for non-2xx responses. + error = errors.APIError( + status_code=401, error_description="Invalid API key" + ) + assert api_error_details(error) == (401, "Invalid API key") + + def test_raw_http_error_structured_body(self): + error = _http_error(400, { + "error": {"code": "AUDIT_INVALID", "message": "bad scopes"} + }) + assert api_error_details(error) == (400, "bad scopes") + + def test_raw_http_error_code_without_message(self): + error = _http_error(400, {"error": {"code": "AUDIT_INVALID"}}) + assert api_error_details(error) == (400, "AUDIT_INVALID") + + def test_raw_http_error_plain_body(self): + error = _http_error(400, {"error": "invalid_key"}) + assert api_error_details(error) == (400, "invalid_key") + + def test_raw_http_error_without_json_body(self): + status, message = api_error_details(_http_error(404)) + assert status == 404 + assert message # degrades to the exception text + + def test_network_error_keeps_exception_text(self): + error = requests.ConnectionError("connection refused") + assert api_error_details(error) == (None, "connection refused") + + +class TestResolveAuditUrl: + """audit URL resolution mirrors campus_python's precedence.""" + + @pytest.fixture + def config(self, tmp_path, monkeypatch): + """A Config isolated from the developer's real config file.""" + monkeypatch.delenv("CAMPUS_AUDIT_URL", raising=False) + monkeypatch.delenv("ENV", raising=False) + monkeypatch.delenv("CAMPUS_ENV", raising=False) + return Config(config_path=tmp_path / "config.json") + + def test_development_default(self, config): + assert config.audit_url == Config.DEFAULT_AUDIT_URL + + def test_staging_env_mapping(self, config, monkeypatch): + monkeypatch.setenv("ENV", "staging") + assert config.audit_url == Config.STAGING_AUDIT_URL + + def test_production_env_mapping(self, config, monkeypatch): + monkeypatch.setenv("ENV", "production") + assert config.audit_url == Config.PRODUCTION_AUDIT_URL + + def test_testing_maps_to_development(self, config, monkeypatch): + monkeypatch.setenv("CAMPUS_ENV", "testing") + assert config.audit_url == Config.DEFAULT_AUDIT_URL + + def test_explicit_env_var_wins(self, config, monkeypatch): + monkeypatch.setenv("ENV", "production") + monkeypatch.setenv("CAMPUS_AUDIT_URL", AUDIT_URL) + assert config.audit_url == AUDIT_URL + + def test_config_file_key_beats_env_mapping(self, config, monkeypatch): + monkeypatch.setenv("ENV", "production") + config.audit_url = AUDIT_URL + assert config.audit_url == AUDIT_URL + + def test_invalid_env_exits_cleanly(self, config, monkeypatch): + monkeypatch.setenv("ENV", "qa") + with ( + mock.patch("campus_cli.audit.client.config", config), + pytest.raises(typer.Exit), + ): + resolve_audit_url() + + +class TestAuditEndpointMismatch: + """Stored keys are bound to the audit URL they were stored for.""" + + def test_no_url_binding_matches(self, monkeypatch): + creds = mock.Mock() + creds.get_audit_key_url.return_value = None + monkeypatch.setattr("campus_cli.audit.client.credentials", creds) + assert audit_endpoint_mismatch() is None + + def test_trailing_slash_normalized(self, monkeypatch): + creds = mock.Mock() + creds.get_audit_key_url.return_value = AUDIT_URL + "/" + monkeypatch.setattr("campus_cli.audit.client.credentials", creds) + monkeypatch.setenv("CAMPUS_AUDIT_URL", AUDIT_URL) + assert audit_endpoint_mismatch() is None + + def test_mismatch_reports_both(self, monkeypatch): + creds = mock.Mock() + creds.get_audit_key_url.return_value = "https://audit-old.test" + monkeypatch.setattr("campus_cli.audit.client.credentials", creds) + monkeypatch.setenv("CAMPUS_AUDIT_URL", AUDIT_URL) + assert audit_endpoint_mismatch() == ("https://audit-old.test", AUDIT_URL) + + +class TestResolveKeyInput: + """The key comes from --api-key, the environment, or a prompt.""" + + def test_option_wins_over_env(self, monkeypatch): + monkeypatch.setenv("CAMPUS_AUDIT_API_KEY", "audit_v1_env") + assert _resolve_key_input("audit_v1_opt") == "audit_v1_opt" + + def test_env_fallback_is_stripped(self, monkeypatch): + monkeypatch.setenv("CAMPUS_AUDIT_API_KEY", " audit_v1_env \n") + assert _resolve_key_input(None) == "audit_v1_env" + + def test_prompted_when_interactive(self, monkeypatch): + monkeypatch.delenv("CAMPUS_AUDIT_API_KEY", raising=False) + monkeypatch.setattr("sys.stdin", mock.Mock(isatty=lambda: True)) + monkeypatch.setattr( + "campus_cli.audit.login.typer.prompt", + lambda *_args, **_kwargs: "audit_v1_prompted", + ) + assert _resolve_key_input(None) == "audit_v1_prompted" + + def test_nothing_given_exits_non_interactive(self, monkeypatch): + monkeypatch.delenv("CAMPUS_AUDIT_API_KEY", raising=False) + monkeypatch.setattr("sys.stdin", mock.Mock(isatty=lambda: False)) + with pytest.raises(typer.Exit): + _resolve_key_input(None) + + def test_blank_option_exits(self, monkeypatch): + monkeypatch.delenv("CAMPUS_AUDIT_API_KEY", raising=False) + monkeypatch.setattr("sys.stdin", mock.Mock(isatty=lambda: False)) + with pytest.raises(typer.Exit): + _resolve_key_input(" ") + + +class TestVerifyApiKey: + """Any authenticated response proves the key; 401 rejects it.""" + + def _verify(self, side_effect): + with mock.patch.multiple( + "campus_python", AuditRoot=mock.DEFAULT, CampusRequest=mock.DEFAULT + ) as mocked: + root = mocked["AuditRoot"].return_value + if side_effect is not None: + root.apikeys.list.side_effect = side_effect + return verify_api_key(AUDIT_URL, "audit_v1_k") + + def test_accepted_key_verifies(self): + assert self._verify(None) == "verified" + + def test_forbidden_key_is_authenticated(self): + # 403: valid key, but no apikeys:read (trace-only student key). + assert self._verify(_http_error(403)) == "verified" + + def test_rejected_key_raises_without_storing(self): + with pytest.raises(typer.Exit): + self._verify(_http_error(401)) + + def test_unreachable_service_stores_unverified(self): + assert self._verify( + requests.ConnectionError("connection refused") + ) == "unverified" + + def test_server_error_does_not_prove_the_key(self): + assert self._verify(_http_error(500)) == "unverified" + + +class TestAuditLoginCmd: + """campus audit login stores (or rejects) the key end to end.""" + + def test_login_stores_key_bound_to_url(self): + creds = mock.Mock() + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch( + "campus_cli.audit.login.verify_api_key", return_value="verified" + ), + mock.patch.dict( + "os.environ", + { + "CAMPUS_AUDIT_URL": AUDIT_URL, + "CAMPUS_AUDIT_API_KEY": "audit_v1_new", + }, + ), + ): + result = CliRunner().invoke(app, ["audit", "login"]) + assert result.exit_code == 0, result.output + creds.set_audit_api_key.assert_called_once_with( + "audit_v1_new", AUDIT_URL + ) + + def test_login_rejected_key_stores_nothing(self): + creds = mock.Mock() + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch( + "campus_cli.audit.login.verify_api_key", + side_effect=typer.Exit(1), + ), + mock.patch.dict( + "os.environ", + { + "CAMPUS_AUDIT_URL": AUDIT_URL, + "CAMPUS_AUDIT_API_KEY": "audit_v1_bad", + }, + ), + ): + result = CliRunner().invoke(app, ["audit", "login"]) + assert result.exit_code == 1 + creds.set_audit_api_key.assert_not_called() + + def test_login_skip_verify_stores_without_network(self): + creds = mock.Mock() + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch("campus_cli.audit.login.verify_api_key") as mock_verify, + mock.patch.dict( + "os.environ", + { + "CAMPUS_AUDIT_URL": AUDIT_URL, + "CAMPUS_AUDIT_API_KEY": "audit_v1_offline", + }, + ), + ): + result = CliRunner().invoke(app, ["audit", "login", "--skip-verify"]) + assert result.exit_code == 0, result.output + mock_verify.assert_not_called() + creds.set_audit_api_key.assert_called_once_with( + "audit_v1_offline", AUDIT_URL + ) + + def test_logout_clears_stored_key_with_confirm(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + with mock.patch("campus_cli.audit.login.credentials", creds): + result = CliRunner().invoke(app, ["audit", "logout", "--confirm"]) + assert result.exit_code == 0, result.output + creds.delete_audit_api_key.assert_called_once() + + def test_logout_refuses_without_confirm_when_not_a_tty(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + with mock.patch("campus_cli.audit.login.credentials", creds): + result = CliRunner().invoke(app, ["audit", "logout"]) + assert result.exit_code == 1 + creds.delete_audit_api_key.assert_not_called() + + def test_logout_when_not_logged_in_is_a_noop(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = None + with mock.patch("campus_cli.audit.login.credentials", creds): + result = CliRunner().invoke(app, ["audit", "logout"]) + assert result.exit_code == 0, result.output + creds.delete_audit_api_key.assert_not_called() + + +class TestTraceFormatting: + """Trace summaries render status, timing, and principal.""" + + def test_status_classes(self): + assert "green" in _status_markup(200) + assert "yellow" in _status_markup(404) + assert "red" in _status_markup(500) + assert "dim" in _status_markup(None) + + def test_summary_shows_request_and_timing(self): + line = _format_trace_summary({ + "trace_id": "a" * 32, + "span_count": 3, + "started_at": "2026-10-20T10:00:00+00:00", + "duration_ms": 42.5, + "root_span": { + "method": "GET", + "path": "/auth/v1/users/", + "status_code": 200, + "user_id": "alice@school.edu", + }, + }) + assert "GET /auth/v1/users/" in line + assert "42.5 ms" in line + assert "3 span(s)" in line + assert "alice@school.edu" in line + + def test_summary_falls_back_to_anonymous(self): + line = _format_trace_summary({ + "trace_id": "a" * 32, + "span_count": 1, + "duration_ms": 5, + "root_span": {"method": "POST", "path": "/x", "status_code": 401}, + }) + assert "anonymous" in line + + +class TestAuditCredentialStorage: + """The audit key rides the shared credential store, URL-bound.""" + + def test_round_trip(self, credential_storage): + credential_storage.set_audit_api_key("audit_v1_k", AUDIT_URL) + assert credential_storage.get_audit_api_key() == "audit_v1_k" + assert credential_storage.get_audit_key_url() == AUDIT_URL + + credential_storage.delete_audit_api_key() + assert credential_storage.get_audit_api_key() is None + assert credential_storage.get_audit_key_url() is None + + def test_delete_without_key_raises(self, credential_storage): + with pytest.raises(CredentialError): + credential_storage.delete_audit_api_key() From 12b62ebf4ea92c064aa27f1b433f764579e72d97 Mon Sep 17 00:00:00 2001 From: JS Date: Tue, 20 Oct 2026 20:52:54 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(cli):=20campus.audit=20review=20fixes?= =?UTF-8?q?=20=E2=80=94=20key=20status,=20store=20errors,=20logout=20note,?= =?UTF-8?q?=20polish=20(#56,=20#57,=20#58,=20#59)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the #55 review. - #56: the key status now distinguishes revoked / expired / active. A key with a future expiry is still active (the server counts it as such), so it no longer renders as "expires"; an expiry in the past renders as "expired". An unparsable expiry never counts as expired. - #57: credential-store failures in the audit group are reported and exit 1 instead of escaping as a traceback — the login write, the client key read, and the URL-binding read. - #58: `auth logout` names the still-stored campus.audit key and the command that clears it, so "logged out" does not read as "every stored credential is gone". - #59: no double full stop when the server's message is already a sentence, `--json` on `keys revoke`, `keys list --dry-run` shows only the arguments the caller passed, and the README leads with the env var / prompt for `campus audit login` and describes `owner_id` as a free-form label (the server accepts any string). Tests: 277 passing (18 new), ruff clean. --- README.md | 17 ++-- campus_cli/audit/client.py | 32 ++++++- campus_cli/audit/keys.py | 82 +++++++++++++---- campus_cli/audit/login.py | 6 +- campus_cli/auth/login.py | 13 +++ tests/unit/test_audit.py | 184 +++++++++++++++++++++++++++++++++++++ tests/unit/test_login.py | 33 +++++++ 7 files changed, 338 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index bf394f1..981a2a3 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,7 @@ campus grant grant --grantee-type user --grantee-id someone@school.edu \ --resource-type users --level write # campus.audit: mint a student's trace-reading key (operator, #54) -campus audit login --api-key +CAMPUS_AUDIT_API_KEY= campus audit login campus audit keys new --name "Alice" --owner alice@school.edu \ --scope traces:read --rate-limit 60 campus audit keys list --owner alice@school.edu @@ -150,9 +150,13 @@ URL it was stored for (per `ENV`/`CAMPUS_ENV` or `CAMPUS_AUDIT_URL`, like every other endpoint): ```bash -# Store a key: --api-key, the CAMPUS_AUDIT_API_KEY env var, or an -# interactive prompt. The key is verified against the service first -# (--skip-verify stores it without the round trip) +# Store a key: the CAMPUS_AUDIT_API_KEY env var, an interactive prompt +# (hidden input), or --api-key. Prefer the first two: --api-key lands +# the key in your shell history and in the process list. The key is +# verified against the service first (--skip-verify stores it without +# the round trip) +CAMPUS_AUDIT_API_KEY=audit_v1_… campus audit login +campus audit login # prompts for the key campus audit login --api-key audit_v1_… # …and the login state is cleared locally by: @@ -177,8 +181,9 @@ Two roles use the group: Plaintext key material is shown exactly once, at creation and at regeneration (the server keeps only the hash) — pass it to the - student, who stores it with `campus audit login`. `owner_id` is an - email-format label used for organization and filtering only; + student, who stores it with `campus audit login`. `owner_id` is a + free-form label used for organization and filtering only (an email + is the convention, but the server accepts any string); campus.audit is user-blind and self-contained by design, so it never resolves to a campus user account. diff --git a/campus_cli/audit/client.py b/campus_cli/audit/client.py index e4209b0..1709604 100644 --- a/campus_cli/audit/client.py +++ b/campus_cli/audit/client.py @@ -11,7 +11,7 @@ import typer from campus_cli.config import config -from campus_cli.credentials import credentials +from campus_cli.credentials import CredentialError, credentials from campus_cli.utils.output import print_error # Key material of the campus.audit service is minted with this prefix; @@ -37,6 +37,21 @@ def resolve_audit_url() -> str: raise typer.Exit(1) from e +def _stored_audit_key_url() -> str | None: + """Read the audit URL the stored key was bound to. + + Raises: + typer.Exit: If the credential store cannot be read (locked + keyring, unreadable fallback file) — the store + failure is reported, not raised as a traceback. + """ + try: + return credentials.get_audit_key_url() + except CredentialError as e: + print_error(f"Failed to read the stored audit API key: {e}") + raise typer.Exit(1) from e + + def audit_endpoint_mismatch() -> tuple[str, str] | None: """ Check whether the stored audit key was stored for the targeted service URL. @@ -46,7 +61,7 @@ def audit_endpoint_mismatch() -> tuple[str, str] | None: stored before URL binding (no stored URL) are treated as matching; they bind at the next 'campus audit login'. """ - stored = credentials.get_audit_key_url() + stored = _stored_audit_key_url() if not stored: return None current = resolve_audit_url() @@ -85,7 +100,11 @@ def get_audit_client(): Raises: typer.Exit: If no key is stored, or it targets another URL. """ - api_key = credentials.get_audit_api_key() + try: + api_key = credentials.get_audit_api_key() + except CredentialError as e: + print_error(f"Failed to read the stored audit API key: {e}") + raise typer.Exit(1) from e if not api_key: print_error( "Not authenticated with campus.audit. Run 'campus audit login'" @@ -155,5 +174,10 @@ def fail_api(action: str, e: Exception) -> None: " The stored key lacks the required scope;" " mint or update it with 'campus audit keys'." ) - print_error(f"Failed to {action}: {message}.{hint}") + # The server's message may already be a sentence: don't double the + # full stop when appending the remedy hint. + detail = message.strip() + if detail and not detail.endswith((".", "!", "?")): + detail += "." + print_error(f"Failed to {action}: {detail}{hint}") raise typer.Exit(1) from e diff --git a/campus_cli/audit/keys.py b/campus_cli/audit/keys.py index 29cc548..c936ad5 100644 --- a/campus_cli/audit/keys.py +++ b/campus_cli/audit/keys.py @@ -1,15 +1,16 @@ """campus.audit API key management (#54). The operator workflow for minting and managing audit API keys, e.g. -one per student for dev-environment access: `owner_id` is an -email-format label (organizational only — campus.audit is user-blind -and self-contained by design), `scopes` gate what the key may call -(e.g. ["traces:read"]), and key management itself requires apikeys:* -scopes, held only by the operator key. Plaintext key material is -shown exactly once, at creation and at regeneration (mirroring the -server). +one per student for dev-environment access: `owner_id` is a free-form +label, conventionally the student's email (organizational only — +campus.audit is user-blind and self-contained by design), `scopes` +gate what the key may call (e.g. ["traces:read"]), and key management +itself requires apikeys:* scopes, held only by the operator key. +Plaintext key material is shown exactly once, at creation and at +regeneration (mirroring the server). """ +from datetime import datetime, timezone from typing import Optional import typer @@ -61,13 +62,42 @@ def _print_reveal_key(key_value: str) -> None: console.print("[dim]-------8<-------[/dim]") +def _parse_timestamp(value: object) -> Optional[datetime]: + """Parse an ISO 8601 timestamp into an aware datetime. + + Returns: + The parsed datetime, or None when the value is missing or + unparsable (an unknown expiry never counts as expired). + """ + if not value or not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def _key_status(record: dict) -> str: + """Status label for a key record: revoked, expired, or active. + + A key with a future expiry is still active — the server counts a + key as active when it is neither revoked nor expired — so the + expiry is reported in the detail line, not as a status of its own. + """ + if record.get("revoked_at"): + return "[red]revoked[/red]" + expires_at = _parse_timestamp(record.get("expires_at")) + if expires_at is not None and expires_at < datetime.now(timezone.utc): + return "[red]expired[/red]" + return "[green]active[/green]" + + def _print_key_record(record: dict) -> None: """Print one API key record in human-readable form.""" - status = "[green]active[/green]" - if record.get("revoked_at"): - status = "[red]revoked[/red]" - elif record.get("expires_at"): - status = "[yellow]expires[/yellow]" + status = _key_status(record) console.print(f"[cyan]{record.get('name')}[/cyan] [dim]{record.get('id')}[/dim]") console.print( f" Owner: {record.get('owner_id')} Status: {status}" @@ -103,14 +133,24 @@ def keys_list( Requires apikeys:read on the stored key (the operator key). """ if dry_run: + # Only the fields the caller actually set: active_only defaults + # to True server-side, so it only appears when --all widened it. + list_kwargs = [] + if owner is not None: + list_kwargs.append(f" owner_id={owner!r},") + if include_inactive: + list_kwargs.append(" active_only=False,") + if limit is not None: + list_kwargs.append(f" limit={limit!r},") + call = ( + ["result = campus_audit.apikeys.list(", *list_kwargs, ")"] + if list_kwargs + else ["result = campus_audit.apikeys.list()"] + ) print_python_api( "campus audit keys list", [ - "result = campus_audit.apikeys.list(", - f" owner_id={owner!r},", - f" active_only={not include_inactive!r},", - f" limit={limit!r},", - ")", + *call, "for key in result['api_keys']:", " print(key['id'], key['name'], key['scopes'])", ], @@ -315,6 +355,7 @@ def keys_update( def keys_revoke( key_id: str = ApiKeyOption, confirm: bool = typer.Option(False, "--confirm", "-y", help="Skip confirmation"), + output_json: bool = typer.Option(False, "--json", help="Output as JSON"), dry_run: bool = dry_run_option(), ) -> None: """ @@ -336,7 +377,12 @@ def keys_revoke( try: api = get_audit_client() api.apikeys[key_id].revoke() - print_success(f"API key '{key_id}' revoked (record kept for audit).") + if output_json: + print_json({"id": key_id, "revoked": True}) + else: + print_success( + f"API key '{key_id}' revoked (record kept for audit)." + ) except typer.Exit: raise diff --git a/campus_cli/audit/login.py b/campus_cli/audit/login.py index f970d67..9a1027e 100644 --- a/campus_cli/audit/login.py +++ b/campus_cli/audit/login.py @@ -138,7 +138,11 @@ def login_cmd( if not skip_verify: verify_api_key(audit_url, api_key) - credentials.set_audit_api_key(api_key, audit_url) + try: + credentials.set_audit_api_key(api_key, audit_url) + except CredentialError as e: + print_error(f"Failed to store the audit API key: {e}") + raise typer.Exit(1) from e print_success("Audit API key stored.") console.print(f"[dim]Audit service: {audit_url}[/dim]") diff --git a/campus_cli/auth/login.py b/campus_cli/auth/login.py index dad83e9..79a6f4e 100644 --- a/campus_cli/auth/login.py +++ b/campus_cli/auth/login.py @@ -537,6 +537,19 @@ def logout_cmd( else: print_success("Not logged in.") + # The campus.audit API key rides the same credential store but is a + # separate credential (#54): say so, so "logged out" does not read + # as "every stored credential is gone". + try: + audit_key_stored = bool(credentials.get_audit_api_key()) + except CredentialError: + audit_key_stored = False + if audit_key_stored: + console.print( + "[dim]Note: the stored campus.audit API key is unchanged;" + " run 'campus audit logout' to clear it.[/dim]" + ) + @login_app.command("refresh") def refresh_cmd( diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py index e177bcd..41dad3d 100644 --- a/tests/unit/test_audit.py +++ b/tests/unit/test_audit.py @@ -11,8 +11,11 @@ from campus_cli.audit.client import ( api_error_details, audit_endpoint_mismatch, + fail_api, + get_audit_client, resolve_audit_url, ) +from campus_cli.audit.keys import _key_status, _parse_timestamp from campus_cli.audit.login import _resolve_key_input, verify_api_key from campus_cli.audit.traces import _format_trace_summary, _status_markup from campus_cli.cli import app @@ -341,3 +344,184 @@ def test_round_trip(self, credential_storage): def test_delete_without_key_raises(self, credential_storage): with pytest.raises(CredentialError): credential_storage.delete_audit_api_key() + + +class TestKeyStatus: + """Key status distinguishes revoked, expired, and still-active keys.""" + + def test_future_expiry_is_still_active(self): + # The server counts a key as active when it is neither revoked + # nor expired, so a future expiry must not read as a status. + record = { + "id": "k1", + "name": "Alice", + "owner_id": "alice@school.edu", + "scopes": ["traces:read"], + "expires_at": "2999-12-31T23:59:59+00:00", + "revoked_at": None, + } + assert "active" in _key_status(record) + + def test_past_expiry_is_expired(self): + record = { + "id": "k2", + "name": "Bob", + "owner_id": "bob@school.edu", + "scopes": ["traces:read"], + "expires_at": "2000-01-01T00:00:00+00:00", + "revoked_at": None, + } + assert "expired" in _key_status(record) + + def test_revoked_wins_over_expiry(self): + record = { + "id": "k3", + "name": "Carol", + "owner_id": "carol@school.edu", + "scopes": ["traces:read"], + "expires_at": "2000-01-01T00:00:00+00:00", + "revoked_at": "2026-10-05T00:00:00+00:00", + } + assert "revoked" in _key_status(record) + + def test_no_expiry_is_active(self): + record = { + "id": "k4", + "name": "Dan", + "owner_id": "dan@school.edu", + "scopes": ["traces:read"], + "expires_at": None, + "revoked_at": None, + } + assert "active" in _key_status(record) + + def test_unparsable_expiry_is_not_expired(self): + record = { + "id": "k5", + "name": "Eve", + "owner_id": "eve@school.edu", + "scopes": ["traces:read"], + "expires_at": "not-a-date", + "revoked_at": None, + } + assert "active" in _key_status(record) + + def test_timestamp_parsing_handles_zulu_and_naive(self): + assert _parse_timestamp("2026-12-31T23:59:59Z") is not None + assert _parse_timestamp("2026-12-31T23:59:59") is not None + assert _parse_timestamp("nonsense") is None + assert _parse_timestamp(None) is None + + +class TestFailApiMessage: + """Failed audit calls report one clean sentence plus the remedy.""" + + def _message(self, error): + with ( + mock.patch("campus_cli.audit.client.print_error") as mock_print, + pytest.raises(typer.Exit), + ): + fail_api("list API keys", error) + return mock_print.call_args[0][0] + + def test_no_double_full_stop(self): + message = self._message( + errors.APIError( + status_code=403, + error_description="Missing required scope: apikeys:read.", + ) + ) + assert "apikeys:read.. " not in message + assert "apikeys:read." in message + + def test_missing_stop_is_added(self): + message = self._message( + errors.APIError( + status_code=400, error_description="bad request" + ) + ) + assert "Failed to list API keys: bad request." in message + + def test_401_carries_the_relogin_remedy(self): + message = self._message( + errors.APIError(status_code=401, error_description="Invalid API key") + ) + assert "Re-run 'campus audit login'" in message + + +class TestCredentialStoreFailures: + """A broken credential store is reported, not raised as a traceback.""" + + def test_login_reports_a_store_failure(self): + creds = mock.Mock() + creds.set_audit_api_key.side_effect = CredentialError("keyring locked") + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch( + "campus_cli.audit.login.verify_api_key", return_value="verified" + ), + mock.patch.dict( + "os.environ", + { + "CAMPUS_AUDIT_URL": AUDIT_URL, + "CAMPUS_AUDIT_API_KEY": "audit_v1_new", + }, + ), + ): + result = CliRunner().invoke(app, ["audit", "login"]) + assert result.exit_code == 1 + assert "Failed to store the audit API key" in result.stderr + + def test_client_reports_an_unreadable_store(self, monkeypatch): + creds = mock.Mock() + creds.get_audit_api_key.side_effect = CredentialError("keyring locked") + monkeypatch.setattr("campus_cli.audit.client.credentials", creds) + with pytest.raises(typer.Exit): + get_audit_client() + + +class TestKeysRevokeJson: + """`keys revoke --json` reports the revocation as JSON.""" + + def test_json_output(self): + api = mock.MagicMock() + with mock.patch("campus_cli.audit.keys.get_audit_client", return_value=api): + result = CliRunner().invoke( + app, ["audit", "keys", "revoke", "--key-id", "k1", "-y", "--json"] + ) + assert result.exit_code == 0, result.output + assert '"revoked": true' in result.stdout + api.apikeys.__getitem__.assert_called_once_with("k1") + + +class TestKeysListDryRun: + """The keys list snippet carries only the fields the caller set.""" + + def test_no_filters_means_no_arguments(self): + result = CliRunner().invoke( + app, ["audit", "keys", "list", "--dry-run"] + ) + assert result.exit_code == 0, result.output + assert "campus_audit.apikeys.list()" in result.stdout + assert "owner_id=" not in result.stdout + assert "active_only=" not in result.stdout + + def test_only_the_given_filters_appear(self): + result = CliRunner().invoke( + app, + [ + "audit", "keys", "list", "--owner", "alice@school.edu", + "--dry-run", + ], + ) + assert result.exit_code == 0, result.output + assert "owner_id='alice@school.edu'" in result.stdout + assert "active_only=" not in result.stdout + assert "limit=" not in result.stdout + + def test_all_flag_widens_active_only(self): + result = CliRunner().invoke( + app, ["audit", "keys", "list", "--all", "--dry-run"] + ) + assert result.exit_code == 0, result.output + assert "active_only=False" in result.stdout diff --git a/tests/unit/test_login.py b/tests/unit/test_login.py index 01d8634..7446140 100644 --- a/tests/unit/test_login.py +++ b/tests/unit/test_login.py @@ -5,6 +5,7 @@ import pytest import requests from campus_python import errors +from typer.testing import CliRunner from campus_cli.auth.common import revoke_token from campus_cli.auth.login import ( @@ -12,6 +13,8 @@ poll_for_token, request_device_code, ) +from campus_cli.cli import app +from campus_cli.credentials import CredentialError def _http_error(error_body): @@ -334,3 +337,33 @@ def test_login_without_scope_response_records_neither(): creds.set_token_scopes.assert_not_called() creds.set_token_user_id.assert_not_called() + + +class TestLogoutAuditKeyNote: + """`auth logout` says when a campus.audit key is still stored (#54).""" + + def test_note_names_the_stored_audit_key(self): + creds = mock.Mock() + creds.get_token.return_value = None + creds.get_audit_api_key.return_value = "audit_v1_k" + with mock.patch("campus_cli.auth.login.credentials", creds): + result = CliRunner().invoke(app, ["auth", "logout"]) + assert result.exit_code == 0, result.output + assert "campus audit logout" in result.output + + def test_no_note_without_a_stored_audit_key(self): + creds = mock.Mock() + creds.get_token.return_value = None + creds.get_audit_api_key.return_value = None + with mock.patch("campus_cli.auth.login.credentials", creds): + result = CliRunner().invoke(app, ["auth", "logout"]) + assert result.exit_code == 0, result.output + assert "campus audit logout" not in result.output + + def test_unreadable_store_does_not_break_logout(self): + creds = mock.Mock() + creds.get_token.return_value = None + creds.get_audit_api_key.side_effect = CredentialError("locked") + with mock.patch("campus_cli.auth.login.credentials", creds): + result = CliRunner().invoke(app, ["auth", "logout"]) + assert result.exit_code == 0, result.output From 09821934500990744a341a96d3ee4d37f45f4ce1 Mon Sep 17 00:00:00 2001 From: JS Date: Tue, 20 Oct 2026 21:10:47 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(cli):=20campus.audit=20review=20follow-?= =?UTF-8?q?ups=20=E2=80=94=20error=20detail,=20JSON=20shapes,=20status=20(?= =?UTF-8?q?#60,=20#61,=20#62)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the three open review issues on #55. - #60: a non-2xx failure with no error_description (proxy HTML 502, empty body, misrouted 404 page) now renders as "HTTP " instead of an empty sentence — the login warning included — so the one breadcrumb an operator can act on actually reaches the message. The server's 422 field-level errors array (parsed by campus_python into APIError.errors) is appended, so a rejected 'keys new' names the offending field instead of a bare "One or more fields are invalid". - #61: 'keys list --json' prints the server's {"api_keys": [...], "count": n} envelope verbatim, matching the traces commands' shape — one shape for the whole group, documented in the README (bodyless actions like revoke report {"id": ..., "revoked": true}). - #62: new 'campus audit status' — a purely offline local view of the stored key: whether one is stored, the URL it is bound to, the URL the CLI currently targets (an invalid ENV/CAMPUS_ENV renders as unknown rather than dying, since that is exactly the state worth diagnosing), and when the key was stored (new audit_key_stored_at in the credential store; keys stored before this report "unknown"). A bound/target mismatch is called out with the same sentence the commands' guard uses (factored into client.mismatch_message). Exit 1 only when no key is stored; a mismatch is informational, mirroring 'campus auth status'. Tests: 293 passing (16 new), ruff clean. --- README.md | 13 +++ campus_cli/audit/__init__.py | 10 +- campus_cli/audit/client.py | 66 +++++++++-- campus_cli/audit/keys.py | 5 +- campus_cli/audit/login.py | 104 ++++++++++++++++- campus_cli/credentials.py | 18 +++ tests/unit/test_audit.py | 211 +++++++++++++++++++++++++++++++++++ 7 files changed, 413 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 981a2a3..bd590e3 100644 --- a/README.md +++ b/README.md @@ -161,8 +161,21 @@ campus audit login --api-key audit_v1_… # …and the login state is cleared locally by: campus audit logout + +# What does this install hold? A purely offline local view (no network +# call): whether a key is stored, the audit URL it is bound to, the URL +# the CLI currently targets (a mismatch is called out the way the +# commands' guard does), and when the key was stored. Exit 1 when no +# key is stored. --json emits the same fields for scripts. +campus audit status ``` +`--json` in this group prints the server response verbatim, so every +command has one shape: list envelopes keep their `count` +(`keys list`) and pagination `cursor` (`traces list/search`); a +bodyless action such as `keys revoke` reports `{"id": …, "revoked": +true}`. + Two roles use the group: - **The operator** holds the key with `apikeys:*` scopes (the one diff --git a/campus_cli/audit/__init__.py b/campus_cli/audit/__init__.py index d1ce2d8..d0194a3 100644 --- a/campus_cli/audit/__init__.py +++ b/campus_cli/audit/__init__.py @@ -4,21 +4,23 @@ authenticating with its own API keys (`audit_v1_...`, Bearer) rather than OAuth user tokens. The command group mirrors its two capabilities: the operator's key lifecycle (campus audit keys) and -read-only trace queries (campus audit traces), plus login/logout over -the stored key — the CLI's only non-OAuth credential. +read-only trace queries (campus audit traces), plus login/logout and +the offline status view over the stored key — the CLI's only +non-OAuth credential. """ import typer from campus_cli.audit.keys import keys_app -from campus_cli.audit.login import login_cmd, logout_cmd +from campus_cli.audit.login import login_cmd, logout_cmd, status_cmd from campus_cli.audit.traces import traces_app audit_app = typer.Typer(help="Campus Audit service commands (API-key auth)") -# login/logout attach directly; keys and traces are sub-groups. +# login/logout/status attach directly; keys and traces are sub-groups. audit_app.command("login")(login_cmd) audit_app.command("logout")(logout_cmd) +audit_app.command("status")(status_cmd) audit_app.add_typer(keys_app, name="keys", help="Audit API key lifecycle (operator)") audit_app.add_typer(traces_app, name="traces", help="Trace queries (read-only)") diff --git a/campus_cli/audit/client.py b/campus_cli/audit/client.py index 1709604..29803e7 100644 --- a/campus_cli/audit/client.py +++ b/campus_cli/audit/client.py @@ -70,6 +70,23 @@ def audit_endpoint_mismatch() -> tuple[str, str] | None: return None +def mismatch_message(stored: str | None, current: str | None) -> str | None: + """The standing binding-mismatch sentence, or None when URLs match. + + Shared by the command guard and 'campus audit status' (#62) so the + two word the problem identically. + """ + if not stored or not current: + return None + if stored.rstrip("/") == current.rstrip("/"): + return None + return ( + f"Stored audit API key was stored for {stored} but the CLI is " + f"targeting {current}. Re-run 'campus audit login' against " + f"{current}." + ) + + def ensure_audit_endpoint_match() -> None: """ Fail fast if the stored audit key targets a different service URL. @@ -82,11 +99,7 @@ def ensure_audit_endpoint_match() -> None: if mismatch is None: return stored, current = mismatch - print_error( - f"Stored audit API key was stored for {stored} but the CLI is " - f"targeting {current}. Re-run 'campus audit login' against " - f"{current}." - ) + print_error(mismatch_message(stored, current)) raise typer.Exit(1) @@ -127,6 +140,30 @@ def get_audit_client(): raise typer.Exit(1) from None +def _render_field_errors(e: Exception) -> str: + """Render the server's field-level validation errors, when present. + + The server's 422 body carries an errors array (field, code, + message) that campus_python parses into APIError.errors; without + this a rejected 'keys new' would report only "One or more fields + are invalid" with no indication of which field was rejected. + """ + parts = [] + for field_error in getattr(e, "errors", None) or []: + piece = str(getattr(field_error, "field", "?")) + code = getattr(field_error, "code", None) + message = getattr(field_error, "message", None) + if code: + piece += f" ({code}" + if message: + piece += f": {message}" + piece += ")" + elif message: + piece += f": {message}" + parts.append(piece) + return "; ".join(parts) + + def api_error_details(e: Exception) -> tuple[int | None, str]: """Best-effort (HTTP status, human message) from a failed audit call. @@ -134,6 +171,12 @@ def api_error_details(e: Exception) -> tuple[int | None, str]: the status code and the server's error_description; raw transport failures (requests.HTTPError or connection errors) keep their exception text, with any structured Campus error body unpacked. + + A non-2xx failure with no description (a proxy's HTML 502, an + empty body, a misrouted 404 page) falls back to "HTTP " — + the status is the one breadcrumb an operator can act on. The 422 + field-level errors array, when present, is appended so a rejected + payload names the offending field. """ status = getattr(e, "status_code", None) description = getattr(e, "error_description", None) @@ -148,9 +191,18 @@ def api_error_details(e: Exception) -> tuple[int | None, str]: description = error_obj.get("message") or error_obj.get("code") elif error_obj: description = error_obj + if description: - return status, str(description) - return status, str(e) + message = str(description) + elif status is not None: + message = f"HTTP {status}" + else: + message = str(e) + + fields = _render_field_errors(e) + if fields: + message = f"{message} (fields: {fields})" + return status, message def api_error_message(e: Exception) -> str: diff --git a/campus_cli/audit/keys.py b/campus_cli/audit/keys.py index c936ad5..8e90324 100644 --- a/campus_cli/audit/keys.py +++ b/campus_cli/audit/keys.py @@ -166,7 +166,10 @@ def keys_list( keys = result.get("api_keys", []) if output_json: - print_json(keys) + # The server response verbatim, like every --json output in + # this group (#61): the {"api_keys": [...], "count": n} + # envelope, not the bare array. + print_json(result) else: if not keys: console.print("[dim]No API keys found.[/dim]") diff --git a/campus_cli/audit/login.py b/campus_cli/audit/login.py index 9a1027e..4969d26 100644 --- a/campus_cli/audit/login.py +++ b/campus_cli/audit/login.py @@ -1,15 +1,19 @@ -"""campus.audit login/logout (#54). +"""campus.audit login/logout/status (#54, #62). The audit service only accepts its own API keys, so 'campus audit login' stores a key from --api-key, CAMPUS_AUDIT_API_KEY, or an interactive prompt instead of running the OAuth device flow. The key is verified against the service (best-effort, never blocking on network errors) and bound to the audit URL it was stored for, like -OAuth tokens are bound to their auth endpoint. +OAuth tokens are bound to their auth endpoint. 'status' reports the +stored key's local state; the audit service has no "who am I" +endpoint, so it is a purely offline view. """ +import json import os import sys +from datetime import datetime, timezone import requests import typer @@ -19,9 +23,12 @@ AUDIT_KEY_PREFIX, api_error_details, api_error_message, + mismatch_message, resolve_audit_url, ) +from campus_cli.audit.keys import _parse_timestamp from campus_cli.auth.common import confirm_destructive +from campus_cli.config import config from campus_cli.credentials import CredentialError, credentials from campus_cli.utils.output import print_error, print_success, print_warning @@ -172,3 +179,96 @@ def logout_cmd( raise typer.Exit(1) from e print_success("Logged out of campus.audit (local key cleared).") + + +def _relative_age(parsed: datetime) -> str: + """Coarse age of a stored key ('3 hours ago').""" + seconds = int((datetime.now(timezone.utc) - parsed).total_seconds()) + if seconds < 0: + # Clock skew between storage and now; don't render a negative. + return "just now" + if seconds < 90: + return f"{seconds} seconds ago" + minutes = seconds // 60 + if minutes < 90: + return f"{minutes} minutes ago" + hours = minutes // 60 + if hours < 36: + return f"{hours} hours ago" + return f"{hours // 24} days ago" + + +def status_cmd( + output_json: bool = typer.Option( + False, "--json", help="Output status as JSON" + ), +) -> None: + """ + Show the stored audit key's local state (#62). + + The audit service has no "who am I" endpoint, so this is a purely + local view: whether a key is stored, which URL it is bound to, the + URL the CLI currently targets (flagging a mismatch the way the + commands' guard does), and when the key was stored. No network + call is made — unlike 'campus auth status', there is no online + check. Exit code: 1 when no key is stored, 0 otherwise (a + bound/target mismatch is reported but does not fail the check; + every API-backed command refuses on its own until re-login). + """ + try: + api_key = credentials.get_audit_api_key() + key_url = credentials.get_audit_key_url() + stored_at = credentials.get_audit_key_stored_at() + except CredentialError as e: + print_error(f"Failed to read the stored audit API key: {e}") + raise typer.Exit(1) from e + + # A local view must not die on an unresolvable target: an invalid + # ENV/CAMPUS_ENV is exactly the state worth diagnosing here. + try: + current_url: str | None = config.audit_url + except ValueError: + current_url = None + + mismatch = mismatch_message(key_url, current_url) + + if output_json: + # Plain stdout write: Rich's word wrap can corrupt JSON for a + # parsing consumer (same reasoning as 'campus auth refresh'). + typer.echo(json.dumps({ + "authenticated": api_key is not None, + "audit_url": current_url, + "key_url": key_url, + "endpoint_match": mismatch is None, + "stored_at": stored_at, + })) + elif not api_key: + print_error( + "Not logged in to campus.audit." + " Run 'campus audit login' first." + ) + if current_url: + console.print(f"Audit service (target): {current_url}") + else: + print_success("Audit API key stored.") + if current_url: + console.print(f"Audit service (target): {current_url}") + else: + console.print( + "Audit service (target): [red]unknown[/red]" + " (invalid ENV/CAMPUS_ENV; set CAMPUS_AUDIT_URL)" + ) + console.print(f"Key bound to: {key_url or '[dim]none[/dim]'}") + parsed_stored_at = _parse_timestamp(stored_at) + if parsed_stored_at is None: + console.print("[dim]Stored: unknown[/dim]") + else: + console.print( + f"Stored: {parsed_stored_at.strftime('%Y-%m-%d %H:%M:%S UTC')}" + f" ({_relative_age(parsed_stored_at)})" + ) + if mismatch: + print_error(mismatch) + + if not api_key: + raise typer.Exit(1) diff --git a/campus_cli/credentials.py b/campus_cli/credentials.py index 5325487..f60b3f6 100644 --- a/campus_cli/credentials.py +++ b/campus_cli/credentials.py @@ -280,13 +280,18 @@ def set_audit_api_key(self, api_key: str, audit_url: str) -> None: """ Store the campus.audit API key, bound to the service URL (#54). + Records the storage time, for 'campus audit status' (#62) to + report the key's age. + Args: api_key: The plaintext API key. audit_url: The audit service base URL the key was stored (and verified) against. """ + stored_at = datetime.now(timezone.utc).isoformat() self.set_password("audit_api_key", api_key) self.set_password("audit_key_url", audit_url) + self.set_password("audit_key_stored_at", stored_at) def get_audit_key_url(self) -> str | None: """ @@ -297,6 +302,16 @@ def get_audit_key_url(self) -> str | None: """ return self.get_password("audit_key_url") + def get_audit_key_stored_at(self) -> str | None: + """ + Get the ISO timestamp the audit API key was stored at (#62). + + Returns: + The timestamp, or None for keys stored before recording + was introduced (unknown, not missing). + """ + return self.get_password("audit_key_stored_at") + def delete_audit_api_key(self) -> None: """ Delete the stored campus.audit API key and its URL binding (#54). @@ -308,6 +323,9 @@ def delete_audit_api_key(self) -> None: with contextlib.suppress(CredentialError): # May not exist (keys stored before URL binding, #54) self.delete_password("audit_key_url") + with contextlib.suppress(CredentialError): + # May not exist (keys stored before timestamp recording, #62) + self.delete_password("audit_key_stored_at") def get_login_session_id(self) -> str | None: """ diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py index 41dad3d..47c3934 100644 --- a/tests/unit/test_audit.py +++ b/tests/unit/test_audit.py @@ -1,5 +1,6 @@ """Unit tests for the campus.audit command group (#54).""" +import json from unittest import mock import pytest @@ -525,3 +526,213 @@ def test_all_flag_widens_active_only(self): ) assert result.exit_code == 0, result.output assert "active_only=False" in result.stdout + + +class TestApiErrorFallbacks: + """Failures with no description still carry their status; 422s name fields.""" + + def test_missing_description_falls_back_to_status(self): + # A proxy's HTML 502 or an empty body: APIError with no + # description must not render as an empty message. + error = errors.APIError(status_code=502) + status, message = api_error_details(error) + assert status == 502 + assert message == "HTTP 502" + + def test_field_level_errors_are_surfaced(self): + error = errors.APIError( + status_code=422, + error_description="One or more fields are invalid", + errors=[ + errors.FieldError( + field="scopes", + code="EMPTY", + message="must be a non-empty list", + ) + ], + ) + status, message = api_error_details(error) + assert status == 422 + assert "One or more fields are invalid" in message + assert "scopes (EMPTY: must be a non-empty list)" in message + + def test_field_errors_append_to_status_fallback(self): + error = errors.APIError( + status_code=422, + errors=[ + errors.FieldError(field="rate_limit", code="INVALID", message=""), + ], + ) + _, message = api_error_details(error) + assert message.startswith("HTTP 422") + assert "rate_limit (INVALID)" in message + + +class TestMismatchMessage: + """The guard and status share one mismatch sentence (#62).""" + + def test_mismatch_sentence_matches_the_guard(self): + from campus_cli.audit.client import mismatch_message + + message = mismatch_message("https://audit-old.test", AUDIT_URL) + assert message == ( + "Stored audit API key was stored for https://audit-old.test" + f" but the CLI is targeting {AUDIT_URL}." + f" Re-run 'campus audit login' against {AUDIT_URL}." + ) + + def test_matching_or_missing_urls_give_none(self): + from campus_cli.audit.client import mismatch_message + + assert mismatch_message(AUDIT_URL, AUDIT_URL + "/") is None + assert mismatch_message(None, AUDIT_URL) is None + assert mismatch_message(AUDIT_URL, None) is None + + +class TestAuditStatusCmd: + """campus audit status is a purely offline local view (#62).""" + + def _invoke(self, creds, *args): + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch.dict( + "os.environ", {"CAMPUS_AUDIT_URL": AUDIT_URL} + ), + ): + return CliRunner().invoke(app, ["audit", "status", *args]) + + def test_makes_no_network_call(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = AUDIT_URL + creds.get_audit_key_stored_at.return_value = None + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch.dict("os.environ", {"CAMPUS_AUDIT_URL": AUDIT_URL}), + mock.patch( + "campus_python.CampusRequest", + side_effect=AssertionError("status must not build a client"), + ), + ): + result = CliRunner().invoke(app, ["audit", "status"]) + assert result.exit_code == 0, result.output + + def test_not_logged_in_exits_1(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = None + creds.get_audit_key_url.return_value = None + creds.get_audit_key_stored_at.return_value = None + result = self._invoke(creds) + assert result.exit_code == 1 + assert "Not logged in" in result.output + + def test_stored_key_reports_binding_and_target(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = AUDIT_URL + creds.get_audit_key_stored_at.return_value = "2026-10-20T04:00:00+00:00" + result = self._invoke(creds) + assert result.exit_code == 0, result.output + assert "Audit API key stored." in result.stdout + assert AUDIT_URL in result.stdout + assert "Stored:" in result.stdout + + def test_mismatch_is_called_out_like_the_guard(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = "https://audit-old.test" + creds.get_audit_key_stored_at.return_value = None + result = self._invoke(creds) + # Informational: exit 0, but the standing mismatch sentence + # (print_error writes stderr; result.output mixes both streams). + assert result.exit_code == 0, result.output + assert "was stored for https://audit-old.test" in result.output + assert "Re-run 'campus audit login'" in result.output + + def test_unknown_stored_at_is_reported_as_unknown(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = AUDIT_URL + creds.get_audit_key_stored_at.return_value = None + result = self._invoke(creds) + assert result.exit_code == 0, result.output + assert "Stored: unknown" in result.stdout + + def test_json_shape(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = AUDIT_URL + creds.get_audit_key_stored_at.return_value = "2026-10-20T04:00:00+00:00" + result = self._invoke(creds, "--json") + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload == { + "authenticated": True, + "audit_url": AUDIT_URL, + "key_url": AUDIT_URL, + "endpoint_match": True, + "stored_at": "2026-10-20T04:00:00+00:00", + } + + def test_json_not_logged_in_still_parses(self): + creds = mock.Mock() + creds.get_audit_api_key.return_value = None + creds.get_audit_key_url.return_value = None + creds.get_audit_key_stored_at.return_value = None + result = self._invoke(creds, "--json") + assert result.exit_code == 1 + payload = json.loads(result.stdout) + assert payload["authenticated"] is False + + def test_invalid_env_still_reports_the_local_state(self, tmp_path): + creds = mock.Mock() + creds.get_audit_api_key.return_value = "audit_v1_k" + creds.get_audit_key_url.return_value = AUDIT_URL + creds.get_audit_key_stored_at.return_value = None + isolated = Config(config_path=tmp_path / "config.json") + with ( + mock.patch("campus_cli.audit.login.credentials", creds), + mock.patch("campus_cli.audit.login.config", isolated), + mock.patch.dict("os.environ", {"ENV": "qa"}), + ): + result = CliRunner().invoke(app, ["audit", "status"]) + assert result.exit_code == 0, result.output + assert "unknown" in result.stdout + + def test_store_failure_is_reported(self): + creds = mock.Mock() + creds.get_audit_api_key.side_effect = CredentialError("keyring locked") + result = self._invoke(creds) + assert result.exit_code == 1 + assert "Failed to read the stored audit API key" in result.stderr + + +class TestAuditJsonShapes: + """Every --json output prints the server response verbatim (#61).""" + + def test_keys_list_json_keeps_the_envelope(self): + server_response = { + "api_keys": [{"id": "k1", "name": "Alice"}], + "count": 1, + } + api = mock.Mock() + api.apikeys.list.return_value = server_response + with mock.patch("campus_cli.audit.keys.get_audit_client", return_value=api): + result = CliRunner().invoke(app, ["audit", "keys", "list", "--json"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == server_response + + def test_traces_list_json_keeps_the_envelope(self): + server_response = { + "traces": [{"trace_id": "a" * 32}], + "cursor": {"next": None, "has_more": False}, + } + api = mock.Mock() + api.traces.list.return_value = server_response + with ( + mock.patch("campus_cli.audit.traces.get_audit_client", return_value=api), + mock.patch.dict("os.environ", {"CAMPUS_AUDIT_URL": AUDIT_URL}), + ): + result = CliRunner().invoke(app, ["audit", "traces", "list", "--json"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == server_response