Files
sales-trainer/backend/app/services/oauth.py
Macky d6e7cffc84 feat(auth): Google + Facebook OAuth login/register
Public social signup into OAUTH_DEFAULT_ORG (role user, seat-checked);
email-match links existing active user instead of duplicating. Server-side
provider token validation via stdlib urllib only (no new dep): Google
tokeninfo (aud + email_verified) and Facebook app/debug-token/me (is_valid,
app_id, me.id==user_id). Fail-closed when creds unconfigured, rate-limited
per-IP + per-email, /oauth/config leaks no secrets. Frontend: login buttons
(only enabled providers), GSI + FB SDK on-demand, monochrome glyphs, TH/EN.

Login page shows social buttons only when backend reports provider enabled.

348 backend tests pass (337 + 11 new OAuth), frontend build + 4/4 unit
clean, manual security review PASS. Not pushed (push auto-deploys).
2026-08-20 10:57:04 +07:00

114 lines
4.6 KiB
Python

"""Server-side OAuth provider token validation (stdlib only — no new deps).
Every function raises :class:`OAuthError` (fail closed) on any network,
transport, or verification failure. The backend NEVER trusts a client-declared
identity: the provider token is validated over HTTPS and the derived email /
subject are the only source of identity. Provider tokens are single-use inputs
and are never stored, logged, or returned.
"""
from __future__ import annotations
import json
import urllib.parse
import urllib.request
from ..config import Config
_GOOGLE_TOKENINFO = "https://oauth2.googleapis.com/tokeninfo"
_GRAPH_BASE = "https://graph.facebook.com"
_TIMEOUT = 8
class OAuthError(Exception):
"""Provider validation failed (network, transport, or verification)."""
def _https_get_json(url: str) -> dict:
"""GET an https URL and return parsed JSON, or raise OAuthError."""
try:
request = urllib.request.Request(
url, headers={"User-Agent": "SalesTrainer/1.0", "Accept": "application/json"}
)
with urllib.request.urlopen(request, timeout=_TIMEOUT) as response:
payload = json.loads(response.read().decode("utf-8"))
except Exception as exc: # noqa: BLE001 - any failure fails closed
raise OAuthError(f"provider unavailable ({type(exc).__name__})") from exc
if not isinstance(payload, dict):
raise OAuthError("provider response is invalid")
return payload
def _string(value: object, *, field: str) -> str:
if not isinstance(value, str) or not value:
raise OAuthError(f"provider validation failed (missing {field})")
return value
def validate_google_token(token: str) -> tuple[str, str, str]:
"""Return ``(email, sub, name)`` for a verified Google ID token.
Verifies via Google's tokeninfo endpoint that the token's audience matches
our configured client id and that the account email is verified.
"""
if not Config.oauth_provider_enabled("google"):
raise OAuthError("provider disabled")
params = urllib.parse.urlencode({"id_token": token})
data = _https_get_json(f"{_GOOGLE_TOKENINFO}?{params}")
if data.get("aud") != Config.OAUTH_GOOGLE_CLIENT_ID:
raise OAuthError("provider validation failed (audience mismatch)")
if data.get("email_verified") != "true":
raise OAuthError("provider validation failed (email not verified)")
email = _string(data.get("email"), field="email")
sub = _string(data.get("sub"), field="sub")
name = data.get("name") or ""
if not isinstance(name, str):
name = ""
return email, sub, name
def validate_facebook_token(token: str) -> tuple[str, str, str]:
"""Return ``(email, id, name)`` for a verified Facebook access token.
Exchanges the app credentials for an app token, fetches the user identity,
then verifies the submitted user token against the app token (is_valid,
app_id match, and user_id matches the identity we fetched).
"""
if not Config.oauth_provider_enabled("facebook"):
raise OAuthError("provider disabled")
# 1. App token from the app credentials.
app_params = urllib.parse.urlencode(
{
"client_id": Config.OAUTH_FACEBOOK_APP_ID,
"client_secret": Config.OAUTH_FACEBOOK_APP_SECRET,
"grant_type": "client_credentials",
}
)
app_data = _https_get_json(f"{_GRAPH_BASE}/oauth/access_token?{app_params}")
app_token = _string(app_data.get("access_token"), field="access_token")
# 2. Identity for the submitted user token.
me_params = urllib.parse.urlencode(
{"fields": "id,email,name", "access_token": token}
)
me_data = _https_get_json(f"{_GRAPH_BASE}/me?{me_params}")
# 3. Verify the submitted token is valid and belongs to our app.
debug_params = urllib.parse.urlencode(
{"input_token": token, "access_token": app_token}
)
debug_data = _https_get_json(f"{_GRAPH_BASE}/debug_token?{debug_params}")
token_data = debug_data.get("data")
if not isinstance(token_data, dict) or token_data.get("is_valid") is not True:
raise OAuthError("provider validation failed (invalid token)")
if token_data.get("app_id") != Config.OAUTH_FACEBOOK_APP_ID:
raise OAuthError("provider validation failed (app mismatch)")
user_id = _string(me_data.get("id"), field="id")
if user_id != token_data.get("user_id"):
raise OAuthError("provider validation failed (user mismatch)")
email = _string(me_data.get("email"), field="email")
name = me_data.get("name") or ""
if not isinstance(name, str):
name = ""
return email, user_id, name