"""initial schema — users, refresh_tokens, api_keys, auth_audit_log Revision ID : 0001 Revises : (none — this is the base migration) Create Date : 2026-06-02 00:00:00 UTC Tables created -------------- users Human user accounts with RBAC roles refresh_tokens JWT refresh-token family tree (rotation + theft detection) api_keys External subscriber HMAC keys + internal cron keys auth_audit_log Immutable security event journal Security decisions documented inline at each DDL statement. Rollback (downgrade) drops tables in reverse dependency order. """ from __future__ import annotations from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql # --------------------------------------------------------------------------- # Alembic revision identifiers # --------------------------------------------------------------------------- revision: str = "0001" down_revision: str | None = None # base — no prior migration branch_labels: str | None = None depends_on: str | None = None # =========================================================================== # UPGRADE — apply schema # =========================================================================== def upgrade() -> None: # ----------------------------------------------------------------------- # 0. Extensions # pgcrypto → gen_random_uuid() (UUID primary keys) # btree_gin → GIN index on scalar columns (future flexibility) # ----------------------------------------------------------------------- op.execute('CREATE EXTENSION IF NOT EXISTS "pgcrypto"') op.execute('CREATE EXTENSION IF NOT EXISTS "btree_gin"') # ----------------------------------------------------------------------- # 1. USERS # # Security notes: # • UUID PK — prevents sequential-ID enumeration attacks. # • password_hash — bcrypt, cost ≥ 12 enforced in app (passlib). # • role CHECK — DB-level guard so a bug cannot insert 'superuser'. # • failed_login_attempts + locked_until — brute-force lockout state. # • is_active — soft-delete avoids referential-integrity gaps while # still blocking the account immediately. # ----------------------------------------------------------------------- op.create_table( "users", sa.Column( "id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()"), nullable=False, comment="UUIDv4 — non-sequential, prevents row-count enumeration", ), sa.Column( "email", sa.Text, nullable=False, comment="Stored lowercased + trimmed; unique constraint below", ), sa.Column( "password_hash", sa.Text, nullable=False, comment="bcrypt hash, cost>=12. NEVER store plaintext password.", ), sa.Column( "role", sa.String(16), nullable=False, server_default="viewer", comment="RBAC role: viewer | analyst | admin", ), sa.Column( "is_active", sa.Boolean, nullable=False, server_default=sa.text("true"), comment="Soft-disable; checked on every login + JWT verify", ), sa.Column( "failed_login_attempts", sa.Integer, nullable=False, server_default="0", comment="Incremented on each bad password; reset on success", ), sa.Column( "locked_until", sa.DateTime(timezone=True), nullable=True, comment="Account locked until this UTC timestamp. NULL = not locked.", ), sa.Column( "last_login_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), comment="Update this via trigger or app layer on every write", ), # DB-level RBAC guard — prevents invalid role strings from sneaking in sa.CheckConstraint( "role IN ('viewer', 'analyst', 'admin')", name="ck_users_role_valid", ), # Minimum length for password hashes (bcrypt output is always 60 chars) sa.CheckConstraint( "length(password_hash) >= 60", name="ck_users_password_hash_len", ), ) # Unique email — case handled in app (always lowercase before INSERT) op.create_unique_constraint("uq_users_email", "users", ["email"]) # Partial index: only active users — login path only ever looks up active rows op.create_index( "ix_users_email_active", "users", ["email"], postgresql_where=sa.text("is_active = true"), ) # Role index: RBAC admin queries list users by role op.create_index("ix_users_role", "users", ["role"]) # ----------------------------------------------------------------------- # 1a. Trigger: auto-update updated_at on every row change # ----------------------------------------------------------------------- op.execute(""" CREATE OR REPLACE FUNCTION trg_set_updated_at() RETURNS TRIGGER LANGUAGE plpgsql AS $$ BEGIN NEW.updated_at = NOW(); RETURN NEW; END; $$ """) op.execute(""" CREATE TRIGGER trg_users_updated_at BEFORE UPDATE ON users FOR EACH ROW EXECUTE FUNCTION trg_set_updated_at() """) # ----------------------------------------------------------------------- # 2. REFRESH TOKENS # # Security notes: # • jti (JWT ID) is the lookup key — UUID embedded in the signed JWT. # • family_id groups the rotation chain. On detecting a reused jti, # revoke ALL rows sharing this family_id → forces full re-login. # • revoked + expires_at are both checked — a token can be valid (not # revoked) but still rejected if expired. # • ip_address / user_agent — fingerprinting to detect theft from a # different device; violations should raise HIGH-severity audit event. # • CASCADE DELETE — revoking a user cleans all their tokens atomically. # • Nightly cleanup: DELETE WHERE expires_at < NOW() - INTERVAL '1 day' # (keep 1-day grace so theft detection logs stay warm after expiry) # ----------------------------------------------------------------------- op.create_table( "refresh_tokens", sa.Column( "jti", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()"), nullable=False, comment="JWT ID claim. Embedded in signed refresh token.", ), sa.Column( "user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "family_id", postgresql.UUID(as_uuid=True), nullable=False, comment="Shared across all rotations from one login. Theft → revoke family.", ), sa.Column( "revoked", sa.Boolean, nullable=False, server_default=sa.text("false"), comment="Set true on: logout, rotation (old token), theft detection", ), sa.Column( "user_agent", sa.Text, nullable=True, comment="Browser/app user-agent captured at token issuance", ), sa.Column( "ip_address", postgresql.INET, nullable=True, comment="Client IP at issuance. inet type supports CIDR subnet queries.", ), sa.Column( "expires_at", sa.DateTime(timezone=True), nullable=False, comment="Hard expiry = issued_at + REFRESH_TOKEN_TTL (default 7 days)", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), # Guard: expires_at must be in the future at creation time sa.CheckConstraint( "expires_at > created_at", name="ck_refresh_tokens_expiry_future", ), ) # Lookup by user (list sessions, revoke all for a user) op.create_index("ix_refresh_tokens_user_id", "refresh_tokens", ["user_id"]) # Theft detection: find all active siblings in a family op.create_index( "ix_refresh_tokens_family_active", "refresh_tokens", ["family_id", "revoked"], ) # Validation hot path: WHERE jti=? AND revoked=false AND expires_at > NOW() op.create_index( "ix_refresh_tokens_jti_active", "refresh_tokens", ["jti", "revoked"], ) # Nightly cleanup: DELETE WHERE expires_at < NOW() - INTERVAL '1 day' op.create_index( "ix_refresh_tokens_expires_at", "refresh_tokens", ["expires_at"], ) # ----------------------------------------------------------------------- # 3. API KEYS (Caller Type 2: HMAC-signed external + Type 3: cron) # # Security notes: # • secret_hash — bcrypt of a randomly-generated 32-byte secret. # The plain secret is displayed ONCE at creation, then discarded. # • scopes — TEXT[] with GIN index. Enables O(1) containment check: # WHERE scopes @> ARRAY['data:pull'] # Valid scope values: data:push | data:pull | feed:subscribe | internal:job # • rate_limit — per-key ceiling stored here; Redis enforces it at runtime. # • is_active — immediate revocation flag. For key rotation, keep the old # key active 24h (grace period) before setting is_active=false. # • last_used_at — written asynchronously (fire-and-forget) so it never # adds latency to the hot request path. # • expires_at — optional for time-bounded keys (e.g. contractor access). # • created_by — audit trail; SET NULL on user deletion keeps key record. # ----------------------------------------------------------------------- op.create_table( "api_keys", sa.Column( "key_id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()"), nullable=False, comment="Sent as X-Service-Key header. Safe to expose publicly.", ), sa.Column( "secret_hash", sa.Text, nullable=False, comment="bcrypt hash (cost>=12) of the 32-byte random secret.", ), sa.Column( "service_name", sa.Text, nullable=False, comment="Human label: analytics-svc, cron-collector, etc.", ), sa.Column( "scopes", postgresql.ARRAY(sa.Text), nullable=False, server_default="{}", comment="Permission list. GIN-indexed for @> containment queries.", ), sa.Column( "is_active", sa.Boolean, nullable=False, server_default=sa.text("true"), comment="False = immediate revocation. Use 24h grace on rotation.", ), sa.Column( "rate_limit", sa.Integer, nullable=False, server_default="600", comment="Max requests/minute for this key. Enforced via Redis.", ), sa.Column( "expires_at", sa.DateTime(timezone=True), nullable=True, comment="NULL = never expires. Set for time-bounded access grants.", ), sa.Column( "created_by", postgresql.UUID(as_uuid=True), # SET NULL so the key record survives admin account deletion sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, comment="Admin who issued the key. NULL after issuer account deleted.", ), sa.Column( "description", sa.Text, nullable=True, comment="Ops-team notes: purpose, owner, rotation schedule", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), sa.Column( "last_used_at", sa.DateTime(timezone=True), nullable=True, comment="Updated async on each request. Never block hot path for this.", ), # Guard: rate limit must be positive sa.CheckConstraint( "rate_limit > 0", name="ck_api_keys_rate_limit_positive", ), # Guard: bcrypt hash minimum length sa.CheckConstraint( "length(secret_hash) >= 60", name="ck_api_keys_secret_hash_len", ), ) # GIN index on scopes array — enables fast @> containment operator op.create_index( "ix_api_keys_scopes_gin", "api_keys", ["scopes"], postgresql_using="gin", ) # Partial index: active-only keys for validation hot path op.create_index( "ix_api_keys_is_active", "api_keys", ["is_active"], postgresql_where=sa.text("is_active = true"), ) # ----------------------------------------------------------------------- # 4. AUTH AUDIT LOG # # Security notes: # • BIGSERIAL primary key (not UUID) — sequential for O(1) append and # fast time-ordered range scans (no UUID fragmentation penalty). # • No FK constraints on user_id / key_id — log rows must survive # account deletion for forensics and compliance requirements. # • metadata JSONB — flexible event-specific context blob. # NEVER put passwords, tokens, or secrets in this column. # • No UPDATE / DELETE in application code — treat as append-only. # Use PostgreSQL Row Security or a WRITE-ONLY role in production. # • Partition by month in production (PARTITION BY RANGE (created_at)) # once volume exceeds ~10M rows/month. # ----------------------------------------------------------------------- op.create_table( "auth_audit_log", sa.Column( "id", sa.BigInteger, primary_key=True, autoincrement=True, comment="Sequential BIGINT for fast append and time-ordered range scans", ), sa.Column( "event_type", sa.String(64), nullable=False, comment=( "login_success | login_failure | token_theft | token_expired | " "logout | key_revoked | replay_attack | invalid_signature | " "scope_denied | account_locked | job_run" ), ), sa.Column( "caller_type", sa.String(16), nullable=True, comment="user | api_key | cron", ), sa.Column( "user_id", postgresql.UUID(as_uuid=True), nullable=True, comment="No FK — log survives user deletion. Index for user history queries.", ), sa.Column( "key_id", postgresql.UUID(as_uuid=True), nullable=True, comment="No FK — log survives key deletion.", ), sa.Column( "ip_address", postgresql.INET, nullable=True, comment="inet type allows CIDR subnet queries: ip << '10.0.0.0/8'", ), sa.Column( "user_agent", sa.Text, nullable=True, comment="Raw User-Agent; truncated to 512 chars in application layer", ), sa.Column( "metadata", postgresql.JSONB, nullable=False, server_default="{}", comment=( "Event detail blob. Examples:\n" " login_failure → {reason: wrong_password, attempt: 3}\n" " token_theft → {family_id: ..., reused_jti: ...}\n" " replay_attack → {delta_sec: 312}\n" " job_run → {job: collect, duration_ms: 450}\n" "NEVER log secrets, passwords, or tokens here." ), ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), # Guard: event_type must be a recognised value sa.CheckConstraint( """event_type IN ( 'login_success', 'login_failure', 'token_theft', 'token_expired', 'logout', 'key_revoked', 'replay_attack', 'invalid_signature', 'scope_denied', 'account_locked', 'job_run' )""", name="ck_audit_event_type_valid", ), # Guard: caller_type must be known or NULL sa.CheckConstraint( "caller_type IS NULL OR caller_type IN ('user', 'api_key', 'cron')", name="ck_audit_caller_type_valid", ), ) # Brute-force detection: COUNT failures per IP in a time window op.create_index( "ix_audit_ip_event_time", "auth_audit_log", ["ip_address", "event_type", "created_at"], ) # User history: list events for a specific account (admin investigation) op.create_index( "ix_audit_user_time", "auth_audit_log", ["user_id", "created_at"], ) # Event stream: pull all events of one type in time order (alerting system) op.create_index( "ix_audit_event_type_time", "auth_audit_log", ["event_type", "created_at"], ) # GIN index on metadata — enables JSON path queries for investigation op.create_index( "ix_audit_metadata_gin", "auth_audit_log", ["metadata"], postgresql_using="gin", ) # ----------------------------------------------------------------------- # 5. Row-level security: prevent application user from deleting audit rows # # In production, the FastAPI app connects as role 'finauth_app'. # A separate 'finauth_audit_reader' role can SELECT but not DELETE. # The GRANT/REVOKE below assumes those roles exist — adjust to your setup. # ----------------------------------------------------------------------- op.execute(""" -- Create a dedicated app role if it doesn't exist yet -- (idempotent; safe to run on repeat) DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'finauth_app') THEN CREATE ROLE finauth_app NOLOGIN; END IF; END $$; """) # asyncpg requires one statement per op.execute() — no multi-statement batches op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON users TO finauth_app") op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON refresh_tokens TO finauth_app") op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON api_keys TO finauth_app") # Audit log: INSERT only — enforces append-only policy at DB permission layer op.execute("GRANT SELECT, INSERT ON auth_audit_log TO finauth_app") op.execute("GRANT USAGE, SELECT ON SEQUENCE auth_audit_log_id_seq TO finauth_app") # ----------------------------------------------------------------------- # 6. Seed data — roles enum check is DB-enforced, no seed needed. # Insert a placeholder "system" user for audit log foreign-key reference # in automated jobs. Password is a bcrypt hash of a random string that # is immediately discarded — this account cannot log in. # ----------------------------------------------------------------------- op.execute(""" INSERT INTO users ( id, email, password_hash, role, is_active ) VALUES ( '00000000-0000-0000-0000-000000000000', 'system@internal.invalid', -- bcrypt hash of random string; login disabled via is_active=false '$2b$12$PLACEHOLDER_DO_NOT_USE_FOR_LOGIN_aaaaaaaaaaaaaaaaaaaaaa', 'admin', false -- cannot log in; used only as FK target for system events ) ON CONFLICT (id) DO NOTHING; """) # =========================================================================== # DOWNGRADE — roll back schema (reverse dependency order) # =========================================================================== def downgrade() -> None: # Drop in reverse FK dependency order # 4. Audit log (no FKs pointing to it — drop first is fine) op.drop_index("ix_audit_metadata_gin", table_name="auth_audit_log") op.drop_index("ix_audit_event_type_time", table_name="auth_audit_log") op.drop_index("ix_audit_user_time", table_name="auth_audit_log") op.drop_index("ix_audit_ip_event_time", table_name="auth_audit_log") op.drop_table("auth_audit_log") # 3. API keys (FK: created_by → users) op.drop_index("ix_api_keys_is_active", table_name="api_keys") op.drop_index("ix_api_keys_scopes_gin", table_name="api_keys") op.drop_table("api_keys") # 2. Refresh tokens (FK: user_id → users) op.drop_index("ix_refresh_tokens_expires_at", table_name="refresh_tokens") op.drop_index("ix_refresh_tokens_jti_active", table_name="refresh_tokens") op.drop_index("ix_refresh_tokens_family_active", table_name="refresh_tokens") op.drop_index("ix_refresh_tokens_user_id", table_name="refresh_tokens") op.drop_table("refresh_tokens") # 1. Users (referenced by refresh_tokens + api_keys — must drop last) op.execute("DROP TRIGGER IF EXISTS trg_users_updated_at ON users") op.execute("DROP FUNCTION IF EXISTS trg_set_updated_at()") op.drop_index("ix_users_role", table_name="users") op.drop_index("ix_users_email_active", table_name="users") op.drop_constraint("uq_users_email", table_name="users", type_="unique") op.drop_table("users") # 0. Extensions — only drop if you are sure nothing else uses them # op.execute('DROP EXTENSION IF EXISTS "btree_gin"') # op.execute('DROP EXTENSION IF EXISTS "pgcrypto"') # (Commented out intentionally — extensions may be shared by other schemas)