Аутентификация OAuth в Python

# Пример запроса с указанием областей действия
oauth = OAuth(app)
oauth.register(
    name='github',
    client_id='YOUR_GITHUB_CLIENT_ID',
    client_secret='YOUR_GITHUB_CLIENT_SECRET',
    authorize_url='https://github.com/login/oauth/authorize',
    access_token_url='https://github.com/login/oauth/access_token',
    client_kwargs={'scope': 'user:email repo'},  # Доступ к email и репозиториям
)

@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)  # Генерируем URI перенаправления
    return oauth.github.authorize_redirect(redirect_uri)

@app.route('/login')
def login():
    state = generate_random_state()  # Генерируем случайное значение
    session['oauth_state'] = state   # Сохраняем в сессии
    return oauth.github.authorize_redirect(redirect_uri, state=state)

Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

# Пример работы с ID-токеном в Python
import jwt
 
# Декодирование ID-токена
decoded_token = jwt.decode(id_token, key=public_key, algorithms=['RS256'])
user_info = {
    'user_id': decoded_token['sub'],
    'name': decoded_token['name'],
    'email': decoded_token['email']
}

# Создание JWT-токена в Python
import jwt
import datetime
 
payload = {
    'user_id': 123,
    'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=1)
}
token = jwt.encode(payload, 'secret_key', algorithm='HS256')

from flask import Flask, redirect, url_for, session, request
from authlib.integrations.flask_client import OAuth
import os
 
app = Flask(__name__)
app.secret_key = os.urandom(24)  # В продакшене используйте настоящий секретный ключ
 
# Инициализация OAuth
oauth = OAuth(app)
 
# Регистрация провайдера OAuth
oauth.register(
    name='github',
    client_id='YOUR_CLIENT_ID',  # Замените на ваш client_id
    client_secret='YOUR_CLIENT_SECRET',  # Замените на ваш client_secret
    authorize_url='https://github.com/login/oauth/authorize',
    authorize_params=None,
    access_token_url='https://github.com/login/oauth/access_token',
    access_token_params=None,
    refresh_token_url=None,
    client_kwargs={'scope': 'user:email'},
)
 
@app.route('/')
def home():
    user = session.get('user')
    if user:
        return f'Привет, {user["login"]}! <a href="/logout">Выйти</a>'
    return 'Добро пожаловать! <a href="/login">Войти через GitHub</a>'
 
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    return oauth.github.authorize_redirect(redirect_uri)
 
@app.route('/callback')
def authorize():
    token = oauth.github.authorize_access_token()
    resp = oauth.github.get('https://api.github.com/user', token=token)
    user_info = resp.json()
    session['user'] = user_info
    return redirect('/')
 
@app.route('/logout')
def logout():
    session.pop('user', None)
    return redirect('/')
 
if __name__ == '__main__':
    app.run(debug=True)

@app.route('/callback')
def authorize():
    try:
        token = oauth.github.authorize_access_token()
        resp = oauth.github.get('https://api.github.com/user', token=token)
        user_info = resp.json()
        
        # Проверка успешного ответа
        if resp.status_code != 200:
            raise Exception(f"Ошибка API: {user_info.get('message')}")
            
        session['user'] = user_info
        return redirect('/')
    except Exception as e:
        print(f"Ошибка авторизации: {str(e)}")
        return redirect('/login')

import secrets
 
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    state = secrets.token_hex(16)
    session['oauth_state'] = state
    return oauth.github.authorize_redirect(redirect_uri, state=state)
 
@app.route('/callback')
def authorize():
    # Проверка state для предотвращения CSRF
    expected_state = session.pop('oauth_state', None)
    state = request.args.get('state')
    
    if state != expected_state:
        return 'Возможная CSRF-атака. Авторизация отменена.', 403
        
    token = oauth.github.authorize_access_token()
    # Остальной код...

# Пример интеграции Authlib с Flask
from authlib.integrations.flask_client import OAuth
 
oauth = OAuth(app)
oauth.register(
    name='github',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_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'},
)

# Пример использования Requests-OAuthlib
from requests_oauthlib import OAuth2Session
 
client_id = 'YOUR_CLIENT_ID'
authorization_base_url = 'https://github.com/login/oauth/authorize'
token_url = 'https://github.com/login/oauth/access_token'
 
github = OAuth2Session(client_id, scope=['user:email'])
authorization_url, state = github.authorization_url(authorization_base_url)
 
# Перенаправляем пользователя на authorization_url
[H2]...[/H2]
 
# После перенаправления обратно:
token = github.fetch_token(
    token_url,
    client_secret='YOUR_CLIENT_SECRET',
    authorization_response=request.url
)
 
# Теперь можно делать запросы
user_data = github.get('https://api.github.com/user').json()

# Пример настройки Python-Social-Auth для Django
INSTALLED_APPS = [
    # ...
    'social_django',
    # ...
]
 
AUTHENTICATION_BACKENDS = [
    'social_core.backends.github.GithubOAuth2',
    'social_core.backends.google.GoogleOAuth2',
    'django.contrib.auth.backends.ModelBackend',
]
 
SOCIAL_AUTH_GITHUB_KEY = 'YOUR_CLIENT_ID'
SOCIAL_AUTH_GITHUB_SECRET = 'YOUR_CLIENT_SECRET'

# Пример использования Flask-Dance
from flask import Flask, redirect, url_for
from flask_dance.contrib.github import make_github_blueprint, github
 
app = Flask(__name__)
app.secret_key = "supersekrit"
blueprint = make_github_blueprint(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET",
)
app.register_blueprint(blueprint, url_prefix="/login")
 
@app.route("/")
def index():
    if not github.authorized:
        return redirect(url_for("github.login"))
    resp = github.get("/user")
    return f"Вы вошли как: {resp.json()['login']}"

# Пример настройки Django OAuth Toolkit
INSTALLED_APPS = [
    # ...
    'oauth2_provider',
    'corsheaders',
    # ...
]
 
MIDDLEWARE = [
    # ...
    'corsheaders.middleware.CorsMiddleware',
    # ...
]
 
# Настройка OAuth2
OAUTH2_PROVIDER = {
    'SCOPES': {'read': 'Read scope', 'write': 'Write scope'},
    'ACCESS_TOKEN_EXPIRE_SECONDS': 3600,
}

# Пример использования Oauthlib (довольно низкоуровневый)
from oauthlib.oauth2 import WebApplicationClient
 
client = WebApplicationClient("YOUR_CLIENT_ID")
authorization_url = client.prepare_request_uri(
    "https://github.com/login/oauth/authorize",
    redirect_uri="http://localhost:5000/callback",
    scope=["user:email"],
)
 
# После получения кода авторизации
token_url, headers, body = client.prepare_token_request(
    "https://github.com/login/oauth/access_token",
    authorization_response=request.url,
    redirect_uri="http://localhost:5000/callback",
    client_secret="YOUR_CLIENT_SECRET"
)

# Реализация Authorization Code Grant с использованием Authlib
from authlib.integrations.flask_client import OAuth
from flask import Flask, redirect, url_for, session
 
app = Flask(__name__)
oauth = OAuth(app)
 
github = oauth.register(
    'github',
    client_id='CLIENT_ID',
    client_secret='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'},
)
 
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    return github.authorize_redirect(redirect_uri)
 
@app.route('/callback')
def authorize():
    token = github.authorize_access_token()
    # Теперь у нас есть токен доступа
    return redirect('/')

# Реализация Implicit Grant (устаревший подход)
from flask import Flask, request
from authlib.integrations.flask_client import OAuth
 
app = Flask(__name__)
oauth = OAuth(app)
 
spa_client = oauth.register(
    'spa_client',
    client_id='CLIENT_ID',
    authorize_url='https://authorization-server.com/authorize',
    client_kwargs={
        'scope': 'profile email',
        'response_type': 'token',  # Ключевое отличие от Authorization Code
    },
)
 
@app.route('/login')
def login():
    redirect_uri = 'https://your-spa-app.com/callback'
    return spa_client.authorize_redirect(redirect_uri)

# Реализация Resource Owner Password Credentials Grant
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import LegacyApplicationClient
 
client_id = 'CLIENT_ID'
client_secret = 'CLIENT_SECRET'
token_url = 'https://authorization-server.com/token'
 
client = OAuth2Session(client=LegacyApplicationClient(client_id=client_id))
token = client.fetch_token(
    token_url=token_url,
    username='[email protected]',
    password='password123',
    client_id=client_id,
    client_secret=client_secret,
)
 
# Теперь можно использовать токен для доступа к API
user_info = client.get('https://api.example.com/user').json()

# Реализация Client Credentials Grant
from authlib.integrations.requests_client import OAuth2Session
 
client_id = 'CLIENT_ID'
client_secret = 'CLIENT_SECRET'
token_url = 'https://authorization-server.com/token'
scope = ['read', 'write']  # Области доступа для клиента
 
session = OAuth2Session(client_id=client_id, client_secret=client_secret, scope=scope)
token = session.fetch_token(token_url)
 
# Использование токена для доступа к API
response = session.get('https://api.example.com/resources')

# Реализация Refresh Token Grant
from authlib.integrations.requests_client import OAuth2Session
 
client_id = 'CLIENT_ID'
client_secret = 'CLIENT_SECRET'
refresh_token = 'PREVIOUS_REFRESH_TOKEN'
token_url = 'https://authorization-server.com/token'
 
session = OAuth2Session(
    client_id=client_id,
    client_secret=client_secret,
    token={'refresh_token': refresh_token}
)
 
# Запрос нового токена доступа с использованием refresh token
new_token = session.refresh_token(token_url)
 
# Сохраняем новый refresh_token для будущего использования
new_refresh_token = new_token.get('refresh_token')

import secrets
import base64
import hashlib
from authlib.integrations.requests_client import OAuth2Session
 
# Шаг 1: Генерация code_verifier
def generate_code_verifier(length=128):
    # Генерируем случайную строку байтов
    token = secrets.token_bytes(length)
    # Кодируем в base64url формат
    code_verifier = base64.urlsafe_b64encode(token).decode('utf-8')
    # Удаляем padding '=' символы
    code_verifier = code_verifier.replace('=', '')
    return code_verifier[:128]  # Ограничиваем длину до 128 символов
 
# Шаг 2: Создание code_challenge из code_verifier
def generate_code_challenge(code_verifier):
    # Хешируем code_verifier используя SHA-256
    sha256 = hashlib.sha256(code_verifier.encode('utf-8')).digest()
    # Кодируем хеш в base64url формат
    code_challenge = base64.urlsafe_b64encode(sha256).decode('utf-8')
    # Удаляем padding '=' символы
    code_challenge = code_challenge.replace('=', '')
    return code_challenge
 
# Инициализация OAuth-сессии с PKCE
def create_oauth_session_with_pkce(client_id, redirect_uri, scope):
    # Генерируем code_verifier и code_challenge
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    
    # Создаём OAuth-сессию
    session = OAuth2Session(
        client_id=client_id,
        redirect_uri=redirect_uri,
        scope=scope
    )
    
    # Сохраняем code_verifier для последующего использования
    session.code_verifier = code_verifier
    
    return session, code_challenge

# Пример использования
client_id = 'YOUR_CLIENT_ID'
redirect_uri = 'com.example.app://callback'
scope = ['profile', 'email']
 
# Создаём OAuth-сессию с PKCE
session, code_challenge = create_oauth_session_with_pkce(client_id, redirect_uri, scope)
 
# Формируем URL авторизации с code_challenge
auth_url = session.authorization_url(
    'https://authorization-server.com/authorize',
    code_challenge=code_challenge,
    code_challenge_method='S256'  # Указываем метод трансформации
)
 
# auth_url теперь можно открыть в браузере или WebView мобильного приложения

# Обработка ответа после авторизации
def handle_callback(session, callback_url, token_url):
    # Извлекаем код из callback_url
    code = session.parse_authorization_response(callback_url).get('code')
    
    # Обмениваем код на токен, предоставляя code_verifier
    token = session.fetch_token(
        token_url,
        code=code,
        code_verifier=session.code_verifier,  # Тот самый code_verifier, сохранённый ранее
    )
    
    return token

# Серверная часть (Python Flask)
@app.route('/get_auth_url')
def get_auth_url():
    session, code_challenge = create_oauth_session_with_pkce(CLIENT_ID, REDIRECT_URI, SCOPE)
    
    # Сохраняем code_verifier в Redis или другом хранилище с ключом session_id
    session_id = str(uuid.uuid4())
    redis_client.set(f"pkce:{session_id}", session.code_verifier, ex=3600)
    
    auth_url = session.authorization_url(
        AUTH_URL,
        code_challenge=code_challenge,
        code_challenge_method='S256',
        state=session_id  # Используем session_id как state
    )
    
    return jsonify({'auth_url': auth_url, 'session_id': session_id})
 
@app.route('/exchange_token')
def exchange_token():
    code = request.args.get('code')
    session_id = request.args.get('state')
    
    # Получаем code_verifier из хранилища
    code_verifier = redis_client.get(f"pkce:{session_id}")
    if not code_verifier:
        return jsonify({'error': 'Invalid or expired session'}), 400
    
    # Создаём новую сессию
    session = OAuth2Session(CLIENT_ID, redirect_uri=REDIRECT_URI)
    
    # Обмениваем код на токен
    token = session.fetch_token(
        TOKEN_URL,
        code=code,
        code_verifier=code_verifier.decode('utf-8')
    )
    
    # Удаляем использованный code_verifier
    redis_client.delete(f"pkce:{session_id}")
    
    return jsonify(token)

# Настройка логирования для библиотеки Requests
import logging
 
# Включаем логирование HTTP-запросов
logging.basicConfig(level=logging.DEBUG)
logging.getLogger('requests').setLevel(logging.DEBUG)
logging.getLogger('urllib3').setLevel(logging.DEBUG)

# Отладочный код для проверки URI перенаправления
@app.route('/debug/redirect')
def debug_redirect():
redirect_uri = url_for('authorize', _external=True)
return f"""
<h1>Redirect URI Debug</h1>
<p>Redirect URI: <code>{redirect_uri}</code></p>
<p>Registered URI: <code>http://localhost:5000/callback</code></p>
"""

@app.route('/debug/token')
def debug_token():
if 'token' not in session:
    return "Токен не найден в сессии"
    
token = session['token']
expiry = token.get('expires_at', 'Не указано')
scope = token.get('scope', 'Не указано')
refresh_token = 'Присутствует' if token.get('refresh_token') else 'Отсутствует'
 
return f"""
<h1>Токен Debug</h1>
<p>Срок действия: {expiry}</p>
<p>Области: {scope}</p>
<p>Refresh токен: {refresh_token}</p>
"""

import unittest
from unittest.mock import patch, MagicMock
from your_app import app, oauth
 
class OAuthTestCase(unittest.TestCase):
    def setUp(self):
        self.client = app.test_client()
        self.client.testing = True
        
    @patch('your_app.oauth.github.authorize_redirect')
    def test_login_redirect(self, mock_auth_redirect):
        mock_auth_redirect.return_value = "Redirecting..."
        response = self.client.get('/login')
        self.assertEqual(response.status_code, 200)
        mock_auth_redirect.assert_called_once()
        
    @patch('your_app.oauth.github.authorize_access_token')
    @patch('your_app.oauth.github.get')
    def test_callback(self, mock_get, mock_token):
        mock_token.return_value = {"access_token": "test_token"}
        user_mock = MagicMock()
        user_mock.json.return_value = {"login": "test_user"}
        mock_get.return_value = user_mock
        
        response = self.client.get('/callback?code=test_code')
        self.assertEqual(response.status_code, 302)  # Redirect after login

import pytest
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
 
@pytest.fixture
def driver():
    driver = webdriver.Chrome()
    yield driver
    driver.quit()
 
def test_oauth_flow(driver):
    # Открываем страницу логина
    driver.get('http://localhost:5000/login')
    
    # Ждем перенаправления на GitHub
    WebDriverWait(driver, 10).until(
        EC.url_contains('github.com/login')
    )
    
    # Вводим тестовые учетные данные
    driver.find_element(By.ID, 'login_field').send_keys('test_user')
    driver.find_element(By.ID, 'password').send_keys('test_password')
    driver.find_element(By.NAME, 'commit').click()
    
    # Подтверждаем авторизацию (если необходимо)
    try:
        authorize_button = WebDriverWait(driver, 5).until(
            EC.element_to_be_clickable((By.ID, 'js-oauth-authorize-btn'))
        )
        authorize_button.click()
    except:
        pass  # Если пользователь уже авторизовал приложение
    
    # Проверяем успешное перенаправление обратно
    WebDriverWait(driver, 10).until(
        EC.url_contains('localhost:5000')
    )
    
    # Проверяем, что пользователь авторизован
    assert 'Привет, test_user' in driver.page_source

from authlib.integrations.flask_oauth2 import AuthorizationServer, ResourceProtector
from authlib.oauth2.rfc6749 import grants
from flask import Flask, jsonify
 
app = Flask(__name__)
query_client = lambda client_id: clients.get(client_id)
authorization = AuthorizationServer(app, query_client=query_client)
 
# Определяем тестовых клиентов
clients = {
    'test_client': {
        'client_id': 'test_client',
        'client_secret': 'test_secret',
        'redirect_uris': ['http://localhost:5000/callback']
    }
}
 
# Регистрируем гранты
authorization.register_grant(grants.AuthorizationCodeGrant)
authorization.register_grant(grants.RefreshTokenGrant)
 
# Endpoint авторизации
@app.route('/authorize', methods=['GET', 'POST'])
def authorize():
    # Упрощенная версия для тестирования
    return authorization.create_authorization_response(grant_user=1)
 
# Endpoint токенов
@app.route('/token', methods=['POST'])
def issue_token():
    return authorization.create_token_response()

# Пример использования Authlib с AIOHTTP
from authlib.integrations.aiohttp_client import OAuth
from aiohttp import web
 
routes = web.RouteTableDef()
app = web.Application()
 
# Инициализация асинхронного OAuth клиента
oauth = OAuth()
oauth.register(
    name='github',
    client_id='CLIENT_ID',
    client_secret='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'},
)
 
@routes.get('/login')
async def login(request):
    redirect_uri = request.app.router['authorize'].url_for()
    return await oauth.github.authorize_redirect(request, redirect_uri)
 
@routes.get('/callback')
async def authorize(request):
    token = await oauth.github.authorize_access_token(request)
    resp = await oauth.github.get('user', token=token)
    user = await resp.json()
    return web.Response(text=f'Hello {user["login"]}!')
 
app.add_routes(routes)

from fastapi import FastAPI, Depends, Request, HTTPException
from fastapi.responses import RedirectResponse
from authlib.integrations.starlette_client import OAuth
from starlette.middleware.sessions import SessionMiddleware
 
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="supersecret")
 
oauth = OAuth()
oauth.register(
    name="github",
    client_id="CLIENT_ID",
    client_secret="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"},
)
 
@app.get("/login")
async def login(request: Request):
    redirect_uri = request.url_for("auth")
    return await oauth.github.authorize_redirect(request, redirect_uri)
 
@app.get("/auth")
async def auth(request: Request):
    try:
        token = await oauth.github.authorize_access_token(request)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    resp = await oauth.github.get("user", token=token)
    user = await resp.json()
    request.session["user"] = user
    return RedirectResponse(url="/")
 
@app.get("/")
async def homepage(request: Request):
    user = request.session.get("user")
    if user:
        return {"message": f"Hello, {user['login']}!"}
    return {"message": "Not logged in", "login_url": "/login"}

@app.get("/auth")
async def auth(request: Request):
    try:
        token = await oauth.github.authorize_access_token(request)
        
        async with oauth.github.get("user", token=token) as resp:
            if resp.status != 200:
                error_detail = await resp.json()
                raise HTTPException(
                    status_code=resp.status,
                    detail=f"API error: {error_detail.get('message')}"
                )
            
            user = await resp.json()
            request.session["user"] = user
            return RedirectResponse(url="/")
            
    except Exception as e:
        # Логирование ошибки
        logger.error(f"OAuth error: {str(e)}")
        raise HTTPException(status_code=400, detail=str(e))

oauth.register(
    name="github",
    # ... другие параметры ...
    client_kwargs={
        "scope": "user:email",
        "timeout": 10.0  # 10 секунд тайм-аут
    },
)

import aioredis
from fastapi import FastAPI, Depends, Request
 
app = FastAPI()
 
# Создаем пул подключений к Redis
async def get_redis():
    redis = await aioredis.create_redis_pool("redis://localhost")
    try:
        yield redis
    finally:
        redis.close()
        await redis.wait_closed()
 
# Сохранение токена в Redis
async def save_token(redis, user_id, token, expires_in=3600):
    token_data = json.dumps(token)
    await redis.setex(f"token:{user_id}", expires_in, token_data)
 
# Получение токена из Redis
async def get_token(redis, user_id):
    token_data = await redis.get(f"token:{user_id}")
    if token_data:
        return json.loads(token_data)
    return None
 
@app.get("/auth")
async def auth(request: Request, redis=Depends(get_redis)):
    token = await oauth.github.authorize_access_token(request)
    # ... получение информации о пользователе ...
    await save_token(redis, user["id"], token)
    return RedirectResponse(url="/")

import asyncio
from fastapi import FastAPI, Depends, Request
from datetime import datetime
 
app = FastAPI()
 
# Словарь семафоров для каждого пользователя
refresh_locks = {}
 
async def get_valid_token(redis, user_id):
    # Получаем текущий токен
    token = await get_token(redis, user_id)
    
    # Если токен отсутствует или истек
    now = datetime.utcnow().timestamp()
    if not token or token.get("expires_at", 0) < now:
        # Получаем или создаем семафор для пользователя
        if user_id not in refresh_locks:
            refresh_locks[user_id] = asyncio.Semaphore(1)
        
        # Пытаемся захватить семафор
        async with refresh_locks[user_id]:
            # Повторно проверяем токен (он мог быть обновлен другим запросом)
            token = await get_token(redis, user_id)
            if not token or token.get("expires_at", 0) < now:
                # Обновляем токен
                refresh_token = token.get("refresh_token")
                if refresh_token:
                    new_token = await oauth.github.refresh_token(refresh_token)
                    await save_token(redis, user_id, new_token)
                    return new_token
                # Если нет refresh_token, возвращаем None
                return None
    
    return token

# Простой in-memory кеш для токенов
token_cache = {}
 
def get_cached_token(user_id):
if user_id in token_cache:
    token_data = token_cache[user_id]
    # Проверяем срок действия токена
    if token_data['expires_at'] > time.time():
        return token_data['token']
return None
 
def cache_token(user_id, token, expires_in):
token_cache[user_id] = {
    'token': token,
    'expires_at': time.time() + expires_in
}

import redis
import json
 
redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
def get_cached_token(user_id):
token_data = redis_client.get(f"oauth_token:{user_id}")
if token_data:
    token_info = json.loads(token_data)
    return token_info
return None
 
def cache_token(user_id, token_info, expires_in):
redis_client.setex(
    f"oauth_token:{user_id}",
    expires_in,  # Автоматическое удаление по истечении срока
    json.dumps(token_info)
)

from sqlalchemy import create_engine, Column, String, Integer, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
 
Base = declarative_base()
engine = create_engine('sqlite:///tokens.db')
Session = sessionmaker(bind=engine)
 
class TokenStorage(Base):
__tablename__ = 'tokens'
user_id = Column(String, primary_key=True)
access_token = Column(String)
refresh_token = Column(String)
expires_at = Column(Float)
 
Base.metadata.create_all(engine)
 
def store_token(user_id, access_token, refresh_token, expires_at):
session = Session()
token = session.query(TokenStorage).filter_by(user_id=user_id).first()
if token:
    token.access_token = access_token
    token.refresh_token = refresh_token
    token.expires_at = expires_at
else:
    token = TokenStorage(
        user_id=user_id,
        access_token=access_token,
        refresh_token=refresh_token,
        expires_at=expires_at
    )
    session.add(token)
session.commit()

def get_or_refresh_token(user_id):
# Сначала проверяем Redis
token_data = redis_client.get(f"oauth_token:{user_id}")
if token_data:
    token_info = json.loads(token_data)
    # Если токен не истёк, используем его
    if token_info['expires_at'] > time.time():
        return token_info['access_token']
 
# Если токен не найден или истёк, ищем refresh token в БД
session = Session()
token_record = session.query(TokenStorage).filter_by(user_id=user_id).first()
if token_record and token_record.refresh_token:
    # Обновляем токен через OAuth-провайдера
    new_token = oauth_client.refresh_token(token_record.refresh_token)
    
    # Обновляем запись в БД
    token_record.access_token = new_token['access_token']
    token_record.expires_at = time.time() + new_token['expires_in']
    if 'refresh_token' in new_token:  # Не все провайдеры возвращают новый refresh_token
        token_record.refresh_token = new_token['refresh_token']
    session.commit()
    
    # Кешируем новый access token в Redis
    cache_token(user_id, new_token, new_token['expires_in'])
    
    return new_token['access_token']
 
# Если ничего не нашли, пользователю нужно авторизоваться заново
return None

import threading
 
# Словарь блокировок для каждого пользователя
token_locks = {}
lock_for_locks = threading.Lock()  # Мета-блокировка для доступа к словарю блокировок
 
def get_user_lock(user_id):
with lock_for_locks:
    if user_id not in token_locks:
        token_locks[user_id] = threading.Lock()
    return token_locks[user_id]
 
def get_or_refresh_token(user_id):
# Сначала пробуем получить токен без блокировки
token = get_cached_token(user_id)
if token and not is_token_expired(token):
    return token
 
# Если токен не найден или истёк, берём блокировку и пробуем обновить
with get_user_lock(user_id):
    # Повторно проверяем кеш (токен мог быть обновлён другим потоком)
    token = get_cached_token(user_id)
    if token and not is_token_expired(token):
        return token
    
    # Обновляем токен
    new_token = refresh_token_from_oauth_provider(user_id)
    cache_token(user_id, new_token, new_token['expires_in'])
    return new_token

import asyncio
 
# Словарь асинхронных блокировок
token_locks = {}
 
async def get_user_lock(user_id):
if user_id not in token_locks:
    token_locks[user_id] = asyncio.Lock()
return token_locks[user_id]
 
async def get_or_refresh_token(user_id):
# Логика аналогична, но с использованием async/await
token = await get_cached_token(user_id)
if token and not is_token_expired(token):
    return token
 
async with await get_user_lock(user_id):
    # Повторная проверка
    token = await get_cached_token(user_id)
    if token and not is_token_expired(token):
        return token
    
    new_token = await refresh_token_from_oauth_provider(user_id)
    await cache_token(user_id, new_token, new_token['expires_in'])
    return new_token

def is_token_near_expiry(token_info, threshold=0.1):
"""Проверяет, приближается ли токен к истечению (по умолчанию 10% от срока)"""
now = time.time()
token_lifetime = token_info['expires_at'] - token_info['created_at']
time_left = token_info['expires_at'] - now
return time_left < (token_lifetime * threshold)
 
def get_token(user_id):
token_info = get_cached_token(user_id)
if not token_info:
    return None
    
if is_token_expired(token_info):
    # Токен истёк, необходимо обновить
    return refresh_token(user_id)
    
if is_token_near_expiry(token_info):
    # Токен скоро истечёт, запускаем обновление в фоновом режиме
    threading.Thread(target=refresh_token, args=(user_id,)).start()
    
return token_info['access_token']

dev_activity_tracker/
│
├── app.py            # Основной файл приложения
├── config.py         # Конфигурация и настройки
├── auth/             # Модуль аутентификации
│   ├── __init__.py
│   ├── oauth.py      # Логика OAuth
│   └── routes.py     # Маршруты для аутентификации
├── models/           # Модели данных
│   ├── __init__.py
│   └── user.py       # Модель пользователя
├── static/           # Статические файлы
└── templates/        # HTML-шаблоны

python -m venv venv
source venv/bin/activate  # или venv\Scripts\activate на Windows
 
pip install flask authlib flask-sqlalchemy flask-login python-dotenv redis

import os
from dotenv import load_dotenv
 
load_dotenv()  # Загружаем переменные окружения из .env файла
 
class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'вы-никогда-не-угадаете-этот-ключ'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///dev_activity.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    
    # Настройки Redis для кеширования токенов
    REDIS_URL = os.environ.get('REDIS_URL') or 'redis://localhost:6379/0'
    
    # OAuth настройки
    GITHUB_CLIENT_ID = os.environ.get('GITHUB_CLIENT_ID')
    GITHUB_CLIENT_SECRET = os.environ.get('GITHUB_CLIENT_SECRET')

SECRET_KEY=ваш-секретный-ключ-здесь
GITHUB_CLIENT_ID=ваш-github-client-id
GITHUB_CLIENT_SECRET=ваш-github-client-secret

from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
import json
 
db = SQLAlchemy()
 
class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(120), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=True)
    github_id = db.Column(db.String(120), unique=True, nullable=True)
    avatar_url = db.Column(db.String(256), nullable=True)
    
    # Информация о токенах будет храниться в Redis, а не в базе данных
    # Это обеспечит быстрый доступ и автоматическое удаление по истечении срока
    
    def __repr__(self):
        return f'<User {self.username}>'

from authlib.integrations.flask_client import OAuth
from flask import current_app, url_for, session, redirect, request
import redis
import json
import time
 
# Инициализируем OAuth
oauth = OAuth()
 
# Подключаемся к Redis
redis_client = None
 
def init_oauth(app):
    global redis_client
    oauth.init_app(app)
    
    # Регистрируем GitHub как провайдера OAuth
    oauth.register(
        name='github',
        client_id=app.config['GITHUB_CLIENT_ID'],
        client_secret=app.config['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 repo'},
    )
    
    # Инициализируем подключение к Redis
    redis_client = redis.from_url(app.config['REDIS_URL'])
    
    return oauth
 
# Функция для сохранения токена в Redis
def save_token(user_id, token, expires_in=3600):
    # Добавляем время истечения
    token['expires_at'] = time.time() + expires_in
    
    # Сохраняем токен в Redis с автоматическим истечением
    redis_client.setex(
        f'oauth_token:{user_id}',
        expires_in,
        json.dumps(token)
    )
 
# Функция для получения токена из Redis
def get_token(user_id):
    token_data = redis_client.get(f'oauth_token:{user_id}')
    if token_data:
        return json.loads(token_data)
    return None
 
# Функция для получения валидного токена с обновлением при необходимости
def get_valid_token(user_id):
    token = get_token(user_id)
    
    # Если токен не найден или истёк
    if not token or token.get('expires_at', 0) < time.time():
        # Если есть refresh_token, пытаемся обновить
        if token and 'refresh_token' in token:
            new_token = oauth.github.refresh_token(token['refresh_token'])
            save_token(user_id, new_token, new_token.get('expires_in', 3600))
            return new_token
        # Иначе требуется повторная аутентификация
        return None
    
    return token

from flask import Blueprint, redirect, url_for, current_app, session, request, flash
from flask_login import current_user, login_user, logout_user, login_required
from .oauth import oauth, save_token
from models.user import User, db
import secrets
 
# Создаем блупринт для аутентификации
auth_bp = Blueprint('auth', __name__)
 
@auth_bp.route('/login')
def login():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))
    
    # Генерируем состояние для предотвращения CSRF
    state = secrets.token_hex(16)
    session['oauth_state'] = state
    
    # Формируем URI перенаправления
    redirect_uri = url_for('auth.github_callback', _external=True)
    
    # Перенаправляем пользователя на GitHub для авторизации
    return oauth.github.authorize_redirect(redirect_uri, state=state)
 
@auth_bp.route('/github/callback')
def github_callback():
    # Проверяем состояние для предотвращения CSRF
    if request.args.get('state') != session.pop('oauth_state', None):
        flash('Ошибка безопасности: несовпадение состояния', 'danger')
        return redirect(url_for('main.index'))
    
    try:
        # Получаем токен доступа
        token = oauth.github.authorize_access_token()
        
        # Получаем информацию о пользователе
        resp = oauth.github.get('user', token=token)
        profile = resp.json()
        
        # Проверяем, существует ли пользователь
        user = User.query.filter_by(github_id=str(profile['id'])).first()
        
        if not user:
            # Создаем нового пользователя
            user = User(
                username=profile['login'],
                github_id=str(profile['id']),
                avatar_url=profile['avatar_url']
            )
            
            # Получаем email пользователя, если он доступен
            email_resp = oauth.github.get('user/emails', token=token)
            emails = email_resp.json()
            primary_email = next((e for e in emails if e['primary']), None)
            
            if primary_email:
                user.email = primary_email['email']
            
            db.session.add(user)
            db.session.commit()
        
        # Сохраняем токен в Redis
        save_token(user.id, token, token.get('expires_in', 3600))
        
        # Логиним пользователя
        login_user(user, remember=True)
        
        return redirect(url_for('main.dashboard'))
        
    except Exception as e:
        flash(f'Ошибка авторизации: {str(e)}', 'danger')
        return redirect(url_for('main.index'))
 
@auth_bp.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('main.index'))

from flask import Flask, render_template, redirect, url_for
from flask_login import LoginManager, current_user
from config import Config
from models.user import db, User
from auth.oauth import init_oauth
from auth.routes import auth_bp
 
def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
    
    # Инициализируем расширения
    db.init_app(app)
    init_oauth(app)
    
    # Настраиваем Flask-Login
    login_manager = LoginManager()
    login_manager.init_app(app)
    login_manager.login_view = 'auth.login'
    
    @login_manager.user_loader
    def load_user(user_id):
        return User.query.get(int(user_id))
    
    # Регистрируем блупринты
    app.register_blueprint(auth_bp, url_prefix='/auth')
    
    # Определяем основные маршруты
    @app.route('/')
    def index():
        if current_user.is_authenticated:
            return redirect(url_for('dashboard'))
        return render_template('index.html')
    
    @app.route('/dashboard')
    def dashboard():
        if not current_user.is_authenticated:
            return redirect(url_for('index'))
        return render_template('dashboard.html', user=current_user)
    
    # Создаем базу данных (для разработки)
    with app.app_context():
        db.create_all()
    
    return app
 
if __name__ == '__main__':
    app = create_app()
    app.run(debug=True)

# Регистрация Google как OAuth-провайдера
oauth.register(
    name='google',
    client_id=app.config['GOOGLE_CLIENT_ID'],
    client_secret=app.config['GOOGLE_CLIENT_SECRET'],
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@auth_bp.route('/login/google')
def google_login():
    redirect_uri = url_for('auth.google_callback', _external=True)
    state = secrets.token_hex(16)
    session['oauth_state'] = state
    return oauth.google.authorize_redirect(redirect_uri, state=state)
 
@auth_bp.route('/google/callback')
def google_callback():
    token = oauth.google.authorize_access_token()
    # Google возвращает информацию о пользователе в id_token
    user_info = oauth.google.parse_id_token(token)
    
    # Дальнейшая обработка...

oauth.register(
    name='facebook',
    client_id=app.config['FACEBOOK_CLIENT_ID'],
    client_secret=app.config['FACEBOOK_CLIENT_SECRET'],
    access_token_url='https://graph.facebook.com/v12.0/oauth/access_token',
    authorize_url='https://www.facebook.com/v12.0/dialog/oauth',
    api_base_url='https://graph.facebook.com/v12.0/',
    client_kwargs={'scope': 'email,public_profile'}
)

@auth_bp.route('/facebook/callback')
def facebook_callback():
    token = oauth.facebook.authorize_access_token()
    resp = oauth.facebook.get('me?fields=id,name,email,picture', token=token)
    profile = resp.json()
    
    # Обработка данных пользователя...

# Словарь с настройками провайдеров
OAUTH_PROVIDERS = {
    'github': {
        'client_id': 'GITHUB_CLIENT_ID',
        'client_secret': 'GITHUB_CLIENT_SECRET',
        'authorize_url': 'https://github.com/login/oauth/authorize',
        'access_token_url': 'https://github.com/login/oauth/access_token',
        'api_base_url': 'https://api.github.com/',
        'client_kwargs': {'scope': 'user:email repo'},
        'userinfo_endpoint': 'user',
        'id_field': 'id',
        'username_field': 'login',
        'email_field': 'email'
    },
    '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'},
        'userinfo_endpoint': 'userinfo',
        'id_field': 'sub',
        'username_field': 'name',
        'email_field': 'email'
    },
    # Другие провайдеры...
}
 
# Функция для регистрации провайдеров
def register_oauth_providers(app, oauth):
    for provider_name, config in OAUTH_PROVIDERS.items():
        provider_config = {k: v for k, v in config.items() if k != 'client_kwargs'}
        
        # Получаем значения из переменных окружения
        for key in ['client_id', 'client_secret']:
            if key in provider_config:
                env_var = app.config.get(provider_config[key])
                provider_config[key] = env_var
        
        # Добавляем client_kwargs, если они есть
        if 'client_kwargs' in config:
            provider_config['client_kwargs'] = config['client_kwargs']
        
        # Регистрируем провайдера
        oauth.register(name=provider_name, **provider_config)

@auth_bp.route('/login/<provider>')
def oauth_login(provider):
    if provider not in OAUTH_PROVIDERS:
        flash(f'Провайдер {provider} не поддерживается', 'danger')
        return redirect(url_for('main.index'))
    
    redirect_uri = url_for('auth.oauth_callback', provider=provider, _external=True)
    state = secrets.token_hex(16)
    session['oauth_state'] = state
    
    return getattr(oauth, provider).authorize_redirect(redirect_uri, state=state)
 
@auth_bp.route('/callback/<provider>')
def oauth_callback(provider):
    if provider not in OAUTH_PROVIDERS:
        flash(f'Провайдер {provider} не поддерживается', 'danger')
        return redirect(url_for('main.index'))
    
    # Проверка state
    if request.args.get('state') != session.pop('oauth_state', None):
        flash('Ошибка безопасности: несовпадение состояния', 'danger')
        return redirect(url_for('main.index'))
    
    try:
        # Получаем токен
        token = getattr(oauth, provider).authorize_access_token()
        
        # Получаем информацию о пользователе
        provider_config = OAUTH_PROVIDERS[provider]
        
        # Для Google используем специальный метод parse_id_token
        if provider == 'google' and 'id_token' in token:
            userinfo = oauth.google.parse_id_token(token)
        else:
            resp = getattr(oauth, provider).get(provider_config['userinfo_endpoint'], token=token)
            userinfo = resp.json()
        
        # Извлекаем нужные поля
        provider_user_id = str(userinfo[provider_config['id_field']])
        username = userinfo.get(provider_config['username_field'])
        email = userinfo.get(provider_config['email_field'])
        
        # Ищем или создаем пользователя
        # ...
        
    except Exception as e:
        flash(f'Ошибка авторизации: {str(e)}', 'danger')
        return redirect(url_for('main.index'))

# Пример CQRS для OAuth-токенов
class TokenWriter:
    def save_token(self, user_id, token_data):
        # Сохраняем в основную БД с полной транзакционностью
        with transaction.atomic():
            Token.objects.update_or_create(
                user_id=user_id,
                defaults={"token_data": token_data}
            )
            # Инвалидируем кеш
            cache.delete(f"token:{user_id}")
 
class TokenReader:
    def get_token(self, user_id):
        # Сначала пробуем получить из кеша
        cached_token = cache.get(f"token:{user_id}")
        if cached_token:
            return cached_token
            
        # Если нет в кеше, читаем из реплики БД
        token = TokenReadReplica.objects.get(user_id=user_id)
        
        # Кешируем для будущих запросов
        cache.set(f"token:{user_id}", token.token_data, timeout=300)
        return token.token_data

def get_shard_for_user(user_id):
    # Простая стратегия шардирования по модулю
    shard_count = 16  # Количество шардов
    return user_id % shard_count
 
def get_token_from_sharded_storage(user_id):
    shard_id = get_shard_for_user(user_id)
    shard_connection = get_connection_for_shard(shard_id)
    
    # Получаем токен из соответствующего шарда
    return shard_connection.get_token(user_id)

# Двухуровневый кеш для токенов
class TwoLevelTokenCache:
    def __init__(self, local_cache_size=1000):
        # Локальный кеш в памяти (LRU)
        self.local_cache = LRUCache(local_cache_size)
        # Соединение с Redis
        self.redis = redis.Redis()
    
    def get_token(self, user_id):
        # Попытка получить из локального кеша
        key = f"token:{user_id}"
        if key in self.local_cache:
            return self.local_cache[key]
        
        # Попытка получить из Redis
        token_data = self.redis.get(key)
        if token_data:
            # Заполняем локальный кеш
            self.local_cache[key] = token_data
            return token_data
            
        return None
    
    def set_token(self, user_id, token_data, expire=3600):
        key = f"token:{user_id}"
        # Сохраняем в Redis
        self.redis.setex(key, expire, token_data)
        # Обновляем локальный кеш
        self.local_cache[key] = token_data

# Пример планирования задачи на обновление токена
from datetime import datetime, timedelta
from celery import shared_task
 
@shared_task
def schedule_token_refresh():
    """Задача для планирования обновления токенов, которые скоро истекут"""
    # Ищем токены, которые истекут в ближайшие 10 минут
    soon = datetime.now() + timedelta(minutes=10)
    expiring_tokens = Token.objects.filter(expires_at__lt=soon)
    
    for token in expiring_tokens:
        # Планируем задачу на обновление для каждого токена
        refresh_token.delay(token.user_id, token.refresh_token)

@app.route('/api/sessions')
@login_required
def get_user_sessions():
    """Эндпоинт для получения всех активных сессий пользователя"""
    sessions = Session.query.filter_by(user_id=current_user.id).all()
    return jsonify([{
        'device': session.user_agent,
        'ip': session.ip_address,
        'last_active': session.last_active,
        'is_current': session.id == session.get('session_id')
    } for session in sessions])
 
@app.route('/api/sessions/<session_id>', methods=['DELETE'])
@login_required
def terminate_session(session_id):
    """Эндпоинт для завершения конкретной сессии"""
    session = Session.query.get(session_id)
    if not session or session.user_id != current_user.id:
        return jsonify({'error': 'Session not found'}), 404
        
    # Отзываем связанный с сессией токен
    revoke_token(session.token_id)
    # Удаляем сессию
    db.session.delete(session)
    db.session.commit()
    
    return jsonify({'success': True})

# Проверка токена с учетом версии
def validate_token(token, required_version=None):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        
        # Проверка версии токена
        token_version = payload.get('version', 1)
        if required_version and token_version < required_version:
            raise InvalidTokenError("Token version is outdated")
            
        return payload
    except jwt.InvalidTokenError:
        return None

class TokenBlacklist:
    def __init__(self):
        self.redis = redis.Redis()
        self.blacklist_key = "oauth:token:blacklist"
    
    def revoke_token(self, jti, expire=86400):
        """Добавляет токен в черный список"""
        self.redis.setex(f"{self.blacklist_key}:{jti}", expire, 1)
    
    def is_revoked(self, jti):
        """Проверяет, отозван ли токен"""
        return bool(self.redis.get(f"{self.blacklist_key}:{jti}"))
    
    def revoke_all_for_user(self, user_id, expire=86400):
        """Отзывает все токены для пользователя"""
        # Получаем все токены пользователя
        tokens = Token.query.filter_by(user_id=user_id).all()
        for token in tokens:
            self.revoke_token(token.jti, expire)

# Пример middleware для API Gateway (FastAPI)
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import OAuth2AuthorizationCodeBearer
import jwt
 
app = FastAPI()
 
oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl="https://auth-server.com/authorize",
    tokenUrl="https://auth-server.com/token"
)
 
async def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, "secret_key", algorithms=["HS256"])
        # Добавляем дополнительные проверки токена при необходимости
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
 
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    # Проверяем путь запроса - некоторые эндпоинты могут быть публичными
    if request.url.path.startswith("/api/"):
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not token:
            return JSONResponse(status_code=401, content={"detail": "Not authenticated"})
        
        try:
            # Проверяем токен
            payload = jwt.decode(token, "secret_key", algorithms=["HS256"])
            # Добавляем информацию о пользователе в заголовки для микросервисов
            request.state.user = payload
        except jwt.PyJWTError:
            return JSONResponse(status_code=401, content={"detail": "Invalid token"})
    
    # Продолжаем обработку запроса
    return await call_next(request)

# Пример сервиса, который вызывает другой микросервис от имени пользователя
import requests
from fastapi import FastAPI, Depends, HTTPException, Header
 
app = FastAPI()
 
def get_current_user(authorization: str = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")
    token = authorization.replace("Bearer ", "")
    # Здесь мы предполагаем, что токен уже проверен API Gateway
    return {"token": token}
 
@app.get("/aggregate-data")
async def aggregate_data(current_user = Depends(get_current_user)):
    # Вызываем другой микросервис, передавая оригинальный токен пользователя
    user_data_response = requests.get(
        "http://user-service/api/user-data",
        headers={"Authorization": f"Bearer {current_user['token']}"}
    )
    
    orders_response = requests.get(
        "http://orders-service/api/orders",
        headers={"Authorization": f"Bearer {current_user['token']}"}
    )
    
    # Объединяем данные и возвращаем результат
    return {
        "user_data": user_data_response.json(),
        "orders": orders_response.json()
    }

# Сервис аутентификации на Flask
from flask import Flask, request, jsonify
import jwt
import redis
 
app = Flask(__name__)
token_cache = redis.Redis()
 
@app.route('/verify', methods=['POST'])
def verify_token():
    token = request.json.get('token')
    if not token:
        return jsonify({"error": "Token is required"}), 400
    
    # Проверяем в кеше
    cached_verification = token_cache.get(f"verified:{token}")
    if cached_verification:
        return jsonify({"valid": True, "payload": jwt.decode(cached_verification)})
    
    # Проверяем токен
    try:
        payload = jwt.decode(token, "secret_key", algorithms=["HS256"])
        # Кешируем результат проверки
        token_cache.setex(f"verified:{token}", 300, jwt.encode(payload, ""))
        return jsonify({"valid": True, "payload": payload})
    except jwt.PyJWTError as e:
        return jsonify({"valid": False, "error": str(e)}), 401

# Клиентский код для микросервиса
import requests
 
def verify_user_token(token):
    response = requests.post(
        "http://auth-service/verify",
        json={"token": token}
    )
    if response.status_code == 200 and response.json()["valid"]:
        return response.json()["payload"]
    return None

# Генерация внутреннего токена
def create_internal_token(original_token):
    # Проверяем оригинальный токен
    payload = jwt.decode(original_token, "oauth_secret", algorithms=["RS256"])
    
    # Создаем "тонкий" внутренний токен
    internal_payload = {
        "sub": payload["sub"],
        "token_id": str(uuid.uuid4()),
        "exp": time.time() + 3600  # 1 час
    }
    
    # Сохраняем связь между внутренним и оригинальным токеном
    token_cache.setex(f"internal:{internal_payload['token_id']}", 3600, original_token)
    
    return jwt.encode(internal_payload, "internal_secret", algorithm="HS256")

from flask import Flask, jsonify, request
from authlib.integrations.flask_oauth2 import AuthorizationServer, ResourceProtector
from authlib.oauth2.rfc6749 import grants
 
app = Flask(__name__)
app.secret_key = 'your-secret-key'
 
# Хранилище клиентов (в реальном приложении это была бы база данных)
oauth_clients = {
    'client1': {
        'client_id': 'client1',
        'client_secret': 'secret1',
        'redirect_uris': ['http://localhost:8000/callback'],
        'grant_types': ['authorization_code', 'refresh_token'],
        'scope': 'profile email'
    }
}
 
# Хранилище токенов (в реальном приложении - база данных или Redis)
oauth_tokens = {}
 
# Функция для поиска клиента
def get_client(client_id):
    client = oauth_clients.get(client_id)
    if client:
        return client
    return None
 
# Инициализация сервера авторизации
authorization = AuthorizationServer(app, query_client=get_client)

class AuthorizationCodeGrant(grants.AuthorizationCodeGrant):
    def save_authorization_code(self, code, request):
        # В реальном приложении - сохранение в БД
        client = request.client
        oauth_tokens[code] = {
            'client_id': client.client_id,
            'redirect_uri': request.redirect_uri,
            'user_id': request.user.id,  # Предполагается, что у нас есть пользователь
            'scope': request.scope
        }
        return code
 
    def query_authorization_code(self, code, client):
        item = oauth_tokens.get(code)
        if item and item['client_id'] == client.client_id:
            return item
        return None
 
    def delete_authorization_code(self, code):
        if code in oauth_tokens:
            del oauth_tokens[code]
 
    def authenticate_user(self, authorization_code):
        return {'id': authorization_code['user_id']}
 
# Регистрируем грант
authorization.register_grant(AuthorizationCodeGrant)

class RefreshTokenGrant(grants.RefreshTokenGrant):
    def authenticate_refresh_token(self, refresh_token):
        item = oauth_tokens.get(refresh_token)
        if item and 'refresh' in item:
            return item
        return None
 
    def authenticate_user(self, refresh_token):
        return {'id': refresh_token['user_id']}
 
    def revoke_old_credential(self, refresh_token):
        if refresh_token in oauth_tokens:
            del oauth_tokens[refresh_token]
 
# Регистрируем грант
authorization.register_grant(RefreshTokenGrant)

@app.route('/oauth/authorize', methods=['GET', 'POST'])
def authorize():
    # В реальном приложении здесь была бы проверка авторизации пользователя
    user = {'id': 1, 'username': 'test_user'}
    
    if request.method == 'GET':
        # Отображаем форму согласия
        return '''
        <form method="post">
            <p>Приложение "Client App" запрашивает доступ к:</p>
            <ul>
                <li>Профилю</li>
                <li>Email</li>
            </ul>
            <button type="submit" name="action" value="allow">Разрешить</button>
            <button type="submit" name="action" value="deny">Отклонить</button>
        </form>
        '''
    
    # Обрабатываем ответ пользователя
    if request.form.get('action') == 'allow':
        # Создаем ответ авторизации
        return authorization.create_authorization_response(grant_user=user)
    # Отклоняем запрос
    return authorization.create_authorization_response(grant_user=None)
 
@app.route('/oauth/token', methods=['POST'])
def issue_token():
    return authorization.create_token_response()
 
@app.route('/oauth/userinfo', methods=['GET'])
def userinfo():
    # В реальном приложении - проверка токена и получение данных пользователя
    # Для простоты возвращаем тестовые данные
    return jsonify({
        'sub': '1',
        'name': 'Test User',
        'email': '[email protected]'
    })

@app.route('/admin/clients', methods=['POST'])
def create_client():
    data = request.json
    client_id = data.get('client_id')
    
    if client_id in oauth_clients:
        return jsonify({'error': 'Client already exists'}), 400
    
    oauth_clients[client_id] = {
        'client_id': client_id,
        'client_secret': data.get('client_secret'),
        'redirect_uris': data.get('redirect_uris', []),
        'grant_types': data.get('grant_types', []),
        'scope': data.get('scope', '')
    }
    
    return jsonify({'client_id': client_id}), 201
 
@app.route('/admin/clients/<client_id>', methods=['GET'])
def get_client(client_id):
    client = oauth_clients.get(client_id)
    if not client:
        return jsonify({'error': 'Client not found'}), 404
    return jsonify(client)

import ldap
 
def authenticate_user_ldap(username, password):
    """Аутентификация пользователя через LDAP"""
    try:
        # Подключаемся к LDAP-серверу
        conn = ldap.initialize('ldap://ldap.example.com')
        conn.simple_bind_s(f'uid={username},ou=people,dc=example,dc=com', password)
        
        # Поиск пользователя для получения атрибутов
        result = conn.search_s(
            'ou=people,dc=example,dc=com',
            ldap.SCOPE_SUBTREE,
            f'(uid={username})',
            ['cn', 'mail']
        )
        
        if result:
            dn, attributes = result[0]
            return {
                'id': username,
                'name': attributes.get('cn', [b''])[0].decode('utf-8'),
                'email': attributes.get('mail', [b''])[0].decode('utf-8')
            }
        return None
    except ldap.INVALID_CREDENTIALS:
        return None
    except Exception as e:
        print(f"LDAP authentication error: {e}")
        return None
 
# Модифицируем эндпоинт авторизации для использования LDAP
@app.route('/oauth/authorize', methods=['GET', 'POST'])
def authorize():
    if request.method == 'GET':
        # Если пользователь не аутентифицирован, показываем форму входа
        if 'user' not in session:
            return render_template('login.html')
        # Иначе показываем форму согласия
        return render_template('consent.html')
    
    # Если это форма входа
    if 'username' in request.form:
        user = authenticate_user_ldap(request.form['username'], request.form['password'])
        if user:
            session['user'] = user
            return render_template('consent.html')
        return render_template('login.html', error='Invalid credentials')
    
    # Если это форма согласия
    if 'action' in request.form and request.form['action'] == 'allow':
        return authorization.create_authorization_response(grant_user=session['user'])
    return authorization.create_authorization_response(grant_user=None)

# Неправильная реализация (уязвимая к CSRF)
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    return oauth.github.authorize_redirect(redirect_uri)

import secrets
 
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    # Генерируем случайный state и сохраняем в сессии
    state = secrets.token_hex(16)
    session['oauth_state'] = state
    return oauth.github.authorize_redirect(redirect_uri, state=state)
 
@app.route('/callback')
def authorize():
    # Проверяем state для предотвращения CSRF
    expected_state = session.pop('oauth_state', None)
    received_state = request.args.get('state')
    
    if received_state != expected_state:
        abort(403, "Возможная CSRF-атака. Авторизация отменена.")
        
    # Продолжаем обработку...

# Опасная реализация с открытым перенаправлением
@app.route('/login')
def login():
    # Получаем redirect_uri из параметров запроса без проверки
    redirect_uri = request.args.get('redirect_uri', url_for('default_page'))
    return oauth.github.authorize_redirect(redirect_uri)