api/oauth_connect.py

"""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_<provider>`, 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: `{<provider>: {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)