P0 security batch: admin-only user mgmt, unified role model, login rate limiting, webhook secret, security headers, pagination caps
- Remove POST /api/auth/register (404); no sign-up UI; users are admin-managed - Add admin-only POST/PATCH/DELETE /api/auth/users (forced canonical roles, self-lockout + reference guards) - Unify role model in app/core/roles.py; reject unknown roles at creation and at login/JWT validation; startup normalizes unambiguous legacy aliases - Login rate limiting ~5 fails/15 min per IP+email -> 429 (in-process, tunable) - WhatsApp webhook requires X-Webhook-Secret; fail-closed when env unset; GET handshake uses constant-time verify token (403 on mismatch) - GET /api/whatsapp/mock-log now requires auth - Security headers middleware: X-Frame-Options DENY, nosniff, CSP on HTML, HSTS behind TLS - Pagination: limit alias for page_size, hard cap enforced, both -> 422
This commit is contained in:
root
+11
-1
@@ -18,16 +18,26 @@ os.environ.setdefault("SECRET_KEY", "test-secret-key-not-for-production-01234567
|
||||
os.environ.setdefault("CORS_ORIGINS", "http://test")
|
||||
|
||||
import pytest_asyncio # noqa: E402 (DATABASE_URL must be set before app imports)
|
||||
import pytest # noqa: E402
|
||||
from httpx import ASGITransport, AsyncClient # noqa: E402
|
||||
|
||||
from app.core.database import Base, async_session_factory, engine # noqa: E402
|
||||
from app.core.ratelimit import login_rate_limiter # noqa: E402
|
||||
from app.main import app # noqa: E402
|
||||
from app.models.ticket import Ticket # noqa: E402
|
||||
from app.services.seed import seed_categories, seed_units, seed_users # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def _reset_login_rate_limiter():
|
||||
"""Isolate login rate-limit state between tests (shared in-process store)."""
|
||||
login_rate_limiter.reset()
|
||||
yield
|
||||
login_rate_limiter.reset()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client():
|
||||
async def client(_reset_login_rate_limiter):
|
||||
"""Async test client with a fresh, seeded database per test."""
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
@@ -0,0 +1,540 @@
|
||||
"""P0 security batch — admin user management, unified role model, login rate
|
||||
limiting, WhatsApp webhook secret, mock-log auth, security headers, pagination.
|
||||
|
||||
Each item in the P0 hardening batch has an executable behavioral test here
|
||||
(plus the register-removal tests living in test_p0_hardening.py).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.database import async_session_factory
|
||||
from app.core.security import hash_password
|
||||
from app.models.user import User
|
||||
from app.services.seed import normalize_legacy_user_roles
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
# ── helpers ───────────────────────────────────────────────────────────
|
||||
async def _login(client, email="wahab@denya.com", password="denya123") -> str:
|
||||
resp = await client.post("/api/auth/login", json={"email": email, "password": password})
|
||||
assert resp.status_code == 200, resp.text
|
||||
return resp.json()["access_token"]
|
||||
|
||||
|
||||
def _auth(token: str) -> dict[str, str]:
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
async def _insert_user(email: str, role: str, *, active: bool = True) -> int:
|
||||
"""Insert a user row directly (bypasses API role validation) — used to
|
||||
simulate legacy bootstrap/registration rows in the DB."""
|
||||
async with async_session_factory() as session:
|
||||
user = User(
|
||||
email=email,
|
||||
password_hash=hash_password("denya123"),
|
||||
full_name=f"Legacy {email}",
|
||||
role=role,
|
||||
active=active,
|
||||
)
|
||||
session.add(user)
|
||||
await session.commit()
|
||||
return user.id
|
||||
|
||||
|
||||
async def _normalize_roles() -> None:
|
||||
async with async_session_factory() as session:
|
||||
await normalize_legacy_user_roles(session)
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def _create_ticket(client, token: str, **overrides) -> dict:
|
||||
payload = {
|
||||
"unit_id": 2,
|
||||
"category_id": 3,
|
||||
"priority": "medium",
|
||||
"reporter": "P0 Batch Test",
|
||||
"reported_via": "walk-in",
|
||||
"description": "p0 batch ticket",
|
||||
**overrides,
|
||||
}
|
||||
resp = await client.post("/api/tickets", json=payload, headers=_auth(token))
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()
|
||||
|
||||
|
||||
# ── Item 2/3: admin user management ───────────────────────────────────
|
||||
async def test_create_user_requires_admin(client: AsyncClient):
|
||||
token = await _login(client, email="bella@denya.com") # CS Rep
|
||||
resp = await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
async def test_create_user_requires_auth(client: AsyncClient):
|
||||
resp = await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
async def test_admin_creates_user_with_forced_role(client: AsyncClient):
|
||||
token = await _login(client) # Admin/Wahab
|
||||
resp = await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": "New.Tech@Example.com", "password": "password1", "full_name": "New Tech", "role": "Tech"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
created = resp.json()
|
||||
assert created["role"] == "Tech"
|
||||
assert created["active"] is True
|
||||
assert created["email"] == "new.tech@example.com" # normalized lower-case
|
||||
|
||||
# The new user can actually log in with their forced role.
|
||||
login = await client.post(
|
||||
"/api/auth/login", json={"email": "new.tech@example.com", "password": "password1"}
|
||||
)
|
||||
assert login.status_code == 200
|
||||
me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
|
||||
assert me.json()["role"] == "Tech"
|
||||
|
||||
|
||||
async def test_admin_create_user_rejects_unknown_roles(client: AsyncClient):
|
||||
"""Unified role model: junk/legacy roles cannot be minted at creation."""
|
||||
token = await _login(client)
|
||||
for role in ("admin", "superadmin", "technician", "root"):
|
||||
resp = await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": f"{role.strip().lower()}@example.com", "password": "password1", "full_name": "X", "role": role},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 422, (role, resp.text)
|
||||
|
||||
|
||||
async def test_admin_create_user_duplicate_email_conflict(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
payload = {"email": "dupe2@example.com", "password": "password1", "full_name": "D", "role": "CS Rep"}
|
||||
assert (await client.post("/api/auth/users", json=payload, headers=_auth(token))).status_code == 201
|
||||
resp = await client.post("/api/auth/users", json=payload, headers=_auth(token))
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
async def test_patch_user_role_change(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
users = (await client.get("/api/auth/users", headers=_auth(token))).json()
|
||||
tech = next(u for u in users if u["role"] == "Tech")
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{tech['id']}",
|
||||
json={"role": "CS Rep"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["role"] == "CS Rep"
|
||||
assert resp.json()["active"] is True
|
||||
|
||||
|
||||
async def test_patch_user_rejects_unknown_role(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
users = (await client.get("/api/auth/users", headers=_auth(token))).json()
|
||||
tech = next(u for u in users if u["role"] == "Tech")
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{tech['id']}",
|
||||
json={"role": "superadmin"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 422, resp.text
|
||||
|
||||
|
||||
async def test_patch_deactivate_blocks_login(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
users = (await client.get("/api/auth/users", headers=_auth(token))).json()
|
||||
tech = next(u for u in users if u["role"] == "Tech")
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{tech['id']}",
|
||||
json={"active": False},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["active"] is False
|
||||
|
||||
login = await client.post(
|
||||
"/api/auth/login", json={"email": tech["email"], "password": "denya123"}
|
||||
)
|
||||
assert login.status_code == 401
|
||||
|
||||
|
||||
async def test_admin_cannot_modify_own_account(client: AsyncClient):
|
||||
token = await _login(client) # wahab
|
||||
me = (await client.get("/api/auth/me", headers=_auth(token))).json()
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{me['id']}", json={"active": False}, headers=_auth(token)
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
async def test_admin_can_demote_other_admin_but_not_self(client: AsyncClient):
|
||||
"""An admin can manage the other admin seat, but self-removal stays blocked,
|
||||
so at least one canonical admin always remains (structural invariant)."""
|
||||
token = await _login(client) # wahab
|
||||
users = (await client.get("/api/auth/users", headers=_auth(token))).json()
|
||||
jerome = next(u for u in users if u["role"] == "Admin/Jerome")
|
||||
wahab = next(u for u in users if u["role"] == "Admin/Wahab")
|
||||
|
||||
# Demote the OTHER admin → allowed, wahab is still the acting admin.
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{jerome['id']}", json={"role": "CEO"}, headers=_auth(token)
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# Demoting/deactivating yourself is always rejected.
|
||||
resp = await client.patch(
|
||||
f"/api/auth/users/{wahab['id']}", json={"active": False}, headers=_auth(token)
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
async def test_delete_user_admin_only_and_works(client: AsyncClient):
|
||||
admin_token = await _login(client)
|
||||
users = (await client.get("/api/auth/users", headers=_auth(admin_token))).json()
|
||||
tech = next(u for u in users if u["role"] == "Tech")
|
||||
|
||||
# Non-admin cannot delete.
|
||||
cs_token = await _login(client, email="bella@denya.com")
|
||||
resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(cs_token))
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Admin deletes → 204, user gone, login fails.
|
||||
resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(admin_token))
|
||||
assert resp.status_code == 204
|
||||
login = await client.post(
|
||||
"/api/auth/login", json={"email": tech["email"], "password": "denya123"}
|
||||
)
|
||||
assert login.status_code == 401
|
||||
|
||||
|
||||
async def test_delete_user_referenced_by_ticket_is_409(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
users = (await client.get("/api/auth/users", headers=_auth(token))).json()
|
||||
tech = next(u for u in users if u["role"] == "Tech")
|
||||
ticket = await _create_ticket(client, token)
|
||||
resp = await client.patch(
|
||||
f"/api/tickets/{ticket['id']}",
|
||||
json={"assigned_to": tech["id"]},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(token))
|
||||
assert resp.status_code == 409
|
||||
assert "related" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
async def test_admin_cannot_delete_self(client: AsyncClient):
|
||||
token = await _login(client)
|
||||
me = (await client.get("/api/auth/me", headers=_auth(token))).json()
|
||||
resp = await client.delete(f"/api/auth/users/{me['id']}", headers=_auth(token))
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
# ── Item 3: unified role model — legacy rows & JWT validation ─────────
|
||||
async def test_legacy_alias_role_normalized_at_startup(client: AsyncClient):
|
||||
"""A legacy 'technician' row is mapped onto canonical 'Tech' at startup."""
|
||||
await _insert_user("legacy-tech@example.com", "technician")
|
||||
await _normalize_roles() # what lifespan does each boot
|
||||
|
||||
login = await client.post(
|
||||
"/api/auth/login", json={"email": "legacy-tech@example.com", "password": "denya123"}
|
||||
)
|
||||
assert login.status_code == 200, login.text
|
||||
me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
|
||||
assert me.json()["role"] == "Tech"
|
||||
|
||||
|
||||
async def test_login_rejects_unknown_legacy_role_fail_closed(client: AsyncClient):
|
||||
"""Lowercase 'superadmin' rows (mintable by the old open register) cannot
|
||||
log in — fail closed, never granted admin powers."""
|
||||
await _insert_user("legacy-admin@example.com", "superadmin")
|
||||
# NOTE: no normalize call — the row is exactly what the live DB holds today.
|
||||
|
||||
login = await client.post(
|
||||
"/api/auth/login", json={"email": "legacy-admin@example.com", "password": "denya123"}
|
||||
)
|
||||
assert login.status_code == 401
|
||||
assert "role" in login.json()["detail"].lower()
|
||||
|
||||
|
||||
async def test_jwt_validation_rejects_unknown_role(client: AsyncClient):
|
||||
"""A token for a user whose row later becomes junk-role must fail closed."""
|
||||
token = await _login(client) # wahab is a canonical admin at token time
|
||||
me = (await client.get("/api/auth/me", headers=_auth(token))).json()
|
||||
|
||||
# Simulate a legacy DB row flip to a non-canonical role.
|
||||
async with async_session_factory() as session:
|
||||
from sqlalchemy import select
|
||||
user = (await session.execute(select(User).where(User.id == me["id"]))).scalar_one()
|
||||
user.role = "admin"
|
||||
await session.commit()
|
||||
|
||||
resp = await client.get("/api/auth/me", headers=_auth(token))
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
async def test_normalize_does_not_map_ambiguous_admin_alias(client: AsyncClient):
|
||||
"""normalize_legacy_user_roles leaves identity-ambiguous 'admin' rows for
|
||||
operator remediation instead of guessing a canonical admin."""
|
||||
await _insert_user("legacy-admin2@example.com", "admin")
|
||||
await _normalize_roles()
|
||||
|
||||
async with async_session_factory() as session:
|
||||
from sqlalchemy import select
|
||||
user = (
|
||||
await session.execute(select(User).where(User.email == "legacy-admin2@example.com"))
|
||||
).scalar_one()
|
||||
assert user.role == "admin"
|
||||
|
||||
|
||||
async def test_admin_only_rbac_gate(client: AsyncClient):
|
||||
"""Canonical admins pass /api/auth/admin-only; everyone else 403."""
|
||||
wahab = await _login(client)
|
||||
assert (await client.get("/api/auth/admin-only", headers=_auth(wahab))).status_code == 200
|
||||
|
||||
bella = await _login(client, email="bella@denya.com")
|
||||
assert (await client.get("/api/auth/admin-only", headers=_auth(bella))).status_code == 403
|
||||
|
||||
|
||||
# ── Item 4: login rate limiting ───────────────────────────────────────
|
||||
async def test_login_rate_limited_after_five_failures(client: AsyncClient):
|
||||
email, password = "rate-limited@example.com", "denya123"
|
||||
# Make sure the account exists with a valid password.
|
||||
token = await _login(client)
|
||||
await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": email, "password": password, "full_name": "Rate", "role": "CS Rep"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
|
||||
for _ in range(5):
|
||||
resp = await client.post(
|
||||
"/api/auth/login", json={"email": email, "password": "wrong-password"}
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
# 6th attempt — even with the CORRECT password — is throttled.
|
||||
resp = await client.post(
|
||||
"/api/auth/login", json={"email": email, "password": password}
|
||||
)
|
||||
assert resp.status_code == 429, resp.text
|
||||
|
||||
|
||||
async def test_rate_limit_is_per_email(client: AsyncClient):
|
||||
"""Failures for one account never lock out another account."""
|
||||
token = await _login(client)
|
||||
for email in ("victim@example.com", "other@example.com"):
|
||||
await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": email, "password": "password1", "full_name": "U", "role": "CS Rep"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
|
||||
for _ in range(6):
|
||||
resp = await client.post(
|
||||
"/api/auth/login", json={"email": "victim@example.com", "password": "bad"}
|
||||
)
|
||||
assert resp.status_code in (401, 429)
|
||||
|
||||
# Unaffected account still logs in fine.
|
||||
resp = await client.post(
|
||||
"/api/auth/login", json={"email": "other@example.com", "password": "password1"}
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
async def test_rate_limit_window_expires(client: AsyncClient, monkeypatch):
|
||||
"""After the 15-minute window passes, the account can log in again."""
|
||||
import app.core.ratelimit as ratelimit_mod
|
||||
import time as _time
|
||||
|
||||
email, password = "window@example.com", "password1"
|
||||
token = await _login(client)
|
||||
await client.post(
|
||||
"/api/auth/users",
|
||||
json={"email": email, "password": password, "full_name": "W", "role": "CS Rep"},
|
||||
headers=_auth(token),
|
||||
)
|
||||
|
||||
# Pin the limiter clock so the window can be fast-forwarded deterministically.
|
||||
clock = {"now": _time.time()}
|
||||
monkeypatch.setattr(ratelimit_mod, "_now", lambda: clock["now"])
|
||||
for _ in range(5):
|
||||
await client.post("/api/auth/login", json={"email": email, "password": "bad"})
|
||||
|
||||
blocked = await client.post("/api/auth/login", json={"email": email, "password": password})
|
||||
assert blocked.status_code == 429, blocked.text
|
||||
|
||||
clock["now"] += settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS + 1
|
||||
resp = await client.post("/api/auth/login", json={"email": email, "password": password})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
# ── Item 5: WhatsApp webhook secret (fail closed) ─────────────────────
|
||||
WEBHOOK_BODY = {
|
||||
"object": "whatsapp_business_account",
|
||||
"entry": [
|
||||
{
|
||||
"id": "1",
|
||||
"changes": [
|
||||
{
|
||||
"id": "wamid.1",
|
||||
"message": {"from": "+233000000000", "id": "wamid.1", "text": {"text": "AC leaking"}},
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
async def test_webhook_fail_closed_when_env_unset(client: AsyncClient):
|
||||
"""WHATSAPP_WEBHOOK_SECRET unset ⇒ every message rejected (403)."""
|
||||
assert settings.WHATSAPP_WEBHOOK_SECRET == "" # test env default is unset
|
||||
resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
|
||||
assert resp.status_code == 403
|
||||
assert "not configured" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
async def test_webhook_rejects_missing_or_wrong_secret(client: AsyncClient, monkeypatch):
|
||||
monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
|
||||
resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
|
||||
assert resp.status_code == 403
|
||||
|
||||
resp = await client.post(
|
||||
"/api/whatsapp/webhook", json=WEBHOOK_BODY, headers={"X-Webhook-Secret": "wrong"}
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
async def test_webhook_accepts_valid_secret_and_creates_ticket(client: AsyncClient, monkeypatch):
|
||||
from app.routers import whatsapp as whatsapp_router
|
||||
|
||||
async def _fake_reply(to_phone: str, text: str):
|
||||
from app.schemas.whatsapp import WhatsAppReplyResponse
|
||||
return WhatsAppReplyResponse(success=True, message="sent")
|
||||
|
||||
monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
|
||||
monkeypatch.setattr(whatsapp_router, "send_whatsapp_reply", _fake_reply)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/whatsapp/webhook",
|
||||
json=WEBHOOK_BODY,
|
||||
headers={"X-Webhook-Secret": "test-webhook-secret"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["status"] == "processed"
|
||||
assert data["ticket_number"].startswith("PAV-")
|
||||
|
||||
# The message is logged and visible to an authenticated mock-log caller.
|
||||
token = await _login(client)
|
||||
log = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
|
||||
assert log.status_code == 200
|
||||
assert len(log.json()) == 1
|
||||
assert log.json()[0]["ticket_number"] == data["ticket_number"]
|
||||
|
||||
|
||||
async def test_webhook_verify_token_mismatch_403(client: AsyncClient, monkeypatch):
|
||||
monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
|
||||
resp = await client.get(
|
||||
"/api/whatsapp/webhook",
|
||||
params={"hub.mode": "subscribe", "hub.verify_token": "nope", "hub.challenge": "1234"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
async def test_webhook_verify_token_match_returns_challenge(client: AsyncClient, monkeypatch):
|
||||
monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
|
||||
resp = await client.get(
|
||||
"/api/whatsapp/webhook",
|
||||
params={"hub.mode": "subscribe", "hub.verify_token": "verify-me", "hub.challenge": "1234"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == {"challenge": "1234"}
|
||||
|
||||
|
||||
# ── Item 6: mock-log requires auth ────────────────────────────────────
|
||||
async def test_mock_log_requires_auth(client: AsyncClient):
|
||||
resp = await client.get("/api/whatsapp/mock-log")
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
token = await _login(client)
|
||||
resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
|
||||
|
||||
# ── Item 7: security headers ──────────────────────────────────────────
|
||||
async def test_security_headers_on_html_page(client: AsyncClient):
|
||||
resp = await client.get("/login")
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers["x-frame-options"] == "DENY"
|
||||
assert resp.headers["x-content-type-options"] == "nosniff"
|
||||
assert "content-security-policy" in resp.headers
|
||||
assert "default-src 'self'" in resp.headers["content-security-policy"]
|
||||
|
||||
|
||||
async def test_csp_only_on_html_not_json_api(client: AsyncClient):
|
||||
resp = await client.get("/api/tickets")
|
||||
assert resp.status_code == 200
|
||||
assert "content-type" in resp.headers and resp.headers["content-type"].startswith("application/json")
|
||||
assert "content-security-policy" not in resp.headers
|
||||
# Frame/type hardening headers apply everywhere.
|
||||
assert resp.headers["x-frame-options"] == "DENY"
|
||||
assert resp.headers["x-content-type-options"] == "nosniff"
|
||||
|
||||
|
||||
async def test_hsts_only_when_tls_terminates(client: AsyncClient):
|
||||
plain = await client.get("/login")
|
||||
assert "strict-transport-security" not in plain.headers
|
||||
|
||||
tls = await client.get("/login", headers={"X-Forwarded-Proto": "https"})
|
||||
assert tls.headers["strict-transport-security"] == "max-age=31536000; includeSubDomains"
|
||||
|
||||
|
||||
# ── Item 8: pagination (page/limit) and sane max page size ────────────
|
||||
async def test_limit_alias_over_cap_rejected(client: AsyncClient, seed_tickets):
|
||||
await seed_tickets(10)
|
||||
resp = await client.get("/api/tickets", params={"limit": 500})
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
async def test_limit_alias_paginates(client: AsyncClient, seed_tickets):
|
||||
await seed_tickets(14)
|
||||
resp = await client.get("/api/tickets", params={"page": 2, "limit": 5})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 14
|
||||
assert len(data["items"]) == 5
|
||||
assert data["page_size"] == 5
|
||||
assert data["page"] == 2
|
||||
|
||||
|
||||
async def test_page_beyond_last_returns_empty_with_total(client: AsyncClient, seed_tickets):
|
||||
await seed_tickets(7)
|
||||
resp = await client.get("/api/tickets", params={"page": 999, "page_size": 10})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["items"] == []
|
||||
assert data["total"] == 7
|
||||
|
||||
|
||||
async def test_page_size_and_limit_conflict_is_422(client: AsyncClient, seed_tickets):
|
||||
await seed_tickets(3)
|
||||
resp = await client.get("/api/tickets", params={"page_size": 10, "limit": 20})
|
||||
assert resp.status_code == 422
|
||||
+33
-33
@@ -1,8 +1,8 @@
|
||||
"""P0 hardening regression tests (HARDENING.md P0.1 / P0.2 / P0.3).
|
||||
|
||||
Covers:
|
||||
- P0.3: self-registration CANNOT mint a privileged role (role field ignored)
|
||||
- P0.3: duplicate email still 409s
|
||||
- P0.3: self-registration is REMOVED — POST /api/auth/register returns 404 and
|
||||
no public path can mint a user at all (users are admin-created only).
|
||||
- P0.1: app fails to import/boot with placeholder or missing SECRET_KEY
|
||||
- P0.2: app fails to boot with CORS_ORIGINS="*"
|
||||
"""
|
||||
@@ -15,18 +15,17 @@ import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from httpx import AsyncClient
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
# ── P0.3: self-registration is removed entirely ───────────────────────
|
||||
|
||||
# ── P0.3: registration role-escalation ────────────────────────────────
|
||||
|
||||
async def test_register_cannot_mint_admin_role(client: AsyncClient):
|
||||
"""A raw unauthenticated register call must NOT be able to mint Admin/*."""
|
||||
async def test_register_endpoint_is_removed(client: AsyncClient):
|
||||
"""A raw unauthenticated register call must 404 — no public signup path."""
|
||||
resp = await client.post(
|
||||
"/api/auth/register",
|
||||
json={
|
||||
@@ -36,41 +35,42 @@ async def test_register_cannot_mint_admin_role(client: AsyncClient):
|
||||
"role": "Admin/Jerome",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
created = resp.json()
|
||||
assert created["role"] == "CS Rep", (
|
||||
f"self-registration minted privileged role: {created['role']}"
|
||||
)
|
||||
assert resp.status_code == 404, resp.text
|
||||
|
||||
|
||||
async def test_register_role_wahab_also_blocked(client: AsyncClient):
|
||||
resp = await client.post(
|
||||
async def test_register_endpoint_removed_regardless_of_role(client: AsyncClient):
|
||||
"""Attempts to mint privileged (or any) roles via register all 404."""
|
||||
for role in ("Admin/Jerome", "Admin/Wahab", "admin", "superadmin", "CS Rep"):
|
||||
resp = await client.post(
|
||||
"/api/auth/register",
|
||||
json={
|
||||
"email": f"attacker-{role.lower().replace('/', '-')}@example.com",
|
||||
"password": "Sup3rSecret!",
|
||||
"full_name": "Attacker",
|
||||
"role": role,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 404, (role, resp.text)
|
||||
|
||||
|
||||
async def test_register_does_not_create_user(client: AsyncClient):
|
||||
"""No user row is ever created through the removed register endpoint."""
|
||||
await client.post(
|
||||
"/api/auth/register",
|
||||
json={
|
||||
"email": "attacker2@example.com",
|
||||
"email": "ghost@example.com",
|
||||
"password": "Sup3rSecret!",
|
||||
"full_name": "Attacker Two",
|
||||
"role": "Admin/Wahab",
|
||||
"full_name": "Ghost",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["role"] == "CS Rep"
|
||||
# The ghost account must not be able to log in.
|
||||
resp = await client.post(
|
||||
"/api/auth/login",
|
||||
json={"email": "ghost@example.com", "password": "Sup3rSecret!"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
async def test_register_duplicate_email_conflict(client: AsyncClient):
|
||||
payload = {
|
||||
"email": "dupe@example.com",
|
||||
"password": "Sup3rSecret!",
|
||||
"full_name": "Dupe",
|
||||
}
|
||||
r1 = await client.post("/api/auth/register", json=payload)
|
||||
assert r1.status_code == 201
|
||||
r2 = await client.post("/api/auth/register", json=payload)
|
||||
assert r2.status_code == 409
|
||||
|
||||
|
||||
# ── P0.1 / P0.2: fail-closed boot validation ──────────────────────────
|
||||
|
||||
def _boot_with_env(env_overrides: dict[str, str]) -> subprocess.CompletedProcess:
|
||||
"""Try importing app.main in a subprocess with the given env; the import
|
||||
must fail (non-zero) when fail-closed validation trips."""
|
||||
|
||||
Reference in New Issue
Block a user