52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""control0005_superadmin_mfa
|
|
|
|
Adds opt-in TOTP two-factor columns to the `superadmins` table:
|
|
- mfa_enabled TINYINT(1) NOT NULL DEFAULT 0
|
|
- mfa_secret VARCHAR(64) NULL — base32 TOTP shared secret
|
|
- mfa_recovery_codes JSON NULL — hashed one-time backup codes
|
|
|
|
All columns are guarded by INFORMATION_SCHEMA existence checks so this
|
|
migration is safe to re-run. Existing superadmins default to disabled, so
|
|
nothing changes until an operator enrolls.
|
|
|
|
Revision ID: control0005_superadmin_mfa
|
|
Revises: control0004_dunning_tracking
|
|
Create Date: 2026-07-04
|
|
"""
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = 'control0005_superadmin_mfa'
|
|
down_revision = 'control0004_dunning_tracking'
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _column_exists(table, column):
|
|
result = op.get_bind().execute(sa.text(
|
|
"SELECT COUNT(*) FROM information_schema.COLUMNS "
|
|
"WHERE TABLE_SCHEMA = DATABASE() "
|
|
" AND TABLE_NAME = :tbl "
|
|
" AND COLUMN_NAME = :col"
|
|
), {'tbl': table, 'col': column})
|
|
return result.scalar() > 0
|
|
|
|
|
|
def upgrade():
|
|
if not _column_exists('superadmins', 'mfa_enabled'):
|
|
op.add_column('superadmins', sa.Column('mfa_enabled', sa.Boolean(),
|
|
nullable=False, server_default='0'))
|
|
if not _column_exists('superadmins', 'mfa_secret'):
|
|
op.add_column('superadmins', sa.Column('mfa_secret', sa.String(64),
|
|
nullable=True))
|
|
if not _column_exists('superadmins', 'mfa_recovery_codes'):
|
|
op.add_column('superadmins', sa.Column('mfa_recovery_codes', sa.JSON(),
|
|
nullable=True))
|
|
|
|
|
|
def downgrade():
|
|
for col in ('mfa_recovery_codes', 'mfa_secret', 'mfa_enabled'):
|
|
if _column_exists('superadmins', col):
|
|
op.drop_column('superadmins', col)
|