api/src/api/v1/auth.py

@@ -2,11 +2,7 @@ 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 (
+from src.core.security import decode_keycloak_token
-    decode_keycloak_token,
-    email_from_keycloak_payload,
-    normalize_bearer_token,
-)
 from src.db.session import get_db
 from src.domain.schemas import LoginRequest, RefreshRequest, Token
 from src.services.auth_service import AuthService
@@ -66,28 +62,27 @@ else:
         login_data: LoginRequest,
         db: AsyncSession = Depends(get_db),
     ):
-        raw_token = normalize_bearer_token(login_data.token)
+        if not login_data.token:
-        if not raw_token:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail=_PROD_TOKEN_REQUIRED,
                 headers={"WWW-Authenticate": "Bearer"},
             )
-        payload = decode_keycloak_token(raw_token)
+        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 email_from_keycloak_payload(payload):
+        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(raw_token)
+        token = await auth_service.authenticate_with_keycloak_token(login_data.token)
         if not token:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,

api/src/api/v1/deps.py

@@ -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 decode_keycloak_token, email_from_keycloak_payload, 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
@@ -58,7 +58,9 @@ async def resolve_user_from_access_token(
         except HTTPException:
             return None
     payload = decode_keycloak_token(token)
-    email = email_from_keycloak_payload(payload)
+    if payload is None:
+        return None
+    email = payload.get("email")
     if not email:
         return None
     user_repo = UserRepository(db)

api/src/core/config.py

@@ -78,11 +78,7 @@ class Settings(BaseSettings):
     JWKS_URL: str = Field(
         default="https://keycloak.raisa.go.rshbank.ru/realms/datalab/protocol/openid-connect/certs",
         description="URL для получения JWKS",
-        validation_alias=AliasChoices(
+        alias="JWKS_URL",
-            "JWKS_URL",
-            "OPENBAO__AUTH__JWKS_URL",
-            f"{prefix}__JWKS_URL",
-        ),
     )
     APP_NAMESPACE: str = Field(
         default="dfip",

api/src/core/security.py

@@ -1,23 +1,16 @@
 from datetime import datetime, timedelta, timezone
-import logging
 from typing import Optional
 
 import httpx
 from argon2 import PasswordHasher
 from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
-from jose import JOSEError, JWTError, jwt
+from jose import JWTError, jwt
 
 from src.core.config import settings
 
 pwd_context = PasswordHasher()
-logger = logging.getLogger(__name__)
 
 _KEYCLOAK_ALGORITHMS = ["RS256"]
-_KEYCLOAK_DECODE_OPTIONS = {
-    "verify_aud": False,
-    "verify_at_hash": False,
-    "leeway": 60,
-}
 _jwks_cache: dict | None = None
 
 
@@ -71,25 +64,6 @@ def clear_jwks_cache() -> None:
     _jwks_cache = None
 
 
-def normalize_bearer_token(token: str | None) -> str | None:
-    if not token:
-        return None
-    value = token.strip()
-    if value.lower().startswith("bearer "):
-        value = value[7:].strip()
-    return value or None
-
-
-def email_from_keycloak_payload(payload: dict | None) -> str | None:
-    if not payload:
-        return None
-    for key in ("email", "preferred_username", "upn"):
-        value = payload.get(key)
-        if isinstance(value, str) and "@" in value:
-            return value
-    return None
-
-
 def _get_jwks(*, force_refresh: bool = False) -> dict:
     global _jwks_cache
     if _jwks_cache is None or force_refresh:
@@ -99,26 +73,25 @@ def _get_jwks(*, force_refresh: bool = False) -> dict:
     return _jwks_cache
 
 
-def _decode_with_jwks(token: str, jwks: dict) -> dict:
-    return jwt.decode(
-        token,
-        jwks,
-        algorithms=_KEYCLOAK_ALGORITHMS,
-        options=_KEYCLOAK_DECODE_OPTIONS,
-    )
-
-
 def decode_keycloak_token(token: str) -> Optional[dict]:
     """Проверяет Keycloak JWT по JWKS и возвращает payload."""
-    raw = normalize_bearer_token(token)
-    if raw is None:
-        return None
     try:
-        return _decode_with_jwks(raw, _get_jwks())
+        header = jwt.get_unverified_header(token)
-    except (JOSEError, httpx.HTTPError, ValueError, TypeError, KeyError) as exc:
+        kid = header.get("kid")
-        logger.warning("Keycloak JWT rejected: %s", exc)
+        jwks = _get_jwks()
-        try:
+        keys = jwks.get("keys") or []
-            return _decode_with_jwks(raw, _get_jwks(force_refresh=True))
+        key_data = next((item for item in keys if item.get("kid") == kid), None)
-        except (JOSEError, httpx.HTTPError, ValueError, TypeError, KeyError) as retry_exc:
+        if key_data is None:
-            logger.warning("Keycloak JWT rejected after JWKS refresh: %s", retry_exc)
+            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

api/src/services/auth_service.py

@@ -6,7 +6,6 @@ from src.core.security import (
     create_access_token,
     create_refresh_token,
     decode_keycloak_token,
-    email_from_keycloak_payload,
     verify_token,
 )
 from src.domain.schemas import Token
@@ -44,7 +43,9 @@ class AuthService:
 
     async def authenticate_with_keycloak_token(self, token: str) -> Optional[Token]:
         payload = decode_keycloak_token(token)
-        email = email_from_keycloak_payload(payload)
+        if payload is None:
+            return None
+        email = payload.get("email")
         if not email:
             return None
         user = await self.user_repo.authenticate_via_email(email)

api/tests/unit/test_auth_service.py

@@ -62,48 +62,6 @@ def test_decode_keycloak_token_accepts_valid_rs256(monkeypatch):
     assert payload["email"] == "admin@example.com"
 
 
-def test_decode_keycloak_token_strips_bearer_and_ignores_at_hash(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(
-        {
-            "preferred_username": "admin@example.com",
-            "at_hash": "not-a-real-hash",
-            "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(f"Bearer {token}")
-
-    assert payload is not None
-    from src.core.security import email_from_keycloak_payload
-
-    assert email_from_keycloak_payload(payload) == "admin@example.com"
-
-
 @pytest.mark.asyncio
 async def test_authenticate_user_returns_token(monkeypatch):
     db = AsyncMock()