412 lines
14 KiB
Python
412 lines
14 KiB
Python
"""OAuth + Telegram auth endpoints.
|
|
|
|
Phase 3B of AUDIT-2026-Q3.md.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
|
|
from app.core.redis import get_redis
|
|
from app.domains.auth.jwt import _create_jwt, _derive_user_id
|
|
from app.domains.auth.schemas import (
|
|
GoogleAuthResponse,
|
|
TelegramAuthRequest,
|
|
WalletAuthResponse,
|
|
)
|
|
from app.domains.auth.store import _get_user, _get_user_by_email
|
|
|
|
router = APIRouter(tags=["auth"])
|
|
|
|
FRONTEND_URL = os.getenv("FRONTEND_URL", "https://rugmunch.io")
|
|
|
|
|
|
@router.get("/google/url", response_model=GoogleAuthResponse)
|
|
async def google_auth_url():
|
|
"""Get Google OAuth URL."""
|
|
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
|
redirect_uri = os.getenv(
|
|
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
|
)
|
|
|
|
if not client_id:
|
|
return {"url": "/auth/callback?error=google_not_configured"}
|
|
|
|
from urllib.parse import urlencode
|
|
|
|
params = {
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"response_type": "code",
|
|
"scope": "openid email profile",
|
|
"access_type": "offline",
|
|
"prompt": "consent",
|
|
}
|
|
url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(params)}"
|
|
return {"url": url}
|
|
|
|
|
|
@router.get("/google/callback")
|
|
async def google_callback(request: Request):
|
|
"""Handle Google OAuth redirect - exchanges code, creates user, redirects to frontend with token."""
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
code = request.query_params.get("code")
|
|
error = request.query_params.get("error")
|
|
|
|
if error:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error={error}")
|
|
|
|
if not code:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=missing_code")
|
|
|
|
import httpx
|
|
|
|
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
|
client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
|
|
redirect_uri = os.getenv(
|
|
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
|
)
|
|
|
|
if not client_id or not client_secret:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
"https://oauth2.googleapis.com/token",
|
|
data={
|
|
"code": code,
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uri": redirect_uri,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=token_exchange_failed")
|
|
|
|
tokens = resp.json()
|
|
access_token = tokens.get("access_token")
|
|
|
|
resp = await client.get(
|
|
"https://www.googleapis.com/oauth2/v2/userinfo",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
if resp.status_code != 200:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=userinfo_failed")
|
|
|
|
google_user = resp.json()
|
|
email = google_user.get("email")
|
|
|
|
user = _get_user_by_email(email)
|
|
if not user:
|
|
user_id = _derive_user_id(email)
|
|
user = {
|
|
"id": user_id,
|
|
"email": email,
|
|
"display_name": google_user.get("name", email),
|
|
"tier": "FREE",
|
|
"role": "USER",
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"xp": 0,
|
|
"level": 1,
|
|
"badges": [],
|
|
"scans_remaining": 5,
|
|
"scans_used": 0,
|
|
}
|
|
r = get_redis()
|
|
r.hset("rmi:users", user_id, json.dumps(user))
|
|
r.hset("rmi:users:email", email.lower(), user_id)
|
|
|
|
jwt_token = _create_jwt(
|
|
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
|
)
|
|
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=google")
|
|
|
|
|
|
@router.post("/google/callback", response_model=WalletAuthResponse)
|
|
async def google_callback_post(request: Request):
|
|
"""Legacy POST endpoint for manual code exchange."""
|
|
body = await request.json()
|
|
code = body.get("code")
|
|
|
|
if not code:
|
|
raise HTTPException(status_code=400, detail="Missing authorization code")
|
|
|
|
import httpx
|
|
|
|
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
|
client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
|
|
redirect_uri = os.getenv(
|
|
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
|
)
|
|
|
|
if not client_id or not client_secret:
|
|
raise HTTPException(status_code=500, detail="Google OAuth not configured")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
"https://oauth2.googleapis.com/token",
|
|
data={
|
|
"code": code,
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uri": redirect_uri,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise HTTPException(status_code=400, detail="Failed to exchange code")
|
|
|
|
tokens = resp.json()
|
|
access_token = tokens.get("access_token")
|
|
|
|
resp = await client.get(
|
|
"https://www.googleapis.com/oauth2/v2/userinfo",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise HTTPException(status_code=400, detail="Failed to get user info")
|
|
|
|
google_user = resp.json()
|
|
email = google_user.get("email")
|
|
|
|
user = _get_user_by_email(email)
|
|
if not user:
|
|
user_id = _derive_user_id(email)
|
|
user = {
|
|
"id": user_id,
|
|
"email": email,
|
|
"display_name": google_user.get("name", email),
|
|
"tier": "FREE",
|
|
"role": "USER",
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"xp": 0,
|
|
"level": 1,
|
|
"badges": [],
|
|
"scans_remaining": 5,
|
|
"scans_used": 0,
|
|
}
|
|
r = get_redis()
|
|
r.hset("rmi:users", user_id, json.dumps(user))
|
|
r.hset("rmi:users:email", email.lower(), user_id)
|
|
|
|
jwt_token = _create_jwt(
|
|
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
|
)
|
|
|
|
return {
|
|
"access_token": jwt_token,
|
|
"refresh_token": jwt_token,
|
|
"user": {
|
|
"id": user["id"],
|
|
"email": user["email"],
|
|
"display_name": user.get("display_name", email),
|
|
"wallet_address": None,
|
|
"wallet_chain": None,
|
|
"tier": user.get("tier", "FREE"),
|
|
"role": user.get("role", "USER"),
|
|
"created_at": user.get("created_at"),
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/telegram", response_model=WalletAuthResponse)
|
|
async def telegram_auth(req: TelegramAuthRequest):
|
|
"""Authenticate via Telegram Web App data."""
|
|
bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
|
if not bot_token:
|
|
raise HTTPException(status_code=500, detail="Telegram auth not configured")
|
|
|
|
data_check = {
|
|
"id": str(req.id),
|
|
"auth_date": str(req.auth_date),
|
|
}
|
|
if req.first_name:
|
|
data_check["first_name"] = req.first_name
|
|
if req.last_name:
|
|
data_check["last_name"] = req.last_name
|
|
if req.username:
|
|
data_check["username"] = req.username
|
|
if req.photo_url:
|
|
data_check["photo_url"] = req.photo_url
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
sorted_data = "\n".join(f"{k}={v}" for k, v in sorted(data_check.items()))
|
|
secret = hashlib.sha256(bot_token.encode()).digest()
|
|
expected_hash = hmac.new(secret, sorted_data.encode(), hashlib.sha256).hexdigest()
|
|
|
|
if not hmac.compare_digest(req.hash, expected_hash):
|
|
raise HTTPException(status_code=401, detail="Invalid Telegram auth data")
|
|
|
|
telegram_user_id = f"tg:{req.id}"
|
|
user = _get_user(telegram_user_id)
|
|
|
|
if not user:
|
|
display_name = (
|
|
f"{req.first_name or ''} {req.last_name or ''}".strip()
|
|
or req.username
|
|
or f"User{req.id}"
|
|
)
|
|
email = f"{req.id}@telegram.rmi"
|
|
|
|
user = {
|
|
"id": telegram_user_id,
|
|
"email": email,
|
|
"display_name": display_name,
|
|
"telegram_id": req.id,
|
|
"telegram_username": req.username,
|
|
"tier": "FREE",
|
|
"role": "USER",
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"xp": 0,
|
|
"level": 1,
|
|
"badges": [],
|
|
"scans_remaining": 5,
|
|
"scans_used": 0,
|
|
}
|
|
r = get_redis()
|
|
r.hset("rmi:users", telegram_user_id, json.dumps(user))
|
|
r.hset("rmi:users:telegram", str(req.id), telegram_user_id)
|
|
|
|
jwt_token = _create_jwt(
|
|
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
|
)
|
|
|
|
return {
|
|
"access_token": jwt_token,
|
|
"refresh_token": jwt_token,
|
|
"user": {
|
|
"id": user["id"],
|
|
"email": user["email"],
|
|
"display_name": user.get("display_name"),
|
|
"wallet_address": None,
|
|
"wallet_chain": None,
|
|
"tier": user.get("tier", "FREE"),
|
|
"role": user.get("role", "USER"),
|
|
"created_at": user.get("created_at"),
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/github/url", response_model=GoogleAuthResponse)
|
|
async def github_auth_url():
|
|
"""Get GitHub OAuth URL."""
|
|
client_id = os.getenv("GITHUB_CLIENT_ID")
|
|
redirect_uri = os.getenv(
|
|
"GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
|
|
)
|
|
|
|
if not client_id:
|
|
return {"url": "/auth/callback?error=github_not_configured"}
|
|
|
|
from urllib.parse import urlencode
|
|
|
|
params = {
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": "user:email",
|
|
}
|
|
url = f"https://github.com/login/oauth/authorize?{urlencode(params)}"
|
|
return {"url": url}
|
|
|
|
|
|
@router.get("/github/callback")
|
|
async def github_callback(request: Request):
|
|
"""Handle GitHub OAuth redirect - exchanges code, creates user, redirects to frontend with token."""
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
code = request.query_params.get("code")
|
|
error = request.query_params.get("error")
|
|
|
|
if error:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error={error}")
|
|
|
|
if not code:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=missing_code")
|
|
|
|
import httpx
|
|
|
|
client_id = os.getenv("GITHUB_CLIENT_ID")
|
|
client_secret = os.getenv("GITHUB_CLIENT_SECRET")
|
|
redirect_uri = os.getenv(
|
|
"GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
|
|
)
|
|
|
|
if not client_id or not client_secret:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
"https://github.com/login/oauth/access_token",
|
|
data={
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if resp.status_code != 200:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=token_exchange_failed")
|
|
|
|
tokens = resp.json()
|
|
access_token = tokens.get("access_token")
|
|
|
|
resp = await client.get(
|
|
"https://api.github.com/user",
|
|
headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
|
|
)
|
|
if resp.status_code != 200:
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=userinfo_failed")
|
|
|
|
github_user = resp.json()
|
|
email = github_user.get("email")
|
|
|
|
if not email:
|
|
resp = await client.get(
|
|
"https://api.github.com/user/emails",
|
|
headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
|
|
)
|
|
if resp.status_code == 200:
|
|
emails = resp.json()
|
|
primary = next((e for e in emails if e.get("primary")), None)
|
|
email = (
|
|
primary.get("email")
|
|
if primary
|
|
else (emails[0].get("email") if emails else None)
|
|
)
|
|
|
|
if not email:
|
|
email = f"{github_user.get('login', 'github_user')}@github.rmi"
|
|
|
|
user = _get_user_by_email(email)
|
|
if not user:
|
|
user_id = _derive_user_id(email)
|
|
user = {
|
|
"id": user_id,
|
|
"email": email,
|
|
"display_name": github_user.get("name", github_user.get("login", email)),
|
|
"tier": "FREE",
|
|
"role": "USER",
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"xp": 0,
|
|
"level": 1,
|
|
"badges": [],
|
|
"scans_remaining": 5,
|
|
"scans_used": 0,
|
|
}
|
|
r = get_redis()
|
|
r.hset("rmi:users", user_id, json.dumps(user))
|
|
r.hset("rmi:users:email", email.lower(), user_id)
|
|
|
|
jwt_token = _create_jwt(
|
|
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
|
)
|
|
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=github")
|