484 lines
20 KiB
Python
484 lines
20 KiB
Python
# tests/test_auth_api.py — Integration tests for auth router endpoints
|
||
import os
|
||
import pytest
|
||
|
||
# The existing test-vault/ and data/users.json already exist
|
||
# with admin:chab30 credentials
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_client():
|
||
"""Create a TestClient with auth enabled, isolated temp data."""
|
||
import tempfile
|
||
import shutil
|
||
from pathlib import Path
|
||
|
||
tmp = Path(tempfile.mkdtemp())
|
||
data_dir = tmp / "data"
|
||
data_dir.mkdir()
|
||
|
||
# Fresh users.json with admin:chab30
|
||
import json
|
||
from backend.auth.password import hash_password
|
||
pw_hash = hash_password("chab30")
|
||
users = {
|
||
"version": 1,
|
||
"users": {
|
||
"admin": {
|
||
"id": "admin-1",
|
||
"username": "admin",
|
||
"display_name": "admin",
|
||
"password_hash": pw_hash,
|
||
"role": "admin",
|
||
"vaults": ["*"],
|
||
"active": True,
|
||
"created_at": "2026-01-01T00:00:00",
|
||
}
|
||
}
|
||
}
|
||
(data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
|
||
|
||
# Copy secret.key if it exists
|
||
src_secret = Path("data/secret.key")
|
||
if src_secret.exists():
|
||
shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
|
||
|
||
# Save original data dir
|
||
orig_cwd = os.getcwd()
|
||
test_vault_path = os.path.abspath("test-vault")
|
||
os.chdir(str(tmp))
|
||
|
||
os.environ["VAULT_1_NAME"] = "TestVault"
|
||
os.environ["VAULT_1_PATH"] = test_vault_path
|
||
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
|
||
os.environ["OBSIGATE_ADMIN_USER"] = "admin"
|
||
os.environ["OBSIGATE_ADMIN_PASSWORD"] = "chab30"
|
||
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
|
||
|
||
import backend.main
|
||
backend.main._load_config = lambda: {"watcher_enabled": False}
|
||
|
||
from backend.main import app
|
||
from backend.indexer import build_index, index
|
||
import asyncio
|
||
|
||
for key in list(index.keys()):
|
||
del index[key]
|
||
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
loop.run_until_complete(build_index())
|
||
|
||
from backend.search import init_inverted_index
|
||
init_inverted_index()
|
||
|
||
from fastapi.testclient import TestClient
|
||
client = TestClient(app)
|
||
yield client
|
||
# Properly close the test client to flush gzip writers
|
||
# before the temp directory is removed
|
||
if hasattr(client, 'close'):
|
||
client.close()
|
||
# Give the event loop a chance to clean up pending callbacks
|
||
loop.run_until_complete(asyncio.sleep(0))
|
||
|
||
os.chdir(orig_cwd)
|
||
shutil.rmtree(str(tmp), ignore_errors=True)
|
||
for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
|
||
"OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED"]:
|
||
os.environ.pop(k, None)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Auth Status
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestAuthStatus:
|
||
def test_auth_enabled(self, auth_client):
|
||
resp = auth_client.get("/api/auth/status")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["auth_enabled"] is True
|
||
|
||
def test_health_works_without_auth(self, auth_client):
|
||
"""Health endpoint should be public"""
|
||
resp = auth_client.get("/api/health")
|
||
assert resp.status_code == 200
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Login
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestLogin:
|
||
def test_login_success(self, auth_client):
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "chab30",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "access_token" in data
|
||
assert data["token_type"] == "bearer"
|
||
assert data["user"]["username"] == "admin"
|
||
assert data["user"]["role"] == "admin"
|
||
|
||
def test_login_wrong_password(self, auth_client):
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "wrongpass",
|
||
})
|
||
assert resp.status_code == 401
|
||
|
||
def test_login_unknown_user(self, auth_client):
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "nobody",
|
||
"password": "pass123",
|
||
})
|
||
assert resp.status_code == 401
|
||
|
||
def test_locked_account_returns_401_not_429(self, auth_client, monkeypatch):
|
||
"""BUG-039: a locked account must be indistinguishable from an unknown one."""
|
||
import backend.auth.router as auth_router
|
||
|
||
monkeypatch.setattr(auth_router, "is_locked", lambda username: True)
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "chab30",
|
||
})
|
||
assert resp.status_code == 401
|
||
assert "verrouill" not in resp.json()["detail"].lower()
|
||
|
||
def test_account_rate_limited_returns_401(self, auth_client, monkeypatch):
|
||
"""BUG-039: per-account throttling must not reveal the account exists."""
|
||
import backend.auth.router as auth_router
|
||
|
||
monkeypatch.setattr(auth_router, "is_account_rate_limited", lambda username: True)
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "chab30",
|
||
})
|
||
assert resp.status_code == 401
|
||
|
||
def test_inactive_account_returns_401(self, auth_client, monkeypatch):
|
||
"""BUG-039: a disabled account answers like an unknown user."""
|
||
import backend.auth.router as auth_router
|
||
|
||
monkeypatch.setattr(auth_router, "get_user", lambda username: {
|
||
"username": username, "active": False, "password_hash": "x",
|
||
})
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "chab30",
|
||
})
|
||
assert resp.status_code == 401
|
||
|
||
def test_login_remember_me(self, auth_client):
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin",
|
||
"password": "chab30",
|
||
"remember_me": True,
|
||
})
|
||
assert resp.status_code == 200
|
||
assert "access_token" in resp.json()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Authenticated requests
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestAuthenticated:
|
||
def _login(self, client):
|
||
resp = client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
return resp.json()["access_token"]
|
||
|
||
def test_vaults_with_auth(self, auth_client):
|
||
token = self._login(auth_client)
|
||
resp = auth_client.get("/api/vaults", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data) >= 1
|
||
|
||
def test_me_endpoint(self, auth_client):
|
||
token = self._login(auth_client)
|
||
resp = auth_client.get("/api/auth/me", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["username"] == "admin"
|
||
|
||
def test_unauthorized_request(self, auth_client):
|
||
resp = auth_client.get("/api/vaults")
|
||
assert resp.status_code in (401, 403)
|
||
|
||
def test_invalid_token(self, auth_client):
|
||
resp = auth_client.get("/api/vaults", headers={
|
||
"Authorization": "Bearer invalidtoken123",
|
||
})
|
||
assert resp.status_code in (401, 403)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Admin endpoints
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestAdmin:
|
||
def _login_admin(self, client):
|
||
resp = client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
return resp.json()["access_token"]
|
||
|
||
def test_list_users(self, auth_client):
|
||
token = self._login_admin(auth_client)
|
||
resp = auth_client.get("/api/auth/admin/users", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data) >= 1
|
||
|
||
def test_create_user(self, auth_client):
|
||
token = self._login_admin(auth_client)
|
||
resp = auth_client.post("/api/auth/admin/users", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
}, json={
|
||
"username": "testuser",
|
||
"password": "testpass",
|
||
"role": "user",
|
||
"vaults": ["TestVault"],
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["username"] == "testuser"
|
||
|
||
def test_patch_user(self, auth_client):
|
||
"""PATCH /api/auth/admin/users/{username} updates fields."""
|
||
token = self._login_admin(auth_client)
|
||
# Create a user first, then patch it.
|
||
auth_client.post("/api/auth/admin/users", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
}, json={
|
||
"username": "patchuser",
|
||
"password": "origpass",
|
||
"role": "user",
|
||
"vaults": ["TestVault"],
|
||
})
|
||
resp = auth_client.patch("/api/auth/admin/users/patchuser", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
}, json={
|
||
"display_name": "Patched User",
|
||
"active": False,
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["display_name"] == "Patched User"
|
||
assert data["active"] is False
|
||
|
||
def test_delete_user(self, auth_client):
|
||
"""DELETE /api/auth/admin/users/{username} removes the user."""
|
||
token = self._login_admin(auth_client)
|
||
# Create a throwaway user, then delete it.
|
||
auth_client.post("/api/auth/admin/users", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
}, json={
|
||
"username": "deleteuser",
|
||
"password": "delpass123",
|
||
"role": "user",
|
||
"vaults": ["TestVault"],
|
||
})
|
||
resp = auth_client.delete("/api/auth/admin/users/deleteuser", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
assert resp.status_code == 200
|
||
# Confirm it's gone via list.
|
||
list_resp = auth_client.get("/api/auth/admin/users", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
usernames = [u["username"] for u in list_resp.json()]
|
||
assert "deleteuser" not in usernames
|
||
|
||
def test_logout(self, auth_client):
|
||
token = self._login_admin(auth_client)
|
||
resp = auth_client.post("/api/auth/logout", headers={
|
||
"Authorization": f"Bearer {token}",
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Profile avatar (#113)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
# 1×1 transparent PNG.
|
||
TINY_PNG = (
|
||
"data:image/png;base64,"
|
||
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8"
|
||
"z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
|
||
)
|
||
|
||
|
||
class TestAvatar:
|
||
def _login(self, client):
|
||
resp = client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
return resp.json()["access_token"]
|
||
|
||
def _headers(self, token):
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
def test_me_returns_avatar_field(self, auth_client):
|
||
"""GET /api/auth/me exposes the avatar key (None by default)."""
|
||
token = self._login(auth_client)
|
||
resp = auth_client.get("/api/auth/me", headers=self._headers(token))
|
||
assert resp.status_code == 200
|
||
assert "avatar" in resp.json()
|
||
assert resp.json()["avatar"] is None
|
||
|
||
def test_patch_me_sets_avatar(self, auth_client):
|
||
"""A valid PNG data-URL is persisted and returned by GET /me."""
|
||
token = self._login(auth_client)
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": TINY_PNG})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["avatar"] == TINY_PNG
|
||
me = auth_client.get("/api/auth/me", headers=self._headers(token)).json()
|
||
assert me["avatar"] == TINY_PNG
|
||
|
||
def test_patch_me_avatar_empty_string_clears(self, auth_client):
|
||
"""Sending an empty string removes the avatar."""
|
||
token = self._login(auth_client)
|
||
auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": TINY_PNG})
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": ""})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["avatar"] is None
|
||
|
||
def test_patch_me_avatar_rejects_svg(self, auth_client):
|
||
"""SVG data-URLs are refused (XSS surface)."""
|
||
token = self._login(auth_client)
|
||
svg = "data:image/svg+xml;base64,PHN2Zz48L3N2Zz4="
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": svg})
|
||
assert resp.status_code == 400
|
||
|
||
def test_patch_me_avatar_rejects_non_image_payload(self, auth_client):
|
||
"""A PNG data-URL header over non-image bytes is refused."""
|
||
token = self._login(auth_client)
|
||
fake = "data:image/png;base64,aGVsbG8gd29ybGQ=" # "hello world"
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": fake})
|
||
assert resp.status_code == 400
|
||
|
||
def test_patch_me_avatar_rejects_too_large(self, auth_client):
|
||
"""Payloads above the size cap are refused before any decode."""
|
||
token = self._login(auth_client)
|
||
huge = "data:image/png;base64," + "A" * 500_000
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": huge})
|
||
assert resp.status_code == 400
|
||
|
||
def test_patch_me_avatar_rejects_invalid_base64(self, auth_client):
|
||
"""Malformed base64 inside a well-formed data-URL is refused."""
|
||
token = self._login(auth_client)
|
||
resp = auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": "data:image/png;base64,@@not-base64@@"})
|
||
assert resp.status_code == 400
|
||
|
||
def test_login_user_payload_includes_avatar(self, auth_client):
|
||
"""The login response user object carries the avatar for the sidebar."""
|
||
token = self._login(auth_client)
|
||
auth_client.patch("/api/auth/me", headers=self._headers(token),
|
||
json={"avatar": TINY_PNG})
|
||
# Re-login to get a fresh user payload.
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["user"]["avatar"] == TINY_PNG
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# Secure cookies (#87 T8)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestSecureCookies:
|
||
"""`Secure` auto par défaut : https → flag, http → pas de flag
|
||
(les navigateurs jettent les cookies Secure sur http)."""
|
||
|
||
@staticmethod
|
||
def _req(scheme="http", forwarded_proto=None):
|
||
from types import SimpleNamespace
|
||
headers = {}
|
||
if forwarded_proto is not None:
|
||
headers["x-forwarded-proto"] = forwarded_proto
|
||
return SimpleNamespace(
|
||
url=SimpleNamespace(scheme=scheme),
|
||
headers=headers,
|
||
)
|
||
|
||
def test_forced_true(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.setenv("OBSIGATE_SECURE_COOKIES", "true")
|
||
assert is_secure_cookies(self._req("http")) is True
|
||
|
||
def test_forced_false(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.setenv("OBSIGATE_SECURE_COOKIES", "false")
|
||
assert is_secure_cookies(self._req("https")) is False
|
||
|
||
def test_auto_http(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
assert is_secure_cookies(self._req("http")) is False
|
||
|
||
def test_auto_https(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
assert is_secure_cookies(self._req("https")) is True
|
||
|
||
def test_auto_forwarded_proto_trusted(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
monkeypatch.setenv("OBSIGATE_TRUST_PROXY", "true")
|
||
assert is_secure_cookies(self._req("http", "https")) is True
|
||
|
||
def test_auto_forwarded_proto_untrusted(self, monkeypatch):
|
||
from backend.auth.router import is_secure_cookies
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
monkeypatch.delenv("OBSIGATE_TRUST_PROXY", raising=False)
|
||
assert is_secure_cookies(self._req("http", "https")) is False
|
||
|
||
def test_login_http_sets_cookie_without_secure(self, auth_client, monkeypatch):
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
resp = auth_client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
assert resp.status_code == 200
|
||
set_cookie = resp.headers.get("set-cookie", "")
|
||
assert "access_token" in set_cookie
|
||
assert "secure" not in set_cookie.lower()
|
||
|
||
def test_login_https_sets_secure_cookie(self, auth_client, monkeypatch):
|
||
"""Même app servie en https → flag Secure présent."""
|
||
from backend.main import app
|
||
from fastapi.testclient import TestClient
|
||
monkeypatch.delenv("OBSIGATE_SECURE_COOKIES", raising=False)
|
||
https_client = TestClient(app, base_url="https://testserver",
|
||
raise_server_exceptions=False)
|
||
try:
|
||
resp = https_client.post("/api/auth/login", json={
|
||
"username": "admin", "password": "chab30",
|
||
})
|
||
assert resp.status_code == 200
|
||
assert "secure" in resp.headers.get("set-cookie", "").lower()
|
||
finally:
|
||
if hasattr(https_client, "close"):
|
||
https_client.close() |