Интеграция с провайдерами, Google, GitHub OAuth
OAuth2 позволяет пользователям входить через Google, GitHub, Facebook и других провайдеров. В этой теме вы научитесь интегрировать социальную аутентификацию в FastAPI.
OAuth 2.0 — стандарт авторизации, позволяющий приложениям получать ограниченный доступ к аккаунтам пользователей через провайдеров.
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Client │ │ Server │ │ Provider │
│ (ваш API) │◄────►│ (ваш API) │◄────►│ (Google) │
└─────────────┘ └─────────────┘ └─────────────┘
1. Пользователь нажимает "Login with Google"
↓
2. Перенаправление на Google OAuth
↓
3. Пользователь подтверждает доступ
↓
4. Google возвращает authorization code
↓
5. Сервер обменивает code на access token
↓
6. Сервер получает данные пользователя
↓
7. Сервер создаёт сессию/JWT для пользователя
pip install authlib httpxauthlib — библиотека для OAuth (поддерживает множество провайдеров)httpx — HTTP-клиент для запросов к провайдерамhttp://localhost:8000/auth/google/callbackhttp://localhost:8000http://localhost:8000/auth/github/callbackfrom fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.responses import RedirectResponse
from authlib.integrations.starlette_client import OAuth
from starlette.config import Config
from starlette.middleware.sessions import SessionMiddleware
import jwt
from jwt.exceptions import InvalidTokenError
from datetime import datetime, timedelta, timezone
import secrets
app = FastAPI()
# === Конфигурация ===
SECRET_KEY = secrets.token_urlsafe(32)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# OAuth провайдеры
oauth = OAuth()
# Google
oauth.register(
name='google',
client_id='YOUR_GOOGLE_CLIENT_ID',
client_secret='YOUR_GOOGLE_CLIENT_SECRET',
server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
client_kwargs={
'scope': 'openid email profile'
}
)
# GitHub
oauth.register(
name='github',
client_id='YOUR_GITHUB_CLIENT_ID',
client_secret='YOUR_GITHUB_CLIENT_SECRET',
access_token_url='https://github.com/login/oauth/access_token',
access_token_params=None,
authorize_url='https://github.com/login/oauth/authorize',
api_base_url='https://api.github.com/',
client_kwargs={
'scope': 'user:email'
}
)
# Middleware для сессий (нужен для state параметра)
app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY)
# === Вспомогательные функции ===
def create_access_token(data: dict, expires_delta: timedelta):
expire = datetime.now(timezone.utc) + expires_delta
to_encode = data.model_copy()
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_or_create_user(provider_user: dict, provider: str):
"""
Находит или создаёт пользователя в БД.
В реальности — SQLAlchemy запросы.
"""
# Имитация БД
users_db = {}
email = provider_user.get('email')
if email in users_db:
return users_db[email]
# Создаём нового пользователя
user = {
'id': len(users_db) + 1,
'email': email,
'name': provider_user.get('name', ''),
'provider': provider,
'provider_id': provider_user.get('id', ''),
'created_at': datetime.now(timezone.utc)
}
users_db[email] = user
return user
# === Endpoints ===
@app.get('/auth/login')
async def auth_login(request: Request):
"""
Начальная точка входа.
Перенаправляет на провайдера OAuth.
"""
redirect_uri = request.url_for('auth_callback', provider='google')
return await oauth.google.authorize_redirect(request, redirect_uri)
@app.get('/auth/{provider}/callback')
async def auth_callback(request: Request, provider: str):
"""
Callback от OAuth провайдера.
Получает токен и данные пользователя.
"""
try:
# Обмениваем code на токен
token = await getattr(oauth, provider).authorize_access_token(request)
# Получаем данные пользователя
user_info = token.get('userinfo')
if not user_info:
# Для GitHub нужно сделать отдельный запрос
if provider == 'github':
resp = await getattr(oauth, provider).get('user', token=token)
user_info = resp.json()
# Находим или создаём пользователя
user = get_or_create_user(user_info, provider)
# Создаём JWT токен
access_token = create_access_token(
data={"sub": user['email'], "provider": provider},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
# Перенаправляем на фронтенд с токеном
# В реальности лучше использовать cookie или отдельный endpoint
return {
"access_token": access_token,
"token_type": "bearer",
"user": user
}
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Authentication failed: {str(e)}"
)
@app.get('/auth/me')
async def get_current_user(token: str = Depends(oauth2_scheme)):
"""
Защищённый endpoint для получения текущего пользователя.
"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise HTTPException(status_code=401, detail="Invalid token")
# Получаем пользователя из БД
# user = get_user_from_db(email)
return {"email": email, "token_valid": True}
except InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")<!DOCTYPE html>
<html>
<head>
<title>Login</title>
</head>
<body>
<h1>Login</h1>
<!-- Google -->
<a href="/auth/login">
<img src="/google-icon.png" alt="Google">
Login with Google
</a>
<!-- GitHub -->
<a href="/auth/github/login">
<img src="/github-icon.png" alt="GitHub">
Login with GitHub
</a>
<script>
// Обработка ответа от сервера
window.addEventListener('message', (event) => {
if (event.data.type === 'auth-success') {
localStorage.setItem('token', event.data.token);
window.location.href = '/dashboard';
}
});
</script>
</body>
</html>from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.responses import RedirectResponse, JSONResponse
from authlib.integrations.starlette_client import OAuth
from starlette.middleware.sessions import SessionMiddleware
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
from datetime import datetime, timedelta, timezone
import jwt
from jwt.exceptions import InvalidTokenError
import secrets
import httpx
from database import get_db
from models import User as DBUser
app = FastAPI()
# === Конфигурация ===
SECRET_KEY = secrets.token_urlsafe(32)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY)
oauth = OAuth()
# Google
oauth.register(
name='google',
client_id='GOOGLE_CLIENT_ID',
client_secret='GOOGLE_CLIENT_SECRET',
server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
client_kwargs={'scope': 'openid email profile'}
)
# GitHub
oauth.register(
name='github',
client_id='GITHUB_CLIENT_ID',
client_secret='GITHUB_CLIENT_SECRET',
access_token_url='https://github.com/login/oauth/access_token',
authorize_url='https://github.com/login/oauth/authorize',
api_base_url='https://api.github.com/',
client_kwargs={'scope': 'user:email'}
)
# Facebook
oauth.register(
name='facebook',
client_id='FACEBOOK_CLIENT_ID',
client_secret='FACEBOOK_CLIENT_SECRET',
access_token_url='https://graph.facebook.com/oauth/access_token',
authorize_url='https://www.facebook.com/dialog/oauth',
api_base_url='https://graph.facebook.com/',
client_kwargs={'scope': 'email'}
)
# === Модели ===
class TokenResponse(BaseModel):
access_token: str
token_type: str
user: dict
# === Вспомогательные функции ===
def create_access_token(data: dict, expires_delta: timedelta):
expire = datetime.now(timezone.utc) + expires_delta
to_encode = data.model_copy()
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_or_create_user(db: Session, email: str, name: str, provider: str, provider_id: str):
user = db.query(DBUser).filter(DBUser.email == email).first()
if user:
return user
user = DBUser(
email=email,
name=name,
provider=provider,
provider_id=provider_id,
is_active=True
)
db.add(user)
db.commit()
db.refresh(user)
return user
# === Endpoints ===
@app.get('/auth/{provider}/login')
async def oauth_login(request: Request, provider: str):
"""Начало OAuth flow"""
if provider not in ['google', 'github', 'facebook']:
raise HTTPException(status_code=400, detail="Unsupported provider")
redirect_uri = request.url_for('oauth_callback', provider=provider)
return await getattr(oauth, provider).authorize_redirect(request, redirect_uri)
@app.get('/auth/{provider}/callback')
async def oauth_callback(request: Request, provider: str, db: Session = Depends(get_db)):
"""Callback от OAuth провайдера"""
try:
token = await getattr(oauth, provider).authorize_access_token(request)
# Получаем данные пользователя
if provider == 'google':
user_info = token.get('userinfo')
email = user_info.get('email')
name = user_info.get('name', '')
provider_id = user_info.get('sub', '')
elif provider == 'github':
async with httpx.AsyncClient() as client:
resp = await client.get(
'https://api.github.com/user',
headers={'Authorization': f'token {token["access_token"]}'}
)
user_info = resp.json()
# Получаем email
email_resp = await client.get(
'https://api.github.com/user/emails',
headers={'Authorization': f'token {token["access_token"]}'}
)
emails = email_resp.json()
email = next((e['email'] for e in emails if e['primary']), None)
name = user_info.get('name', user_info.get('login', ''))
provider_id = str(user_info.get('id', ''))
elif provider == 'facebook':
async with httpx.AsyncClient() as client:
resp = await client.get(
'https://graph.facebook.com/me',
params={
'fields': 'id,name,email',
'access_token': token['access_token']
}
)
user_info = resp.json()
email = user_info.get('email')
name = user_info.get('name', '')
provider_id = user_info.get('id', '')
if not email:
raise HTTPException(status_code=400, detail="Email not provided by provider")
# Создаём или находим пользователя
user = get_or_create_user(db, email, name, provider, provider_id)
# Создаём JWT
access_token = create_access_token(
data={"sub": email, "provider": provider},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {
"access_token": access_token,
"token_type": "bearer",
"user": {
"id": user.id,
"email": user.email,
"name": user.name,
"provider": user.provider
}
}
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Authentication failed: {str(e)}"
)
@app.get('/auth/me')
async def get_current_user(
authorization: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Получить текущего пользователя"""
token = authorization.replace('Bearer ', '')
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise HTTPException(status_code=401, detail="Invalid token")
user = db.query(DBUser).filter(DBUser.email == email).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {
"id": user.id,
"email": user.email,
"name": user.name,
"provider": user.provider
}
except InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")State защищает от CSRF-атак:
@app.get('/auth/{provider}/login')
async def oauth_login(request: Request, provider: str):
# Authlib автоматически генерирует и проверяет state
redirect_uri = request.url_for('oauth_callback', provider=provider)
return await getattr(oauth, provider).authorize_redirect(
request,
redirect_uri
)Запрашивайте минимальные необходимые права:
# Только email
client_kwargs={'scope': 'email'}
# Email и профиль
client_kwargs={'scope': 'openid email profile'}
# GitHub: только публичные репозитории
client_kwargs={'scope': 'public_repo'}Проблема: URI в настройках провайдера не совпадает с кодом.
Решение: Убедитесь, что redirect_uri в коде точно совпадает с настройками провайдера (включая http/https, порты).
client_secret = 'hardcoded_secret' # Плохо!Решение: Используйте переменные окружения:
import os
client_secret = os.environ['GITHUB_CLIENT_SECRET']token = await oauth.google.authorize_access_token(request)
# Если пользователь отменил — ошибка!Решение:
try:
token = await oauth.google.authorize_access_token(request)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Auth failed: {e}")Лаборатория
«OAuth2PasswordBearer, tokenUrl, проверка scope»
доступна в публичном GitLab. Она проверяет OAuth2PasswordBearer с
tokenUrl, интеграцию с JWT, проверку scope через Security(scopes=...)
и обработку истёкших и невалидных токенов.
Склонируйте репозиторий и запустите тесты из его корня:
git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest oauth2/testsДалее: Загрузка файлов