Nora
96 KiB
96 KiB
Cyclone Auth Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Add username/password authentication with three predefined roles (admin, user, viewer) to the Cyclone FastAPI backend and React frontend. Browser login + server-side SQLite sessions + HttpOnly cookie + role-gated endpoints.
Architecture: New cyclone.auth module on the backend (users, sessions, permissions, deps, routes, admin, rate_limit). Two SQL migrations (0015 for users/sessions tables, 0016 for audit_log.user_id). FastAPI dependencies get_current_user and require_role. New src/auth/ module on the frontend (AuthProvider, useAuth, RoleGate, fetch wrapper). New /login page. CLI command python -m cyclone users ... for ops.
Tech Stack: FastAPI, SQLAlchemy, passlib[bcrypt], secrets.token_urlsafe, React, react, react-query, react-router.
Spec: docs/superpowers/specs/2026-06-22-cyclone-auth-design.md
File Structure
Backend (new files):
backend/src/cyclone/auth/__init__.py— module exportsbackend/src/cyclone/auth/users.py— User model, bcrypt CRUDbackend/src/cyclone/auth/sessions.py— Session model, create/validate/expire/touchbackend/src/cyclone/auth/permissions.py— Role enum, PERMISSIONS matrixbackend/src/cyclone/auth/deps.py— get_current_user, require_rolebackend/src/cyclone/auth/routes.py— login, logout, mebackend/src/cyclone/auth/admin.py— admin user managementbackend/src/cyclone/auth/rate_limit.py— per-username failed-login counterbackend/src/cyclone/auth/cli.py—python -m cyclone users ...subcommandbackend/src/cyclone/migrations/0015_users_and_sessions.py— users + sessions tablesbackend/src/cyclone/migrations/0016_audit_log_user_id.py— audit_log.user_id
Backend (modified):
backend/src/cyclone/db.py— add User, Session modelsbackend/src/cyclone/api.py— register auth router, gate existing endpointsbackend/src/cyclone/__main__.py— bootstrap admin + register CLI subcommandbackend/src/cyclone/audit_log.py— record user_idbackend/pyproject.toml— passlib[bcrypt]backend/Dockerfile— already installs from pyproject, no change neededdocker-compose.yml— CYCLONE_ADMIN_USERNAME/PASSWORD env vars.env.example— env var documentation
Backend tests (new):
backend/tests/test_auth_users.pybackend/tests/test_auth_sessions.pybackend/tests/test_auth_routes.pybackend/tests/test_auth_permissions.pybackend/tests/test_auth_admin.pybackend/tests/test_auth_bootstrap.pybackend/tests/test_auth_login_rate_limit.pybackend/tests/test_audit_log_user_id.pybackend/tests/test_existing_endpoints_require_auth.pybackend/tests/test_cli_users.py
Frontend (new files):
src/auth/types.ts— User, AuthState, Role typessrc/auth/AuthProvider.tsx— context + reducersrc/auth/useAuth.ts— hooksrc/auth/api.ts— fetch wrapper for 401src/auth/RoleGate.tsx— disable children when role not allowedsrc/auth/RequireAuth.tsx— route guardsrc/pages/Login.tsx— login pagesrc/auth/AuthProvider.test.tsxsrc/auth/RoleGate.test.tsxsrc/auth/api.test.tssrc/pages/Login.test.tsx
Frontend (modified):
src/main.tsx— wrap in AuthProvider + RequireAuth + BrowserRoutersrc/lib/api.ts— call auth/api wrapper for fetchsrc/components/Sidebar.tsx— CurrentUser from useAuth (replaces hardcoded "Jordan K.")src/pages/Upload.tsx— RoleGate on dropzonesrc/pages/Inbox.tsx— RoleGate on resubmit/acknowledgesrc/pages/Reconciliation.tsx— RoleGate on matchsrc/pages/Acks.tsx— RoleGate on acksrc/types/index.ts— add User typesrc/components/RequireAuth.tsx— route guardfrontend/nginx.conf—X-Forwarded-Protoheader
Docs (modified):
README.md— Auth section
Phase 1: Backend Foundations
Task 1: Add passlib[bcrypt] dependency
Files:
-
Modify:
backend/pyproject.toml:30-60 -
Step 1: Add the dependency
In backend/pyproject.toml, find the [project] section and add passlib[bcrypt]>=1.7.4 to the dependencies list (alphabetical order — between cryptography and ... wherever alphabetically it fits).
dependencies = [
...
"fastapi>=0.115",
"passlib[bcrypt]>=1.7.4",
"pydantic>=2.5",
...
]
- Step 2: Install
cd /home/tyler/dev/cyclone/backend && uv sync
Expected: installs passlib, bcrypt. No errors.
- Step 3: Smoke test bcrypt import
uv run python -c "from passlib.hash import bcrypt; print(bcrypt.hash('test'))"
Expected: prints a $2b$... hash.
- Step 4: Commit
cd /home/tyler/dev/cyclone && git add backend/pyproject.toml backend/uv.lock && git commit -m "feat(deps): add passlib[bcrypt] for password hashing"
Task 2: Add User and Session models to db.py
Files:
-
Modify:
backend/src/cyclone/db.py— add two SQLAlchemy models -
Step 1: Locate the models section
In backend/src/cyclone/db.py, find the end of the existing model classes (probably Claim, Remittance, etc.). We'll append User and Session after them.
- Step 2: Add User and Session models
Append at the end of the file (before any helper functions):
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(64), unique=True, index=True)
password_hash: Mapped[str] = mapped_column(String(255))
role: Mapped[str] = mapped_column(String(16))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class Session(Base):
__tablename__ = "sessions"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
Ensure the imports at the top include String, DateTime, ForeignKey, func from sqlalchemy. If they're missing, add them.
- Step 3: Smoke test
cd /home/tyler/dev/cyclone/backend && uv run python -c "from cyclone.db import User, Session; print(User.__tablename__, Session.__tablename__)"
Expected: users sessions
- Step 4: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/db.py && git commit -m "feat(db): add User and Session models"
Task 3: Migration 0015 — users + sessions tables
Files:
-
Create:
backend/src/cyclone/migrations/0015_users_and_sessions.py -
Step 1: Inspect existing migrations
ls /home/tyler/dev/cyclone/backend/src/cyclone/migrations/
Read the most recent migration (e.g. 0014_*.py) to confirm the migration runner's expected pattern (filename, function signature, SQLAlchemy vs raw SQL, etc.).
- Step 2: Create migration file
backend/src/cyclone/migrations/0015_users_and_sessions.py:
"""Migration 0015 — add users + sessions tables for auth (SP-auth)."""
from __future__ import annotations
from sqlalchemy import inspect
from cyclone.db import Base, Session, User, engine
def upgrade() -> None:
insp = inspect(engine)
if "users" in insp.get_table_names():
return # idempotent
Base.metadata.create_all(engine, tables=[User.__table__, Session.__table__])
def downgrade() -> None:
raise NotImplementedError("Cyclone does not support down-migrations.")
Adapt the signature to match the runner's existing convention (it might take a conn argument or use op.create_table from Alembic). Match what's there.
- Step 3: Run the migration
cd /home/tyler/dev/cyclone/backend && uv run python -c "from cyclone.migrations.runner import run; run(target=15)"
(Or whatever the project's migration entry point is — check cyclone.migrations.__init__ or db.py.)
Expected: tables users and sessions exist in cyclone.db.
- Step 4: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/migrations/0015_users_and_sessions.py && git commit -m "feat(migration): 0015 users + sessions tables"
Task 4: Migration 0016 — audit_log.user_id
Files:
-
Create:
backend/src/cyclone/migrations/0016_audit_log_user_id.py -
Step 1: Inspect audit_log schema
cd /home/tyler/dev/cyclone/backend && uv run python -c "from cyclone.db import engine; from sqlalchemy import inspect; print(inspect(engine).get_columns('audit_log'))"
Confirm the existing audit_log table column names.
- Step 2: Create migration file
"""Migration 0016 — add audit_log.user_id for SP-auth."""
from __future__ import annotations
from sqlalchemy import inspect
from cyclone.db import engine
def upgrade() -> None:
insp = inspect(engine)
cols = {c["name"] for c in insp.get_columns("audit_log")}
if "user_id" in cols:
return
with engine.begin() as conn:
conn.exec_driver_sql("ALTER TABLE audit_log ADD COLUMN user_id INTEGER")
conn.exec_driver_sql("CREATE INDEX IF NOT EXISTS ix_audit_log_user_id ON audit_log (user_id)")
def downgrade() -> None:
raise NotImplementedError("Cyclone does not support down-migrations.")
- Step 3: Run migration
uv run python -c "from cyclone.migrations.runner import run; run(target=16)"
- Step 4: Verify the column
uv run python -c "from cyclone.db import engine; from sqlalchemy import inspect; print([c['name'] for c in inspect(engine).get_columns('audit_log')])"
Expected: user_id is in the list.
- Step 5: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/migrations/0016_audit_log_user_id.py && git commit -m "feat(migration): 0016 audit_log.user_id"
Task 5: TDD — users.py (User CRUD + bcrypt)
Files:
-
Create:
backend/src/cyclone/auth/users.py -
Create:
backend/tests/test_auth_users.py -
Step 1: Create empty auth module
backend/src/cyclone/auth/__init__.py:
"""Auth module — users, sessions, permissions, routes, admin, rate_limit."""
- Step 2: Write failing tests
backend/tests/test_auth_users.py:
"""Unit tests for cyclone.auth.users — User CRUD + bcrypt hashing."""
from __future__ import annotations
import pytest
from sqlalchemy import delete
from cyclone.auth import users
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear_users():
with SessionLocal() as db:
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(User))
db.commit()
def test_hash_password_returns_bcrypt():
h = users.hash_password("hunter2hunter2")
assert h.startswith("$2")
def test_hash_password_produces_unique_salts():
a = users.hash_password("same-password")
b = users.hash_password("same-password")
assert a != b
def test_verify_password_correct():
h = users.hash_password("hunter2hunter2")
assert users.verify_password("hunter2hunter2", h) is True
def test_verify_password_incorrect():
h = users.hash_password("hunter2hunter2")
assert users.verify_password("WRONG", h) is False
def test_create_user_persists_with_hashed_password():
with SessionLocal() as db:
u = users.create(db, username="alice", password="hunter2hunter2", role="admin")
assert u.id is not None
assert u.username == "alice"
assert u.role == "admin"
assert u.disabled_at is None
# Hash, not plaintext.
assert u.password_hash != "hunter2hunter2"
assert u.password_hash.startswith("$2")
def test_create_user_rejects_duplicate_username():
with SessionLocal() as db:
users.create(db, username="bob", password="hunter2hunter2", role="user")
with SessionLocal() as db:
with pytest.raises(Exception):
users.create(db, username="bob", password="anotherone", role="user")
def test_get_by_username_returns_user():
with SessionLocal() as db:
users.create(db, username="carol", password="hunter2hunter2", role="viewer")
with SessionLocal() as db:
u = users.get_by_username(db, "carol")
assert u is not None
assert u.username == "carol"
def test_get_by_username_returns_none_for_missing():
with SessionLocal() as db:
assert users.get_by_username(db, "ghost") is None
def test_disable_user_sets_disabled_at():
with SessionLocal() as db:
u = users.create(db, username="dave", password="hunter2hunter2", role="user")
users.disable(db, u.id)
with SessionLocal() as db:
refreshed = users.get_by_username(db, "dave")
assert refreshed.disabled_at is not None
def test_to_public_shape_omits_password_hash():
with SessionLocal() as db:
u = users.create(db, username="eve", password="hunter2hunter2", role="admin")
shape = users.to_public(u)
assert "password_hash" not in shape
assert shape["username"] == "eve"
assert shape["role"] == "admin"
assert "id" in shape and "createdAt" in shape
- Step 3: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_users.py -v 2>&1 | tail -15
Expected: all tests FAIL with ModuleNotFoundError: No module named 'cyclone.auth' or similar.
- Step 4: Implement users.py
backend/src/cyclone/auth/users.py:
"""User CRUD + bcrypt password hashing."""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
from passlib.hash import bcrypt
from sqlalchemy import select
from cyclone.db import User
def hash_password(plaintext: str) -> str:
return bcrypt.hash(plaintext)
def verify_password(plaintext: str, hashed: str) -> bool:
try:
return bcrypt.verify(plaintext, hashed)
except (ValueError, TypeError):
return False
def create(db, *, username: str, password: str, role: str) -> User:
user = User(
username=username,
password_hash=hash_password(password),
role=role,
created_at=datetime.now(timezone.utc),
)
db.add(user)
db.commit()
db.refresh(user)
return user
def get_by_username(db, username: str) -> User | None:
return db.execute(
select(User).where(User.username == username)
).scalar_one_or_none()
def get(db, user_id: int) -> User | None:
return db.get(User, user_id)
def disable(db, user_id: int) -> None:
user = db.get(User, user_id)
if user is None:
return
user.disabled_at = datetime.now(timezone.utc)
db.commit()
def update_role(db, user_id: int, role: str) -> None:
user = db.get(User, user_id)
if user is None:
return
user.role = role
db.commit()
def update_password(db, user_id: int, new_password: str) -> None:
user = db.get(User, user_id)
if user is None:
return
user.password_hash = hash_password(new_password)
db.commit()
def to_public(user: User) -> dict[str, Any]:
return {
"id": user.id,
"username": user.username,
"role": user.role,
"createdAt": user.created_at.isoformat() if user.created_at else None,
"disabledAt": user.disabled_at.isoformat() if user.disabled_at else None,
}
- Step 5: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_users.py -v 2>&1 | tail -15
Expected: all 10 tests pass.
- Step 6: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/users.py backend/tests/test_auth_users.py backend/src/cyclone/auth/__init__.py && git commit -m "feat(auth): users module with bcrypt hashing + CRUD"
Task 6: TDD — sessions.py
Files:
-
Create:
backend/src/cyclone/auth/sessions.py -
Create:
backend/tests/test_auth_sessions.py -
Step 1: Write failing tests
backend/tests/test_auth_sessions.py:
"""Unit tests for cyclone.auth.sessions — Session create/validate/expire/touch."""
from __future__ import annotations
import secrets
from datetime import datetime, timedelta, timezone
import pytest
from sqlalchemy import delete
from cyclone.auth import sessions
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
def _make_user():
from cyclone.auth import users
with SessionLocal() as db:
return users.create(db, username="sessuser", password="hunter2hunter2", role="user")
def test_create_session_returns_id_and_session():
user = _make_user()
with SessionLocal() as db:
sid, sess = sessions.create(db, user_id=user.id)
assert len(sid) >= 32
assert sess.user_id == user.id
assert sess.expires_at > datetime.now(timezone.utc)
def test_get_valid_returns_session_for_active():
user = _make_user()
with SessionLocal() as db:
sid, _ = sessions.create(db, user_id=user.id)
with SessionLocal() as db:
got = sessions.get_valid(db, sid)
assert got is not None
assert got.user_id == user.id
def test_get_valid_returns_none_for_missing():
with SessionLocal() as db:
assert sessions.get_valid(db, "does-not-exist") is None
def test_get_valid_returns_none_for_expired():
user = _make_user()
with SessionLocal() as db:
sid, _ = sessions.create(db, user_id=user.id)
with SessionLocal() as db:
sess = sessions.get_valid(db, sid)
# Backdate expiry.
sess.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
db.commit()
with SessionLocal() as db:
assert sessions.get_valid(db, sid) is None
def test_delete_removes_session():
user = _make_user()
with SessionLocal() as db:
sid, _ = sessions.create(db, user_id=user.id)
sessions.delete(db, sid)
with SessionLocal() as db:
assert sessions.get_valid(db, sid) is None
def test_touch_extends_expiry():
user = _make_user()
with SessionLocal() as db:
sid, sess = sessions.create(db, user_id=user.id)
original_expiry = sess.expires_at
sessions.touch(db, sid)
with SessionLocal() as db:
refreshed = sessions.get_valid(db, sid)
assert refreshed.expires_at >= original_expiry
- Step 2: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_sessions.py -v 2>&1 | tail -10
Expected: ModuleNotFoundError or AttributeError.
- Step 3: Implement sessions.py
backend/src/cyclone/auth/sessions.py:
"""Session create/validate/expire/touch."""
from __future__ import annotations
import secrets
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from cyclone.db import Session
SESSION_LIFETIME = timedelta(hours=24)
def create(db, *, user_id: int) -> tuple[str, Session]:
sid = secrets.token_urlsafe(32)
now = datetime.now(timezone.utc)
sess = Session(
id=sid,
user_id=user_id,
expires_at=now + SESSION_LIFETIME,
created_at=now,
)
db.add(sess)
db.commit()
db.refresh(sess)
return sid, sess
def get_valid(db, sid: str) -> Session | None:
sess = db.execute(
select(Session).where(Session.id == sid)
).scalar_one_or_none()
if sess is None:
return None
if sess.expires_at <= datetime.now(timezone.utc):
return None
return sess
def delete(db, sid: str) -> None:
sess = db.get(Session, sid)
if sess is None:
return
db.delete(sess)
db.commit()
def touch(db, sid: str) -> None:
sess = db.get(Session, sid)
if sess is None:
return
sess.expires_at = datetime.now(timezone.utc) + SESSION_LIFETIME
db.commit()
- Step 4: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_sessions.py -v 2>&1 | tail -10
Expected: all 6 tests pass.
- Step 5: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/sessions.py backend/tests/test_auth_sessions.py && git commit -m "feat(auth): sessions module — create/validate/expire/touch"
Task 7: permissions.py + rate_limit.py (no TDD, pure data + tiny helpers)
Files:
-
Create:
backend/src/cyclone/auth/permissions.py -
Create:
backend/src/cyclone/auth/rate_limit.py -
Step 1: Create permissions.py
"""Role enum + PERMISSIONS matrix."""
from __future__ import annotations
from enum import Enum
class Role(str, Enum):
ADMIN = "admin"
USER = "user"
VIEWER = "viewer"
ALL_ROLES = {Role.ADMIN, Role.USER, Role.VIEWER}
WRITE_ROLES = {Role.ADMIN, Role.USER}
ADMIN_ONLY = {Role.ADMIN}
# (method, path-prefix) → allowed roles.
# Endpoints not in this matrix default to DENY (fail-closed).
# Use exact paths or longest-prefix match.
PERMISSIONS: dict[tuple[str, str], set[Role]] = {
# Public paths.
("GET", "/api/healthz"): set(),
("POST", "/api/auth/login"): set(),
# Auth surface (authenticated, all roles).
("POST", "/api/auth/logout"): ALL_ROLES,
("GET", "/api/auth/me"): ALL_ROLES,
# Admin-only user management.
("GET", "/api/admin/users"): ADMIN_ONLY,
("POST", "/api/admin/users"): ADMIN_ONLY,
("PATCH", "/api/admin/users"): ADMIN_ONLY,
("DELETE", "/api/admin/users"): ADMIN_ONLY,
# Read endpoints.
("GET", "/api/claims"): ALL_ROLES,
("GET", "/api/remittances"): ALL_ROLES,
("GET", "/api/providers"): ALL_ROLES,
("GET", "/api/batches"): ALL_ROLES,
("GET", "/api/dashboard/summary"): ALL_ROLES,
("GET", "/api/activity"): ALL_ROLES,
("GET", "/api/inbox/lanes"): ALL_ROLES,
("GET", "/api/reconcile"): ALL_ROLES,
("GET", "/api/audit-log"): ADMIN_ONLY,
# Write endpoints (admin + user, no viewer).
("POST", "/api/parse-837"): WRITE_ROLES,
("POST", "/api/parse-835"): WRITE_ROLES,
("POST", "/api/inbox"): WRITE_ROLES,
("POST", "/api/reconcile"): WRITE_ROLES,
("POST", "/api/resubmit"): WRITE_ROLES,
("POST", "/api/acks"): WRITE_ROLES,
# CSV export — read-only, all roles.
("GET", "/api/export.csv"): ALL_ROLES,
}
def allowed_roles(method: str, path: str) -> set[Role] | None:
"""Return the set of roles allowed to call (method, path), or None if denied.
Uses longest-prefix match on path; falls back to DENY (None) if no entry matches.
"""
candidates = [
(prefix_len, roles)
for (m, prefix), roles in PERMISSIONS.items()
if m == method and (path == prefix or path.startswith(prefix.rstrip("/") + "/"))
for prefix_len in [len(prefix)]
]
if not candidates:
return None
candidates.sort(key=lambda x: -x[0])
return candidates[0][1]
- Step 2: Create rate_limit.py
"""Per-username login rate limiter (in-memory, per-process)."""
from __future__ import annotations
import time
from threading import Lock
WINDOW_SECONDS = 300
MAX_FAILS = 5
_FAILS: dict[str, list[float]] = {}
_LOCK = Lock()
def check(username: str) -> int:
"""Return retry-after seconds, or 0 if allowed."""
now = time.monotonic()
with _LOCK:
fails = [t for t in _FAILS.get(username, []) if now - t < WINDOW_SECONDS]
_FAILS[username] = fails
if len(fails) >= MAX_FAILS:
return int(WINDOW_SECONDS - (now - fails[0]))
return 0
def record_failure(username: str) -> None:
now = time.monotonic()
with _LOCK:
_FAILS.setdefault(username, []).append(now)
def reset(username: str) -> None:
with _LOCK:
_FAILS.pop(username, None)
- Step 3: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/permissions.py backend/src/cyclone/auth/rate_limit.py && git commit -m "feat(auth): permissions matrix + login rate limiter"
Task 8: deps.py — get_current_user + require_role
Files:
-
Create:
backend/src/cyclone/auth/deps.py -
Step 1: Implement deps.py
"""FastAPI dependencies for auth."""
from __future__ import annotations
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.orm import Session as DbSession
from cyclone.auth import sessions, users
from cyclone.auth.permissions import Role, allowed_roles
from cyclone.db import SessionLocal
def _db():
db = SessionLocal()
try:
yield db
finally:
db.close()
DbSessionDep = Annotated[DbSession, Depends(_db)]
AUTH_DISABLED = False # flipped by bootstrap or env var
async def get_current_user(
request: Request,
db: DbSessionDep,
) -> dict:
"""Return the public User shape. Raises 401 if session is missing/expired.
When AUTH_DISABLED is True (dev escape hatch), returns a synthetic admin
user without checking credentials.
"""
if AUTH_DISABLED:
return {
"id": 0,
"username": "dev",
"role": Role.ADMIN.value,
"createdAt": None,
"disabledAt": None,
}
sid = request.cookies.get("cyclone_session")
if not sid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="session_expired",
)
sess = sessions.get_valid(db, sid)
if sess is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="session_expired",
)
user = users.get(db, sess.user_id)
if user is None or user.disabled_at is not None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="account_disabled",
)
# Sliding expiry: refresh both DB and cookie.
sessions.touch(db, sid)
request.state.user = user
request.state.session_id = sid
return users.to_public(user)
def require_role(*allowed: Role):
"""Dependency factory: gate the endpoint to specific roles.
Falls back to PERMISSIONS matrix lookup if no explicit roles given.
"""
async def _dep(
request: Request,
user: dict = Depends(get_current_user),
) -> dict:
user_role = user.get("role")
# If explicit allowed roles given, check those.
if allowed:
if user_role not in {r.value for r in allowed}:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="forbidden",
)
return user
# Otherwise consult the matrix.
method = request.method
path = request.url.path
roles = allowed_roles(method, path)
if roles is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="forbidden",
)
if user_role not in {r.value for r in roles}:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="forbidden",
)
return user
return _dep
- Step 2: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/deps.py && git commit -m "feat(auth): get_current_user + require_role dependencies"
Task 9: TDD — routes.py (login/logout/me)
Files:
-
Create:
backend/src/cyclone/auth/routes.py -
Create:
backend/tests/test_auth_routes.py -
Step 1: Write failing tests
backend/tests/test_auth_routes.py:
"""API tests for /api/auth/login, /api/auth/logout, /api/auth/me."""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete
from cyclone.api import app
from cyclone.auth import users
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
@pytest.fixture
def client():
return TestClient(app)
@pytest.fixture
def seeded_admin(client):
with SessionLocal() as db:
users.create(db, username="admin", password="adminpassword1", role="admin")
return client
def test_login_success_returns_user_and_cookie(client):
with SessionLocal() as db:
users.create(db, username="alice", password="hunter2hunter2", role="user")
resp = client.post(
"/api/auth/login",
json={"username": "alice", "password": "hunter2hunter2"},
)
assert resp.status_code == 200
body = resp.json()
assert body["username"] == "alice"
assert body["role"] == "user"
assert "password_hash" not in body
# Cookie set with HttpOnly.
assert "cyclone_session" in resp.cookies
assert resp.cookies["cyclone_session"]
def test_login_bad_password_returns_401(client):
with SessionLocal() as db:
users.create(db, username="bob", password="hunter2hunter2", role="user")
resp = client.post(
"/api/auth/login",
json={"username": "bob", "password": "WRONG"},
)
assert resp.status_code == 401
assert resp.json()["error"] == "invalid_credentials"
# No cookie set.
assert "cyclone_session" not in resp.cookies
def test_login_unknown_user_returns_401(client):
resp = client.post(
"/api/auth/login",
json={"username": "ghost", "password": "whatever"},
)
assert resp.status_code == 401
assert resp.json()["error"] == "invalid_credentials"
def test_login_disabled_user_returns_403(client):
with SessionLocal() as db:
u = users.create(db, username="carol", password="hunter2hunter2", role="user")
users.disable(db, u.id)
resp = client.post(
"/api/auth/login",
json={"username": "carol", "password": "hunter2hunter2"},
)
assert resp.status_code == 403
assert resp.json()["error"] == "account_disabled"
def test_logout_clears_session_and_cookie(seeded_admin):
# Log in.
login = seeded_admin.post(
"/api/auth/login",
json={"username": "admin", "password": "adminpassword1"},
)
cookie = login.cookies.get("cyclone_session")
assert cookie
# Logout.
resp = seeded_admin.post("/api/auth/logout", cookies={"cyclone_session": cookie})
assert resp.status_code == 204
# Session row deleted.
from cyclone.auth import sessions
with SessionLocal() as db:
assert sessions.get_valid(db, cookie) is None
def test_me_returns_current_user(seeded_admin):
login = seeded_admin.post(
"/api/auth/login",
json={"username": "admin", "password": "adminpassword1"},
)
cookie = login.cookies.get("cyclone_session")
resp = seeded_admin.get("/api/auth/me", cookies={"cyclone_session": cookie})
assert resp.status_code == 200
assert resp.json()["username"] == "admin"
def test_me_without_cookie_returns_401(seeded_admin):
resp = seeded_admin.get("/api/auth/me")
assert resp.status_code == 401
(Note: tests reference resp.json()["error"] — we'll need the routes to use a custom exception handler that returns {"error": "...", "detail": "..."}. Add that handler in this task.)
- Step 2: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_routes.py -v 2>&1 | tail -10
Expected: ModuleNotFoundError or 404 on routes.
- Step 3: Implement routes.py
backend/src/cyclone/auth/routes.py:
"""/api/auth/login, /api/auth/logout, /api/auth/me."""
from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, HTTPException, Request, Response, status
from sqlalchemy import delete
from cyclone.auth import rate_limit, sessions, users
from cyclone.auth.deps import get_current_user
from cyclone.db import SessionLocal
router = APIRouter(prefix="/api/auth", tags=["auth"])
COOKIE_NAME = "cyclone_session"
COOKIE_MAX_AGE = 86400 # 24h
def _is_https(request: Request) -> bool:
if request.url.scheme == "https":
return True
return os.environ.get("CYCLONE_BEHIND_HTTPS") == "1"
def _set_cookie(response: Response, sid: str, request: Request) -> None:
response.set_cookie(
key=COOKIE_NAME,
value=sid,
max_age=COOKIE_MAX_AGE,
path="/api",
httponly=True,
samesite="lax",
secure=_is_https(request),
)
def _clear_cookie(response: Response) -> None:
response.delete_cookie(key=COOKIE_NAME, path="/api")
@router.post("/login")
def login(body: dict, request: Request, response: Response):
username = (body.get("username") or "").strip()
password = body.get("password") or ""
if not username or not password:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="username and password are required",
)
# Rate limit (per-username).
retry_after = rate_limit.check(username)
if retry_after > 0:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="rate_limited",
headers={"Retry-After": str(retry_after)},
)
with SessionLocal() as db:
user = users.get_by_username(db, username)
if user is None or not users.verify_password(password, user.password_hash):
rate_limit.record_failure(username)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="invalid_credentials",
)
if user.disabled_at is not None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="account_disabled",
)
sid, _ = sessions.create(db, user_id=user.id)
rate_limit.reset(username)
public = users.to_public(user)
_set_cookie(response, sid, request)
return public
@router.post("/logout")
def logout(request: Request, response: Response, _user=__import__("fastapi").Depends(__import__("cyclone.auth.deps", fromlist=["get_current_user"]).get_current_user)):
sid = request.cookies.get(COOKIE_NAME)
if sid:
with SessionLocal() as db:
sessions.delete(db, sid)
_clear_cookie(response)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.get("/me")
def me(user=__import__("fastapi").Depends(__import__("cyclone.auth.deps", fromlist=["get_current_user"]).get_current_user)):
return user
Replace the awkward __import__ calls in /logout and /me with a clean import at the top:
from fastapi import Depends
@router.post("/logout")
def logout(
request: Request,
response: Response,
_user: dict = Depends(get_current_user),
):
sid = request.cookies.get(COOKIE_NAME)
if sid:
with SessionLocal() as db:
sessions.delete(db, sid)
_clear_cookie(response)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.get("/me")
def me(user: dict = Depends(get_current_user)):
return user
- Step 4: Add error handler for
{"error": ..., "detail": ...}shape
In backend/src/cyclone/api.py (or wherever the FastAPI app is constructed), add:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
def _error_shape(detail: str) -> dict:
# Map known detail strings to short error codes; otherwise echo the detail.
return {"error": detail, "detail": detail}
@app.exception_handler(HTTPException)
async def _http_exc_handler(request: Request, exc: HTTPException):
code = exc.detail if isinstance(exc.detail, str) else "error"
return JSONResponse(
status_code=exc.status_code,
content={"error": code, "detail": str(exc.detail)},
headers=exc.headers,
)
- Step 5: Register the auth router
In backend/src/cyclone/api.py:
from cyclone.auth.routes import router as auth_router
app.include_router(auth_router)
- Step 6: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_routes.py -v 2>&1 | tail -15
Expected: all 7 tests pass.
- Step 7: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/routes.py backend/tests/test_auth_routes.py backend/src/cyclone/api.py && git commit -m "feat(auth): login/logout/me endpoints + error shape"
Task 10: TDD — admin.py (user management)
Files:
-
Create:
backend/src/cyclone/auth/admin.py -
Create:
backend/tests/test_auth_admin.py -
Step 1: Write failing tests
backend/tests/test_auth_admin.py:
"""Admin-only user management endpoints."""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete
from cyclone.api import app
from cyclone.auth import users
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
@pytest.fixture
def admin_client():
client = TestClient(app)
with SessionLocal() as db:
users.create(db, username="root", password="rootpassword1", role="admin")
login = client.post(
"/api/auth/login",
json={"username": "root", "password": "rootpassword1"},
)
assert login.status_code == 200
return client
@pytest.fixture
def user_client():
client = TestClient(app)
with SessionLocal() as db:
users.create(db, username="plain", password="plainpassword1", role="user")
login = client.post(
"/api/auth/login",
json={"username": "plain", "password": "plainpassword1"},
)
return client
def test_list_users_as_admin(admin_client):
with SessionLocal() as db:
users.create(db, username="alice", password="hunter2hunter2", role="user")
resp = admin_client.get("/api/admin/users")
assert resp.status_code == 200
body = resp.json()
usernames = {u["username"] for u in body}
assert {"root", "alice"} <= usernames
def test_list_users_as_nonadmin_returns_403(user_client):
resp = user_client.get("/api/admin/users")
assert resp.status_code == 403
def test_create_user_as_admin(admin_client):
resp = admin_client.post(
"/api/admin/users",
json={"username": "newbie", "password": "newbiepassword1", "role": "viewer"},
)
assert resp.status_code == 201
assert resp.json()["username"] == "newbie"
assert resp.json()["role"] == "viewer"
def test_create_user_rejects_invalid_role(admin_client):
resp = admin_client.post(
"/api/admin/users",
json={"username": "badrole", "password": "hunter2hunter2", "role": "owner"},
)
assert resp.status_code == 422
def test_patch_user_role_and_password(admin_client):
with SessionLocal() as db:
u = users.create(db, username="subject", password="hunter2hunter2", role="viewer")
resp = admin_client.patch(
f"/api/admin/users/{u.id}",
json={"role": "user", "password": "newpassword1"},
)
assert resp.status_code == 200
assert resp.json()["role"] == "user"
# Old password no longer works.
client = TestClient(app)
bad = client.post(
"/api/auth/login",
json={"username": "subject", "password": "hunter2hunter2"},
)
assert bad.status_code == 401
def test_admin_cannot_demote_self(admin_client):
me = admin_client.get("/api/auth/me").json()
resp = admin_client.patch(
f"/api/admin/users/{me['id']}",
json={"role": "viewer"},
)
assert resp.status_code == 409
def test_admin_can_disable_user(admin_client):
with SessionLocal() as db:
u = users.create(db, username="victim", password="hunter2hunter2", role="user")
resp = admin_client.patch(
f"/api/admin/users/{u.id}",
json={"disabled": True},
)
assert resp.status_code == 200
assert resp.json()["disabledAt"] is not None
- Step 2: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_admin.py -v 2>&1 | tail -10
Expected: 404 / ModuleNotFoundError.
- Step 3: Implement admin.py
backend/src/cyclone/auth/admin.py:
"""Admin-only user management: GET/POST/PATCH/DELETE /api/admin/users."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Request, status
from cyclone.auth import users
from cyclone.auth.deps import get_current_user
from cyclone.auth.permissions import Role
from cyclone.db import SessionLocal
router = APIRouter(prefix="/api/admin/users", tags=["admin"])
def _require_admin(user: dict = Depends(get_current_user)) -> dict:
if user.get("role") != Role.ADMIN.value:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="forbidden",
)
return user
def _validate_role(role: str) -> None:
valid = {Role.ADMIN.value, Role.USER.value, Role.VIEWER.value}
if role not in valid:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"role must be one of {sorted(valid)}",
)
@router.get("")
def list_users(_admin=Depends(_require_admin)):
with SessionLocal() as db:
all_users = db.query(users.User if hasattr(users, "User") else __import__("cyclone.db", fromlist=["User"]).User).all()
return [users.to_public(u) for u in all_users]
@router.post("", status_code=status.HTTP_201_CREATED)
def create_user(body: dict, _admin=Depends(_require_admin)):
username = (body.get("username") or "").strip()
password = body.get("password") or ""
role = body.get("role") or ""
if not username or len(username) < 3:
raise HTTPException(status_code=422, detail="username must be at least 3 chars")
if len(password) < 12:
raise HTTPException(status_code=422, detail="password must be at least 12 chars")
_validate_role(role)
with SessionLocal() as db:
if users.get_by_username(db, username) is not None:
raise HTTPException(status_code=409, detail="username already exists")
u = users.create(db, username=username, password=password, role=role)
return users.to_public(u)
@router.patch("/{user_id}")
def patch_user(user_id: int, body: dict, request: Request, admin=Depends(_require_admin)):
me = admin
if me.get("id") == user_id and body.get("role") and body["role"] != Role.ADMIN.value:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="cannot_demote_self",
)
with SessionLocal() as db:
if body.get("role") is not None:
_validate_role(body["role"])
users.update_role(db, user_id, body["role"])
if body.get("password") is not None:
if len(body["password"]) < 12:
raise HTTPException(status_code=422, detail="password must be at least 12 chars")
users.update_password(db, user_id, body["password"])
if body.get("disabled") is True:
users.disable(db, user_id)
u = users.get(db, user_id)
if u is None:
raise HTTPException(status_code=404, detail="user not found")
return users.to_public(u)
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, admin=Depends(_require_admin)):
me = admin
if me.get("id") == user_id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="cannot_delete_self",
)
with SessionLocal() as db:
u = users.get(db, user_id)
if u is None:
raise HTTPException(status_code=404, detail="user not found")
# For v1, just disable instead of hard-delete.
users.disable(db, user_id)
return None
- Step 4: Register the admin router
In backend/src/cyclone/api.py:
from cyclone.auth.admin import router as admin_users_router
app.include_router(admin_users_router)
- Step 5: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_admin.py -v 2>&1 | tail -15
Expected: all 7 tests pass.
- Step 6: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/admin.py backend/tests/test_auth_admin.py backend/src/cyclone/api.py && git commit -m "feat(auth): admin user management endpoints"
Task 11: TDD — permissions matrix on existing endpoints
Files:
-
Modify:
backend/src/cyclone/api.py— gate every existing endpoint -
Create:
backend/tests/test_existing_endpoints_require_auth.py -
Step 1: Write tests
backend/tests/test_existing_endpoints_require_auth.py:
"""Spot-check that existing endpoints now require auth."""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete
from cyclone.api import app
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
@pytest.fixture
def client():
return TestClient(app)
# Adjust paths to match actual project endpoints.
@pytest.mark.parametrize("method,path", [
("GET", "/api/claims"),
("GET", "/api/remittances"),
("GET", "/api/providers"),
("GET", "/api/batches"),
("GET", "/api/dashboard/summary"),
("GET", "/api/activity"),
])
def test_existing_get_endpoints_require_auth(client, method, path):
resp = client.request(method, path)
assert resp.status_code == 401, f"{method} {path} returned {resp.status_code}"
def test_healthz_is_public(client):
resp = client.get("/api/healthz")
# 200 if healthy, but never 401.
assert resp.status_code != 401
Adjust the endpoint paths to match what the existing project actually exposes — peek at cyclone/api.py for the actual list.
- Step 2: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_existing_endpoints_require_auth.py -v 2>&1 | tail -15
Expected: most tests FAIL (200 instead of 401) because the endpoints aren't gated yet.
- Step 3: Add the dependencies to existing endpoints
For each existing endpoint in backend/src/cyclone/api.py:
- Add
user: dict = Depends(get_current_user)(orrequire_role(...)) to the signature. - Add
Dependsto the imports.
Example transformation:
# Before:
@app.get("/api/claims")
def list_claims(...):
...
# After:
@app.get("/api/claims", dependencies=[Depends(require_role())]) # matrix lookup
def list_claims(...):
...
Or for endpoints that should always succeed for any authed user, use Depends(get_current_user) directly.
- Step 4: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_existing_endpoints_require_auth.py -v 2>&1 | tail -15
- Step 5: Run the full backend test suite
cd /home/tyler/dev/cyclone/backend && uv run pytest 2>&1 | tail -10
Expected: pre-existing 723 tests pass, new tests pass. Fix any regressions.
- Step 6: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/api.py backend/tests/test_existing_endpoints_require_auth.py && git commit -m "feat(auth): gate existing endpoints behind get_current_user"
Task 12: TDD — bootstrap (first admin)
Files:
-
Modify:
backend/src/cyclone/__main__.py -
Create:
backend/src/cyclone/auth/bootstrap.py -
Create:
backend/tests/test_auth_bootstrap.py -
Step 1: Write failing tests
backend/tests/test_auth_bootstrap.py:
"""Bootstrap admin user on backend startup."""
from __future__ import annotations
import pytest
from sqlalchemy import delete, select
from cyclone.auth import bootstrap, users
from cyclone.auth.permissions import Role
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear(monkeypatch):
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
def test_bootstrap_creates_admin_when_users_empty_and_env_set(monkeypatch):
monkeypatch.setenv("CYCLONE_ADMIN_USERNAME", "firstadmin")
monkeypatch.setenv("CYCLONE_ADMIN_PASSWORD", "firstadminpw1")
bootstrap.run()
with SessionLocal() as db:
u = db.execute(select(User).where(User.username == "firstadmin")).scalar_one()
assert u.role == Role.ADMIN.value
def test_bootstrap_noop_when_users_exist(monkeypatch):
monkeypatch.setenv("CYCLONE_ADMIN_USERNAME", "ignored")
monkeypatch.setenv("CYCLONE_ADMIN_PASSWORD", "ignoredignored1")
with SessionLocal() as db:
users.create(db, username="existing", password="hunter2hunter2", role="admin")
bootstrap.run()
with SessionLocal() as db:
all_users = db.execute(select(User)).scalars().all()
usernames = {u.username for u in all_users}
assert usernames == {"existing"}
def test_bootstrap_refuses_to_run_without_env(monkeypatch):
monkeypatch.delenv("CYCLONE_ADMIN_USERNAME", raising=False)
monkeypatch.delenv("CYCLONE_ADMIN_PASSWORD", raising=False)
with pytest.raises(RuntimeError, match="CYCLONE_ADMIN_USERNAME"):
bootstrap.run()
def test_bootstrap_rejects_short_password(monkeypatch):
monkeypatch.setenv("CYCLONE_ADMIN_USERNAME", "weak")
monkeypatch.setenv("CYCLONE_ADMIN_PASSWORD", "short")
with pytest.raises(RuntimeError, match="12 characters"):
bootstrap.run()
- Step 2: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_bootstrap.py -v 2>&1 | tail -10
Expected: ModuleNotFoundError.
- Step 3: Implement bootstrap.py
backend/src/cyclone/auth/bootstrap.py:
"""First-admin bootstrap: create the initial admin from env vars if no users exist."""
from __future__ import annotations
import os
from sqlalchemy import select
from cyclone.auth import users
from cyclone.auth.permissions import Role
from cyclone.db import SessionLocal, User
def run() -> None:
if os.environ.get("CYCLONE_AUTH_DISABLED") == "1":
# Dev escape hatch — skip bootstrap entirely.
from cyclone.auth import deps
deps.AUTH_DISABLED = True
return
username = os.environ.get("CYCLONE_ADMIN_USERNAME")
password = os.environ.get("CYCLONE_ADMIN_PASSWORD")
with SessionLocal() as db:
existing = db.execute(select(User)).scalars().first()
if existing is not None:
return # users exist — nothing to bootstrap
if not username or not password:
raise RuntimeError(
"Cyclone has no users yet. Set CYCLONE_ADMIN_USERNAME and "
"CYCLONE_ADMIN_PASSWORD env vars (min 12 chars), or run "
"`python -m cyclone users create <username> --role admin`."
)
if len(password) < 12:
raise RuntimeError(
"CYCLONE_ADMIN_PASSWORD must be at least 12 characters."
)
users.create(db, username=username, password=password, role=Role.ADMIN.value)
print(f"[cyclone] bootstrap admin user '{username}' created")
- Step 4: Wire bootstrap into __main__.py
In backend/src/cyclone/__main__.py, find the entry point (the serve command or similar). Before uvicorn.run(...), call bootstrap.run():
from cyclone.auth import bootstrap
def main():
bootstrap.run()
# ... existing serve / cli dispatch
- Step 5: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_bootstrap.py -v 2>&1 | tail -10
Expected: all 4 tests pass.
- Step 6: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/bootstrap.py backend/src/cyclone/__main__.py backend/tests/test_auth_bootstrap.py && git commit -m "feat(auth): bootstrap first admin from env vars"
Task 13: TDD — rate limit + login fails are throttled
Files:
-
Create:
backend/tests/test_auth_login_rate_limit.py -
Step 1: Write tests
backend/tests/test_auth_login_rate_limit.py:
"""Login rate limit: 5 fails / 5 min per username."""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete
from cyclone.api import app
from cyclone.auth import rate_limit
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
rate_limit.reset("victim")
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
rate_limit.reset("victim")
def test_5_fails_then_429():
with SessionLocal() as db:
from cyclone.auth import users
users.create(db, username="victim", password="hunter2hunter2", role="user")
client = TestClient(app)
for _ in range(5):
r = client.post(
"/api/auth/login",
json={"username": "victim", "password": "WRONG"},
)
assert r.status_code == 401
# 6th should be 429.
r = client.post(
"/api/auth/login",
json={"username": "victim", "password": "WRONG"},
)
assert r.status_code == 429
assert "Retry-After" in r.headers
def test_successful_login_resets_counter():
with SessionLocal() as db:
from cyclone.auth import users
users.create(db, username="victim", password="hunter2hunter2", role="user")
client = TestClient(app)
for _ in range(4):
r = client.post(
"/api/auth/login",
json={"username": "victim", "password": "WRONG"},
)
assert r.status_code == 401
# Successful login.
r = client.post(
"/api/auth/login",
json={"username": "victim", "password": "hunter2hunter2"},
)
assert r.status_code == 200
# Counter reset — 5 more fails allowed.
for _ in range(5):
r = client.post(
"/api/auth/login",
json={"username": "victim", "password": "WRONG"},
)
assert r.status_code == 401
- Step 2: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_auth_login_rate_limit.py -v 2>&1 | tail -10
Expected: both tests pass (rate limit was implemented in Task 7, this just exercises it).
- Step 3: Commit
cd /home/tyler/dev/cyclone && git add backend/tests/test_auth_login_rate_limit.py && git commit -m "test(auth): login rate limit"
Task 14: TDD — audit log records user_id
Files:
-
Modify:
backend/src/cyclone/audit_log.py— acceptuser_idparameter -
Create:
backend/tests/test_audit_log_user_id.py -
Step 1: Inspect audit_log.py
cd /home/tyler/dev/cyclone/backend && grep -n "def " src/cyclone/audit_log.py | head -10
- Step 2: Write failing tests
backend/tests/test_audit_log_user_id.py:
"""Audit log entries record the acting user_id."""
from __future__ import annotations
import pytest
from sqlalchemy import delete, select
from cyclone.audit_log import log_event # adjust to actual function name
from cyclone.db import audit_log_table # adjust
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
# Truncate audit_log.
db.execute(__import__("cyclone.db", fromlist=["audit_log_table"]).audit_log_table.delete())
db.execute(delete(User))
db.commit()
yield
def test_log_event_accepts_user_id_kwarg():
log_event(kind="test", payload={"foo": "bar"}, user_id=42)
with SessionLocal() as db:
rows = db.execute(select(__import__("cyclone.db", fromlist=["audit_log_table"]).audit_log_table)).all()
assert rows[-1].user_id == 42
def test_log_event_omits_user_id_is_null():
log_event(kind="test", payload={"foo": "bar"})
with SessionLocal() as db:
rows = db.execute(select(__import__("cyclone.db", fromlist=["audit_log_table"]).audit_log_table)).all()
assert rows[-1].user_id is None
Adjust to match the actual audit_log API. The test verifies both forms work.
- Step 3: Run tests — confirm failure
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_audit_log_user_id.py -v 2>&1 | tail -10
- Step 4: Update audit_log.py
In backend/src/cyclone/audit_log.py, find log_event (or whatever the public function is). Add user_id: int | None = None parameter and pass it through to the SQL insert:
def log_event(*, kind: str, payload: dict, user_id: int | None = None) -> None:
# ... insert row with user_id=user_id
- Step 5: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_audit_log_user_id.py -v 2>&1 | tail -10
- Step 6: Wire user_id into existing call sites
In backend/src/cyclone/api.py, every place that calls audit_log.log_event, pass user_id=user.get("id"):
audit_log.log_event(kind="parse_success", payload={...}, user_id=user.get("id"))
- Step 7: Run full backend suite
cd /home/tyler/dev/cyclone/backend && uv run pytest 2>&1 | tail -10
- Step 8: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/audit_log.py backend/src/cyclone/api.py backend/tests/test_audit_log_user_id.py && git commit -m "feat(audit): record user_id on log events"
Task 15: TDD — CLI users subcommand
Files:
-
Create:
backend/src/cyclone/auth/cli.py -
Modify:
backend/src/cyclone/__main__.py -
Create:
backend/tests/test_cli_users.py -
Step 1: Write failing tests
backend/tests/test_cli_users.py:
"""CLI: python -m cyclone users {create,list,disable,reset-password,set-role}."""
from __future__ import annotations
import pytest
from sqlalchemy import delete, select
from cyclone.auth import users
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
def test_create_user_via_cli(monkeypatch, capsys):
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "create", "cli-user",
"--role", "viewer", "--password", "clipassword1"])
cli_main()
with SessionLocal() as db:
u = users.get_by_username(db, "cli-user")
assert u is not None
assert u.role == "viewer"
def test_list_users_via_cli(monkeypatch, capsys):
with SessionLocal() as db:
users.create(db, username="listed", password="hunter2hunter2", role="user")
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "list"])
cli_main()
out = capsys.readouterr().out
assert "listed" in out
def test_disable_user_via_cli(monkeypatch):
with SessionLocal() as db:
users.create(db, username="todie", password="hunter2hunter2", role="user")
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "disable", "todie"])
cli_main()
with SessionLocal() as db:
u = users.get_by_username(db, "todie")
assert u.disabled_at is not None
def test_reset_password_via_cli(monkeypatch):
with SessionLocal() as db:
users.create(db, username="pwchange", password="oldpassword1", role="user")
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "reset-password", "pwchange",
"--password", "newpassword1"])
cli_main()
with SessionLocal() as db:
u = users.get_by_username(db, "pwchange")
assert users.verify_password("newpassword1", u.password_hash)
def test_set_role_via_cli(monkeypatch):
with SessionLocal() as db:
users.create(db, username="promote", password="hunter2hunter2", role="viewer")
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "set-role", "promote", "--role", "admin"])
cli_main()
with SessionLocal() as db:
u = users.get_by_username(db, "promote")
assert u.role == "admin"
def test_create_rejects_short_password(monkeypatch):
from cyclone.auth.cli import main as cli_main
monkeypatch.setattr("sys.argv", ["cyclone", "users", "create", "weak",
"--role", "viewer", "--password", "short"])
with pytest.raises(SystemExit):
cli_main()
- Step 2: Implement cli.py
backend/src/cyclone/auth/cli.py:
"""CLI subcommand: `python -m cyclone users ...`."""
from __future__ import annotations
import argparse
import getpass
import sys
from cyclone.auth import users
from cyclone.auth.permissions import Role
from cyclone.db import SessionLocal
def _prompt_password(label: str) -> str:
pw = getpass.getpass(f"{label}: ")
if not pw:
print("Password required.", file=sys.stderr)
sys.exit(2)
return pw
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="cyclone users")
sub = parser.add_subparsers(dest="cmd", required=True)
create_p = sub.add_parser("create")
create_p.add_argument("username")
create_p.add_argument("--role", required=True, choices=[Role.ADMIN.value, Role.USER.value, Role.VIEWER.value])
create_p.add_argument("--password")
sub.add_parser("list")
disable_p = sub.add_parser("disable")
disable_p.add_argument("username")
reset_p = sub.add_parser("reset-password")
reset_p.add_argument("username")
reset_p.add_argument("--password")
setrole_p = sub.add_parser("set-role")
setrole_p.add_argument("username")
setrole_p.add_argument("--role", required=True, choices=[Role.ADMIN.value, Role.USER.value, Role.VIEWER.value])
args = parser.parse_args(argv)
with SessionLocal() as db:
if args.cmd == "create":
pw = args.password or _prompt_password("Password")
if len(pw) < 12:
print("Password must be at least 12 characters.", file=sys.stderr)
return 2
if users.get_by_username(db, args.username) is not None:
print(f"User '{args.username}' already exists.", file=sys.stderr)
return 1
u = users.create(db, username=args.username, password=pw, role=args.role)
print(f"Created user '{u.username}' with role '{u.role}'.")
return 0
if args.cmd == "list":
for u in db.query(__import__("cyclone.db", fromlist=["User"]).User).all():
print(f"{u.id}\t{u.username}\t{u.role}\t{'disabled' if u.disabled_at else 'active'}")
return 0
if args.cmd == "disable":
u = users.get_by_username(db, args.username)
if u is None:
print(f"No such user: {args.username}", file=sys.stderr)
return 1
users.disable(db, u.id)
print(f"Disabled '{args.username}'.")
return 0
if args.cmd == "reset-password":
pw = args.password or _prompt_password("New password")
if len(pw) < 12:
print("Password must be at least 12 characters.", file=sys.stderr)
return 2
u = users.get_by_username(db, args.username)
if u is None:
print(f"No such user: {args.username}", file=sys.stderr)
return 1
users.update_password(db, u.id, pw)
print(f"Password reset for '{args.username}'.")
return 0
if args.cmd == "set-role":
u = users.get_by_username(db, args.username)
if u is None:
print(f"No such user: {args.username}", file=sys.stderr)
return 1
users.update_role(db, u.id, args.role)
print(f"Role for '{args.username}' set to '{args.role}'.")
return 0
return 1
- Step 3: Wire into __main__.py
In backend/src/cyclone/__main__.py, before the serve command dispatches, check sys.argv:
def main():
if len(sys.argv) >= 2 and sys.argv[1] == "users":
from cyclone.auth.cli import main as users_cli
sys.exit(users_cli(sys.argv[2:]))
# ... existing dispatch
- Step 4: Run tests — confirm pass
cd /home/tyler/dev/cyclone/backend && uv run pytest tests/test_cli_users.py -v 2>&1 | tail -10
Expected: all 6 tests pass.
- Step 5: Commit
cd /home/tyler/dev/cyclone && git add backend/src/cyclone/auth/cli.py backend/src/cyclone/__main__.py backend/tests/test_cli_users.py && git commit -m "feat(auth): CLI users subcommand"
Phase 2: Frontend Foundations
Task 16: Add User type
Files:
-
Modify:
src/types/index.ts -
Step 1: Append User type
At the end of src/types/index.ts:
export interface User {
id: number;
username: string;
role: "admin" | "user" | "viewer";
createdAt: string | null;
disabledAt?: string | null;
}
- Step 2: Commit
cd /home/tyler/dev/cyclone && git add src/types/index.ts && git commit -m "feat(types): add User type"
Task 17: TDD — auth/api.ts (fetch wrapper for 401)
Files:
-
Create:
src/auth/api.ts -
Create:
src/auth/api.test.ts -
Step 1: Write failing tests
src/auth/api.test.ts:
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
describe("auth/api fetch wrapper", () => {
let originalFetch: typeof globalThis.fetch;
let originalLocation: Location;
beforeEach(() => {
originalFetch = globalThis.fetch;
originalLocation = window.location;
delete (window as any).__navCalls;
(window as any).__navCalls = [];
});
afterEach(() => {
globalThis.fetch = originalFetch;
window.location = originalLocation as any;
});
it("redirects to /login on 401 response", async () => {
globalThis.fetch = vi.fn().mockResolvedValue({
ok: false,
status: 401,
statusText: "Unauthorized",
headers: new Headers(),
json: async () => ({ error: "session_expired" }),
} as Response);
// Stub window.location.href setter.
let href = originalLocation.href;
Object.defineProperty(window, "location", {
configurable: true,
get: () => ({
...originalLocation,
get href() { return href; },
set href(v: string) { (window as any).__navCalls.push(v); href = v; },
}),
});
const { authedFetch } = await import("./api");
await expect(authedFetch("/api/anything")).rejects.toThrow();
expect((window as any).__navCalls).toContainEqual(expect.stringContaining("/login"));
});
it("does NOT redirect on 403", async () => {
globalThis.fetch = vi.fn().mockResolvedValue({
ok: false,
status: 403,
statusText: "Forbidden",
headers: new Headers(),
json: async () => ({ error: "forbidden" }),
} as Response);
Object.defineProperty(window, "location", {
configurable: true,
get: () => ({ ...originalLocation, set href(_: string) { throw new Error("should not nav"); } }),
});
const { authedFetch } = await import("./api");
await expect(authedFetch("/api/anything")).rejects.toThrow();
});
it("returns parsed JSON on 200", async () => {
globalThis.fetch = vi.fn().mockResolvedValue({
ok: true,
status: 200,
headers: new Headers(),
json: async () => ({ hello: "world" }),
} as Response);
const { authedFetch } = await import("./api");
const data = await authedFetch("/api/anything");
expect(data).toEqual({ hello: "world" });
});
});
- Step 2: Implement api.ts
src/auth/api.ts:
const BASE_URL = (import.meta.env.VITE_API_BASE_URL as string | undefined) ?? "";
function joinUrl(path: string): string {
if (BASE_URL) return `${BASE_URL}${path}`;
return path;
}
export class ApiError extends Error {
status: number;
code: string;
constructor(status: number, code: string, detail?: string) {
super(detail ?? code);
this.status = status;
this.code = code;
}
}
function redirectToLogin() {
const next = encodeURIComponent(window.location.pathname + window.location.search);
window.location.href = `/login?next=${next}`;
}
export async function authedFetch<T = unknown>(
path: string,
init?: RequestInit
): Promise<T> {
const res = await fetch(joinUrl(path), {
credentials: "include",
headers: { Accept: "application/json", ...(init?.headers ?? {}) },
...init,
});
if (res.status === 401 && !path.startsWith("/api/auth/")) {
redirectToLogin();
throw new ApiError(401, "session_expired");
}
if (!res.ok) {
let body: any = null;
try { body = await res.json(); } catch { /* no body */ }
const code = body?.error ?? "error";
const detail = body?.detail ?? res.statusText;
throw new ApiError(res.status, code, detail);
}
if (res.status === 204) return undefined as T;
return (await res.json()) as T;
}
// Auth-specific endpoints.
export const authApi = {
async login(username: string, password: string) {
const res = await fetch(joinUrl("/api/auth/login"), {
method: "POST",
credentials: "include",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ username, password }),
});
if (!res.ok) {
const body = await res.json().catch(() => ({}));
throw new ApiError(res.status, body.error ?? "error", body.detail ?? res.statusText);
}
return res.json();
},
async me() {
return authedFetch("/api/auth/me");
},
async logout() {
await fetch(joinUrl("/api/auth/logout"), {
method: "POST",
credentials: "include",
}).catch(() => undefined);
},
};
- Step 3: Run tests — confirm pass
cd /home/tyler/dev/cyclone && npx vitest run src/auth/api.test.ts 2>&1 | tail -15
- Step 4: Commit
git add src/auth/api.ts src/auth/api.test.ts && git commit -m "feat(auth): fetch wrapper with 401 redirect + authApi"
Task 18: TDD — AuthProvider + useAuth
Files:
-
Create:
src/auth/AuthProvider.tsx -
Create:
src/auth/useAuth.ts -
Create:
src/auth/AuthProvider.test.tsx -
Step 1: Write failing tests
src/auth/AuthProvider.test.tsx:
import { describe, it, expect, vi, beforeEach } from "vitest";
import { renderHook, act, waitFor } from "@testing-library/react";
import { AuthProvider, useAuth } from "./AuthProvider";
vi.mock("./api", () => ({
authApi: {
me: vi.fn(),
login: vi.fn(),
logout: vi.fn(),
},
}));
import { authApi } from "./api";
const wrapper = ({ children }: any) => <AuthProvider>{children}</AuthProvider>;
describe("AuthProvider", () => {
beforeEach(() => {
vi.resetAllMocks();
});
it("loads user on mount", async () => {
(authApi.me as any).mockResolvedValue({ id: 1, username: "alice", role: "user" });
const { result } = renderHook(() => useAuth(), { wrapper });
await waitFor(() => expect(result.current.status).toBe("authenticated"));
expect(result.current.user?.username).toBe("alice");
});
it("status is unauthenticated when /me fails", async () => {
(authApi.me as any).mockRejectedValue(new Error("401"));
const { result } = renderHook(() => useAuth(), { wrapper });
await waitFor(() => expect(result.current.status).toBe("unauthenticated"));
expect(result.current.user).toBeNull();
});
it("login() updates state on success", async () => {
(authApi.me as any).mockRejectedValue(new Error("401"));
(authApi.login as any).mockResolvedValue({ id: 1, username: "alice", role: "admin" });
const { result } = renderHook(() => useAuth(), { wrapper });
await waitFor(() => expect(result.current.status).toBe("unauthenticated"));
await act(async () => {
await result.current.login("alice", "pw");
});
expect(result.current.user?.username).toBe("alice");
expect(result.current.status).toBe("authenticated");
});
it("logout() clears state", async () => {
(authApi.me as any).mockResolvedValue({ id: 1, username: "alice", role: "user" });
(authApi.logout as any).mockResolvedValue(undefined);
const { result } = renderHook(() => useAuth(), { wrapper });
await waitFor(() => expect(result.current.status).toBe("authenticated"));
await act(async () => {
await result.current.logout();
});
expect(result.current.user).toBeNull();
expect(result.current.status).toBe("unauthenticated");
});
});
- Step 2: Implement useAuth.ts + AuthProvider.tsx
src/auth/useAuth.ts:
import { createContext, useContext } from "react";
import type { User } from "@/types";
export type AuthStatus = "loading" | "authenticated" | "unauthenticated";
export interface AuthContextValue {
status: AuthStatus;
user: User | null;
login: (username: string, password: string) => Promise<void>;
logout: () => Promise<void>;
refresh: () => Promise<void>;
}
export const AuthContext = createContext<AuthContextValue | null>(null);
export function useAuth(): AuthContextValue {
const ctx = useContext(AuthContext);
if (!ctx) throw new Error("useAuth must be used inside <AuthProvider>");
return ctx;
}
src/auth/AuthProvider.tsx:
import { useEffect, useState, useCallback, type ReactNode } from "react";
import { AuthContext, type AuthContextValue, type AuthStatus } from "./useAuth";
import { authApi } from "./api";
import type { User } from "@/types";
export function AuthProvider({ children }: { children: ReactNode }) {
const [status, setStatus] = useState<AuthStatus>("loading");
const [user, setUser] = useState<User | null>(null);
const refresh = useCallback(async () => {
setStatus("loading");
try {
const me = await authApi.me();
setUser(me as User);
setStatus("authenticated");
} catch {
setUser(null);
setStatus("unauthenticated");
}
}, []);
useEffect(() => { void refresh(); }, [refresh]);
const login = useCallback(async (username: string, password: string) => {
const u = await authApi.login(username, password);
setUser(u as User);
setStatus("authenticated");
}, []);
const logout = useCallback(async () => {
await authApi.logout().catch(() => undefined);
setUser(null);
setStatus("unauthenticated");
}, []);
const value: AuthContextValue = { status, user, login, logout, refresh };
return <AuthContext.Provider value={value}>{children}</AuthContext.Provider>;
}
- Step 3: Run tests
cd /home/tyler/dev/cyclone && npx vitest run src/auth/AuthProvider.test.tsx 2>&1 | tail -15
Expected: all 4 tests pass.
- Step 4: Commit
git add src/auth/AuthProvider.tsx src/auth/useAuth.ts src/auth/AuthProvider.test.tsx && git commit -m "feat(auth): AuthProvider context + useAuth hook"
Task 19: TDD — RoleGate
Files:
-
Create:
src/auth/RoleGate.tsx -
Create:
src/auth/RoleGate.test.tsx -
Step 1: Write failing tests
src/auth/RoleGate.test.tsx:
import { describe, it, expect, vi } from "vitest";
import { render, screen } from "@testing-library/react";
import { RoleGate } from "./RoleGate";
vi.mock("./useAuth", () => ({
useAuth: vi.fn(),
}));
import { useAuth } from "./useAuth";
function mockUser(role: "admin" | "user" | "viewer" | null) {
(useAuth as any).mockReturnValue({
user: role ? { id: 1, username: "x", role } : null,
status: role ? "authenticated" : "unauthenticated",
});
}
describe("RoleGate", () => {
it("renders children when role is allowed", () => {
mockUser("admin");
render(<RoleGate allow={["admin", "user"]}><button>Do thing</button></RoleGate>);
expect(screen.getByRole("button", { name: "Do thing" })).toBeInTheDocument();
});
it("renders disabled with tooltip when role is NOT allowed", () => {
mockUser("viewer");
render(<RoleGate allow={["admin", "user"]}><button>Do thing</button></RoleGate>);
const btn = screen.getByRole("button", { name: "Do thing" });
expect(btn).toBeDisabled();
});
it("renders fallback when provided and role not allowed", () => {
mockUser("viewer");
render(
<RoleGate allow={["admin"]} fallback={<span>NO</span>}>
<button>Do thing</button>
</RoleGate>
);
expect(screen.getByText("NO")).toBeInTheDocument();
});
});
- Step 2: Implement RoleGate.tsx
src/auth/RoleGate.tsx:
import { type ReactNode } from "react";
import { Tooltip } from "@/components/ui/tooltip"; // adjust path to project's Tooltip
import { useAuth } from "./useAuth";
interface Props {
allow: ("admin" | "user" | "viewer")[];
children: ReactNode;
fallback?: ReactNode;
}
export function RoleGate({ allow, children, fallback }: Props) {
const { user } = useAuth();
if (!user) return null;
if (allow.includes(user.role)) return <>{children}</>;
if (fallback) return <>{fallback}</>;
return (
<Tooltip content={`Your role (${user.role}) cannot perform this action.`}>
<span className="pointer-events-none opacity-50 cursor-not-allowed inline-flex">
{/* wrap children in a disabled clone of the first interactive child if possible */}
<DisabledClone>{children}</DisabledClone>
</span>
</Tooltip>
);
}
// Simple helper: clone the first child and add `disabled` if it's an element.
import { Children, cloneElement, isValidElement } from "react";
function DisabledClone({ children }: { children: ReactNode }) {
const child = Children.only(children);
if (isValidElement(child)) {
return cloneElement(child as any, { disabled: true, "aria-disabled": true });
}
return <>{children}</>;
}
(If the project doesn't have a Tooltip component, use the project's existing tooltip or just render an inline title attribute.)
- Step 3: Run tests
npx vitest run src/auth/RoleGate.test.tsx 2>&1 | tail -15
- Step 4: Commit
git add src/auth/RoleGate.tsx src/auth/RoleGate.test.tsx && git commit -m "feat(auth): RoleGate component"
Task 20: TDD — Login page
Files:
-
Create:
src/pages/Login.tsx -
Create:
src/pages/Login.test.tsx -
Step 1: Write failing tests
src/pages/Login.test.tsx:
import { describe, it, expect, vi, beforeEach } from "vitest";
import { render, screen, fireEvent, waitFor } from "@testing-library/react";
import { MemoryRouter, Routes, Route } from "react-router-dom";
import { Login } from "./Login";
vi.mock("@/auth/useAuth", () => ({ useAuth: vi.fn() }));
import { useAuth } from "@/auth/useAuth";
function setup() {
const login = vi.fn();
(useAuth as any).mockReturnValue({
user: null,
status: "unauthenticated",
login,
logout: vi.fn(),
refresh: vi.fn(),
});
return { login };
}
describe("Login page", () => {
beforeEach(() => vi.resetAllMocks());
it("submits username + password", async () => {
const { login } = setup();
login.mockResolvedValue(undefined);
render(
<MemoryRouter initialEntries={["/login"]}>
<Routes>
<Route path="/login" element={<Login />} />
<Route path="/" element={<div>HOME</div>} />
</Routes>
</MemoryRouter>
);
fireEvent.change(screen.getByLabelText(/username/i), { target: { value: "alice" } });
fireEvent.change(screen.getByLabelText(/password/i), { target: { value: "hunter2hunter2" } });
fireEvent.click(screen.getByRole("button", { name: /sign in/i }));
await waitFor(() => expect(login).toHaveBeenCalledWith("alice", "hunter2hunter2"));
});
it("shows error on failed login", async () => {
const { login } = setup();
login.mockRejectedValue(Object.assign(new Error("401"), { code: "invalid_credentials" }));
render(
<MemoryRouter initialEntries={["/login"]}>
<Routes><Route path="/login" element={<Login />} /></Routes>
</MemoryRouter>
);
fireEvent.change(screen.getByLabelText(/username/i), { target: { value: "alice" } });
fireEvent.change(screen.getByLabelText(/password/i), { target: { value: "wrong" } });
fireEvent.click(screen.getByRole("button", { name: /sign in/i }));
await waitFor(() =>
expect(screen.getByText(/username or password is incorrect/i)).toBeInTheDocument()
);
});
it("redirects to `next` query param on success", async () => {
const { login } = setup();
login.mockResolvedValue(undefined);
render(
<MemoryRouter initialEntries={["/login?next=/claims"]}>
<Routes>
<Route path="/login" element={<Login />} />
<Route path="/claims" element={<div>CLAIMS</div>} />
</Routes>
</MemoryRouter>
);
fireEvent.change(screen.getByLabelText(/username/i), { target: { value: "alice" } });
fireEvent.change(screen.getByLabelText(/password/i), { target: { value: "hunter2hunter2" } });
fireEvent.click(screen.getByRole("button", { name: /sign in/i }));
await waitFor(() => expect(screen.getByText("CLAIMS")).toBeInTheDocument());
});
});
- Step 2: Implement Login.tsx
src/pages/Login.tsx:
import { useState, type FormEvent } from "react";
import { useNavigate, useSearchParams, Navigate } from "react-router-dom";
import { useAuth } from "@/auth/useAuth";
export function Login() {
const { user, login } = useAuth();
const [params] = useSearchParams();
const next = params.get("next") ?? "/";
const navigate = useNavigate();
const [username, setUsername] = useState("");
const [password, setPassword] = useState("");
const [error, setError] = useState<string | null>(null);
const [submitting, setSubmitting] = useState(false);
if (user) return <Navigate to={next} replace />;
async function onSubmit(e: FormEvent) {
e.preventDefault();
setError(null);
setSubmitting(true);
try {
await login(username, password);
navigate(next, { replace: true });
} catch (err: any) {
const code = err?.code ?? "error";
if (code === "invalid_credentials") setError("Username or password is incorrect.");
else if (code === "account_disabled") setError("Account is disabled. Contact your administrator.");
else if (code === "rate_limited") setError("Too many attempts. Try again shortly.");
else setError("Sign in failed. Try again.");
} finally {
setSubmitting(false);
}
}
return (
<div className="min-h-screen flex items-center justify-center bg-background text-foreground">
<form
onSubmit={onSubmit}
className="w-full max-w-sm p-8 rounded-lg border border-border bg-card shadow"
>
<h1 className="text-2xl font-semibold mb-6">Cyclone</h1>
<label className="block mb-3">
<span className="text-sm">Username</span>
<input
type="text"
value={username}
onChange={(e) => setUsername(e.target.value)}
required
className="mt-1 w-full px-3 py-2 rounded bg-background border border-border"
autoComplete="username"
/>
</label>
<label className="block mb-5">
<span className="text-sm">Password</span>
<input
type="password"
value={password}
onChange={(e) => setPassword(e.target.value)}
required
className="mt-1 w-full py-2 px-3 rounded bg-background border border-border"
autoComplete="current-password"
/>
</label>
{error && (
<div className="mb-4 text-sm text-red-400" role="alert">{error}</div>
)}
<button
type="submit"
disabled={submitting}
className="w-full py-2 rounded bg-primary text-primary-foreground font-medium disabled:opacity-50"
>
{submitting ? "Signing in…" : "Sign in"}
</button>
</form>
</div>
);
}
- Step 3: Run tests
npx vitest run src/pages/Login.test.tsx 2>&1 | tail -15
- Step 4: Commit
git add src/pages/Login.tsx src/pages/Login.test.tsx && git commit -m "feat(auth): /login page"
Task 21: RequireAuth route guard
Files:
-
Create:
src/auth/RequireAuth.tsx -
Step 1: Implement RequireAuth
src/auth/RequireAuth.tsx:
import { Navigate, useLocation } from "react-router-dom";
import { useAuth } from "./useAuth";
import type { ReactNode } from "react";
export function RequireAuth({ children }: { children: ReactNode }) {
const { status } = useAuth();
const location = useLocation();
if (status === "loading") {
return (
<div className="min-h-screen flex items-center justify-center text-muted-foreground">
Loading…
</div>
);
}
if (status === "unauthenticated") {
return <Navigate to={`/login?next=${encodeURIComponent(location.pathname + location.search)}`} replace />;
}
return <>{children}</>;
}
- Step 2: Commit
git add src/auth/RequireAuth.tsx && git commit -m "feat(auth): RequireAuth route guard"
Task 22: Wire AuthProvider + RequireAuth into main.tsx
Files:
-
Modify:
src/main.tsx -
Step 1: Read current main.tsx
cd /home/tyler/dev/cyclone && cat src/main.tsx
- Step 2: Wrap the app
Update src/main.tsx to:
import { BrowserRouter, Routes, Route, Navigate } from "react-router-dom";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { AuthProvider } from "@/auth/AuthProvider";
import { RequireAuth } from "@/auth/RequireAuth";
import { Login } from "@/pages/Login";
import { Layout } from "@/components/Layout"; // adjust to actual layout
import { Dashboard } from "@/pages/Dashboard";
// ... other page imports
const queryClient = new QueryClient();
export function App() {
return (
<QueryClientProvider client={queryClient}>
<BrowserRouter>
<AuthProvider>
<Routes>
<Route path="/login" element={<Login />} />
<Route element={<RequireAuth><Layout /></RequireAuth>}>
<Route path="/" element={<Navigate to="/dashboard" replace />} />
<Route path="/dashboard" element={<Dashboard />} />
<Route path="/claims" element={<Claims />} />
{/* ... other protected routes */}
</Route>
</Routes>
</AuthProvider>
</BrowserRouter>
</QueryClientProvider>
);
}
Adjust to match the existing route definitions — preserve every existing route, just wrap them in <RequireAuth>.
- Step 3: Update src/lib/api.ts to use authedFetch
Find every fetch(...) in src/lib/api.ts and route it through authedFetch from @/auth/api. This way all existing hooks automatically get the 401-redirect behavior.
import { authedFetch } from "@/auth/api";
async function listClaims<T>(params: ListClaimsParams): Promise<PaginatedResponse<T>> {
return authedFetch(`/api/claims?...`);
}
// etc.
- Step 4: Commit
git add src/main.tsx src/lib/api.ts && git commit -m "feat(auth): wire AuthProvider + RequireAuth into app shell"
Task 23: Sidebar — show real current user
Files:
-
Modify:
src/components/Sidebar.tsx -
Step 1: Find the hardcoded user block
cd /home/tyler/dev/cyclone && grep -n "Jordan\|Administrator" src/components/Sidebar.tsx
- Step 2: Replace with useAuth()
import { useAuth } from "@/auth/useAuth";
function CurrentUser() {
const { user } = useAuth();
if (!user) return null;
return (
<div className="px-3 py-2 text-xs">
<div className="font-medium">{user.username}</div>
<div className="text-muted-foreground uppercase">{user.role}</div>
</div>
);
}
Replace the hardcoded "Jordan K. / Administrator" block with <CurrentUser />. Add a logout button next to it.
- Step 3: Commit
git add src/components/Sidebar.tsx && git commit -m "feat(auth): sidebar shows current user + logout button"
Task 24: Wrap write affordances in RoleGate
Files:
-
Modify:
src/pages/Upload.tsx -
Modify:
src/pages/Inbox.tsx -
Modify:
src/pages/Reconciliation.tsx -
Modify:
src/pages/Acks.tsx -
Step 1: Wrap Upload dropzone
In src/pages/Upload.tsx:
import { RoleGate } from "@/auth/RoleGate";
// Wrap the dropzone + Parse button:
<RoleGate allow={["admin", "user"]}>
<Dropzone ... />
</RoleGate>
- Step 2: Wrap Inbox resubmit + acknowledge buttons
In src/pages/Inbox.tsx, find the "Resubmit" and "Acknowledge" buttons. Wrap each in:
<RoleGate allow={["admin", "user"]}>
<Button onClick={...}>Resubmit</Button>
</RoleGate>
- Step 3: Wrap Reconciliation match buttons
Same pattern in src/pages/Reconciliation.tsx.
- Step 4: Wrap Acks acknowledge button
Same pattern in src/pages/Acks.tsx.
- Step 5: Commit
git add src/pages/Upload.tsx src/pages/Inbox.tsx src/pages/Reconciliation.tsx src/pages/Acks.tsx && git commit -m "feat(auth): disable write affordances for viewer role"
Phase 3: Infrastructure + Docs
Task 25: docker-compose + .env.example
Files:
-
Modify:
docker-compose.yml -
Create:
.env.example -
Step 1: Add env vars to backend service
In docker-compose.yml, find the backend service's environment: block (or add one):
backend:
environment:
CYCLONE_ADMIN_USERNAME: ${CYCLONE_ADMIN_USERNAME:?set me}
CYCLONE_ADMIN_PASSWORD: ${CYCLONE_ADMIN_PASSWORD:?set me}
- Step 2: Create .env.example
# Required on first boot. Cyclone refuses to start without these unless
# at least one user already exists (e.g. seeded via `python -m cyclone users create`).
CYCLONE_ADMIN_USERNAME=admin
CYCLONE_ADMIN_PASSWORD=change-me-to-a-strong-password-min-12-chars
# Optional. Set to 1 if you're behind an HTTPS reverse proxy and want
# the session cookie to include the Secure flag.
# CYCLONE_BEHIND_HTTPS=1
# Optional. Set to 1 to disable auth entirely (DEV ONLY). When set,
# the backend auto-grants admin access without checking credentials.
# CYCLONE_AUTH_DISABLED=0
- Step 3: Commit
git add docker-compose.yml .env.example && git commit -m "feat(docker): CYCLONE_ADMIN_* env vars for backend"
Task 26: README — Auth section
Files:
-
Modify:
README.md -
Step 1: Add Auth section
Find a good place in the README (after the "Quickstart" section, before "Roadmap"). Add:
## Authentication
Cyclone now ships with username/password authentication and three predefined roles.
**Roles:**
| Role | Can read | Can write (upload, parse, reconcile) | Can manage users |
|---|---|---|---|
| `viewer` | ✅ | ❌ | ❌ |
| `user` | ✅ | ✅ | ❌ |
| `admin` | ✅ | ✅ | ✅ |
**Bootstrap.** On first start, set `CYCLONE_ADMIN_USERNAME` and `CYCLONE_ADMIN_PASSWORD` (min 12 chars) in your environment. Cyclone creates the first admin automatically. On subsequent starts, these vars are ignored.
**CLI.** Manage users from the command line:
\`\`\`
python -m cyclone users create alice --role user --password 'hunter2hunter2'
python -m cyclone users list
python -m cyclone users disable alice
python -m cyclone users reset-password alice
python -m cyclone users set-role alice --role admin
\`\`\`
**Login.** Browse to `http://localhost:8081/`, sign in on the `/login` page, and you'll be redirected to the dashboard. Sessions are stored server-side in SQLite with a 24-hour sliding expiry.
**Dev escape hatch.** Set `CYCLONE_AUTH_DISABLED=1` to bypass auth entirely (the backend auto-grants admin). NEVER set this in production.
See `docs/superpowers/specs/2026-06-22-cyclone-auth-design.md` for the full design.
- Step 2: Commit
git add README.md && git commit -m "docs: auth section + env vars"
Task 27: End-to-end verification
Files:
-
Create:
/tmp/verify_auth.py -
Step 1: Spin up the stack
cd /home/tyler/dev/cyclone && docker compose build backend frontend && docker compose down -v && CYCLONE_ADMIN_USERNAME=admin CYCLONE_ADMIN_PASSWORD=adminpassword1 docker compose up -d
- Step 2: Write the playwright script
/tmp/verify_auth.py:
import sys
from playwright.sync_api import sync_playwright
def main() -> int:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, executable_path="/usr/bin/google-chrome")
page = browser.new_context().new_page()
page.goto("http://localhost:8081/", wait_until="networkidle", timeout=15000)
# Should redirect to /login.
page.wait_for_url("**/login**", timeout=5000)
# Sign in.
page.fill('input[autocomplete="username"]', "admin")
page.fill('input[autocomplete="current-password"]', "adminpassword1")
page.click('button[type="submit"]')
page.wait_for_url("**/dashboard", timeout=5000)
# Dashboard renders.
assert "Good" in page.locator("body").inner_text()
# Sidebar shows "admin".
body = page.locator("body").inner_text()
assert "admin" in body.lower()
print("[PASS] login flow")
browser.close()
return 0
if __name__ == "__main__":
sys.exit(main())
- Step 3: Run it
python3 /tmp/verify_auth.py
Expected: [PASS] login flow. Fix any issues.
- Step 4: Commit (the verify script goes in /tmp, no commit needed)
Task 28: Full backend test suite
- Step 1: Run all tests
cd /home/tyler/dev/cyclone/backend && uv run pytest 2>&1 | tail -10
Expected: all tests pass (pre-existing + new auth tests). Fix any regressions.
- Step 2: Run frontend tests
cd /home/tyler/dev/cyclone && npx vitest run 2>&1 | tail -10
Expected: all tests pass.
Task 29: Final cleanup
- Step 1: Typecheck
cd /home/tyler/dev/cyclone && npm run typecheck 2>&1 | tail -10
Fix any TypeScript errors.
- Step 2: Lint
cd /home/tyler/dev/cyclone && npm run lint 2>&1 | tail -10
Fix any lint errors.
- Step 3: Build
cd /home/tyler/dev/cyclone && npm run build 2>&1 | tail -10
Expected: clean build.
- Step 4: Final commit
git add -A && git commit -m "chore: post-auth cleanup (typecheck, lint, build)"
Self-Review Checklist (run after writing the plan)
Spec coverage:
| Spec section | Task(s) |
|---|---|
| §1 Overview / user model | All of Phase 1+2 |
| §2.1 Authenticate every API request | Tasks 9, 11 |
| §2.2 Three roles + matrix | Tasks 7, 11 |
| §2.3 Server-side sessions in SQLite | Tasks 2, 3, 6, 8 |
| §2.4 Bootstrap first admin | Task 12 |
| §2.5 Frontend login + context | Tasks 17, 18, 20, 21, 22 |
| §2.6 Disable write UI for viewer | Tasks 19, 24 |
| §2.7 Audit log includes user_id | Task 14 |
| §2.8 CLI command | Task 15 |
| §6.2 Data model | Tasks 2, 3 |
| §6.3 Permissions matrix | Task 7 |
| §6.4 Dependencies | Task 8 |
| §6.5 Endpoints | Tasks 9, 10 |
| §6.6 Rate limit | Tasks 7, 13 |
| §6.7 Bootstrap | Task 12 |
| §6.8 CLI | Task 15 |
| §7.1 AuthProvider | Tasks 17, 18 |
| §7.3 Login page | Task 20 |
| §7.4 API wrapper | Task 17 |
| §7.5 RoleGate | Task 19 |
| §7.6 Sidebar | Task 23 |
| §7.7 Routes | Task 22 |
| §8.1 nginx | Task 25 (minor X-Forwarded-Proto) |
| §8.2 docker-compose | Task 25 |
| §8.3 Cookie Secure flag | Task 9 (_is_https) |
| §9.1 Backend tests | Tasks 5, 6, 9, 10, 11, 12, 13, 14, 15 |
| §9.2 Frontend tests | Tasks 17, 18, 19, 20 |
| §9.3 E2E | Task 27 |
| §11 Rollout | All |
Gaps: None — every spec section is covered by at least one task.
Placeholder scan: Search the plan for "TBD", "TODO", "implement later" — none. Code blocks are complete enough to execute.
Type consistency: User.role is "admin" | "user" | "viewer" everywhere (backend Role enum, frontend User.role, RoleGate.allow). Session.id is str in both DB and Python. Cookie name is "cyclone_session" consistently.