fix: critical security hardening (Phase A permissions audit)

- Remove role field from UserCreate schema, hardcode 'engineer' at registration
- Escape all user content in HTML export with html.escape() (XSS fix)
- Add field_validator to reject default SECRET_KEY when DEBUG=False
- Add CHECK constraint on users.role ('engineer'|'viewer') + migration 011
- Fix test_admin fixture to properly grant is_super_admin via ORM
- Fix circular FK (users↔invite_codes) in test DB setup with DROP SCHEMA CASCADE
- Add 5 new security tests (role validation + XSS prevention)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chihlasm
2026-02-05 22:04:37 -05:00
parent fd8fab97bd
commit 3e0fb92012
10 changed files with 236 additions and 48 deletions

View File

@@ -7,6 +7,7 @@ Provides test database setup, client fixtures, and authentication helpers.
import asyncio
from typing import AsyncGenerator, Generator
import pytest
import sqlalchemy as sa
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool
@@ -35,9 +36,10 @@ async def test_db() -> AsyncGenerator[AsyncSession, None]:
This fixture:
1. Creates a test database engine
2. Creates all tables
3. Yields a session for the test
4. Drops all tables after the test
2. Drops all existing tables (CASCADE to handle circular FKs)
3. Creates all tables
4. Yields a session for the test
5. Drops all tables after the test
"""
# Create async engine for tests (with NullPool to avoid connection reuse issues)
engine = create_async_engine(
@@ -46,8 +48,11 @@ async def test_db() -> AsyncGenerator[AsyncSession, None]:
echo=False
)
# Create all tables
# Drop and recreate all tables (use raw SQL CASCADE to handle circular FKs
# between users <-> invite_codes)
async with engine.begin() as conn:
await conn.execute(sa.text("DROP SCHEMA public CASCADE"))
await conn.execute(sa.text("CREATE SCHEMA public"))
await conn.run_sync(Base.metadata.create_all)
# Create async session maker
@@ -61,9 +66,10 @@ async def test_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session
# Drop all tables after test
# Drop all tables after test (CASCADE for circular FKs)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.execute(sa.text("DROP SCHEMA public CASCADE"))
await conn.execute(sa.text("CREATE SCHEMA public"))
await engine.dispose()
@@ -99,8 +105,7 @@ async def test_user(client):
user_data = {
"email": "test@example.com",
"password": "TestPassword123!",
"name": "Test User",
"role": "engineer"
"name": "Test User"
}
response = await client.post("/api/v1/auth/register", json=user_data)
@@ -181,23 +186,32 @@ async def test_tree(client, auth_headers):
@pytest.fixture
async def test_admin(client):
async def test_admin(client, test_db):
"""
Create a test admin user and return their credentials.
Create a test super-admin user.
Returns:
dict with email, password, and user_data
Registers as engineer (the only role available at registration),
then promotes to super_admin directly via the DB session.
"""
from uuid import UUID as PyUUID
from sqlalchemy import select
from app.models.user import User
admin_data = {
"email": "admin@example.com",
"password": "AdminPassword123!",
"name": "Test Admin",
"role": "admin"
"name": "Test Admin"
}
response = await client.post("/api/v1/auth/register", json=admin_data)
assert response.status_code == 200 or response.status_code == 201
user_id = PyUUID(response.json()["id"])
result = await test_db.execute(select(User).where(User.id == user_id))
user = result.scalar_one()
user.is_super_admin = True
await test_db.commit()
return {
"email": admin_data["email"],
"password": admin_data["password"],