293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""Tests for authentication router (/api/auth)."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from src.backend.tests.conftest import _create_user, auth_headers
|
|
|
|
|
|
class TestLogin:
|
|
"""POST /api/auth/login tests."""
|
|
|
|
async def test_login_success(self, client: AsyncClient, db_session):
|
|
"""Successful login returns user data and API key."""
|
|
user = await _create_user(
|
|
db_session, username="loginuser", password="SecurePass1"
|
|
)
|
|
resp = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "loginuser", "password": "SecurePass1"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["user"]["username"] == "loginuser"
|
|
assert "api_key" in data
|
|
assert len(data["api_key"]) > 0
|
|
|
|
async def test_login_wrong_password(self, client: AsyncClient, db_session):
|
|
"""Login with wrong password returns 401."""
|
|
await _create_user(
|
|
db_session, username="wrongpw", password="CorrectPass1"
|
|
)
|
|
resp = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "wrongpw", "password": "WrongPass1"},
|
|
)
|
|
assert resp.status_code == 401
|
|
assert "Invalid" in resp.json()["detail"]
|
|
|
|
async def test_login_unknown_user(self, client: AsyncClient):
|
|
"""Login with non-existent username returns 401."""
|
|
resp = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "nonexistent", "password": "Anything1"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
async def test_login_inactive_user(self, client: AsyncClient, db_session):
|
|
"""Login with inactive user returns 401."""
|
|
await _create_user(
|
|
db_session,
|
|
username="inactive_user",
|
|
password="InactivePass1",
|
|
active=False,
|
|
)
|
|
resp = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "inactive_user", "password": "InactivePass1"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
async def test_login_missing_fields(self, client: AsyncClient):
|
|
"""Login with missing fields returns 422."""
|
|
resp = await client.post("/api/auth/login", json={"username": "only"})
|
|
assert resp.status_code == 422
|
|
|
|
async def test_login_returns_api_key(self, client: AsyncClient, db_session):
|
|
"""Login response contains a valid api_key string."""
|
|
await _create_user(
|
|
db_session, username="apikey_user", password="KeyPass123"
|
|
)
|
|
resp = await client.post(
|
|
"/api/auth/login",
|
|
json={"username": "apikey_user", "password": "KeyPass123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
api_key = resp.json()["api_key"]
|
|
# API key should be a non-empty string (base64url, ~64 chars)
|
|
assert isinstance(api_key, str)
|
|
assert len(api_key) >= 32
|
|
|
|
|
|
class TestLoginKeyPerSession:
|
|
"""Each login issues its own key; none revokes another (#25, #41)."""
|
|
|
|
async def test_two_logins_same_user_get_distinct_valid_keys(
|
|
self, client: AsyncClient, db_session
|
|
):
|
|
await _create_user(db_session, username="twostations", password="TwoStations1")
|
|
creds = {"username": "twostations", "password": "TwoStations1"}
|
|
resp1 = await client.post("/api/auth/login", json=creds)
|
|
resp2 = await client.post("/api/auth/login", json=creds)
|
|
k1, k2 = resp1.json()["api_key"], resp2.json()["api_key"]
|
|
assert k1 != k2
|
|
for k in (k1, k2):
|
|
resp = await client.get("/api/auth/me", headers={"X-API-Key": k})
|
|
assert resp.status_code == 200
|
|
|
|
async def test_db_stores_hash_never_clear_key(
|
|
self, client: AsyncClient, db_session
|
|
):
|
|
from sqlalchemy import select
|
|
from src.backend.models.orm.api_key import ApiKey
|
|
from src.backend.services.auth_service import hash_api_key
|
|
|
|
await _create_user(db_session, username="hashed", password="HashedPass1")
|
|
resp = await client.post(
|
|
"/api/auth/login", json={"username": "hashed", "password": "HashedPass1"}
|
|
)
|
|
key = resp.json()["api_key"]
|
|
hashes = (await db_session.execute(select(ApiKey.key_hash))).scalars().all()
|
|
assert hash_api_key(key) in hashes
|
|
assert key not in hashes
|
|
assert all(len(h) == 64 for h in hashes)
|
|
|
|
async def test_expired_key_gives_401(self, client: AsyncClient, db_session):
|
|
from datetime import timedelta
|
|
from sqlalchemy import update
|
|
from src.backend.models.orm.api_key import ApiKey
|
|
from src.backend.services.auth_service import utcnow
|
|
|
|
user = await _create_user(db_session, username="expired", password="ExpiredPass1")
|
|
await db_session.execute(
|
|
update(ApiKey).values(expires_at=utcnow() - timedelta(minutes=1))
|
|
)
|
|
resp = await client.get("/api/auth/me", headers=auth_headers(user))
|
|
assert resp.status_code == 401
|
|
assert "expired" in resp.json()["detail"].lower()
|
|
|
|
async def test_use_renews_expiry_only_after_interval(
|
|
self, client: AsyncClient, db_session
|
|
):
|
|
from datetime import timedelta
|
|
from sqlalchemy import select, update
|
|
from src.backend.models.orm.api_key import ApiKey
|
|
from src.backend.services.auth_service import utcnow
|
|
|
|
user = await _create_user(db_session, username="renew", password="RenewPass1")
|
|
headers = auth_headers(user)
|
|
old = utcnow() - timedelta(hours=10)
|
|
|
|
# Recent use: no write, expiry untouched.
|
|
recent = utcnow() - timedelta(minutes=1)
|
|
await db_session.execute(
|
|
update(ApiKey).values(expires_at=old + timedelta(hours=24), last_used_at=recent)
|
|
)
|
|
await client.get("/api/auth/me", headers=headers)
|
|
row = (await db_session.execute(select(ApiKey))).scalar_one()
|
|
await db_session.refresh(row)
|
|
assert row.last_used_at == recent
|
|
|
|
# Stale last use: expiry is pushed forward.
|
|
await db_session.execute(
|
|
update(ApiKey).values(last_used_at=utcnow() - timedelta(minutes=10))
|
|
)
|
|
before = utcnow()
|
|
resp = await client.get("/api/auth/me", headers=headers)
|
|
assert resp.status_code == 200
|
|
await db_session.refresh(row)
|
|
assert row.expires_at >= before + timedelta(hours=23, minutes=59)
|
|
|
|
|
|
class TestLogout:
|
|
"""Logout deletes only this session's key (#41)."""
|
|
|
|
async def test_logout_removes_only_own_key(self, client: AsyncClient, db_session):
|
|
await _create_user(db_session, username="shared_logout", password="SharedPass1")
|
|
creds = {"username": "shared_logout", "password": "SharedPass1"}
|
|
k1 = (await client.post("/api/auth/login", json=creds)).json()["api_key"]
|
|
k2 = (await client.post("/api/auth/login", json=creds)).json()["api_key"]
|
|
resp = await client.post("/api/auth/logout", headers={"X-API-Key": k1})
|
|
assert resp.status_code == 204
|
|
|
|
assert (await client.get("/api/auth/me", headers={"X-API-Key": k1})).status_code == 401
|
|
assert (await client.get("/api/auth/me", headers={"X-API-Key": k2})).status_code == 200
|
|
|
|
async def test_logout_all_sessions_removes_every_key(
|
|
self, client: AsyncClient, db_session
|
|
):
|
|
await _create_user(db_session, username="shared_logout_all", password="SharedPass1")
|
|
creds = {"username": "shared_logout_all", "password": "SharedPass1"}
|
|
k1 = (await client.post("/api/auth/login", json=creds)).json()["api_key"]
|
|
k2 = (await client.post("/api/auth/login", json=creds)).json()["api_key"]
|
|
resp = await client.post(
|
|
"/api/auth/logout", params={"all_sessions": "true"},
|
|
headers={"X-API-Key": k1},
|
|
)
|
|
assert resp.status_code == 204
|
|
for k in (k1, k2):
|
|
assert (await client.get("/api/auth/me", headers={"X-API-Key": k})).status_code == 401
|
|
|
|
|
|
class TestAuthenticateUserTiming:
|
|
"""authenticate_user pays the bcrypt cost even for an unknown username (#25)."""
|
|
|
|
async def test_unknown_username_still_returns_none(self, db_session):
|
|
from src.backend.services.auth_service import authenticate_user
|
|
|
|
result = await authenticate_user(db_session, "no-such-user-at-all", "whatever")
|
|
assert result is None
|
|
|
|
def test_dummy_hash_is_a_valid_bcrypt_hash(self):
|
|
from src.backend.services.auth_service import (
|
|
_DUMMY_PASSWORD_HASH,
|
|
verify_password,
|
|
)
|
|
|
|
assert verify_password(
|
|
"tiemeasureflow-dummy-hash-for-timing", _DUMMY_PASSWORD_HASH
|
|
)
|
|
|
|
|
|
class TestHealthAndAuth:
|
|
"""Health check and auth requirement tests."""
|
|
|
|
async def test_health_check_no_auth(self, client: AsyncClient):
|
|
"""Health check endpoint works without authentication."""
|
|
resp = await client.get("/api/health")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "ok"
|
|
assert "version" in data
|
|
|
|
async def test_endpoint_requires_api_key(self, client: AsyncClient):
|
|
"""Protected endpoint returns 401 without API key."""
|
|
resp = await client.get("/api/users")
|
|
assert resp.status_code == 401
|
|
assert "API key" in resp.json()["detail"] or "Missing" in resp.json()["detail"]
|
|
|
|
|
|
class TestRevokeAllKeys:
|
|
"""Password change, deactivation and regenerate invalidate every key (#41)."""
|
|
|
|
async def _two_keys(self, client, db_session, name):
|
|
user = await _create_user(db_session, username=name, password="KeysPass123")
|
|
creds = {"username": name, "password": "KeysPass123"}
|
|
keys = [
|
|
(await client.post("/api/auth/login", json=creds)).json()["api_key"]
|
|
for _ in range(2)
|
|
]
|
|
return user, keys
|
|
|
|
async def _all_401(self, client, keys):
|
|
for k in keys:
|
|
resp = await client.get("/api/auth/me", headers={"X-API-Key": k})
|
|
assert resp.status_code == 401
|
|
|
|
async def test_password_change_invalidates_all(self, client, db_session, admin_user):
|
|
user, keys = await self._two_keys(client, db_session, "pwkeys")
|
|
resp = await client.put(
|
|
f"/api/users/{user.id}/password",
|
|
headers=auth_headers(admin_user), json={"password": "NewSecurePass1"},
|
|
)
|
|
assert resp.status_code == 200
|
|
await self._all_401(client, keys)
|
|
|
|
async def test_deactivation_invalidates_all(self, client, db_session, admin_user):
|
|
user, keys = await self._two_keys(client, db_session, "offkeys")
|
|
resp = await client.delete(
|
|
f"/api/users/{user.id}", headers=auth_headers(admin_user)
|
|
)
|
|
assert resp.status_code == 204
|
|
await self._all_401(client, keys)
|
|
|
|
async def test_regenerate_invalidates_old_keys(self, client, db_session, admin_user):
|
|
user, keys = await self._two_keys(client, db_session, "regenkeys")
|
|
resp = await client.post(
|
|
f"/api/users/{user.id}/regenerate-key", headers=auth_headers(admin_user)
|
|
)
|
|
assert resp.status_code == 200
|
|
await self._all_401(client, keys)
|
|
new = resp.json()["api_key"]
|
|
assert (await client.get("/api/auth/me", headers={"X-API-Key": new})).status_code == 200
|
|
|
|
|
|
class TestExpiredKeyCleanup:
|
|
async def test_expired_key_removed_at_next_login(self, client, db_session):
|
|
from datetime import timedelta
|
|
from sqlalchemy import select, update
|
|
from src.backend.models.orm.api_key import ApiKey
|
|
from src.backend.services.auth_service import hash_api_key, utcnow
|
|
|
|
user = await _create_user(db_session, username="cleanup", password="CleanPass123")
|
|
old_hash = hash_api_key(user.plain_key)
|
|
await db_session.execute(
|
|
update(ApiKey).values(expires_at=utcnow() - timedelta(minutes=1))
|
|
)
|
|
resp = await client.post(
|
|
"/api/auth/login", json={"username": "cleanup", "password": "CleanPass123"}
|
|
)
|
|
assert resp.status_code == 200
|
|
hashes = (await db_session.execute(select(ApiKey.key_hash))).scalars().all()
|
|
assert old_hash not in hashes
|
|
assert len(hashes) == 1
|