counterweight/arbret
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests passing

Browse source
This commit is contained in:
counterweight 2025-12-18 23:33:32 +01:00
parent 322bdd3e6e
commit b173b47925
Signed by: counterweight
GPG key ID: 883EDBAA726BD96C
18 changed files with 1414 additions and 93 deletions
Download patch file Download diff file Expand all files Collapse all files

67
backend/auth.py
View file

@ -1,6 +1,6 @@
import os
from datetime import datetime, timedelta, timezone
from typing import Optional
from typing import List, Optional
import bcrypt
from fastapi import Depends, HTTPException, Request, status
@ -10,7 +10,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from models import User
from models import User, Permission
SECRET_KEY = os.environ["SECRET_KEY"] # Required - see .env.example
ALGORITHM = "HS256"
@ -30,6 +30,8 @@ UserLogin = UserCredentials
class UserResponse(BaseModel):
id: int
email: str
roles: List[str]
permissions: List[str]
class TokenResponse(BaseModel):
@ -99,3 +101,64 @@ async def get_current_user(
raise credentials_exception
return user
def require_permission(*required_permissions: Permission):
"""
Dependency factory that checks if user has ALL of the required permissions.
Usage:
@app.get("/api/counter")
async def get_counter(user: User = Depends(require_permission(Permission.VIEW_COUNTER))):
...
"""
async def permission_checker(
request: Request,
db: AsyncSession = Depends(get_db),
) -> User:
user = await get_current_user(request, db)
user_permissions = await user.get_permissions(db)
missing = [p for p in required_permissions if p not in user_permissions]
if missing:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Missing required permissions: {', '.join(p.value for p in missing)}",
)
return user
return permission_checker
def require_any_permission(*required_permissions: Permission):
"""
Dependency factory that checks if user has ANY of the required permissions.
Usage:
@app.get("/api/resource")
async def get_resource(user: User = Depends(require_any_permission(Permission.VIEW, Permission.ADMIN))):
...
"""
async def permission_checker(
request: Request,
db: AsyncSession = Depends(get_db),
) -> User:
user = await get_current_user(request, db)
user_permissions = await user.get_permissions(db)
if not any(p in user_permissions for p in required_permissions):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Requires one of: {', '.join(p.value for p in required_permissions)}",
)
return user
return permission_checker
async def build_user_response(user: User, db: AsyncSession) -> UserResponse:
"""Build a UserResponse with roles and permissions."""
permissions = await user.get_permissions(db)
return UserResponse(
id=user.id,
email=user.email,
roles=user.role_names,
permissions=[p.value for p in permissions],
)

7
backend/env.example
View file

@ -9,3 +9,10 @@ SECRET_KEY=
# Database URL
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/arbret
# Dev user credentials (regular user)
DEV_USER_EMAIL=
DEV_USER_PASSWORD=
# Dev admin credentials
DEV_ADMIN_EMAIL=
DEV_ADMIN_PASSWORD=

47
backend/main.py
View file

@ -1,12 +1,15 @@
from contextlib import asynccontextmanager
from datetime import datetime
from typing import List
from fastapi import FastAPI, Depends, HTTPException, Response, status, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from sqlalchemy import select, func, desc
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from database import engine, get_db, Base
from models import Counter, User, SumRecord, CounterRecord
from models import Counter, User, SumRecord, CounterRecord, Permission, Role
from auth import (
ACCESS_TOKEN_EXPIRE_MINUTES,
COOKIE_NAME,
@ -18,6 +21,8 @@ from auth import (
authenticate_user,
create_access_token,
get_current_user,
require_permission,
build_user_response,
)
@ -50,6 +55,12 @@ def set_auth_cookie(response: Response, token: str) -> None:
)
async def get_default_role(db: AsyncSession) -> Role | None:
"""Get the default 'regular' role for new users."""
result = await db.execute(select(Role).where(Role.name == "regular"))
return result.scalar_one_or_none()
# Auth endpoints
@app.post("/api/auth/register", response_model=UserResponse)
async def register(
@ -68,13 +79,19 @@ async def register(
email=user_data.email,
hashed_password=get_password_hash(user_data.password),
)
# Assign default role if it exists
default_role = await get_default_role(db)
if default_role:
user.roles.append(default_role)
db.add(user)
await db.commit()
await db.refresh(user)
access_token = create_access_token(data={"sub": str(user.id)})
set_auth_cookie(response, access_token)
return UserResponse(id=user.id, email=user.email)
return await build_user_response(user, db)
@app.post("/api/auth/login", response_model=UserResponse)
@ -92,7 +109,7 @@ async def login(
access_token = create_access_token(data={"sub": str(user.id)})
set_auth_cookie(response, access_token)
return UserResponse(id=user.id, email=user.email)
return await build_user_response(user, db)
@app.post("/api/auth/logout")
@ -102,8 +119,11 @@ async def logout(response: Response):
@app.get("/api/auth/me", response_model=UserResponse)
async def get_me(current_user: User = Depends(get_current_user)):
return UserResponse(id=current_user.id, email=current_user.email)
async def get_me(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await build_user_response(current_user, db)
# Counter endpoints
@ -121,7 +141,7 @@ async def get_or_create_counter(db: AsyncSession) -> Counter:
@app.get("/api/counter")
async def get_counter(
db: AsyncSession = Depends(get_db),
_current_user: User = Depends(get_current_user),
_current_user: User = Depends(require_permission(Permission.VIEW_COUNTER)),
):
counter = await get_or_create_counter(db)
return {"value": counter.value}
@ -130,7 +150,7 @@ async def get_counter(
@app.post("/api/counter/increment")
async def increment_counter(
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission(Permission.INCREMENT_COUNTER)),
):
counter = await get_or_create_counter(db)
value_before = counter.value
@ -162,7 +182,7 @@ class SumResponse(BaseModel):
async def calculate_sum(
data: SumRequest,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission(Permission.USE_SUM)),
):
result = data.a + data.b
record = SumRecord(
@ -177,10 +197,6 @@ async def calculate_sum(
# Audit endpoints
from datetime import datetime
from typing import List
class CounterRecordResponse(BaseModel):
id: int
user_email: str
@ -219,7 +235,7 @@ async def get_counter_records(
page: int = Query(1, ge=1),
per_page: int = Query(10, ge=1, le=100),
db: AsyncSession = Depends(get_db),
_current_user: User = Depends(get_current_user),
_current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)),
):
# Get total count
count_result = await db.execute(select(func.count(CounterRecord.id)))
@ -263,7 +279,7 @@ async def get_sum_records(
page: int = Query(1, ge=1),
per_page: int = Query(10, ge=1, le=100),
db: AsyncSession = Depends(get_db),
_current_user: User = Depends(get_current_user),
_current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)),
):
# Get total count
count_result = await db.execute(select(func.count(SumRecord.id)))
@ -301,4 +317,3 @@ async def get_sum_records(
per_page=per_page,
total_pages=total_pages,
)

134
backend/models.py
View file

@ -1,14 +1,92 @@
from datetime import datetime
from sqlalchemy import Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime, UTC
from enum import Enum as PyEnum
from typing import List, Set
from sqlalchemy import Integer, String, Float, DateTime, ForeignKey, Table, Column, Enum, select
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import AsyncSession
from database import Base
class Counter(Base):
__tablename__ = "counter"
class Permission(str, PyEnum):
"""All available permissions in the system."""
# Counter permissions
VIEW_COUNTER = "view_counter"
INCREMENT_COUNTER = "increment_counter"
# Sum permissions
USE_SUM = "use_sum"
# Audit permissions
VIEW_AUDIT = "view_audit"
id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
value: Mapped[int] = mapped_column(Integer, default=0)
# Role definitions with their permissions
ROLE_DEFINITIONS = {
"admin": {
"description": "Administrator with audit access",
"permissions": [
Permission.VIEW_AUDIT,
],
},
"regular": {
"description": "Regular user with counter and sum access",
"permissions": [
Permission.VIEW_COUNTER,
Permission.INCREMENT_COUNTER,
Permission.USE_SUM,
],
},
}
# Association table: Role <-> Permission (many-to-many)
role_permissions = Table(
"role_permissions",
Base.metadata,
Column("role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
Column("permission", Enum(Permission), primary_key=True),
)
# Association table: User <-> Role (many-to-many)
user_roles = Table(
"user_roles",
Base.metadata,
Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
Column("role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
)
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
description: Mapped[str] = mapped_column(String(255), nullable=True)
# Relationship to users
users: Mapped[List["User"]] = relationship(
"User",
secondary=user_roles,
back_populates="roles",
)
async def get_permissions(self, db: AsyncSession) -> Set[Permission]:
"""Get all permissions for this role."""
result = await db.execute(
select(role_permissions.c.permission).where(role_permissions.c.role_id == self.id)
)
return {row[0] for row in result.fetchall()}
async def add_permission(self, db: AsyncSession, permission: Permission) -> None:
"""Add a permission to this role."""
await db.execute(role_permissions.insert().values(role_id=self.id, permission=permission))
async def set_permissions(self, db: AsyncSession, permissions: List[Permission]) -> None:
"""Set all permissions for this role (replaces existing)."""
await db.execute(role_permissions.delete().where(role_permissions.c.role_id == self.id))
for perm in permissions:
await db.execute(role_permissions.insert().values(role_id=self.id, permission=perm))
class User(Base):
@ -17,6 +95,39 @@ class User(Base):
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
# Relationship to roles
roles: Mapped[List[Role]] = relationship(
"Role",
secondary=user_roles,
back_populates="users",
lazy="selectin",
)
async def get_permissions(self, db: AsyncSession) -> Set[Permission]:
"""Get all permissions from all roles."""
permissions: Set[Permission] = set()
for role in self.roles:
role_perms = await role.get_permissions(db)
permissions.update(role_perms)
return permissions
async def has_permission(self, db: AsyncSession, permission: Permission) -> bool:
"""Check if user has a specific permission through any of their roles."""
permissions = await self.get_permissions(db)
return permission in permissions
@property
def role_names(self) -> List[str]:
"""Get list of role names for API responses."""
return [role.name for role in self.roles]
class Counter(Base):
__tablename__ = "counter"
id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
value: Mapped[int] = mapped_column(Integer, default=0)
class SumRecord(Base):
@ -27,7 +138,9 @@ class SumRecord(Base):
a: Mapped[float] = mapped_column(Float, nullable=False)
b: Mapped[float] = mapped_column(Float, nullable=False)
result: Mapped[float] = mapped_column(Float, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(UTC)
)
class CounterRecord(Base):
@ -37,5 +150,6 @@ class CounterRecord(Base):
user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False, index=True)
value_before: Mapped[int] = mapped_column(Integer, nullable=False)
value_after: Mapped[int] = mapped_column(Integer, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(UTC)
)

89
backend/seed.py
View file

@ -1,14 +1,67 @@
"""Seed the database with a dev user."""
"""Seed the database with roles, permissions, and dev users."""
import asyncio
import os
from typing import List
from sqlalchemy import select
from database import engine, async_session, Base
from models import User
from models import User, Role, Permission, role_permissions, ROLE_DEFINITIONS
from auth import get_password_hash
DEV_USER_EMAIL = os.environ["DEV_USER_EMAIL"]
DEV_USER_PASSWORD = os.environ["DEV_USER_PASSWORD"]
DEV_ADMIN_EMAIL = os.environ["DEV_ADMIN_EMAIL"]
DEV_ADMIN_PASSWORD = os.environ["DEV_ADMIN_PASSWORD"]
async def upsert_role(db, name: str, description: str, permissions: List[Permission]) -> Role:
"""Create or update a role with the given permissions."""
result = await db.execute(select(Role).where(Role.name == name))
role = result.scalar_one_or_none()
if role:
role.description = description
print(f"Updated role: {name}")
else:
role = Role(name=name, description=description)
db.add(role)
await db.flush() # Get the role ID
print(f"Created role: {name}")
# Set permissions for the role
await role.set_permissions(db, permissions)
print(f" Permissions: {', '.join(p.value for p in permissions)}")
return role
async def upsert_user(db, email: str, password: str, role_names: List[str]) -> User:
"""Create or update a user with the given credentials and roles."""
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one_or_none()
# Get roles
roles = []
for role_name in role_names:
result = await db.execute(select(Role).where(Role.name == role_name))
role = result.scalar_one_or_none()
if role:
roles.append(role)
if user:
user.hashed_password = get_password_hash(password)
user.roles = roles
print(f"Updated user: {email} with roles: {role_names}")
else:
user = User(
email=email,
hashed_password=get_password_hash(password),
roles=roles,
)
db.add(user)
print(f"Created user: {email} with roles: {role_names}")
return user
async def seed():
@ -16,23 +69,25 @@ async def seed():
await conn.run_sync(Base.metadata.create_all)
async with async_session() as db:
result = await db.execute(select(User).where(User.email == DEV_USER_EMAIL))
user = result.scalar_one_or_none()
if user:
user.hashed_password = get_password_hash(DEV_USER_PASSWORD)
await db.commit()
print(f"Updated dev user: {DEV_USER_EMAIL} / {DEV_USER_PASSWORD}")
else:
user = User(
email=DEV_USER_EMAIL,
hashed_password=get_password_hash(DEV_USER_PASSWORD),
print("\n=== Seeding Roles ===")
for role_name, role_config in ROLE_DEFINITIONS.items():
await upsert_role(
db,
role_name,
role_config["description"],
role_config["permissions"],
)
db.add(user)
await db.commit()
print(f"Created dev user: {DEV_USER_EMAIL} / {DEV_USER_PASSWORD}")
print("\n=== Seeding Users ===")
# Create regular dev user
await upsert_user(db, DEV_USER_EMAIL, DEV_USER_PASSWORD, ["regular"])
# Create admin dev user
await upsert_user(db, DEV_ADMIN_EMAIL, DEV_ADMIN_PASSWORD, ["admin"])
await db.commit()
print("\n=== Seeding Complete ===\n")
if __name__ == "__main__":
asyncio.run(seed())

143
backend/tests/conftest.py
View file

@ -1,15 +1,19 @@
import os
from contextlib import asynccontextmanager
from typing import List
# Set required env vars before importing app
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from database import Base, get_db
from main import app
from models import User, Role, Permission, ROLE_DEFINITIONS
from auth import get_password_hash
TEST_DATABASE_URL = os.getenv(
"TEST_DATABASE_URL",
@ -20,9 +24,10 @@ TEST_DATABASE_URL = os.getenv(
class ClientFactory:
"""Factory for creating httpx clients with optional cookies."""
def __init__(self, transport, base_url):
def __init__(self, transport, base_url, session_factory):
self._transport = transport
self._base_url = base_url
self._session_factory = session_factory
@asynccontextmanager
async def create(self, cookies: dict | None = None):
@ -45,6 +50,59 @@ class ClientFactory:
async def post(self, url: str, **kwargs):
return await self.request("POST", url, **kwargs)
@asynccontextmanager
async def get_db_session(self):
"""Get a database session for direct DB operations in tests."""
async with self._session_factory() as session:
yield session
async def setup_roles(db: AsyncSession) -> dict[str, Role]:
"""Create all roles with their permissions from ROLE_DEFINITIONS."""
roles = {}
for role_name, config in ROLE_DEFINITIONS.items():
# Check if role exists
result = await db.execute(select(Role).where(Role.name == role_name))
role = result.scalar_one_or_none()
if not role:
role = Role(name=role_name, description=config["description"])
db.add(role)
await db.flush()
# Set permissions
await role.set_permissions(db, config["permissions"])
roles[role_name] = role
await db.commit()
return roles
async def create_user_with_roles(
db: AsyncSession,
email: str,
password: str,
role_names: List[str],
) -> User:
"""Create a user with specified roles."""
# Get roles
roles = []
for role_name in role_names:
result = await db.execute(select(Role).where(Role.name == role_name))
role = result.scalar_one_or_none()
if role:
roles.append(role)
user = User(
email=email,
hashed_password=get_password_hash(password),
roles=roles,
)
db.add(user)
await db.commit()
await db.refresh(user)
return user
@pytest.fixture(scope="function")
async def client_factory():
@ -57,6 +115,10 @@ async def client_factory():
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
# Setup roles
async with session_factory() as db:
await setup_roles(db)
async def override_get_db():
async with session_factory() as session:
yield session
@ -64,7 +126,7 @@ async def client_factory():
app.dependency_overrides[get_db] = override_get_db
transport = ASGITransport(app=app)
factory = ClientFactory(transport, "http://test")
factory = ClientFactory(transport, "http://test", session_factory)
yield factory
@ -77,3 +139,78 @@ async def client(client_factory):
"""Fixture for a simple client without cookies (backwards compatible)."""
async with client_factory.create() as c:
yield c
@pytest.fixture(scope="function")
async def regular_user(client_factory):
"""Create a regular user and return their credentials and cookies."""
from tests.helpers import unique_email
email = unique_email("regular")
password = "password123"
async with client_factory.get_db_session() as db:
await create_user_with_roles(db, email, password, ["regular"])
# Login to get cookies
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": password},
)
return {
"email": email,
"password": password,
"cookies": dict(response.cookies),
"response": response,
}
@pytest.fixture(scope="function")
async def admin_user(client_factory):
"""Create an admin user and return their credentials and cookies."""
from tests.helpers import unique_email
email = unique_email("admin")
password = "password123"
async with client_factory.get_db_session() as db:
await create_user_with_roles(db, email, password, ["admin"])
# Login to get cookies
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": password},
)
return {
"email": email,
"password": password,
"cookies": dict(response.cookies),
"response": response,
}
@pytest.fixture(scope="function")
async def user_no_roles(client_factory):
"""Create a user with NO roles and return their credentials and cookies."""
from tests.helpers import unique_email
email = unique_email("noroles")
password = "password123"
async with client_factory.get_db_session() as db:
await create_user_with_roles(db, email, password, [])
# Login to get cookies
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": password},
)
return {
"email": email,
"password": password,
"cookies": dict(response.cookies),
"response": response,
}

8
backend/tests/test_auth.py
View file

@ -16,6 +16,10 @@ async def test_register_success(client):
data = response.json()
assert data["email"] == email
assert "id" in data
assert "roles" in data
assert "permissions" in data
# New users get regular role by default
assert "regular" in data["roles"]
# Cookie should be set
assert COOKIE_NAME in response.cookies
@ -83,6 +87,8 @@ async def test_login_success(client):
assert response.status_code == 200
data = response.json()
assert data["email"] == email
assert "roles" in data
assert "permissions" in data
assert COOKIE_NAME in response.cookies
@ -146,6 +152,8 @@ async def test_get_me_success(client_factory):
data = response.json()
assert data["email"] == email
assert "id" in data
assert "roles" in data
assert "permissions" in data
@pytest.mark.asyncio

461
backend/tests/test_permissions.py Normal file
View file

@ -0,0 +1,461 @@
"""
Permission and Role-Based Access Control Tests
These tests verify that:
1. Users can only access endpoints they have permission for
2. Users without proper roles are denied access (403)
3. Unauthenticated users are denied access (401)
4. The permission system cannot be bypassed
"""
import pytest
from models import Permission
# =============================================================================
# Role Assignment Tests
# =============================================================================
class TestRoleAssignment:
"""Test that roles are properly assigned and returned."""
@pytest.mark.asyncio
async def test_regular_user_has_correct_roles(self, client_factory, regular_user):
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/auth/me")
assert response.status_code == 200
data = response.json()
assert "regular" in data["roles"]
assert "admin" not in data["roles"]
@pytest.mark.asyncio
async def test_admin_user_has_correct_roles(self, client_factory, admin_user):
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/auth/me")
assert response.status_code == 200
data = response.json()
assert "admin" in data["roles"]
assert "regular" not in data["roles"]
@pytest.mark.asyncio
async def test_regular_user_has_correct_permissions(self, client_factory, regular_user):
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/auth/me")
data = response.json()
permissions = data["permissions"]
# Should have counter and sum permissions
assert Permission.VIEW_COUNTER.value in permissions
assert Permission.INCREMENT_COUNTER.value in permissions
assert Permission.USE_SUM.value in permissions
# Should NOT have audit permission
assert Permission.VIEW_AUDIT.value not in permissions
@pytest.mark.asyncio
async def test_admin_user_has_correct_permissions(self, client_factory, admin_user):
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/auth/me")
data = response.json()
permissions = data["permissions"]
# Should have audit permission
assert Permission.VIEW_AUDIT.value in permissions
# Should NOT have counter/sum permissions
assert Permission.VIEW_COUNTER.value not in permissions
assert Permission.INCREMENT_COUNTER.value not in permissions
assert Permission.USE_SUM.value not in permissions
@pytest.mark.asyncio
async def test_user_with_no_roles_has_no_permissions(self, client_factory, user_no_roles):
async with client_factory.create(cookies=user_no_roles["cookies"]) as client:
response = await client.get("/api/auth/me")
data = response.json()
assert data["roles"] == []
assert data["permissions"] == []
# =============================================================================
# Counter Endpoint Access Tests
# =============================================================================
class TestCounterAccess:
"""Test access control for counter endpoints."""
@pytest.mark.asyncio
async def test_regular_user_can_view_counter(self, client_factory, regular_user):
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/counter")
assert response.status_code == 200
assert "value" in response.json()
@pytest.mark.asyncio
async def test_regular_user_can_increment_counter(self, client_factory, regular_user):
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.post("/api/counter/increment")
assert response.status_code == 200
assert "value" in response.json()
@pytest.mark.asyncio
async def test_admin_cannot_view_counter(self, client_factory, admin_user):
"""Admin users should be forbidden from counter endpoints."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/counter")
assert response.status_code == 403
assert "permission" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_admin_cannot_increment_counter(self, client_factory, admin_user):
"""Admin users should be forbidden from incrementing counter."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.post("/api/counter/increment")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_user_without_roles_cannot_view_counter(self, client_factory, user_no_roles):
"""Users with no roles should be forbidden."""
async with client_factory.create(cookies=user_no_roles["cookies"]) as client:
response = await client.get("/api/counter")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_unauthenticated_cannot_view_counter(self, client):
"""Unauthenticated requests should get 401."""
response = await client.get("/api/counter")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_unauthenticated_cannot_increment_counter(self, client):
"""Unauthenticated requests should get 401."""
response = await client.post("/api/counter/increment")
assert response.status_code == 401
# =============================================================================
# Sum Endpoint Access Tests
# =============================================================================
class TestSumAccess:
"""Test access control for sum endpoint."""
@pytest.mark.asyncio
async def test_regular_user_can_use_sum(self, client_factory, regular_user):
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.post(
"/api/sum",
json={"a": 5, "b": 3},
)
assert response.status_code == 200
data = response.json()
assert data["result"] == 8
@pytest.mark.asyncio
async def test_admin_cannot_use_sum(self, client_factory, admin_user):
"""Admin users should be forbidden from sum endpoint."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.post(
"/api/sum",
json={"a": 5, "b": 3},
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_user_without_roles_cannot_use_sum(self, client_factory, user_no_roles):
async with client_factory.create(cookies=user_no_roles["cookies"]) as client:
response = await client.post(
"/api/sum",
json={"a": 5, "b": 3},
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_unauthenticated_cannot_use_sum(self, client):
response = await client.post(
"/api/sum",
json={"a": 5, "b": 3},
)
assert response.status_code == 401
# =============================================================================
# Audit Endpoint Access Tests
# =============================================================================
class TestAuditAccess:
"""Test access control for audit endpoints."""
@pytest.mark.asyncio
async def test_admin_can_view_counter_audit(self, client_factory, admin_user):
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/audit/counter")
assert response.status_code == 200
data = response.json()
assert "records" in data
assert "total" in data
@pytest.mark.asyncio
async def test_admin_can_view_sum_audit(self, client_factory, admin_user):
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/audit/sum")
assert response.status_code == 200
data = response.json()
assert "records" in data
assert "total" in data
@pytest.mark.asyncio
async def test_regular_user_cannot_view_counter_audit(self, client_factory, regular_user):
"""Regular users should be forbidden from audit endpoints."""
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/audit/counter")
assert response.status_code == 403
assert "permission" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_regular_user_cannot_view_sum_audit(self, client_factory, regular_user):
"""Regular users should be forbidden from audit endpoints."""
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/audit/sum")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_user_without_roles_cannot_view_audit(self, client_factory, user_no_roles):
async with client_factory.create(cookies=user_no_roles["cookies"]) as client:
response = await client.get("/api/audit/counter")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_unauthenticated_cannot_view_counter_audit(self, client):
response = await client.get("/api/audit/counter")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_unauthenticated_cannot_view_sum_audit(self, client):
response = await client.get("/api/audit/sum")
assert response.status_code == 401
# =============================================================================
# Offensive Security Tests - Bypass Attempts
# =============================================================================
class TestSecurityBypassAttempts:
"""
Offensive tests that attempt to bypass security controls.
These simulate potential attack vectors.
"""
@pytest.mark.asyncio
async def test_cannot_access_audit_with_forged_role_claim(self, client_factory, regular_user):
"""
Attempt to access audit by somehow claiming admin role.
The server should verify roles from DB, not trust client claims.
"""
# Regular user tries to access audit endpoint
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/audit/counter")
# Should be denied regardless of any manipulation attempts
assert response.status_code == 403
@pytest.mark.asyncio
async def test_cannot_access_counter_with_expired_session(self, client_factory):
"""Test that invalid/expired tokens are rejected."""
fake_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI5OTk5IiwiZXhwIjoxfQ.invalid"
async with client_factory.create(cookies={"auth_token": fake_token}) as client:
response = await client.get("/api/counter")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_cannot_access_with_tampered_token(self, client_factory, regular_user):
"""Test that tokens signed with wrong key are rejected."""
# Take a valid token and modify it
original_token = regular_user["cookies"].get("auth_token", "")
if original_token:
tampered_token = original_token[:-5] + "XXXXX"
async with client_factory.create(cookies={"auth_token": tampered_token}) as client:
response = await client.get("/api/counter")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_cannot_escalate_to_admin_via_registration(self, client_factory):
"""
Test that new registrations cannot claim admin role.
New users should only get 'regular' role by default.
"""
from tests.helpers import unique_email
response = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "password123"},
)
assert response.status_code == 200
data = response.json()
# Should only have regular role, not admin
assert "admin" not in data["roles"]
assert Permission.VIEW_AUDIT.value not in data["permissions"]
# Try to access audit with this new user
async with client_factory.create(cookies=dict(response.cookies)) as client:
audit_response = await client.get("/api/audit/counter")
assert audit_response.status_code == 403
@pytest.mark.asyncio
async def test_deleted_user_token_is_invalid(self, client_factory):
"""
If a user is deleted, their token should no longer work.
This tests that tokens are validated against current DB state.
"""
from tests.helpers import unique_email
from sqlalchemy import delete
from models import User
email = unique_email("deleted")
# Create and login user
async with client_factory.get_db_session() as db:
from tests.conftest import create_user_with_roles
user = await create_user_with_roles(db, email, "password123", ["regular"])
user_id = user.id
login_response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "password123"},
)
cookies = dict(login_response.cookies)
# Delete the user from DB
async with client_factory.get_db_session() as db:
await db.execute(delete(User).where(User.id == user_id))
await db.commit()
# Try to use the old token
async with client_factory.create(cookies=cookies) as client:
response = await client.get("/api/auth/me")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_role_change_reflected_immediately(self, client_factory):
"""
If a user's role is changed, the change should be reflected
in subsequent requests (no stale permission cache).
"""
from tests.helpers import unique_email
from sqlalchemy import select
from models import User, Role
email = unique_email("rolechange")
# Create regular user
async with client_factory.get_db_session() as db:
from tests.conftest import create_user_with_roles
await create_user_with_roles(db, email, "password123", ["regular"])
login_response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "password123"},
)
cookies = dict(login_response.cookies)
# Verify can access counter but not audit
async with client_factory.create(cookies=cookies) as client:
assert (await client.get("/api/counter")).status_code == 200
assert (await client.get("/api/audit/counter")).status_code == 403
# Change user's role from regular to admin
async with client_factory.get_db_session() as db:
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one()
result = await db.execute(select(Role).where(Role.name == "admin"))
admin_role = result.scalar_one()
result = await db.execute(select(Role).where(Role.name == "regular"))
regular_role = result.scalar_one()
user.roles = [admin_role] # Remove regular, add admin
await db.commit()
# Now should have audit access but not counter access
async with client_factory.create(cookies=cookies) as client:
assert (await client.get("/api/audit/counter")).status_code == 200
assert (await client.get("/api/counter")).status_code == 403
# =============================================================================
# Audit Record Tests
# =============================================================================
class TestAuditRecords:
"""Test that actions are properly recorded in audit logs."""
@pytest.mark.asyncio
async def test_counter_increment_creates_audit_record(
self, client_factory, regular_user, admin_user
):
"""Verify that counter increments are recorded and visible in audit."""
# Regular user increments counter
async with client_factory.create(cookies=regular_user["cookies"]) as client:
await client.post("/api/counter/increment")
# Admin checks audit
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/audit/counter")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
# Find record for our user
records = data["records"]
user_records = [r for r in records if r["user_email"] == regular_user["email"]]
assert len(user_records) >= 1
@pytest.mark.asyncio
async def test_sum_operation_creates_audit_record(
self, client_factory, regular_user, admin_user
):
"""Verify that sum operations are recorded and visible in audit."""
# Regular user uses sum
async with client_factory.create(cookies=regular_user["cookies"]) as client:
await client.post("/api/sum", json={"a": 10, "b": 20})
# Admin checks audit
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/audit/sum")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
# Find record with our values
records = data["records"]
matching = [r for r in records if r["a"] == 10 and r["b"] == 20 and r["result"] == 30]
assert len(matching) >= 1