This commit is contained in:
Raykov-MS 2026-09-03 11:24:56 +03:00
parent ae6640796d
commit 919440d62f
4 changed files with 48 additions and 12 deletions

View File

@ -3,8 +3,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from src.core.security import (
decode_keycloak_token,
email_from_keycloak_payload,
inspect_keycloak_token,
normalize_bearer_token,
)
from src.db.session import get_db
@ -73,11 +73,14 @@ else:
detail=_PROD_TOKEN_REQUIRED,
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_keycloak_token(raw_token)
payload, reason = inspect_keycloak_token(raw_token)
if payload is None:
detail = _PROD_TOKEN_INVALID
if reason:
detail = f"{_PROD_TOKEN_INVALID}: {reason}"
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=_PROD_TOKEN_INVALID,
detail=detail,
headers={"WWW-Authenticate": "Bearer"},
)
if not email_from_keycloak_payload(payload):

View File

@ -110,15 +110,33 @@ def _decode_with_jwks(token: str, jwks: dict) -> dict:
def decode_keycloak_token(token: str) -> Optional[dict]:
"""Проверяет Keycloak JWT по JWKS и возвращает payload."""
payload, _reason = inspect_keycloak_token(token)
return payload
def inspect_keycloak_token(token: str) -> tuple[Optional[dict], str | None]:
"""Как decode_keycloak_token, плюс текст причины отказа (для отладки логина)."""
raw = normalize_bearer_token(token)
if raw is None:
return None
return None, "empty token"
alg = None
kid = None
try:
return _decode_with_jwks(raw, _get_jwks())
header = jwt.get_unverified_header(raw)
alg = header.get("alg")
kid = header.get("kid")
except Exception as exc:
return None, f"malformed jwt: {exc}"
try:
return _decode_with_jwks(raw, _get_jwks()), None
except (JOSEError, httpx.HTTPError, ValueError, TypeError, KeyError) as exc:
logger.warning("Keycloak JWT rejected: %s", exc)
try:
return _decode_with_jwks(raw, _get_jwks(force_refresh=True))
return _decode_with_jwks(raw, _get_jwks(force_refresh=True)), None
except (JOSEError, httpx.HTTPError, ValueError, TypeError, KeyError) as retry_exc:
logger.warning("Keycloak JWT rejected after JWKS refresh: %s", retry_exc)
return None
reason = (
f"{retry_exc.__class__.__name__}: {retry_exc}; "
f"alg={alg}; kid={kid}; jwks_url={settings.JWKS_URL}"
)
return None, reason

View File

@ -72,8 +72,8 @@ def test_auth_login_non_debug_branch_success_and_unauthorized(monkeypatch):
)
m.setattr(
auth_module,
"decode_keycloak_token",
lambda _token: {"email": "prod-user@example.com"},
"inspect_keycloak_token",
lambda _token: ({"email": "prod-user@example.com"}, None),
)
client = TestClient(app)
@ -142,7 +142,11 @@ def test_auth_login_non_debug_rejects_missing_email(monkeypatch):
app.dependency_overrides = {get_db: mock_get_db}
with monkeypatch.context() as m:
m.setattr(auth_module, "decode_keycloak_token", lambda _token: {"sub": "x"})
m.setattr(
auth_module,
"inspect_keycloak_token",
lambda _token: ({"sub": "x"}, None),
)
client = TestClient(app)
response = client.post("/auth/login", json={"token": "kc-token"})
assert response.status_code == 401
@ -166,11 +170,18 @@ def test_auth_login_non_debug_rejects_invalid_token(monkeypatch):
app.dependency_overrides = {get_db: mock_get_db}
with monkeypatch.context() as m:
m.setattr(auth_module, "decode_keycloak_token", lambda _token: None)
m.setattr(
auth_module,
"inspect_keycloak_token",
lambda _token: (None, "Signature verification failed; alg=RS256"),
)
client = TestClient(app)
response = client.post("/auth/login", json={"token": "bad"})
assert response.status_code == 401
assert response.json()["message"] == "Недействительный токен Keycloak"
message = response.json()["message"]
assert message.startswith("Недействительный токен Keycloak")
assert "Signature verification failed" in message
assert "alg=RS256" in message
finally:
_restore_module(monkeypatch, "src.api.v1.auth", original_debug, previous)

View File

@ -10,6 +10,7 @@ from src.core.security import (
clear_jwks_cache,
create_refresh_token,
decode_keycloak_token,
inspect_keycloak_token,
verify_token,
)
from src.services.auth_service import AuthService
@ -24,6 +25,9 @@ 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
_payload, reason = inspect_keycloak_token("not-a-jwt")
assert _payload is None
assert reason
def test_decode_keycloak_token_accepts_valid_rs256(monkeypatch):