feat: ponte SSO entre Keycloak e a sessao do LearnHouse

Sidecar que roda com o codigo do proprio LearnHouse. Em /sso/login confirma a
identidade perguntando ao oauth2-proxy (nunca confia em cabecalho do cliente),
acha ou cria o usuario pelo e-mail verificado e emite LH_access, LH_refresh e
LH_session, que o frontend exige para considerar o usuario logado. /sso/logout
limpa os cookies e encerra a sessao no Keycloak.

Redirecionamento pos-login restrito a caminhos internos (padrao: home).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

sso_bridge/bridge.py

@@ -0,0 +1,128 @@
+"""Ponte Keycloak -> sessão do LearnHouse.
+
+Roda no mesmo image do LearnHouse (usa o código e o segredo dele). Quem chega em
+/sso/login passou pelo Keycloak; a ponte confirma isso perguntando ao oauth2-proxy
+(nunca confia em cabeçalho vindo do cliente), acha ou cria o usuário pelo e-mail
+verificado e emite os mesmos cookies que o /auth/login emite.
+"""
+import logging
+import os
+import random
+from urllib.parse import quote
+
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import PlainTextResponse, RedirectResponse
+from sqlmodel import func, select
+
+from sso_bridge.helpers import (
+    identity_from_proxy_response,
+    keycloak_logout_url,
+    safe_redirect,
+    username_from_email,
+)
+
+log = logging.getLogger("sso_bridge")
+
+PROXY_AUTH_URL = os.environ.get("SSO_PROXY_AUTH_URL", "http://oauth2-proxy:4180/oauth2/auth")
+ORG_SLUG = os.environ.get("SSO_ORG_SLUG", "overia")
+KEYCLOAK_ISSUER = os.environ.get("SSO_KEYCLOAK_ISSUER", "https://login.overia.com.br/realms/overia")
+KEYCLOAK_CLIENT_ID = os.environ.get("SSO_KEYCLOAK_CLIENT_ID", "cursos")
+SITE_URL = os.environ.get("SSO_SITE_URL", "https://cursos.overia.com.br/")
+
+# Logins antigos podem ter gravado LH_access/LH_refresh com Domain=; o cookie novo é host-only e
+# conviveria com o velho (o servidor leria o inválido). Expira os com domínio antes.
+STALE_COOKIE_DOMAINS = (".overia.com.br", "overia.com.br", "cursos.overia.com.br")
+SESSION_COOKIES = ("LH_access", "LH_refresh", "LH_session")
+SESSION_FLAG_MAX_AGE = 30 * 24 * 3600
+
+
+def _expire_stale_domain_cookies(response) -> None:
+    for domain in STALE_COOKIE_DOMAINS:
+        for name in SESSION_COOKIES:
+            response.delete_cookie(key=name, domain=domain, path="/")
+
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+
+async def _verified_email(request: Request) -> str | None:
+    cookie = request.headers.get("cookie")
+    if not cookie:
+        return None
+    try:
+        async with httpx.AsyncClient(timeout=5) as client:
+            resp = await client.get(PROXY_AUTH_URL, headers={"Cookie": cookie})
+    except httpx.HTTPError:
+        log.exception("oauth2-proxy indisponível")
+        return None
+    return identity_from_proxy_response(resp.status_code, resp.headers)
+
+
+def _restart_sso(rd: str) -> RedirectResponse:
+    return RedirectResponse(f"/oauth2/start?rd={quote('/sso/login?rd=' + rd, safe='')}", status_code=302)
+
+
+@app.get("/sso/health")
+async def health():
+    return PlainTextResponse("ok")
+
+
+@app.get("/sso/login")
+async def sso_login(request: Request, rd: str | None = None):
+    from src.core.events.database import _async_session_factory
+    from src.db.organizations import Organization
+    from src.db.users import AnonymousUser, User, UserCreate
+    from src.routers.auth import is_request_secure, set_auth_cookies
+    from src.security.session_context import AUTH_METHOD_SSO
+    from src.services.auth.session import mint_session_tokens
+    from src.services.users.users import create_user
+
+    target = safe_redirect(rd)
+    email = await _verified_email(request)
+    if email is None:
+        return _restart_sso(target)
+
+    async with _async_session_factory() as db:
+        org = (await db.execute(select(Organization).where(Organization.slug == ORG_SLUG))).scalars().first()
+        if org is None:
+            log.error("organização %s não encontrada", ORG_SLUG)
+            return PlainTextResponse("organização indisponível", status_code=503)
+
+        user = (await db.execute(select(User).where(func.lower(User.email) == email))).scalars().first()
+        if user is None:
+            user_object = UserCreate(
+                email=email,
+                username=username_from_email(email, random.randint(100000, 999999)),
+                password="",
+                first_name="",
+                last_name="",
+            )
+            user = await create_user(
+                request, db, AnonymousUser(), user_object, org.id, is_oauth=True, signup_provider="sso"
+            )
+            log.info("usuário provisionado via SSO: %s", email)
+
+        tokens = mint_session_tokens(user.email, AUTH_METHOD_SSO, org.id)
+
+    response = RedirectResponse(target, status_code=302)
+    _expire_stale_domain_cookies(response)
+    set_auth_cookies(response, tokens.access_token, tokens.refresh_token, request)
+    # O frontend decide "logado" por este cookie legível (o /api/auth/token-exchange do Next o grava no login normal).
+    response.set_cookie(
+        key="LH_session", value="1", httponly=False, secure=is_request_secure(request),
+        samesite="lax", path="/", max_age=SESSION_FLAG_MAX_AGE,
+    )
+    return response
+
+
+@app.get("/sso/logout")
+async def sso_logout(request: Request):
+    from src.routers.auth import unset_auth_cookies
+
+    keycloak = keycloak_logout_url(KEYCLOAK_ISSUER, KEYCLOAK_CLIENT_ID, SITE_URL)
+    response = RedirectResponse(f"/oauth2/sign_out?rd={quote(keycloak, safe='')}", status_code=302)
+    unset_auth_cookies(response, request)
+    response.delete_cookie(key="LH_session", path="/")
+    _expire_stale_domain_cookies(response)
+    return response

sso_bridge/helpers.py

@@ -0,0 +1,45 @@
+import re
+from urllib.parse import quote, urlencode
+
+DEFAULT_REDIRECT = "/"
+
+_ALLOWED_PREFIXES = ("/dash", "/courses", "/course/")
+_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+
+
+def safe_redirect(rd: str | None) -> str:
+    if not rd or not rd.startswith("/") or rd.startswith("//"):
+        return DEFAULT_REDIRECT
+    if any(c in rd for c in ("\\", "\r", "\n", "\t", " ")):
+        return DEFAULT_REDIRECT
+    path = rd.split("?", 1)[0]
+    if path == "/" or any(path == p or path.startswith(p.rstrip("/") + "/") for p in _ALLOWED_PREFIXES):
+        return rd
+    return DEFAULT_REDIRECT
+
+
+def normalize_email(raw: str | None) -> str | None:
+    if not raw:
+        return None
+    email = raw.strip().lower()
+    if not _EMAIL_RE.match(email):
+        return None
+    return email
+
+
+def identity_from_proxy_response(status: int, headers) -> str | None:
+    if status != 202:
+        return None
+    lowered = {k.lower(): v for k, v in headers.items()}
+    return normalize_email(lowered.get("x-auth-request-email"))
+
+
+def username_from_email(email: str, suffix: int) -> str:
+    local = email.split("@", 1)[0]
+    cleaned = re.sub(r"[^a-z0-9]", "", local.lower())[:30] or "user"
+    return f"{cleaned}{suffix}"
+
+
+def keycloak_logout_url(issuer: str, client_id: str, return_url: str) -> str:
+    query = urlencode({"client_id": client_id, "post_logout_redirect_uri": return_url}, quote_via=quote)
+    return f"{issuer.rstrip('/')}/protocol/openid-connect/logout?{query}"

tests/test_sso_bridge_helpers.py

@@ -0,0 +1,123 @@
+import pytest
+
+from sso_bridge.helpers import (
+    DEFAULT_REDIRECT,
+    identity_from_proxy_response,
+    normalize_email,
+    safe_redirect,
+    username_from_email,
+)
+
+
+@pytest.mark.parametrize(
+    "rd",
+    ["/dash", "/dash/courses", "/", "/courses", "/course/abc-123", "/dash?x=1"],
+)
+def test_safe_redirect_accepts_internal_paths(rd):
+    assert safe_redirect(rd) == rd
+
+
+@pytest.mark.parametrize(
+    "rd",
+    [
+        None,
+        "",
+        "dash",
+        "//evil.com",
+        "///evil.com",
+        "http://evil.com",
+        "https://evil.com/dash",
+        "/\\evil.com",
+        "/dash\r\nSet-Cookie: a=b",
+        "/dash\nLocation: http://evil.com",
+        "javascript:alert(1)",
+        "/login",
+        "/signup",
+        "/sso/login",
+        "/oauth2/start",
+    ],
+)
+def test_safe_redirect_falls_back_to_default(rd):
+    assert safe_redirect(rd) == DEFAULT_REDIRECT
+
+
+def test_safe_redirect_does_not_allow_authority_via_path():
+    assert safe_redirect("/@evil.com") == DEFAULT_REDIRECT
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        ("Ana@Example.com", "ana@example.com"),
+        ("  ana@example.com  ", "ana@example.com"),
+        ("OVERLEAF@overleaf.com.br", "overleaf@overleaf.com.br"),
+    ],
+)
+def test_normalize_email_lowercases_and_strips(raw, expected):
+    assert normalize_email(raw) == expected
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [None, "", "   ", "sem-arroba", "a@", "@b.com", "a b@c.com", "a@b.com\nbcc: x@y.com", "a@@b.com"],
+)
+def test_normalize_email_rejects_invalid(raw):
+    assert normalize_email(raw) is None
+
+
+def test_identity_requires_202():
+    assert identity_from_proxy_response(202, {"X-Auth-Request-Email": "a@b.com"}) == "a@b.com"
+    assert identity_from_proxy_response(401, {"X-Auth-Request-Email": "a@b.com"}) is None
+    assert identity_from_proxy_response(200, {"X-Auth-Request-Email": "a@b.com"}) is None
+    assert identity_from_proxy_response(500, {}) is None
+
+
+def test_identity_header_lookup_is_case_insensitive():
+    assert identity_from_proxy_response(202, {"x-auth-request-email": "A@B.com"}) == "a@b.com"
+
+
+def test_identity_without_or_with_bad_email_is_none():
+    assert identity_from_proxy_response(202, {}) is None
+    assert identity_from_proxy_response(202, {"X-Auth-Request-Email": "nao-e-email"}) is None
+
+
+def test_username_from_email_uses_sanitized_local_part_and_suffix():
+    name = username_from_email("Ana.Silva+curso@example.com", suffix=123456)
+    assert name == "anasilvacurso123456"
+
+
+def test_username_from_email_falls_back_when_local_part_empty_after_sanitizing():
+    assert username_from_email("+++@example.com", suffix=111111) == "user111111"
+
+
+def test_username_from_email_is_bounded():
+    name = username_from_email("a" * 200 + "@example.com", suffix=222222)
+    assert len(name) <= 40
+    assert name.endswith("222222")
+
+
+def test_keycloak_logout_url_sends_client_and_encoded_return():
+    from sso_bridge.helpers import keycloak_logout_url
+
+    url = keycloak_logout_url(
+        "https://login.overia.com.br/realms/overia", "cursos", "https://cursos.overia.com.br/"
+    )
+    assert url == (
+        "https://login.overia.com.br/realms/overia/protocol/openid-connect/logout"
+        "?client_id=cursos&post_logout_redirect_uri=https%3A%2F%2Fcursos.overia.com.br%2F"
+    )
+
+
+def test_keycloak_logout_url_tolerates_trailing_slash_on_issuer():
+    from sso_bridge.helpers import keycloak_logout_url
+
+    url = keycloak_logout_url("https://k/realms/r/", "c", "https://x/")
+    assert url.startswith("https://k/realms/r/protocol/openid-connect/logout?")
+
+
+def test_default_redirect_is_home_not_the_admin_dashboard():
+    from sso_bridge.helpers import DEFAULT_REDIRECT
+
+    assert DEFAULT_REDIRECT == "/"
+    assert safe_redirect(None) == "/"
+    assert safe_redirect("//evil.com") == "/"