Dev HS256 включается только при AUTH_DEV_ISSUER и DEBUG одновременно; РАИСА — RS256/ES256 по публичному ключу или JWKS с кэшем. PyJWT[crypto] добавлен в requirements.txt как прод-зависимость. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
555 lines
23 KiB
Python
555 lines
23 KiB
Python
"""Ядро аутентификации: проверяющие токены и фабрика по настройкам (без БД, без внешней сети)."""
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import threading
|
||
from datetime import UTC, datetime, timedelta
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import jwt
|
||
import pytest
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import ec, rsa
|
||
from jwt.exceptions import PyJWKClientConnectionError
|
||
|
||
from src.core.auth import (
|
||
AuthError,
|
||
Claims,
|
||
DevHs256Verifier,
|
||
JwksVerifier,
|
||
TokenExpiredError,
|
||
TokenSignatureError,
|
||
build_verifier,
|
||
)
|
||
from src.core.config import Settings
|
||
|
||
DEV_SECRET = "dev-secret-for-unit-tests-0123456789abcdef"
|
||
OTHER_SECRET = "another-secret-for-unit-tests-0123456789"
|
||
EMAIL = "ka_rf@armka.local"
|
||
|
||
|
||
class _LogSpy:
|
||
"""Подмена logger модуля auth: собирает сообщения, не пишет в stderr."""
|
||
|
||
def __init__(self) -> None:
|
||
self.messages: list[str] = []
|
||
|
||
def warning(self, message: str, *args: Any) -> None:
|
||
self.messages.append(message.format(*args))
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def log_messages(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
||
"""Перехватывает логи модуля auth (шум не попадает в вывод тестов) и отдаёт их для проверок."""
|
||
spy = _LogSpy()
|
||
monkeypatch.setattr("src.core.auth.logger", spy)
|
||
return spy.messages
|
||
|
||
|
||
def _payload(**overrides: Any) -> dict[str, Any]:
|
||
now = datetime.now(UTC)
|
||
data: dict[str, Any] = {"sub": EMAIL, "email": EMAIL, "iat": now, "exp": now + timedelta(hours=1)}
|
||
data.update(overrides)
|
||
return {k: v for k, v in data.items() if v is not None}
|
||
|
||
|
||
def _pem_public(key: Any) -> str:
|
||
return (
|
||
key.public_key()
|
||
.public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
|
||
.decode()
|
||
)
|
||
|
||
|
||
def _pem_private(key: Any) -> str:
|
||
return key.private_bytes(
|
||
serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()
|
||
).decode()
|
||
|
||
|
||
def _b64(data: bytes) -> bytes:
|
||
return base64.urlsafe_b64encode(data).rstrip(b"=")
|
||
|
||
|
||
def _forge_hs(payload: dict[str, Any], secret: bytes, alg: str = "HS256") -> str:
|
||
"""Собирает HS-токен вручную (PyJWT не даст подписать HMAC асимметричным PEM — атаки подмены ключа)."""
|
||
header = _b64(json.dumps({"alg": alg, "typ": "JWT"}).encode())
|
||
body = _b64(json.dumps(payload, default=lambda v: int(v.timestamp())).encode())
|
||
digest = {"HS256": hashlib.sha256, "HS384": hashlib.sha384}[alg]
|
||
sig = _b64(hmac.new(secret, header + b"." + body, digest).digest())
|
||
return (header + b"." + body + b"." + sig).decode()
|
||
|
||
|
||
def _unsigned(payload: dict[str, Any]) -> str:
|
||
header = _b64(json.dumps({"alg": "none", "typ": "JWT"}).encode())
|
||
body = _b64(json.dumps(payload, default=lambda v: int(v.timestamp())).encode())
|
||
return (header + b"." + body + b".").decode()
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def rsa_key() -> rsa.RSAPrivateKey:
|
||
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def other_rsa_key() -> rsa.RSAPrivateKey:
|
||
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def ec_key() -> ec.EllipticCurvePrivateKey:
|
||
return ec.generate_private_key(ec.SECP256R1())
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def other_ec_key() -> ec.EllipticCurvePrivateKey:
|
||
return ec.generate_private_key(ec.SECP256R1())
|
||
|
||
|
||
# --- DevHs256Verifier -------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def dev() -> DevHs256Verifier:
|
||
return DevHs256Verifier(DEV_SECRET)
|
||
|
||
|
||
async def test_dev_valid_token(dev: DevHs256Verifier) -> None:
|
||
exp = datetime.now(UTC) + timedelta(hours=2)
|
||
token = jwt.encode(_payload(exp=exp), DEV_SECRET, algorithm="HS256")
|
||
claims = await dev.verify(token)
|
||
assert isinstance(claims, Claims)
|
||
assert claims.subject == EMAIL
|
||
assert claims.email == EMAIL
|
||
assert claims.exp == exp.replace(microsecond=0)
|
||
assert claims.exp.tzinfo is not None
|
||
|
||
|
||
async def test_dev_expired(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(minutes=5)), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(TokenExpiredError, match="срок действия"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_foreign_secret(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(), OTHER_SECRET, algorithm="HS256")
|
||
with pytest.raises(TokenSignatureError, match="подпись"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_no_email(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(email=None), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError, match="нет email"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_empty_email(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(email=" "), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError, match="нет email"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_no_exp(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(exp=None), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError, match="exp"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_no_sub(dev: DevHs256Verifier) -> None:
|
||
token = jwt.encode(_payload(sub=None), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError, match="sub"):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_alg_none(dev: DevHs256Verifier) -> None:
|
||
with pytest.raises(AuthError):
|
||
await dev.verify(_unsigned(_payload()))
|
||
|
||
|
||
async def test_dev_alg_hs384(dev: DevHs256Verifier) -> None:
|
||
"""Токен с корректной подписью, но другим алгоритмом (HS384) не принимается: алгоритм зафиксирован."""
|
||
with pytest.raises(AuthError, match="алгоритм"):
|
||
await dev.verify(_forge_hs(_payload(), DEV_SECRET.encode(), alg="HS384"))
|
||
|
||
|
||
@pytest.mark.parametrize("token", ["", "abc", "a.b.c", "не-токен"])
|
||
async def test_dev_garbage(dev: DevHs256Verifier, token: str) -> None:
|
||
with pytest.raises(AuthError):
|
||
await dev.verify(token)
|
||
|
||
|
||
async def test_dev_issue_token_roundtrip(dev: DevHs256Verifier) -> None:
|
||
before = datetime.now(UTC)
|
||
token = dev.issue_token(EMAIL)
|
||
claims = await dev.verify(token)
|
||
assert claims.subject == EMAIL
|
||
assert claims.email == EMAIL
|
||
assert timedelta(hours=8) - timedelta(seconds=5) <= claims.exp - before <= timedelta(hours=8, seconds=5)
|
||
header = jwt.get_unverified_header(token)
|
||
assert header["alg"] == "HS256"
|
||
assert "iat" in jwt.decode(token, options={"verify_signature": False})
|
||
|
||
|
||
async def test_dev_issue_token_custom_ttl_and_claim() -> None:
|
||
dev = DevHs256Verifier(DEV_SECRET, email_claim="mail")
|
||
token = dev.issue_token(EMAIL, ttl=timedelta(minutes=10))
|
||
assert jwt.decode(token, options={"verify_signature": False})["mail"] == EMAIL
|
||
claims = await dev.verify(token)
|
||
assert claims.email == EMAIL
|
||
assert claims.exp - datetime.now(UTC) < timedelta(minutes=11)
|
||
|
||
|
||
def test_dev_empty_secret_forbidden() -> None:
|
||
with pytest.raises(ValueError, match="секрет"):
|
||
DevHs256Verifier("")
|
||
|
||
|
||
# --- JwksVerifier (статический публичный ключ) -------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture(params=["rs256", "es256"])
|
||
def asym(
|
||
request: pytest.FixtureRequest,
|
||
rsa_key: rsa.RSAPrivateKey,
|
||
other_rsa_key: rsa.RSAPrivateKey,
|
||
ec_key: ec.EllipticCurvePrivateKey,
|
||
other_ec_key: ec.EllipticCurvePrivateKey,
|
||
) -> SimpleNamespace:
|
||
"""Пара алгоритм/ключи: подписывающий, публичный PEM для проверяющего и чужой подписывающий ключ."""
|
||
if request.param == "rs256":
|
||
return SimpleNamespace(
|
||
alg="RS256", priv=_pem_private(rsa_key), pub=_pem_public(rsa_key), other=_pem_private(other_rsa_key)
|
||
)
|
||
return SimpleNamespace(
|
||
alg="ES256", priv=_pem_private(ec_key), pub=_pem_public(ec_key), other=_pem_private(other_ec_key)
|
||
)
|
||
|
||
|
||
async def test_jwks_static_valid(asym: SimpleNamespace) -> None:
|
||
exp = datetime.now(UTC) + timedelta(hours=1)
|
||
token = jwt.encode(_payload(exp=exp), asym.priv, algorithm=asym.alg)
|
||
claims = await JwksVerifier(public_key=asym.pub).verify(token)
|
||
assert (claims.subject, claims.email) == (EMAIL, EMAIL)
|
||
assert claims.exp == exp.replace(microsecond=0)
|
||
|
||
|
||
async def test_jwks_static_pem_with_escaped_newlines(asym: SimpleNamespace) -> None:
|
||
"""В .env PEM удобно хранить одной строкой с литеральными \\n."""
|
||
token = jwt.encode(_payload(), asym.priv, algorithm=asym.alg)
|
||
claims = await JwksVerifier(public_key=asym.pub.replace("\n", "\\n")).verify(token)
|
||
assert claims.email == EMAIL
|
||
|
||
|
||
async def test_jwks_expired(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(minutes=5)), asym.priv, algorithm=asym.alg)
|
||
with pytest.raises(TokenExpiredError):
|
||
await JwksVerifier(public_key=asym.pub).verify(token)
|
||
|
||
|
||
async def test_jwks_leeway_tolerates_small_clock_skew(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(seconds=3)), asym.priv, algorithm=asym.alg)
|
||
claims = await JwksVerifier(public_key=asym.pub).verify(token)
|
||
assert claims.email == EMAIL
|
||
|
||
|
||
async def test_jwks_foreign_key(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(), asym.other, algorithm=asym.alg)
|
||
with pytest.raises(TokenSignatureError):
|
||
await JwksVerifier(public_key=asym.pub).verify(token)
|
||
|
||
|
||
async def test_jwks_no_exp(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(exp=None), asym.priv, algorithm=asym.alg)
|
||
with pytest.raises(AuthError, match="exp"):
|
||
await JwksVerifier(public_key=asym.pub).verify(token)
|
||
|
||
|
||
async def test_jwks_no_sub(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(sub=None), asym.priv, algorithm=asym.alg)
|
||
with pytest.raises(AuthError, match="sub"):
|
||
await JwksVerifier(public_key=asym.pub).verify(token)
|
||
|
||
|
||
async def test_jwks_issuer_checked_when_configured(asym: SimpleNamespace) -> None:
|
||
verifier = JwksVerifier(public_key=asym.pub, issuer="https://sso.raisa")
|
||
ok = jwt.encode(_payload(iss="https://sso.raisa"), asym.priv, algorithm=asym.alg)
|
||
assert (await verifier.verify(ok)).email == EMAIL
|
||
for bad in (_payload(iss="https://evil"), _payload()):
|
||
with pytest.raises(AuthError):
|
||
await verifier.verify(jwt.encode(bad, asym.priv, algorithm=asym.alg))
|
||
|
||
|
||
async def test_jwks_issuer_ignored_when_not_configured(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(iss="whatever"), asym.priv, algorithm=asym.alg)
|
||
assert (await JwksVerifier(public_key=asym.pub).verify(token)).email == EMAIL
|
||
|
||
|
||
async def test_jwks_audience_checked_when_configured(asym: SimpleNamespace) -> None:
|
||
verifier = JwksVerifier(public_key=asym.pub, audience="arm-core")
|
||
ok = jwt.encode(_payload(aud="arm-core"), asym.priv, algorithm=asym.alg)
|
||
assert (await verifier.verify(ok)).email == EMAIL
|
||
for bad in (_payload(aud="other"), _payload()):
|
||
with pytest.raises(AuthError):
|
||
await verifier.verify(jwt.encode(bad, asym.priv, algorithm=asym.alg))
|
||
|
||
|
||
async def test_jwks_audience_in_token_ignored_when_not_configured(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(aud="anything"), asym.priv, algorithm=asym.alg)
|
||
assert (await JwksVerifier(public_key=asym.pub).verify(token)).email == EMAIL
|
||
|
||
|
||
async def test_jwks_custom_email_claim(asym: SimpleNamespace) -> None:
|
||
verifier = JwksVerifier(public_key=asym.pub, email_claim="preferred_username")
|
||
token = jwt.encode(_payload(preferred_username="ivan@rshb.ru"), asym.priv, algorithm=asym.alg)
|
||
assert (await verifier.verify(token)).email == "ivan@rshb.ru"
|
||
# claim «email» при настроенном другом имени не подходит
|
||
with pytest.raises(AuthError, match="нет email"):
|
||
await verifier.verify(jwt.encode(_payload(), asym.priv, algorithm=asym.alg))
|
||
|
||
|
||
async def test_jwks_alg_none(asym: SimpleNamespace) -> None:
|
||
with pytest.raises(AuthError):
|
||
await JwksVerifier(public_key=asym.pub).verify(_unsigned(_payload()))
|
||
|
||
|
||
async def test_jwks_key_confusion_hs256_signed_with_public_key(asym: SimpleNamespace) -> None:
|
||
"""Атака подмены алгоритма: HS256, подписанный байтами публичного ключа как секретом, не принимается."""
|
||
for secret in (asym.pub.encode(), asym.pub.strip().encode()):
|
||
with pytest.raises(AuthError):
|
||
await JwksVerifier(public_key=asym.pub).verify(_forge_hs(_payload(), secret))
|
||
|
||
|
||
async def test_jwks_rejects_dev_token(asym: SimpleNamespace) -> None:
|
||
token = jwt.encode(_payload(), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError):
|
||
await JwksVerifier(public_key=asym.pub).verify(token)
|
||
|
||
|
||
async def test_jwks_rs256_key_does_not_verify_es256_token(
|
||
rsa_key: rsa.RSAPrivateKey, ec_key: ec.EllipticCurvePrivateKey
|
||
) -> None:
|
||
"""Алгоритм выводится из типа ключа: RSA-ключ принимает только RS256, EC-ключ — только ES256."""
|
||
es_token = jwt.encode(_payload(), _pem_private(ec_key), algorithm="ES256")
|
||
rs_token = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256")
|
||
with pytest.raises(AuthError):
|
||
await JwksVerifier(public_key=_pem_public(rsa_key)).verify(es_token)
|
||
with pytest.raises(AuthError):
|
||
await JwksVerifier(public_key=_pem_public(ec_key)).verify(rs_token)
|
||
|
||
|
||
def test_jwks_requires_a_key_source() -> None:
|
||
with pytest.raises(ValueError, match="AUTH_JWKS_URL|ключ"):
|
||
JwksVerifier()
|
||
|
||
|
||
def test_jwks_bad_pem() -> None:
|
||
with pytest.raises(ValueError, match="PEM"):
|
||
JwksVerifier(public_key="not a pem")
|
||
|
||
|
||
def test_jwks_rejects_both_sources(rsa_key: rsa.RSAPrivateKey) -> None:
|
||
with pytest.raises(ValueError, match="одновременно"):
|
||
JwksVerifier(public_key=_pem_public(rsa_key), jwks_url="http://127.0.0.1:1/jwks")
|
||
|
||
|
||
# --- JwksVerifier (JWKS: заглушка клиента и реальный PyJWKClient через локальный HTTP-сервер) --------------------
|
||
|
||
|
||
class _FakeJwkClient:
|
||
def __init__(self, key: Any = None, error: Exception | None = None) -> None:
|
||
self._key, self._error, self.calls = key, error, 0
|
||
|
||
def get_signing_key_from_jwt(self, token: str) -> Any:
|
||
self.calls += 1
|
||
if self._error is not None:
|
||
raise self._error
|
||
return SimpleNamespace(key=self._key)
|
||
|
||
|
||
async def test_jwks_client_valid(rsa_key: rsa.RSAPrivateKey) -> None:
|
||
client = _FakeJwkClient(key=rsa_key.public_key())
|
||
verifier = JwksVerifier(jwk_client=client)
|
||
token = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256")
|
||
assert (await verifier.verify(token)).email == EMAIL
|
||
assert client.calls == 1
|
||
|
||
|
||
async def test_jwks_client_foreign_key(rsa_key: rsa.RSAPrivateKey, other_rsa_key: rsa.RSAPrivateKey) -> None:
|
||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=rsa_key.public_key()))
|
||
token = jwt.encode(_payload(), _pem_private(other_rsa_key), algorithm="RS256")
|
||
with pytest.raises(TokenSignatureError):
|
||
await verifier.verify(token)
|
||
|
||
|
||
async def test_jwks_client_es256(ec_key: ec.EllipticCurvePrivateKey) -> None:
|
||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=ec_key.public_key()))
|
||
token = jwt.encode(_payload(), _pem_private(ec_key), algorithm="ES256")
|
||
assert (await verifier.verify(token)).email == EMAIL
|
||
|
||
|
||
async def test_jwks_client_key_confusion(rsa_key: rsa.RSAPrivateKey) -> None:
|
||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=rsa_key.public_key()))
|
||
with pytest.raises(AuthError):
|
||
await verifier.verify(_forge_hs(_payload(), _pem_public(rsa_key).encode()))
|
||
|
||
|
||
async def test_jwks_client_failure_is_auth_error_without_internals(log_messages: list[str]) -> None:
|
||
secret_detail = "connection to http://internal-host:8443/keys refused"
|
||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(error=PyJWKClientConnectionError(secret_detail)))
|
||
token = jwt.encode(_payload(), DEV_SECRET, algorithm="HS256")
|
||
with pytest.raises(AuthError) as exc:
|
||
await verifier.verify(token)
|
||
assert "internal-host" not in str(exc.value)
|
||
assert "ключ" in str(exc.value)
|
||
assert any("internal-host" in m for m in log_messages) # деталь — только в лог
|
||
|
||
|
||
class _JwksHandler(BaseHTTPRequestHandler):
|
||
body = b""
|
||
|
||
def do_GET(self) -> None:
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.end_headers()
|
||
self.wfile.write(self.body)
|
||
|
||
def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - сигнатура базового класса
|
||
return
|
||
|
||
|
||
async def test_jwks_real_pyjwkclient_over_loopback(rsa_key: rsa.RSAPrivateKey) -> None:
|
||
"""Реальный PyJWKClient: выбор ключа по kid из JWKS, отдаваемого локальным сервером (только loopback)."""
|
||
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(rsa_key.public_key()))
|
||
jwk.update(kid="key-1", use="sig", alg="RS256")
|
||
_JwksHandler.body = json.dumps({"keys": [jwk]}).encode()
|
||
server = HTTPServer(("127.0.0.1", 0), _JwksHandler)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
url = f"http://127.0.0.1:{server.server_port}/jwks"
|
||
verifier = JwksVerifier(jwks_url=url)
|
||
good = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256", headers={"kid": "key-1"})
|
||
assert (await verifier.verify(good)).email == EMAIL
|
||
unknown_kid = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256", headers={"kid": "nope"})
|
||
with pytest.raises(AuthError, match="ключ"):
|
||
await verifier.verify(unknown_kid)
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
|
||
|
||
# --- Фабрика --------------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _settings(monkeypatch: pytest.MonkeyPatch, **auth: Any) -> Settings:
|
||
"""Settings без env-файла и без AUTH_*/DEBUG из окружения процесса; обязательные OPENBAO__* — заглушки."""
|
||
for key in list(os.environ):
|
||
if key.startswith(("OPENBAO__", "AUTH_")) or key == "DEBUG":
|
||
monkeypatch.delenv(key)
|
||
required = {
|
||
"OPENBAO__DATABASE__DATABASE_URL": "postgresql+asyncpg://u:p@localhost:5432/d",
|
||
"OPENBAO__MINIO__MINIO_URL": "localhost:9000",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "login",
|
||
"OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "password",
|
||
"OPENBAO__MINIO__MINIO_BUCKET": "bucket",
|
||
"OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
|
||
"OPENBAO__SMB__SMB_HOST": "host",
|
||
"OPENBAO__SMB__SMB_PATH": "path",
|
||
"OPENBAO__SMB__SMB_LOGIN": "login",
|
||
"OPENBAO__SMB__SMB_PASSWORD": "password",
|
||
}
|
||
for key, value in required.items():
|
||
monkeypatch.setenv(key, value)
|
||
return Settings(_env_file=None, **auth) # type: ignore[call-arg]
|
||
|
||
|
||
def test_factory_dev_off_and_no_jwks_gives_none(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
s = _settings(monkeypatch, AUTH_DEV_ISSUER=False, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True)
|
||
assert build_verifier(s) is None
|
||
|
||
|
||
def test_factory_dev_flag_without_debug_gives_no_dev(monkeypatch: pytest.MonkeyPatch, log_messages: list[str]) -> None:
|
||
s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False)
|
||
assert build_verifier(s) is None
|
||
assert any("AUTH_DEV_ISSUER=true проигнорирован" in m for m in log_messages)
|
||
|
||
|
||
def test_factory_dev_flag_without_debug_falls_back_to_jwks(
|
||
monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey
|
||
) -> None:
|
||
s = _settings(
|
||
monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False, AUTH_PUBLIC_KEY=_pem_public(rsa_key)
|
||
)
|
||
verifier = build_verifier(s)
|
||
assert isinstance(verifier, JwksVerifier)
|
||
assert not isinstance(verifier, DevHs256Verifier)
|
||
|
||
|
||
def test_factory_both_flags_gives_dev(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True, AUTH_EMAIL_CLAIM="mail")
|
||
verifier = build_verifier(s)
|
||
assert isinstance(verifier, DevHs256Verifier)
|
||
token = verifier.issue_token(EMAIL)
|
||
assert jwt.decode(token, options={"verify_signature": False})["mail"] == EMAIL
|
||
|
||
|
||
def test_factory_dev_wins_over_jwks_when_both_flags(
|
||
monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey
|
||
) -> None:
|
||
s = _settings(
|
||
monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True, AUTH_PUBLIC_KEY=_pem_public(rsa_key)
|
||
)
|
||
assert isinstance(build_verifier(s), DevHs256Verifier)
|
||
|
||
|
||
def test_factory_dev_flags_without_secret_is_config_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, DEBUG=True)
|
||
with pytest.raises(RuntimeError, match="AUTH_DEV_SECRET"):
|
||
build_verifier(s)
|
||
|
||
|
||
def test_factory_public_key_gives_jwks(monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey) -> None:
|
||
s = _settings(monkeypatch, AUTH_PUBLIC_KEY=_pem_public(rsa_key))
|
||
assert isinstance(build_verifier(s), JwksVerifier)
|
||
|
||
|
||
def test_factory_jwks_url_gives_jwks(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
s = _settings(monkeypatch, AUTH_JWKS_URL="https://sso.example/.well-known/jwks.json")
|
||
assert isinstance(build_verifier(s), JwksVerifier)
|
||
|
||
|
||
async def test_factory_passes_issuer_audience_and_email_claim(
|
||
monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey
|
||
) -> None:
|
||
s = _settings(
|
||
monkeypatch,
|
||
AUTH_PUBLIC_KEY=_pem_public(rsa_key),
|
||
AUTH_ISSUER="https://sso.raisa",
|
||
AUTH_AUDIENCE="arm-core",
|
||
AUTH_EMAIL_CLAIM="mail",
|
||
)
|
||
verifier = build_verifier(s)
|
||
assert verifier is not None
|
||
priv = _pem_private(rsa_key)
|
||
ok = jwt.encode(_payload(iss="https://sso.raisa", aud="arm-core", mail=EMAIL), priv, algorithm="RS256")
|
||
assert (await verifier.verify(ok)).email == EMAIL
|
||
with pytest.raises(AuthError):
|
||
await verifier.verify(jwt.encode(_payload(mail=EMAIL), priv, algorithm="RS256"))
|
||
|
||
|
||
async def test_dev_token_rejected_when_debug_off(monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey) -> None:
|
||
"""Токен, подписанный dev-секретом, не принимается ни пустой конфигурацией, ни JWKS-проверяющим (DEBUG=false)."""
|
||
token = DevHs256Verifier(DEV_SECRET).issue_token(EMAIL)
|
||
prod = _settings(
|
||
monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False, AUTH_PUBLIC_KEY=_pem_public(rsa_key)
|
||
)
|
||
verifier = build_verifier(prod)
|
||
assert verifier is not None
|
||
with pytest.raises(AuthError):
|
||
await verifier.verify(token)
|