Dmitry
600496048f
get_current_user навешивается на роутер целиком, публичными остаются только health и auth (список закреплён tests/api/test_route_auth.py). API не стартует с плейсхолдером или коротким JWT_SECRET без ALLOW_DEV_SECRET=1. Лимитер логинов ограничен по числу ключей. Клиентский IP берётся из последнего хопа X-Forwarded-For.
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
import pytest
|
|
|
|
|
|
async def test_login_refresh_logout_cycle(client, user):
|
|
r = await client.post("/api/v1/auth/login", json=user)
|
|
assert r.status_code == 200, r.text
|
|
pair = r.json()
|
|
assert pair["token_type"] == "bearer"
|
|
|
|
me = await client.get(
|
|
"/api/v1/auth/me", headers={"Authorization": f"Bearer {pair['access_token']}"}
|
|
)
|
|
assert me.status_code == 200
|
|
assert me.json()["email"] == user["email"]
|
|
|
|
# rotation: the refresh token is single-use
|
|
r2 = await client.post("/api/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]})
|
|
assert r2.status_code == 200
|
|
r3 = await client.post("/api/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]})
|
|
assert r3.status_code == 401
|
|
assert r3.headers["content-type"].startswith("application/problem+json")
|
|
|
|
new_refresh = r2.json()["refresh_token"]
|
|
assert (
|
|
await client.post("/api/v1/auth/logout", json={"refresh_token": new_refresh})
|
|
).status_code == 204
|
|
assert (
|
|
await client.post("/api/v1/auth/refresh", json={"refresh_token": new_refresh})
|
|
).status_code == 401
|
|
|
|
|
|
async def test_wrong_password_and_rate_limit(client, user):
|
|
bad = {"email": user["email"], "password": "nope"}
|
|
for _ in range(5):
|
|
r = await client.post("/api/v1/auth/login", json=bad)
|
|
assert r.status_code == 401
|
|
r = await client.post("/api/v1/auth/login", json=bad)
|
|
assert r.status_code == 429
|
|
assert "retry-after" in r.headers
|
|
|
|
|
|
async def test_protected_routes_need_token(client):
|
|
r = await client.get("/api/v1/auth/me")
|
|
assert r.status_code == 401
|
|
r = await client.get("/api/v1/auth/me", headers={"Authorization": "Bearer garbage"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_client_ip_trusts_only_the_last_forwarded_hop():
|
|
from starlette.requests import Request
|
|
|
|
from fintracker.api.deps import client_ip
|
|
|
|
def req(xff: str) -> Request:
|
|
return Request({"type": "http", "headers": [(b"x-forwarded-for", xff.encode())]})
|
|
|
|
assert client_ip(req("1.2.3.4")) == "1.2.3.4"
|
|
assert client_ip(req("6.6.6.6, 1.2.3.4")) == "1.2.3.4"
|
|
|
|
|
|
def test_limiter_evicts_instead_of_growing_without_bound():
|
|
from fintracker.api.ratelimit import SlidingWindowLimiter
|
|
|
|
limiter = SlidingWindowLimiter(limit=1, max_keys=3)
|
|
for i in range(10):
|
|
assert limiter.hit(f"ip-{i}") is None
|
|
assert len(limiter._hits) <= 3
|
|
assert limiter.hit("ip-9") is not None # the freshest key keeps its state
|
|
|
|
|
|
async def test_weak_jwt_secret_refuses_to_start(monkeypatch):
|
|
from fintracker.api.app import _lifespan
|
|
from fintracker.config import get_settings
|
|
|
|
monkeypatch.setenv("JWT_SECRET", "change-me")
|
|
monkeypatch.delenv("ALLOW_DEV_SECRET", raising=False)
|
|
get_settings.cache_clear()
|
|
try:
|
|
with pytest.raises(RuntimeError, match="JWT_SECRET"):
|
|
async with _lifespan(None): # type: ignore[arg-type]
|
|
pass
|
|
finally:
|
|
monkeypatch.undo()
|
|
get_settings.cache_clear()
|