RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix auth

Browse Source
This commit is contained in:
Raykov-MS 2026-09-02 15:28:29 +03:00
parent faaf50694f
commit 57290a8d03
10 changed files with 440 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
api/src/api/v1/auth.py
View File

@ -2,37 +2,19 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from src.core.security import decode_keycloak_token
from src.db.session import get_db
from src.domain.schemas import LoginRequest, RefreshRequest, Token
from src.services.auth_service import AuthService
if not settings.DEBUG:
from raisa_fastapi_protected_api import UserInfo, get_user_dependency
router = APIRouter(prefix="/auth", tags=["auth"])
_PROD_USER_NOT_FOUND = "Пользователь не найден в локальной базе данных"
_PROD_EMAIL_MISSING = (
"Не удалось идентифицировать пользователя: в токене отсутствует поле email"
)
async def _login_via_keycloak_email(db: AsyncSession, user: "UserInfo") -> Token:
if not getattr(user, "email", None):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_EMAIL_MISSING,
headers={"WWW-Authenticate": "Bearer"},
)
auth_service = AuthService(db)
token = await auth_service.authenticate_user_via_email(user.email)
if not token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_USER_NOT_FOUND,
headers={"WWW-Authenticate": "Bearer"},
)
return token
_PROD_TOKEN_INVALID = "Недействительный токен Keycloak"
_PROD_TOKEN_REQUIRED = "Требуется access token Keycloak"
if settings.DEBUG:
@ -79,16 +61,35 @@ else:
async def login(
login_data: LoginRequest,
db: AsyncSession = Depends(get_db),
user: UserInfo = Depends(get_user_dependency),
):
return await _login_via_keycloak_email(db, user)
@router.post("/login-form", response_model=Token)
async def login_form(
db: AsyncSession = Depends(get_db),
user: UserInfo = Depends(get_user_dependency),
):
return await _login_via_keycloak_email(db, user)
if not login_data.token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_TOKEN_REQUIRED,
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_keycloak_token(login_data.token)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_TOKEN_INVALID,
headers={"WWW-Authenticate": "Bearer"},
)
if not payload.get("email"):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_EMAIL_MISSING,
headers={"WWW-Authenticate": "Bearer"},
)
auth_service = AuthService(db)
token = await auth_service.authenticate_with_keycloak_token(login_data.token)
if not token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_USER_NOT_FOUND,
headers={"WWW-Authenticate": "Bearer"},
)
return token
@router.post("/refresh", response_model=Token)

27
api/src/api/v1/deps.py
View File

@ -3,7 +3,7 @@ from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from src.core.security import verify_token
from src.core.security import decode_keycloak_token, verify_token
from src.db.session import get_db
from src.db.models.app_user import AppUser
from src.db.models.role import UserRoleEnum
@ -45,6 +45,31 @@ async def get_user_by_token(
return user
async def resolve_user_from_access_token(
token: str | None,
db: AsyncSession,
) -> AppUser | None:
"""Map access token to app_user without raising. Used by WebSocket login."""
if not token:
return None
if settings.DEBUG:
try:
return await get_user_by_token(token=token, db=db)
except HTTPException:
return None
payload = decode_keycloak_token(token)
if payload is None:
return None
email = payload.get("email")
if not email:
return None
user_repo = UserRepository(db)
user = await user_repo.get_by_email(email)
if user is None or not user.is_active:
return None
return user
if settings.DEBUG:
async def get_current_user(

7
api/src/api/v1/websocket.py
View File

@ -18,7 +18,7 @@ from src.services.rf_project_report_service import RfProjectReportService
from src.services.rf_project_report_line_service import RfProjectReportLineService
from src.repository.user_repository import UserRepository
from src.services.budget_line_service import BudgetLineService
from src.api.v1.deps import get_user_by_token
from src.api.v1.deps import resolve_user_from_access_token
from src.db.models.app_user import AppUser
from src.db.models.form_type import FormTypeEnum
from src.services.budget_form_service import BudgetFormService
@ -755,7 +755,10 @@ async def login(websocket: WebSocket, **kwargs) -> int:
return None
async with get_db_session() as db:
user = await get_user_by_token(token=user_data["data"].get("token"), db=db)
user = await resolve_user_from_access_token(
token=user_data["data"].get("token"),
db=db,
)
if not user:
return None
manager.set_user(

4
api/src/core/public_paths.py
View File

@ -1,4 +1,4 @@
"""Публичные пути для AuthorizationMiddleware: SPA и infra, без /api."""
"""Публичные пути для AuthorizationMiddleware: SPA, infra и POST /api/v1/auth/login."""
ABSOLUTE_PUBLIC_PATHS = (
"",
@ -7,6 +7,7 @@ ABSOLUTE_PUBLIC_PATHS = (
"/healthcheck2",
"/healthz",
"/readyz",
"/api/v1/auth/login",
)
START_PUBLIC_PREFIXES = (
@ -15,6 +16,7 @@ START_PUBLIC_PREFIXES = (
"/openapi.json",
"/docs",
"/docs-local",
"/api/v1/ws",
"/login",
"/task",
"/project",

42
api/src/core/security.py
View File

@ -1,6 +1,7 @@
from datetime import datetime, timedelta, timezone
from typing import Optional
import httpx
from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
from jose import JWTError, jwt
@ -9,6 +10,9 @@ from src.core.config import settings
pwd_context = PasswordHasher()
_KEYCLOAK_ALGORITHMS = ["RS256"]
_jwks_cache: dict | None = None
def verify_password(plain_password: str, hashed_password: Optional[str]) -> bool:
if not hashed_password:
@ -53,3 +57,41 @@ def verify_token(token: str) -> Optional[dict]:
)
except JWTError:
return None
def clear_jwks_cache() -> None:
global _jwks_cache
_jwks_cache = None
def _get_jwks(*, force_refresh: bool = False) -> dict:
global _jwks_cache
if _jwks_cache is None or force_refresh:
response = httpx.get(settings.JWKS_URL, timeout=10.0)
response.raise_for_status()
_jwks_cache = response.json()
return _jwks_cache
def decode_keycloak_token(token: str) -> Optional[dict]:
"""Проверяет Keycloak JWT по JWKS и возвращает payload."""
try:
header = jwt.get_unverified_header(token)
kid = header.get("kid")
jwks = _get_jwks()
keys = jwks.get("keys") or []
key_data = next((item for item in keys if item.get("kid") == kid), None)
if key_data is None:
jwks = _get_jwks(force_refresh=True)
keys = jwks.get("keys") or []
key_data = next((item for item in keys if item.get("kid") == kid), None)
if key_data is None:
return None
return jwt.decode(
token,
key_data,
algorithms=_KEYCLOAK_ALGORITHMS,
options={"verify_aud": False},
)
except (JWTError, httpx.HTTPError, ValueError, KeyError, StopIteration):
return None

5
api/src/domain/schemas.py
View File

@ -227,8 +227,9 @@ class Token(BaseModel):
class LoginRequest(BaseModel):
username: str
password: str
username: str = ""
password: str = ""
token: Optional[str] = None
class RefreshRequest(BaseModel):

23
api/src/services/auth_service.py
View File

@ -2,7 +2,12 @@ from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.security import create_access_token, create_refresh_token, verify_token
from src.core.security import (
create_access_token,
create_refresh_token,
decode_keycloak_token,
verify_token,
)
from src.domain.schemas import Token
from src.repository.user_repository import UserRepository
@ -36,6 +41,22 @@ class AuthService:
token_type="bearer",
)
async def authenticate_with_keycloak_token(self, token: str) -> Optional[Token]:
payload = decode_keycloak_token(token)
if payload is None:
return None
email = payload.get("email")
if not email:
return None
user = await self.user_repo.authenticate_via_email(email)
if not user or not user.is_active:
return None
return Token(
access_token=token,
refresh_token=None,
token_type="bearer",
)
async def refresh_token(self, refresh_token: str) -> Optional[Token]:
payload = verify_token(refresh_token)
if payload is None:

179
api/tests/unit/test_auth_api_prod_branch.py
View File

@ -3,6 +3,7 @@ import sys
import types
from unittest.mock import AsyncMock, MagicMock
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession
@ -69,30 +70,30 @@ def test_auth_login_non_debug_branch_success_and_unauthorized(monkeypatch):
"src.api.v1.auth.AuthService",
MagicMock(return_value=mock_auth_service),
)
m.setattr(
auth_module,
"decode_keycloak_token",
lambda _token: {"email": "prod-user@example.com"},
)
client = TestClient(app)
mock_auth_service.authenticate_user_via_email = AsyncMock(
mock_auth_service.authenticate_with_keycloak_token = AsyncMock(
return_value={
"access_token": "a",
"refresh_token": "r",
"access_token": "kc-token",
"token_type": "bearer",
}
)
response = client.post(
"/auth/login",
json={"username": "ignored", "password": "ignored"},
)
response = client.post("/auth/login", json={"token": "kc-token"})
assert response.status_code == 200
assert response.json()["access_token"] == "a"
mock_auth_service.authenticate_user_via_email.assert_awaited_once_with(
"prod-user@example.com"
assert response.json()["access_token"] == "kc-token"
mock_auth_service.authenticate_with_keycloak_token.assert_awaited_once_with(
"kc-token"
)
mock_auth_service.authenticate_user_via_email = AsyncMock(return_value=None)
response = client.post(
"/auth/login",
json={"username": "ignored", "password": "ignored"},
mock_auth_service.authenticate_with_keycloak_token = AsyncMock(
return_value=None
)
response = client.post("/auth/login", json={"token": "kc-token"})
assert response.status_code == 401
assert response.json()["message"] == (
"Пользователь не найден в локальной базе данных"
@ -101,52 +102,11 @@ def test_auth_login_non_debug_branch_success_and_unauthorized(monkeypatch):
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
def test_auth_login_form_non_debug_uses_keycloak_email(monkeypatch):
def test_auth_login_non_debug_requires_token(monkeypatch):
original_debug, previous, auth_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.auth"
)
try:
mock_auth_service = MagicMock()
mock_auth_service.authenticate_user = AsyncMock(return_value=None)
mock_auth_service.authenticate_user_via_email = AsyncMock(
return_value={
"access_token": "a",
"refresh_token": "r",
"token_type": "bearer",
}
)
app = FastAPI()
register_exception_handlers(app)
app.include_router(auth_module.router)
async def mock_get_db():
return AsyncMock(spec=AsyncSession)
app.dependency_overrides = {get_db: mock_get_db}
with monkeypatch.context() as m:
m.setattr(
"src.api.v1.auth.AuthService",
MagicMock(return_value=mock_auth_service),
)
client = TestClient(app)
response = client.post("/auth/login-form")
assert response.status_code == 200
mock_auth_service.authenticate_user_via_email.assert_awaited_once_with(
"prod-user@example.com"
)
mock_auth_service.authenticate_user.assert_not_awaited()
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
def test_auth_login_non_debug_rejects_missing_email(monkeypatch):
original_debug, previous, auth_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.auth", email=""
)
try:
app = FastAPI()
register_exception_handlers(app)
@ -159,10 +119,74 @@ def test_auth_login_non_debug_rejects_missing_email(monkeypatch):
client = TestClient(app)
response = client.post(
"/auth/login",
json={"username": "ignored", "password": "ignored"},
json={"username": "admin", "password": "admin123"},
)
assert response.status_code == 401
assert "email" in response.json()["message"]
assert "Keycloak" in response.json()["message"]
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
def test_auth_login_non_debug_rejects_missing_email(monkeypatch):
original_debug, previous, auth_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.auth"
)
try:
app = FastAPI()
register_exception_handlers(app)
app.include_router(auth_module.router)
async def mock_get_db():
return AsyncMock(spec=AsyncSession)
app.dependency_overrides = {get_db: mock_get_db}
with monkeypatch.context() as m:
m.setattr(auth_module, "decode_keycloak_token", lambda _token: {"sub": "x"})
client = TestClient(app)
response = client.post("/auth/login", json={"token": "kc-token"})
assert response.status_code == 401
assert "email" in response.json()["message"]
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
def test_auth_login_non_debug_rejects_invalid_token(monkeypatch):
original_debug, previous, auth_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.auth"
)
try:
app = FastAPI()
register_exception_handlers(app)
app.include_router(auth_module.router)
async def mock_get_db():
return AsyncMock(spec=AsyncSession)
app.dependency_overrides = {get_db: mock_get_db}
with monkeypatch.context() as m:
m.setattr(auth_module, "decode_keycloak_token", lambda _token: None)
client = TestClient(app)
response = client.post("/auth/login", json={"token": "bad"})
assert response.status_code == 401
assert response.json()["message"] == "Недействительный токен Keycloak"
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
def test_auth_login_form_not_available_in_prod(monkeypatch):
original_debug, previous, auth_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.auth"
)
try:
app = FastAPI()
register_exception_handlers(app)
app.include_router(auth_module.router)
client = TestClient(app)
response = client.post("/auth/login-form")
assert response.status_code == 404
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)
@ -230,3 +254,42 @@ def test_get_current_user_non_debug_rejects_inactive(monkeypatch):
assert response.json()["message"] == "Пользователь не найден"
finally:
_restore_module(monkeypatch, "src.api.v1.deps", original_debug, previous)
@pytest.mark.asyncio
async def test_resolve_user_from_access_token_prod_maps_email(monkeypatch):
original_debug, previous, deps_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.deps"
)
try:
db_user = MagicMock(email="prod-user@example.com", is_active=True)
mock_repo = MagicMock()
mock_repo.get_by_email = AsyncMock(return_value=db_user)
monkeypatch.setattr(
deps_module, "decode_keycloak_token", lambda _token: {"email": "prod-user@example.com"}
)
monkeypatch.setattr(deps_module, "UserRepository", MagicMock(return_value=mock_repo))
user = await deps_module.resolve_user_from_access_token(
"kc-token", AsyncMock(spec=AsyncSession)
)
assert user is db_user
mock_repo.get_by_email.assert_awaited_once_with("prod-user@example.com")
finally:
_restore_module(monkeypatch, "src.api.v1.deps", original_debug, previous)
@pytest.mark.asyncio
async def test_resolve_user_from_access_token_prod_rejects_hs256(monkeypatch):
original_debug, previous, deps_module = _reload_with_prod_auth(
monkeypatch, "src.api.v1.deps"
)
try:
monkeypatch.setattr(deps_module, "decode_keycloak_token", lambda _token: None)
user = await deps_module.resolve_user_from_access_token(
"local-hs256", AsyncMock(spec=AsyncSession)
)
assert user is None
finally:
_restore_module(monkeypatch, "src.api.v1.deps", original_debug, previous)

98
api/tests/unit/test_auth_service.py
View File

@ -1,11 +1,67 @@
import base64
from unittest.mock import AsyncMock, MagicMock
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from jose import jwt as jose_jwt
from src.core.security import create_refresh_token, verify_token
from src.core.security import (
clear_jwks_cache,
create_refresh_token,
decode_keycloak_token,
verify_token,
)
from src.services.auth_service import AuthService
def _b64u_int(value: int) -> str:
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
def test_decode_keycloak_token_rejects_garbage(monkeypatch):
clear_jwks_cache()
monkeypatch.setattr("src.core.security._get_jwks", lambda **_kwargs: {"keys": []})
assert decode_keycloak_token("not-a-jwt") is None
def test_decode_keycloak_token_accepts_valid_rs256(monkeypatch):
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
numbers = private_key.public_key().public_numbers()
kid = "test-key"
jwks = {
"keys": [
{
"kty": "RSA",
"kid": kid,
"use": "sig",
"alg": "RS256",
"n": _b64u_int(numbers.n),
"e": _b64u_int(numbers.e),
}
]
}
pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
token = jose_jwt.encode(
{"email": "admin@example.com", "sub": "admin"},
pem,
algorithm="RS256",
headers={"kid": kid},
)
clear_jwks_cache()
monkeypatch.setattr("src.core.security._get_jwks", lambda **_kwargs: jwks)
payload = decode_keycloak_token(token)
assert payload is not None
assert payload["email"] == "admin@example.com"
@pytest.mark.asyncio
async def test_authenticate_user_returns_token(monkeypatch):
db = AsyncMock()
@ -105,3 +161,43 @@ async def test_authenticate_user_via_email_rejects_inactive():
token = await service.authenticate_user_via_email("admin@example.com")
assert token is None
@pytest.mark.asyncio
async def test_authenticate_with_keycloak_token_passthrough(monkeypatch):
db = AsyncMock()
service = AuthService(db)
service.user_repo = MagicMock()
service.user_repo.authenticate_via_email = AsyncMock(
return_value=MagicMock(
username="admin", email="admin@example.com", is_active=True
)
)
monkeypatch.setattr(
"src.services.auth_service.decode_keycloak_token",
lambda _token: {"email": "admin@example.com"},
)
raw = "keycloak-access-token"
token = await service.authenticate_with_keycloak_token(raw)
assert token is not None
assert token.access_token == raw
assert token.refresh_token is None
service.user_repo.authenticate_via_email.assert_awaited_once_with(
"admin@example.com"
)
@pytest.mark.asyncio
async def test_authenticate_with_keycloak_token_rejects_invalid(monkeypatch):
db = AsyncMock()
service = AuthService(db)
monkeypatch.setattr(
"src.services.auth_service.decode_keycloak_token",
lambda _token: None,
)
token = await service.authenticate_with_keycloak_token("bad")
assert token is None

14
api/tests/unit/test_public_paths.py
View File

@ -5,16 +5,22 @@ def test_login_page_is_public_without_root():
specs = public_endpoint_specs("")
assert ("start", "/login") in specs
assert ("absolute", "/") in specs
assert ("absolute", "/api/v1/auth/login") in specs
def test_login_page_is_public_with_root_path():
specs = public_endpoint_specs("/aurora/apps/fastapi-tsygankov-test")
assert ("start", "/login") in specs
assert ("start", "/aurora/apps/fastapi-tsygankov-test/login") in specs
assert ("absolute", "/aurora/apps/fastapi-tsygankov-test/api/v1/auth/login") in specs
def test_api_routes_are_not_public():
def test_api_routes_except_login_and_ws_are_not_public():
specs = public_endpoint_specs("/aurora/apps/fastapi-tsygankov-test")
paths = [path for _, path in specs]
assert not any(path == "/api" or path.startswith("/api/") for path in paths)
assert not any("/api/v1" in path for path in paths)
api_paths = [path for _, path in specs if "/api/" in path]
assert api_paths == [
"/api/v1/auth/login",
"/aurora/apps/fastapi-tsygankov-test/api/v1/auth/login",
"/api/v1/ws",
"/aurora/apps/fastapi-tsygankov-test/api/v1/ws",
]