"""oauth_connect.py — the generic OAuth CONNECTOR framework (wave 22, contract C5 + A2 / R12). WHAT THIS IS, and what it is deliberately not. This connects a data scope to a PERSON — "read *my* Gmail" — which is a different animal from `core/auth_google.py` (LOGIN — proving who you are). The two share one Google Cloud app (one client id, per R12) and nothing else: login stays the untouched template it has been since wave 7, and this module never mints a session. ⭐ REGISTRY-DRIVEN (C5 amendment A2 — the SOURCES-registry law applied to auth). A provider is declared ONCE in `PROVIDERS` — endpoints, scopes, env names, the authorize extras — and the generic flow below reads the declaration. Google is the only entry THIS wave; Slack/Airtable/ Outlook/QuickBooks later are one entry + secrets each, zero new code paths. ⛔ Nango is the STUDY reference for this shape and is Elastic-licensed: no line of `providers.yaml` is copied — every entry here sources from the provider's own public documentation. THE FOUR RULES, each load-bearing: 1. **PER-USER SLOTS (R7, the QM seam).** A refresh token is identity, not infrastructure — it lands in `core.keychain`'s `user_secrets` sub-bucket keyed by USERNAME under the slot `oauth_`, Fernet at rest, invisible to the admin keychain table. An automation polls through its CREATOR's slot, nobody else's. 2. **FAIL-CLOSED, never an error page.** No client id ⇒ the provider is not `configured` ⇒ the email trigger reads `ready: false` and the connect door answers a sentence — the flow is simply not offered. Same in every failure direction: an expired/revoked refresh token becomes a LOUD `reconnect: true` on the status (Google's Testing-mode refresh tokens die every ~7 days — silence here would read as a trigger that just stopped). 3. **PKCE + state, single-use, short-lived, user-bound.** The authorize URL carries S256(code_verifier) and an unguessable `state`; the callback consumes the state exactly once, checks it was minted for THIS user, and refuses anything older than ten minutes. The state also carries the RETURN PATH (A3: the user lands back where they left, never at the root) — relative paths only, so the callback can never be walked off-origin. 4. **Tokens travel in headers/bodies to the provider's REGISTERED endpoints only** — never a URL derived from input — and nothing here logs one: errors quote a STATUS and a sentence, the `bd_call` discipline. """ from __future__ import annotations import base64 import datetime as _dt import hashlib import json import os import re import secrets import threading import time from urllib.parse import urlencode import requests #: ⭐ THE PROVIDER REGISTRY (C5-A2). Everything the generic flow needs to speak one provider, #: declared as data. `authorize_extra` is the per-provider consent dialect — Google's #: `access_type=offline&prompt=consent` is what makes a REFRESH token arrive on every connect; #: without it a re-connect answers access-only and the poll dies within the hour, which reads #: exactly like a broken trigger. PROVIDERS = { "google": { "label": "Google", "authorize_url": "https://accounts.google.com/o/oauth2/v2/auth", "token_url": "https://oauth2.googleapis.com/token", "userinfo_url": "https://openidconnect.googleapis.com/v1/userinfo", "scopes": "openid email https://www.googleapis.com/auth/gmail.readonly", "authorize_extra": {"access_type": "offline", "prompt": "consent"}, "env_id": "GOOGLE_OAUTH_CLIENT_ID", "env_secret": "GOOGLE_OAUTH_CLIENT_SECRET", # ⛔⛔ WAVE 32 · OWNER ITEM 12 / RULING R6 — NOT OFFERED. See `offered()` below. "offered": False, }, } def offered(provider): """Does the PRODUCT offer this provider today? (W32-T14 / owner item 12 / R6.) ⛔⛔ THE TOMBSTONE, PUT WHERE THE NEXT READER WILL LOOK — *"why did Gmail disappear?"* The owner reported the Google and Gmail connectors as broken: the card offered a Connect button, and the button answered `503 oauth_unavailable` in raw JSON. R6: *"Google and Gmail are HIDDEN until they are real. Remove the rows from the connectors directory rather than shipping a door that returns `oauth_unavailable`. No spend, no verification."* ⚠ AND IT IS NOT A BUG TO GO AND FIX. The blocker is **D-45**: Google will not grant the `gmail.readonly` scope to a production app without CASA verification, which is a paid annual security assessment (measured at roughly **$540–1,800/yr**) plus a privacy policy on an owned domain. Until somebody decides to spend that, `GOOGLE_OAUTH_CLIENT_ID` is unset on every deployment and every honest state this flow can reach is "you cannot use this". ⛔ THE REGISTRY ENTRY STAYS, and that is deliberate: `status()` still walks it, so `routes_automation`'s email-trigger readiness keeps its answer instead of turning a provider-shaped question into a KeyError. What changes is that the DIRECTORY does not list it and the start/callback/disconnect doors 404 — the flow is not offered rather than offered and broken. Flip this one key to `True` the day the verification is paid for. Defaults to True: a provider added to the registry is offered unless it says otherwise, so this flag can never silently hide the NEXT connector somebody wires up. """ return bool((provider_def(provider) or {}).get("offered", True)) GMAIL_BASE = "https://gmail.googleapis.com/gmail/v1/users/me" STATE_TTL_SECONDS = 600 HTTP_TIMEOUT = 12.0 def _kc(): import core.keychain as keychain return keychain def provider_def(provider): return PROVIDERS.get(str(provider or "").strip().lower()) def slot_name(provider): return f"oauth_{str(provider or '').strip().lower()}" def client(provider): """`(client_id, client_secret)` from the env names the registry declares.""" p = provider_def(provider) or {} return ((os.environ.get(p.get("env_id", "")) or "").strip(), (os.environ.get(p.get("env_secret", "")) or "").strip()) def configured(provider="google"): cid, secret = client(provider) return bool(cid and secret) def safe_next(raw): """A RETURN PATH the callback may redirect to: same-origin RELATIVE only. Anything absolute, protocol-relative or unparseable degrades to the shell root — a redirect target that arrived in a query string is exactly the thing an open-redirect rides in on.""" nxt = str(raw or "").strip() if not nxt or not nxt.startswith(("/", "#")) or nxt.startswith("//") or "\\" in nxt \ or ":" in nxt.split("?", 1)[0].split("#", 1)[0]: return "/#/" return nxt[:300] # --------------------------------------------------------------------------------------------- # PKCE + state # --------------------------------------------------------------------------------------------- _STATE_LOCK = threading.Lock() _STATES = {} # state -> {"provider", "verifier", "uname", "redirect", "next", "ts"} def _prune_states(now=None): now = now if now is not None else time.time() with _STATE_LOCK: for k in [k for k, v in _STATES.items() if now - v.get("ts", 0) > STATE_TTL_SECONDS]: _STATES.pop(k, None) def _challenge(verifier): """S256: BASE64URL(SHA256(verifier)), no padding — RFC 7636's one worthwhile method.""" digest = hashlib.sha256(verifier.encode("ascii")).digest() return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") def start(provider, uname, redirect_uri, next_path=""): """Mint state+verifier for THIS user and return the provider's consent URL. Returns `(url, error)` — an unknown or unconfigured provider is a sentence, never an exception (rule 2).""" p = provider_def(provider) if p is None: return "", f"{provider!r} is not a connectable provider" if not configured(provider): return "", (f"{p['label']} OAuth is not configured on this deployment — the owner adds " f"{p['env_id']} / {p['env_secret']} first") _prune_states() cid, _secret = client(provider) state = secrets.token_urlsafe(24) verifier = secrets.token_urlsafe(64) # 86 chars — inside RFC 7636's 43..128 with _STATE_LOCK: _STATES[state] = {"provider": str(provider), "verifier": verifier, "uname": str(uname or ""), "redirect": str(redirect_uri or ""), "next": safe_next(next_path), "ts": time.time()} q = { "client_id": cid, "redirect_uri": redirect_uri, "response_type": "code", "scope": p["scopes"], "state": state, "code_challenge": _challenge(verifier), "code_challenge_method": "S256", } q.update(p.get("authorize_extra") or {}) return f"{p['authorize_url']}?{urlencode(q)}", "" def _post_token(url, body): """One POST to a REGISTERED token endpoint. `(payload, error_sentence)` — never raises, never echoes a secret; a non-200 quotes the status and the provider's `error` field only.""" try: r = requests.post(url, data=body, timeout=HTTP_TIMEOUT, headers={"Accept": "application/json"}) except Exception as e: # noqa: BLE001 return None, f"the token endpoint did not answer ({type(e).__name__})" try: payload = r.json() if r.content else {} except Exception: # noqa: BLE001 payload = {} if r.status_code != 200: why = str((payload or {}).get("error") or r.status_code) return None, f"the token exchange was refused ({why})" return payload or {}, "" def _userinfo_email(provider, access_token): p = provider_def(provider) or {} url = p.get("userinfo_url") if not url: return "" try: r = requests.get(url, timeout=HTTP_TIMEOUT, headers={"Authorization": f"Bearer {access_token}"}) if r.status_code == 200: return str((r.json() or {}).get("email") or "") except Exception: # noqa: BLE001 pass return "" def callback(rt, uname, state, code): """The exchange half: verify the state belongs to THIS user, trade the code for tokens, store `{access, refresh, expiry, email}` in the user's per-provider slot. Returns `(email, next_path, error)` — `next_path` is where the browser goes home to (A3).""" _prune_states() with _STATE_LOCK: entry = _STATES.pop(str(state or ""), None) # single-use, whatever happens next home = safe_next((entry or {}).get("next")) if not entry: return "", home, "that connect attempt is unknown or expired — start again" if entry.get("uname") != str(uname or ""): # A state minted for one session consumed by another is exactly the CSRF the state # exists to refuse. Fail closed; the honest user just clicks Connect again. return "", home, "that connect attempt belongs to a different session — start again" provider = entry.get("provider") or "google" p = provider_def(provider) if p is None or not configured(provider): return "", home, f"{provider!r} is not connectable on this deployment" cid, secret = client(provider) payload, err = _post_token(p["token_url"], { "client_id": cid, "client_secret": secret, "grant_type": "authorization_code", "code": str(code or ""), "redirect_uri": entry.get("redirect") or "", "code_verifier": entry.get("verifier") or "", }) if err: return "", home, err access = str(payload.get("access_token") or "") refresh = str(payload.get("refresh_token") or "") if not access: return "", home, "the provider answered without an access token — nothing was stored" if not refresh: # Google answers access-only when consent was silently reused — an access-only slot # dies within the hour, silently, so it is refused with the way out named. return "", home, ("no refresh token came back — disconnect the app on the provider's " "permissions page and connect again") expiry = time.time() + float(payload.get("expires_in") or 3600) - 60 email = _userinfo_email(provider, access) _kc().put_user_secret(rt, uname, slot_name(provider), { "access": access, "refresh": refresh, "expiry": f"{expiry:.0f}", "email": email, }) return email, home, "" def _refresh(rt, uname, provider, slot): """Trade the refresh token for a fresh access token and re-store. `(slot|None, error)`. A refused refresh marks the slot `reconnect` LOUDLY rather than deleting it — the status endpoint then says "reconnect" instead of "not connected", which are different instructions to a person (the Testing-mode 7-day expiry makes this path ROUTINE, not rare).""" p = provider_def(provider) or {} cid, secret = client(provider) payload, err = _post_token(p.get("token_url", ""), { "client_id": cid, "client_secret": secret, "grant_type": "refresh_token", "refresh_token": str(slot.get("refresh") or ""), }) if err: marked = dict(slot) marked["reconnect"] = "1" try: _kc().put_user_secret(rt, uname, slot_name(provider), marked) except Exception: # noqa: BLE001 pass return None, err fresh = dict(slot) fresh["access"] = str(payload.get("access_token") or slot.get("access") or "") fresh["expiry"] = f"{time.time() + float(payload.get('expires_in') or 3600) - 60:.0f}" fresh.pop("reconnect", None) _kc().put_user_secret(rt, uname, slot_name(provider), fresh) return fresh, "" def creds(rt, uname, provider="google"): """THE seam a poller reads through: a live access token for this user, refreshed transparently on expiry. Returns `(access_token, error_sentence)` — `("", why)` covers not-connected, locked keychain and reconnect-needed alike, each with its own sentence.""" kc = _kc() p = provider_def(provider) or {"label": str(provider)} try: slot = kc.read_user_secret(rt, uname, slot_name(provider)) except kc.KeychainLocked as e: return "", f"the keychain is locked ({e}) — no token can be read" if not slot: return "", (f"no {p['label']} connection for this user — connect it in Settings") if slot.get("reconnect"): return "", f"the {p['label']} connection expired — reconnect it in Settings" try: expiry = float(slot.get("expiry") or 0) except (TypeError, ValueError): expiry = 0.0 if time.time() >= expiry: slot, err = _refresh(rt, uname, provider, slot) if err: return "", f"the {p['label']} connection could not refresh — {err}" return str(slot.get("access") or ""), "" def google_creds(rt, uname): """The email trigger's named seam (contract C5's original spelling) — `creds` on the one provider this wave ships.""" return creds(rt, uname, "google") def status(rt, uname): """C5's status shape, registry-driven: `{: {connected, email, reconnect, configured}}` for the session user — one entry per registry row, so today it reads exactly `{google: {...}}`. Never raises — a locked keychain reads as not-connected WITH the note.""" kc = _kc() out = {} for slug in PROVIDERS: row = {"connected": False, "email": "", "reconnect": False, "configured": configured(slug)} try: slot = kc.read_user_secret(rt, uname, slot_name(slug)) except kc.KeychainLocked as e: row["note"] = str(e) out[slug] = row continue if slot: row["connected"] = True row["email"] = str(slot.get("email") or "") row["reconnect"] = bool(slot.get("reconnect")) out[slug] = row return out def disconnect(rt, uname, provider="google"): _kc().drop_user_secret(rt, uname, slot_name(provider)) return True # --------------------------------------------------------------------------------------------- # GMAIL READS — the one data surface this wave (the email trigger's substrate) # --------------------------------------------------------------------------------------------- def gmail_list(access_token, query, max_results=25): """Message ids matching `query`, newest first. `(ids, error)` — bounded, timeboxed, and a non-200 is a sentence (the poll runs on the tick thread; it may be slow ONCE, never hung).""" q = urlencode({"q": str(query or "").strip() or "in:inbox", "maxResults": int(max_results)}) try: r = requests.get(f"{GMAIL_BASE}/messages?{q}", timeout=HTTP_TIMEOUT, headers={"Authorization": f"Bearer {access_token}"}) except Exception as e: # noqa: BLE001 return [], f"Gmail did not answer ({type(e).__name__})" if r.status_code != 200: return [], f"Gmail answered {r.status_code}" try: msgs = (r.json() or {}).get("messages") or [] except Exception: # noqa: BLE001 return [], "Gmail answered with something that is not JSON" return [str(m.get("id")) for m in msgs if isinstance(m, dict) and m.get("id")], "" def gmail_message(access_token, message_id): """One message's headers + snippet → our row shape. `(row|None, error)`.""" try: r = requests.get( f"{GMAIL_BASE}/messages/{message_id}" f"?format=metadata&metadataHeaders=From&metadataHeaders=Subject" f"&metadataHeaders=Date", timeout=HTTP_TIMEOUT, headers={"Authorization": f"Bearer {access_token}"}) except Exception as e: # noqa: BLE001 return None, f"Gmail did not answer ({type(e).__name__})" if r.status_code != 200: return None, f"Gmail answered {r.status_code}" try: payload = r.json() or {} except Exception: # noqa: BLE001 return None, "Gmail answered with something that is not JSON" headers = {str(h.get("name") or "").lower(): str(h.get("value") or "") for h in ((payload.get("payload") or {}).get("headers") or []) if isinstance(h, dict)} stamp = _dt.datetime.now().astimezone().isoformat(timespec="seconds") return { "email_id": str(payload.get("id") or message_id), "email_from": headers.get("from", "")[:200], "email_subject": headers.get("subject", "")[:300], "email_date": headers.get("date", "")[:80], "email_snippet": str(payload.get("snippet") or "")[:500], "email_seen_at": stamp, }, "" #: `re` is imported for the slot-name law in core.keychain (see USER_SECRET_PROVIDERS there); #: keeping the reference here so a linter cannot 'clean up' what the registry contract uses. _SLOT_SHAPE = re.compile(r"^oauth_[a-z0-9_]{1,24}$") # ═════════════════════════════════════════════════════════════════════════════════════════════ # ⭐ W31-T41 — THE NANGO PROBE. Inert on import; reached only by `python oauth_connect.py --nango`. # ═════════════════════════════════════════════════════════════════════════════════════════════ # # R5 makes Nango a production dependency ("Meta connects by clicking Connect Facebook; Nango owns # that flow"), and T41 is its GO/NO-GO: does the key authenticate, does the Facebook integration # carry Marketing API scopes (`ads_read`), and what does the free tier allow against our connection # count. ⛔ It lives HERE because `::creds` is the seam every poller reads and therefore the seam # Nango slots into — the probe belongs beside the thing it would replace, not in a scratch file # that dies with the session. # # ⚠ NOTHING BELOW IS WIRED INTO THE FLOW. `PROVIDERS`, `start`, `callback` and `creds` are # untouched by this block: a measurement that changes the live auth path before its own verdict is # in is the shape R5 was careful to avoid. NANGO_API = os.environ.get("NANGO_API_BASE") or "https://api.nango.dev" def nango_key(): """`NANGO_SECRET_KEY` from the environment or gitignored `platform/.env`; "" when absent.""" from pathlib import Path as _Path tok = os.environ.get("NANGO_SECRET_KEY") or "" if tok: return tok.strip() env = _Path(__file__).resolve().parents[2] / "platform" / ".env" if env.exists(): for line in env.read_text(encoding="utf-8", errors="replace").splitlines(): if line.strip().startswith("NANGO_SECRET_KEY"): _, _, v = line.partition("=") return v.strip().strip('"').strip("'") return "" def nango_probe(): """`{authenticated, integrations, connections, errors}` — read-only GETs against Nango Cloud. ⛔ NEVER PRINTS THE KEY. It rides an Authorization header and the report carries integration ids and scope strings only. ⚠ The scope question is the GO/NO-GO: an integration row's own config states which scopes it requests, and `ads_read` is what R2's Insights reads need. If it is absent, the fallback is the keychain token path — which `connectors_meta.py` has already built — and that has to be said BEFORE any OAuth code is written, per the ticket. """ import base64 import json as _json import urllib.error import urllib.request key = nango_key() out = {"base": NANGO_API, "authenticated": False, "integrations": [], "connections": [], "errors": []} if not key: out["errors"].append( "NANGO_SECRET_KEY is not set in the environment or in platform/.env. Nothing was " "measured. The PRD records it as 'recorded in platform/.env' and it is not there.") return out # ⛔ BEARER, NOT BASIC — measured 2026-08-12 by A, against the owner's real key, once it landed. # Basic (`base64(key + ":")`, the OLD Nango style) answers **401 `invalid_secret_key_format`: # "The provided secret key is not a UUID v4."** — which reads exactly like a bad credential and # is not: Nango takes the whole base64 blob as the key and correctly observes it is not a UUID. # The SAME key over `Bearer` answers **200 with 1 integration**. ⭐ The lesson is the error # message: it described the VALUE when the fault was in the ENVELOPE, so the obvious next step # ("ask the owner for a valid key") would have been wrong and would have read as their mistake. auth = f"Bearer {key}" def _get(path): req = urllib.request.Request(f"{NANGO_API}{path}", headers={"Authorization": auth, "User-Agent": "aios-nango-probe/1"}) with urllib.request.urlopen(req, timeout=45) as r: return _json.loads(r.read().decode("utf-8", "replace")) for label, path, sink in (("integrations", "/integrations", "integrations"), ("connections", "/connection", "connections")): try: body = _get(path) rows = body if isinstance(body, list) else (body.get("configs") or body.get("connections") or body.get("data") or []) out[sink] = rows out["authenticated"] = True except urllib.error.HTTPError as e: out["errors"].append(f"{label}: HTTP {e.code} — " f"{e.read().decode('utf-8', 'replace')[:200]}") except Exception as e: # noqa: BLE001 out["errors"].append(f"{label}: {type(e).__name__}: {e}") return out if __name__ == "__main__": # pragma: no cover import json as _j import sys as _sys if "--nango" in _sys.argv: rep = nango_probe() print(f"Nango key present: {'yes' if nango_key() else 'NO'} (never printed)") print(f"authenticated: {rep['authenticated']} · integrations: " f"{len(rep['integrations'])} · connections: {len(rep['connections'])}") for e in rep["errors"]: print(f" ERROR {e}") for i in rep["integrations"]: if isinstance(i, dict): print(" integration:", _j.dumps({k: v for k, v in i.items() if k in ("unique_key", "provider", "scopes", "oauth_scopes")})) _sys.exit(0 if rep["authenticated"] else 2) print("usage: python oauth_connect.py --nango") _sys.exit(2)