From f8a835e476a22b47ba507941953132bcfa564eef Mon Sep 17 00:00:00 2001 From: teknium1 Date: Sat, 14 Mar 2026 23:32:45 -0700 Subject: [PATCH] fix: harden google workspace oauth setup UX Reduce auth scopes to the requested services, add JSON-mode auth URL output, regenerate fresh auth URLs on stale/expired code failures, and document the headless copy-paste flow more clearly. --- skills/productivity/google-workspace/SKILL.md | 41 ++- .../google-workspace/scripts/google_api.py | 13 +- .../google-workspace/scripts/setup.py | 315 +++++++++++++++--- tests/skills/test_google_oauth_setup.py | 117 ++++++- 4 files changed, 406 insertions(+), 80 deletions(-) diff --git a/skills/productivity/google-workspace/SKILL.md b/skills/productivity/google-workspace/SKILL.md index dbbc54cc04..b86e329bbd 100644 --- a/skills/productivity/google-workspace/SKILL.md +++ b/skills/productivity/google-workspace/SKILL.md @@ -55,8 +55,15 @@ Calendar/Drive/Sheets/Docs?"** Passwords) and takes 2 minutes to set up. No Google Cloud project needed. Load the himalaya skill and follow its setup instructions. -- **Calendar, Drive, Sheets, Docs (or email + these)** → Continue with this - skill's OAuth setup below. +- **Email + Calendar** → Continue with this skill, but use + `--services email,calendar` during auth so the consent screen only asks for + the scopes they actually need. + +- **Calendar/Drive/Sheets/Docs only** → Continue with this skill and use a + narrower `--services` set like `calendar,drive,sheets,docs`. + +- **Full Workspace access** → Continue with this skill and use the default + `all` service set. **Question 2: "Does your Google account use Advanced Protection (hardware security keys required to sign in)? If you're not sure, you probably don't @@ -96,18 +103,29 @@ Once they provide the path: $GSETUP --client-secret /path/to/client_secret.json ``` +If they paste the raw client ID / client secret values instead of a file path, +write a valid Desktop OAuth JSON file for them yourself, save it somewhere +explicit (for example `~/Downloads/hermes-google-client-secret.json`), then run +`--client-secret` against that file. + ### Step 3: Get authorization URL +Use the service set chosen in Step 1. Examples: + ```bash -$GSETUP --auth-url +$GSETUP --auth-url --services email,calendar --format json +$GSETUP --auth-url --services calendar,drive,sheets,docs --format json +$GSETUP --auth-url --services all --format json ``` -This prints a URL. **Send the URL to the user** and tell them: +This returns JSON with an `auth_url` field and also saves the exact URL to +`~/.hermes/google_oauth_last_url.txt`. -> Open this link in your browser, sign in with your Google account, and -> authorize access. After authorizing, you'll be redirected to a page that -> may show an error — that's expected. Copy the ENTIRE URL from your -> browser's address bar and paste it back to me. +Agent rules for this step: +- Extract the `auth_url` field and send that exact URL to the user as a single line. +- Tell the user that the browser will likely fail on `http://localhost:1` after approval, and that this is expected. +- Tell them to copy the ENTIRE redirected URL from the browser address bar. +- If the user gets `Error 403: access_denied`, send them directly to `https://console.cloud.google.com/auth/audience` to add themselves as a test user. ### Step 4: Exchange the code @@ -117,9 +135,14 @@ pending OAuth session locally so `--auth-code` can complete the PKCE exchange later, even on headless systems: ```bash -$GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED" +$GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED" --format json ``` +If `--auth-code` fails because the code expired, was already used, or came from +an older browser tab, it now returns a fresh `fresh_auth_url`. In that case, +immediately send the new URL to the user and have them retry with the newest +browser redirect only. + ### Step 5: Verify ```bash diff --git a/skills/productivity/google-workspace/scripts/google_api.py b/skills/productivity/google-workspace/scripts/google_api.py index 13affb849d..788e8b9f03 100644 --- a/skills/productivity/google-workspace/scripts/google_api.py +++ b/skills/productivity/google-workspace/scripts/google_api.py @@ -54,6 +54,17 @@ def _ensure_authenticated(): sys.exit(1) +def _stored_token_scopes() -> list[str]: + try: + data = json.loads(TOKEN_PATH.read_text()) + except Exception: + return list(SCOPES) + scopes = data.get("scopes") + if isinstance(scopes, list) and scopes: + return scopes + return list(SCOPES) + + def _gws_binary() -> str | None: override = os.getenv("HERMES_GWS_BIN") if override: @@ -156,7 +167,7 @@ def get_credentials(): from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request - creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES) + creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), _stored_token_scopes()) if creds.expired and creds.refresh_token: creds.refresh(Request()) TOKEN_PATH.write_text(creds.to_json()) diff --git a/skills/productivity/google-workspace/scripts/setup.py b/skills/productivity/google-workspace/scripts/setup.py index 14f9c6bf30..792e9bb030 100644 --- a/skills/productivity/google-workspace/scripts/setup.py +++ b/skills/productivity/google-workspace/scripts/setup.py @@ -21,28 +21,50 @@ Agent workflow: 6. Run --check to verify. Done. """ +from __future__ import annotations + import argparse import json import os import subprocess import sys from pathlib import Path +from typing import Iterable HERMES_HOME = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) TOKEN_PATH = HERMES_HOME / "google_token.json" CLIENT_SECRET_PATH = HERMES_HOME / "google_client_secret.json" PENDING_AUTH_PATH = HERMES_HOME / "google_oauth_pending.json" +LAST_AUTH_URL_PATH = HERMES_HOME / "google_oauth_last_url.txt" -SCOPES = [ - "https://www.googleapis.com/auth/gmail.readonly", - "https://www.googleapis.com/auth/gmail.send", - "https://www.googleapis.com/auth/gmail.modify", - "https://www.googleapis.com/auth/calendar", - "https://www.googleapis.com/auth/drive.readonly", - "https://www.googleapis.com/auth/contacts.readonly", - "https://www.googleapis.com/auth/spreadsheets", - "https://www.googleapis.com/auth/documents.readonly", -] +SERVICE_SCOPE_GROUPS = { + "email": [ + "https://www.googleapis.com/auth/gmail.readonly", + "https://www.googleapis.com/auth/gmail.send", + "https://www.googleapis.com/auth/gmail.modify", + ], + "calendar": ["https://www.googleapis.com/auth/calendar"], + "drive": ["https://www.googleapis.com/auth/drive.readonly"], + "contacts": ["https://www.googleapis.com/auth/contacts.readonly"], + "sheets": ["https://www.googleapis.com/auth/spreadsheets"], + "docs": ["https://www.googleapis.com/auth/documents.readonly"], +} +SERVICE_ALIASES = { + "all": "all", + "email": "email", + "gmail": "email", + "mail": "email", + "calendar": "calendar", + "cal": "calendar", + "drive": "drive", + "contacts": "contacts", + "people": "contacts", + "sheets": "sheets", + "docs": "docs", + "documents": "docs", +} +DEFAULT_SERVICES = ["email", "calendar", "drive", "contacts", "sheets", "docs"] +ALL_SCOPES = [scope for service in DEFAULT_SERVICES for scope in SERVICE_SCOPE_GROUPS[service]] REQUIRED_PACKAGES = ["google-api-python-client", "google-auth-oauthlib", "google-auth-httplib2"] @@ -50,6 +72,10 @@ REQUIRED_PACKAGES = ["google-api-python-client", "google-auth-oauthlib", "google # Google deprecated OOB, so we use a localhost redirect and tell the user to # copy the code from the browser's URL bar (or the page body). REDIRECT_URI = "http://localhost:1" +AUDIENCE_URL = "https://console.cloud.google.com/auth/audience" +PROJECT_SELECTOR_URL = "https://console.cloud.google.com/projectselector2/home/dashboard" +API_LIBRARY_URL = "https://console.cloud.google.com/apis/library" +CREDENTIALS_URL = "https://console.cloud.google.com/apis/credentials" def install_deps(): @@ -86,18 +112,98 @@ def _ensure_deps(): sys.exit(1) -def check_auth(): +def _dedupe(items: Iterable[str]) -> list[str]: + seen = set() + result = [] + for item in items: + if item not in seen: + seen.add(item) + result.append(item) + return result + + +def _resolve_services(services_text: str | None) -> tuple[list[str], list[str]]: + """Resolve a comma/space separated service list to canonical service names + scopes.""" + text = (services_text or "all").strip().lower() + if not text or text == "all": + services = list(DEFAULT_SERVICES) + return services, list(ALL_SCOPES) + + raw_parts = [part.strip() for part in text.replace(" ", ",").split(",") if part.strip()] + canonical = [] + unknown = [] + for part in raw_parts: + alias = SERVICE_ALIASES.get(part) + if alias == "all": + return list(DEFAULT_SERVICES), list(ALL_SCOPES) + if not alias: + unknown.append(part) + continue + canonical.append(alias) + + if unknown: + print( + "ERROR: Unknown Google service(s): " + + ", ".join(sorted(set(unknown))) + + ". Supported values: all, email, calendar, drive, contacts, sheets, docs." + ) + sys.exit(1) + + canonical = _dedupe(canonical) + scopes = [scope for service in canonical for scope in SERVICE_SCOPE_GROUPS[service]] + return canonical, scopes + + +def _stored_token_scopes() -> list[str] | None: + if not TOKEN_PATH.exists(): + return None + try: + data = json.loads(TOKEN_PATH.read_text()) + except Exception: + return None + scopes = data.get("scopes") + if isinstance(scopes, list) and scopes: + return scopes + return None + + +def _credentials_scopes(services_text: str | None = None) -> list[str]: + requested_scopes = None + if services_text: + _, requested_scopes = _resolve_services(services_text) + + stored_scopes = _stored_token_scopes() + if stored_scopes: + if requested_scopes: + missing = [scope for scope in requested_scopes if scope not in stored_scopes] + if missing: + print("TOKEN_MISSING_SCOPES: Stored token does not include the requested services.") + print("Missing scopes:") + for scope in missing: + print(f" - {scope}") + print("Re-run setup with a fresh auth URL for the services you need.") + return [] + return stored_scopes + + return requested_scopes or list(ALL_SCOPES) + + +def check_auth(services_text: str | None = None): """Check if stored credentials are valid. Prints status, exits 0 or 1.""" if not TOKEN_PATH.exists(): print(f"NOT_AUTHENTICATED: No token at {TOKEN_PATH}") return False + scopes = _credentials_scopes(services_text) + if not scopes: + return False + _ensure_deps() from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request try: - creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES) + creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), scopes) except Exception as e: print(f"TOKEN_CORRUPT: {e}") return False @@ -135,14 +241,14 @@ def store_client_secret(path: str): if "installed" not in data and "web" not in data: print("ERROR: Not a Google OAuth client secret file (missing 'installed' key).") - print("Download the correct file from: https://console.cloud.google.com/apis/credentials") + print(f"Download the correct file from: {CREDENTIALS_URL}") sys.exit(1) CLIENT_SECRET_PATH.write_text(json.dumps(data, indent=2)) print(f"OK: Client secret saved to {CLIENT_SECRET_PATH}") -def _save_pending_auth(*, state: str, code_verifier: str): +def _save_pending_auth(*, state: str, code_verifier: str, scopes: list[str], services: list[str], auth_url: str): """Persist the OAuth session bits needed for a later token exchange.""" PENDING_AUTH_PATH.write_text( json.dumps( @@ -150,10 +256,14 @@ def _save_pending_auth(*, state: str, code_verifier: str): "state": state, "code_verifier": code_verifier, "redirect_uri": REDIRECT_URI, + "scopes": scopes, + "services": services, + "auth_url": auth_url, }, indent=2, ) ) + LAST_AUTH_URL_PATH.write_text(auth_url) def _load_pending_auth() -> dict: @@ -174,6 +284,8 @@ def _load_pending_auth() -> dict: print("Run --auth-url again to start a fresh OAuth session.") sys.exit(1) + data.setdefault("scopes", list(ALL_SCOPES)) + data.setdefault("services", list(DEFAULT_SERVICES)) return data @@ -188,37 +300,96 @@ def _extract_code_and_state(code_or_url: str) -> tuple[str, str | None]: params = parse_qs(parsed.query) if "code" not in params: print("ERROR: No 'code' parameter found in URL.") + print("When the browser lands on the localhost error page, copy the FULL address bar URL.") sys.exit(1) state = params.get("state", [None])[0] return params["code"][0], state -def get_auth_url(): +def _build_flow(scopes: list[str], *, state: str | None = None, code_verifier: str | None = None, autogenerate_code_verifier: bool = False): + _ensure_deps() + from google_auth_oauthlib.flow import Flow + + return Flow.from_client_secrets_file( + str(CLIENT_SECRET_PATH), + scopes=scopes, + redirect_uri=REDIRECT_URI, + state=state, + code_verifier=code_verifier, + autogenerate_code_verifier=autogenerate_code_verifier, + ) + + +def _create_auth_session(services_text: str | None, *, output_format: str = "plain", emit_output: bool = True) -> str: + services, scopes = _resolve_services(services_text) + flow = _build_flow(scopes, autogenerate_code_verifier=True) + auth_url, state = flow.authorization_url(access_type="offline", prompt="consent") + _save_pending_auth( + state=state, + code_verifier=flow.code_verifier, + scopes=scopes, + services=services, + auth_url=auth_url, + ) + + if emit_output: + if output_format == "json": + print( + json.dumps( + { + "success": True, + "auth_url": auth_url, + "auth_url_file": str(LAST_AUTH_URL_PATH), + "services": services, + "scopes": scopes, + "project_selector_url": PROJECT_SELECTOR_URL, + "api_library_url": API_LIBRARY_URL, + "credentials_url": CREDENTIALS_URL, + "audience_url": AUDIENCE_URL, + "instructions": [ + "Open auth_url in your browser.", + "If the browser lands on a localhost error page, that is expected.", + "Copy the FULL redirected URL from the address bar and pass it to --auth-code.", + ], + }, + indent=2, + ) + ) + else: + print(auth_url) + return auth_url + + +def get_auth_url(services_text: str | None = None, *, output_format: str = "plain"): """Print the OAuth authorization URL. User visits this in a browser.""" if not CLIENT_SECRET_PATH.exists(): print("ERROR: No client secret stored. Run --client-secret first.") sys.exit(1) - _ensure_deps() - from google_auth_oauthlib.flow import Flow - - flow = Flow.from_client_secrets_file( - str(CLIENT_SECRET_PATH), - scopes=SCOPES, - redirect_uri=REDIRECT_URI, - autogenerate_code_verifier=True, - ) - auth_url, state = flow.authorization_url( - access_type="offline", - prompt="consent", - ) - _save_pending_auth(state=state, code_verifier=flow.code_verifier) - # Print just the URL so the agent can extract it cleanly - print(auth_url) + _create_auth_session(services_text, output_format=output_format) -def exchange_auth_code(code: str): +def _print_recovery_url(auth_url: str, output_format: str): + if output_format == "json": + print( + json.dumps( + { + "success": False, + "fresh_auth_url": auth_url, + "auth_url_file": str(LAST_AUTH_URL_PATH), + "audience_url": AUDIENCE_URL, + }, + indent=2, + ) + ) + else: + print("A fresh auth URL has been generated. Use this exact URL:") + print(auth_url) + print(f"If Google blocks access, add your account as a test user here: {AUDIENCE_URL}") + + +def exchange_auth_code(code: str, *, output_format: str = "plain"): """Exchange the authorization code for a token and save it.""" if not CLIENT_SECRET_PATH.exists(): print("ERROR: No client secret stored. Run --client-secret first.") @@ -227,16 +398,18 @@ def exchange_auth_code(code: str): pending_auth = _load_pending_auth() code, returned_state = _extract_code_and_state(code) if returned_state and returned_state != pending_auth["state"]: - print("ERROR: OAuth state mismatch. Run --auth-url again to start a fresh session.") + auth_url = _create_auth_session( + ",".join(pending_auth.get("services", DEFAULT_SERVICES)), + output_format=output_format, + emit_output=False, + ) + if output_format != "json": + print("ERROR: OAuth state mismatch. Your browser redirect came from an older auth session.") + _print_recovery_url(auth_url, output_format) sys.exit(1) - _ensure_deps() - from google_auth_oauthlib.flow import Flow - - flow = Flow.from_client_secrets_file( - str(CLIENT_SECRET_PATH), - scopes=SCOPES, - redirect_uri=pending_auth.get("redirect_uri", REDIRECT_URI), + flow = _build_flow( + pending_auth.get("scopes", list(ALL_SCOPES)), state=pending_auth["state"], code_verifier=pending_auth["code_verifier"], ) @@ -244,14 +417,33 @@ def exchange_auth_code(code: str): try: flow.fetch_token(code=code) except Exception as e: - print(f"ERROR: Token exchange failed: {e}") - print("The code may have expired. Run --auth-url to get a fresh URL.") + auth_url = _create_auth_session( + ",".join(pending_auth.get("services", DEFAULT_SERVICES)), + output_format=output_format, + emit_output=False, + ) + if output_format != "json": + print(f"ERROR: Token exchange failed: {e}") + print("The code may have expired or already been used.") + _print_recovery_url(auth_url, output_format) sys.exit(1) creds = flow.credentials TOKEN_PATH.write_text(creds.to_json()) PENDING_AUTH_PATH.unlink(missing_ok=True) - print(f"OK: Authenticated. Token saved to {TOKEN_PATH}") + if output_format == "json": + print( + json.dumps( + { + "success": True, + "token_path": str(TOKEN_PATH), + "services": pending_auth.get("services", DEFAULT_SERVICES), + }, + indent=2, + ) + ) + else: + print(f"OK: Authenticated. Token saved to {TOKEN_PATH}") def revoke(): @@ -260,16 +452,19 @@ def revoke(): print("No token to revoke.") return + scopes = _stored_token_scopes() or list(ALL_SCOPES) + _ensure_deps() from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request try: - creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES) + creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), scopes) if creds.expired and creds.refresh_token: creds.refresh(Request()) import urllib.request + urllib.request.urlopen( urllib.request.Request( f"https://oauth2.googleapis.com/revoke?token={creds.token}", @@ -283,11 +478,23 @@ def revoke(): TOKEN_PATH.unlink(missing_ok=True) PENDING_AUTH_PATH.unlink(missing_ok=True) + LAST_AUTH_URL_PATH.unlink(missing_ok=True) print(f"Deleted {TOKEN_PATH}") def main(): parser = argparse.ArgumentParser(description="Google Workspace OAuth setup for Hermes") + parser.add_argument( + "--services", + default="all", + help="Comma-separated services to authorize: all, email, calendar, drive, contacts, sheets, docs", + ) + parser.add_argument( + "--format", + choices=["plain", "json"], + default="plain", + help="Output format. Use json for agent-friendly parsing.", + ) group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--check", action="store_true", help="Check if auth is valid (exit 0=yes, 1=no)") group.add_argument("--client-secret", metavar="PATH", help="Store OAuth client_secret.json") @@ -298,16 +505,20 @@ def main(): args = parser.parse_args() if args.check: - sys.exit(0 if check_auth() else 1) - elif args.client_secret: + sys.exit(0 if check_auth(args.services) else 1) + if args.client_secret: store_client_secret(args.client_secret) - elif args.auth_url: - get_auth_url() - elif args.auth_code: - exchange_auth_code(args.auth_code) - elif args.revoke: + return + if args.auth_url: + get_auth_url(args.services, output_format=args.format) + return + if args.auth_code: + exchange_auth_code(args.auth_code, output_format=args.format) + return + if args.revoke: revoke() - elif args.install_deps: + return + if args.install_deps: sys.exit(0 if install_deps() else 1) diff --git a/tests/skills/test_google_oauth_setup.py b/tests/skills/test_google_oauth_setup.py index 361bb7e28c..43498714c4 100644 --- a/tests/skills/test_google_oauth_setup.py +++ b/tests/skills/test_google_oauth_setup.py @@ -4,6 +4,8 @@ These tests cover the headless/manual auth-code flow where the browser step and code exchange happen in separate process invocations. """ +from __future__ import annotations + import importlib.util import json import sys @@ -110,6 +112,7 @@ def setup_module(monkeypatch, tmp_path): monkeypatch.setattr(module, "CLIENT_SECRET_PATH", tmp_path / "google_client_secret.json") monkeypatch.setattr(module, "TOKEN_PATH", tmp_path / "google_token.json") monkeypatch.setattr(module, "PENDING_AUTH_PATH", tmp_path / "google_oauth_pending.json", raising=False) + monkeypatch.setattr(module, "LAST_AUTH_URL_PATH", tmp_path / "google_oauth_last_url.txt", raising=False) client_secret = { "installed": { @@ -123,16 +126,38 @@ def setup_module(monkeypatch, tmp_path): return module -class TestGetAuthUrl: - def test_persists_state_and_code_verifier_for_later_exchange(self, setup_module, capsys): - setup_module.get_auth_url() +class TestResolveServices: + def test_reduces_to_requested_services(self, setup_module): + services, scopes = setup_module._resolve_services("email,calendar") + assert services == ["email", "calendar"] + assert scopes == [ + "https://www.googleapis.com/auth/gmail.readonly", + "https://www.googleapis.com/auth/gmail.send", + "https://www.googleapis.com/auth/gmail.modify", + "https://www.googleapis.com/auth/calendar", + ] - out = capsys.readouterr().out.strip() - assert out == "https://auth.example/authorize?state=generated-state" + +class TestGetAuthUrl: + def test_persists_state_verifier_scopes_and_last_url(self, setup_module, capsys): + setup_module.get_auth_url("email,calendar", output_format="json") + + out = json.loads(capsys.readouterr().out) + assert out["success"] is True + assert out["auth_url"] == "https://auth.example/authorize?state=generated-state" + assert out["services"] == ["email", "calendar"] + assert Path(out["auth_url_file"]).read_text() == out["auth_url"] saved = json.loads(setup_module.PENDING_AUTH_PATH.read_text()) assert saved["state"] == "generated-state" assert saved["code_verifier"] == "generated-code-verifier" + assert saved["services"] == ["email", "calendar"] + assert saved["scopes"] == [ + "https://www.googleapis.com/auth/gmail.readonly", + "https://www.googleapis.com/auth/gmail.send", + "https://www.googleapis.com/auth/gmail.modify", + "https://www.googleapis.com/auth/calendar", + ] flow = FakeFlow.created[-1] assert flow.autogenerate_code_verifier is True @@ -142,7 +167,14 @@ class TestGetAuthUrl: class TestExchangeAuthCode: def test_reuses_saved_pkce_material_for_plain_code(self, setup_module): setup_module.PENDING_AUTH_PATH.write_text( - json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"}) + json.dumps( + { + "state": "saved-state", + "code_verifier": "saved-verifier", + "services": ["email", "calendar"], + "scopes": ["scope-a", "scope-b"], + } + ) ) setup_module.exchange_auth_code("4/test-auth-code") @@ -150,13 +182,21 @@ class TestExchangeAuthCode: flow = FakeFlow.created[-1] assert flow.state == "saved-state" assert flow.code_verifier == "saved-verifier" + assert flow.scopes == ["scope-a", "scope-b"] assert flow.fetch_token_calls == [{"code": "4/test-auth-code"}] assert json.loads(setup_module.TOKEN_PATH.read_text())["token"] == "access-token" assert not setup_module.PENDING_AUTH_PATH.exists() def test_extracts_code_from_redirect_url_and_checks_state(self, setup_module): setup_module.PENDING_AUTH_PATH.write_text( - json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"}) + json.dumps( + { + "state": "saved-state", + "code_verifier": "saved-verifier", + "services": ["email"], + "scopes": ["scope-a"], + } + ) ) setup_module.exchange_auth_code( @@ -166,19 +206,34 @@ class TestExchangeAuthCode: flow = FakeFlow.created[-1] assert flow.fetch_token_calls == [{"code": "4/extracted-code"}] - def test_rejects_state_mismatch(self, setup_module, capsys): + def test_state_mismatch_regenerates_fresh_url(self, setup_module, capsys): setup_module.PENDING_AUTH_PATH.write_text( - json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"}) + json.dumps( + { + "state": "saved-state", + "code_verifier": "saved-verifier", + "services": ["email", "calendar"], + "scopes": ["scope-a", "scope-b"], + } + ) ) + FakeFlow.default_state = "replacement-state" + FakeFlow.default_verifier = "replacement-verifier" with pytest.raises(SystemExit): setup_module.exchange_auth_code( - "http://localhost:1/?code=4/extracted-code&state=wrong-state" + "http://localhost:1/?code=4/extracted-code&state=wrong-state", + output_format="json", ) - out = capsys.readouterr().out - assert "state mismatch" in out.lower() - assert not setup_module.TOKEN_PATH.exists() + out = json.loads(capsys.readouterr().out) + assert out["success"] is False + assert out["fresh_auth_url"] == "https://auth.example/authorize?state=replacement-state" + + saved = json.loads(setup_module.PENDING_AUTH_PATH.read_text()) + assert saved["state"] == "replacement-state" + assert saved["code_verifier"] == "replacement-verifier" + assert saved["services"] == ["email", "calendar"] def test_requires_pending_auth_session(self, setup_module, capsys): with pytest.raises(SystemExit): @@ -188,16 +243,42 @@ class TestExchangeAuthCode: assert "run --auth-url first" in out.lower() assert not setup_module.TOKEN_PATH.exists() - def test_keeps_pending_auth_session_when_exchange_fails(self, setup_module, capsys): + def test_failed_exchange_regenerates_fresh_url(self, setup_module, capsys): setup_module.PENDING_AUTH_PATH.write_text( - json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"}) + json.dumps( + { + "state": "saved-state", + "code_verifier": "saved-verifier", + "services": ["email"], + "scopes": ["scope-a"], + } + ) ) + FakeFlow.default_state = "replacement-state" + FakeFlow.default_verifier = "replacement-verifier" FakeFlow.fetch_error = Exception("invalid_grant: Missing code verifier") with pytest.raises(SystemExit): - setup_module.exchange_auth_code("4/test-auth-code") + setup_module.exchange_auth_code("4/test-auth-code", output_format="json") - out = capsys.readouterr().out - assert "token exchange failed" in out.lower() + out = json.loads(capsys.readouterr().out) + assert out["success"] is False + assert out["fresh_auth_url"] == "https://auth.example/authorize?state=replacement-state" assert setup_module.PENDING_AUTH_PATH.exists() assert not setup_module.TOKEN_PATH.exists() + + def test_check_auth_rejects_missing_requested_scopes(self, setup_module, capsys): + setup_module.TOKEN_PATH.write_text( + json.dumps( + { + "token": "access-token", + "refresh_token": "refresh-token", + "scopes": ["https://www.googleapis.com/auth/gmail.readonly"], + } + ) + ) + + ok = setup_module.check_auth("email,calendar") + out = capsys.readouterr().out + assert ok is False + assert "missing scopes" in out.lower()