OAuth login: refuse emails the provider marks unverified

enlace_auth/auth/oauth.py

@@ -12,16 +12,20 @@
 
 from __future__ import annotations
 
+import logging
 import os
 import time
 from typing import Any, Callable, Optional
 
 from fastapi import APIRouter, HTTPException, Request, Response
+from fastapi.responses import JSONResponse
 
 from enlace_auth.auth.cookies import sign_cookie
 from enlace_auth.auth.sessions import SessionStore
 from enlace_auth.config import OAuthProviderConfig
 
+_logger = logging.getLogger("enlace_auth.oauth")
+
 _PROVIDER_PRESETS = {
     "google": {
         "server_metadata_url": "https://accounts.google.com/.well-known/openid-configuration",
@@ -46,6 +50,32 @@ def _import_authlib():
     return OAuth
 
 
+#: Presets whose userinfo exposes only verified emails, without a claim saying so.
+_EMAIL_VERIFIED_BY_PRESET = frozenset({"github"})
+
+
+def _verified(flag) -> bool:
+    """True only when an ``email_verified`` claim affirms the address.
+
+    >>> _verified(True), _verified("true"), _verified(False), _verified(None)
+    (True, True, False, False)
+    >>> _verified(1)
+    False
+    """
+    return flag is True or (isinstance(flag, str) and flag.strip().lower() == "true")
+
+
+def _email_trusted(provider: str, cfg, claims: dict) -> bool:
+    """Whether an email returned by *provider* may identify an account here."""
+    if _verified(claims.get("email_verified")):
+        return True
+    if claims.get("email_verified") is not None:
+        return False  # the provider said "not verified" (in whatever form)
+    return provider in _EMAIL_VERIFIED_BY_PRESET or bool(
+        getattr(cfg, "trust_unverified_email", False)
+    )
+
+
 def _build_oauth_registry(providers: dict[str, OAuthProviderConfig]):
     OAuth = _import_authlib()
     oauth = OAuth()
@@ -133,23 +163,38 @@ async def callback(provider: str, request: Request):
         try:
             token = await client.authorize_access_token(request)
         except Exception as e:
-            raise HTTPException(status_code=401, detail=f"OAuth failed: {e}") from e
+            # The error text can echo attacker-supplied callback parameters.
+            _logger.warning("OAuth callback failed for %s: %s", provider, e)
+            raise HTTPException(status_code=401, detail="OAuth sign-in failed") from e
 
         email = None
+        claims: dict = {}
         userinfo = token.get("userinfo") if isinstance(token, dict) else None
         if userinfo and isinstance(userinfo, dict):
+            claims = userinfo
             email = userinfo.get("email")
 
         if not email and hasattr(client, "userinfo"):
             try:
                 info = await client.userinfo(token=token)
                 if isinstance(info, dict):
+                    claims = info
                     email = info.get("email")
             except Exception:
                 pass
 
         if not email:
             raise HTTPException(status_code=401, detail="No email from OAuth provider")
+        # Accounts are keyed by email, so an address the provider has not
+        # verified would let someone sign in as whoever owns it here --
+        # including an existing password account. Require the provider to
+        # affirm it (OIDC ``email_verified``); providers that never send the
+        # claim pass only if known to verify (github preset) or opted in.
+        if not _email_trusted(provider, providers.get(provider), claims):
+            raise HTTPException(
+                status_code=401,
+                detail="The OAuth provider has not verified this email address",
+            )
 
         email = email.lower()
         if email not in user_store:
@@ -167,10 +212,7 @@ async def callback(provider: str, request: Request):
                 "oauth_provider": provider,
             }
         session_id = session_store.create(user_id=email, email=email)
-        resp = Response(
-            content=f'{{"ok":true,"email":"{email}"}}',
-            media_type="application/json",
-        )
+        resp = JSONResponse({"ok": True, "email": email})
         _set_session_cookie(resp, session_id)
         return resp
 

enlace_auth/config.py

@@ -38,6 +38,12 @@ class OAuthProviderConfig(BaseModel):
     token_url: Optional[str] = None
     userinfo_url: Optional[str] = None
     server_metadata_url: Optional[str] = None
+    # Accounts are matched by email, so by default a provider must affirm the
+    # address (``email_verified`` true). Set this only for a provider that
+    # verifies emails but does not send the claim; the ``github`` preset is
+    # trusted without it (GitHub exposes only verified emails on /user).
+    # Never set it for Microsoft Entra ID: its ``email`` is tenant-editable.
+    trust_unverified_email: bool = False
 
 
 class OAuthServerConfig(BaseModel):

tests/test_oauth.py

@@ -3,6 +3,7 @@
 import os
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from starlette.responses import RedirectResponse
@@ -14,7 +15,7 @@
 SIGNING_KEY = "oauth-signing-key-32bytes-minlen"
 
 
-def _make_app(providers, user_store, session_store):
+def _make_app(providers, user_store, session_store, *, userinfo=None):
     """Build a FastAPI app with the OAuth router and env vars set."""
     os.environ["G_ID"] = "fake-client-id"
     os.environ["G_SECRET"] = "fake-client-secret"
@@ -26,7 +27,11 @@ def _make_app(providers, user_store, session_store):
         return_value=RedirectResponse("http://example.com/fake-authorize")
     )
     fake_client.authorize_access_token = AsyncMock(
-        return_value={"userinfo": {"email": "alice@example.com"}}
+        return_value={
+            # Google's ID token always carries email_verified.
+            "userinfo": userinfo
+            or {"email": "alice@example.com", "email_verified": True}
+        }
     )
     fake_registry = MagicMock()
     fake_registry.google = fake_client
@@ -98,3 +103,75 @@ def test_no_providers_returns_none():
         )
         is None
     )
+
+
+def test_unverified_provider_email_is_refused():
+    """An unverified address must not sign in as (or create) that account."""
+    providers = {
+        "google": OAuthProviderConfig(
+            client_id_env="G_ID", client_secret_env="G_SECRET"
+        )
+    }
+    users = {"alice@example.com": {"password_hash": "x", "created_at": 0}}
+    sessions = SessionStore({})
+    app, _ = _make_app(
+        providers,
+        users,
+        sessions,
+        userinfo={"email": "alice@example.com", "email_verified": False},
+    )
+    r = TestClient(app).get("/auth/callback/google")
+    assert r.status_code == 401
+    assert "enlace_session=" not in r.headers.get("set-cookie", "")
+    assert sessions.list_all() == []
+
+
+@pytest.mark.parametrize(
+    "provider_cfg, claims, ok",
+    [
+        ({}, {"email": "alice@example.com"}, False),  # claim omitted, not trusted
+        ({"trust_unverified_email": True}, {"email": "alice@example.com"}, True),
+        ({}, {"email": "alice@example.com", "email_verified": "true"}, True),
+        ({}, {"email": "alice@example.com", "email_verified": "false"}, False),
+        (
+            {"trust_unverified_email": True},
+            {"email": "a@x.io", "email_verified": False},
+            False,
+        ),
+    ],
+)
+def test_email_must_be_affirmed_unless_trusted(provider_cfg, claims, ok):
+    providers = {
+        "google": OAuthProviderConfig(
+            client_id_env="G_ID", client_secret_env="G_SECRET", **provider_cfg
+        )
+    }
+    app, _ = _make_app(providers, {}, SessionStore({}), userinfo=claims)
+    r = TestClient(app).get("/auth/callback/google")
+    assert (r.status_code == 200) is ok
+
+
+def test_github_preset_is_trusted_without_the_claim():
+    from enlace_auth.auth.oauth import _email_trusted
+
+    assert _email_trusted("github", None, {"email": "a@x.io"})
+    assert not _email_trusted(
+        "github", None, {"email": "a@x.io", "email_verified": False}
+    )
+    assert not _email_trusted("azure", None, {"email": "a@x.io"})
+
+
+def test_verified_provider_email_signs_in():
+    providers = {
+        "google": OAuthProviderConfig(
+            client_id_env="G_ID", client_secret_env="G_SECRET"
+        )
+    }
+    users: dict = {}
+    app, _ = _make_app(
+        providers,
+        users,
+        SessionStore({}),
+        userinfo={"email": "alice@example.com", "email_verified": True},
+    )
+    assert TestClient(app).get("/auth/callback/google").status_code == 200