49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""phase35 — user two-factor (TOTP) columns
|
|
|
|
Adds opt-in TOTP MFA to user accounts (enforced for admin/director at login
|
|
when enabled). Additive + nullable; existing users default to mfa_enabled=0
|
|
so nothing changes until a user enrolls.
|
|
|
|
Guarded with INFORMATION_SCHEMA column-existence checks — safe to re-run
|
|
across every tenant DB (CLAUDE.md rule 14).
|
|
"""
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
|
|
revision = 'phase35_user_mfa'
|
|
down_revision = 'phase34_inspection_schedules'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _col_exists(bind, table: str, column: str) -> bool:
|
|
result = bind.execute(sa.text(
|
|
"SELECT COUNT(*) FROM information_schema.columns "
|
|
"WHERE table_schema = DATABASE() AND table_name = :t AND column_name = :c"
|
|
), {'t': table, 'c': column})
|
|
return result.scalar() > 0
|
|
|
|
|
|
def upgrade():
|
|
bind = op.get_bind()
|
|
if not _col_exists(bind, 'users', 'mfa_enabled'):
|
|
op.execute(sa.text(
|
|
"ALTER TABLE users ADD COLUMN mfa_enabled TINYINT(1) NOT NULL DEFAULT 0"
|
|
))
|
|
if not _col_exists(bind, 'users', 'mfa_secret'):
|
|
op.execute(sa.text(
|
|
"ALTER TABLE users ADD COLUMN mfa_secret VARCHAR(64) NULL"
|
|
))
|
|
if not _col_exists(bind, 'users', 'mfa_recovery_codes'):
|
|
op.execute(sa.text(
|
|
"ALTER TABLE users ADD COLUMN mfa_recovery_codes JSON NULL"
|
|
))
|
|
|
|
|
|
def downgrade():
|
|
bind = op.get_bind()
|
|
for col in ('mfa_recovery_codes', 'mfa_secret', 'mfa_enabled'):
|
|
if _col_exists(bind, 'users', col):
|
|
op.execute(sa.text(f"ALTER TABLE users DROP COLUMN {col}"))
|