A R R R Associates
29 lines
4.6 KiB
Python
29 lines
4.6 KiB
Python
"""Phase 5 secure credential vault.
|
|
|
|
Revision ID: 20260723_phase5_secure_credential_vault
|
|
Revises: 20260723_phase4_registration_lifecycle
|
|
"""
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "20260723_phase5_secure_credential_vault"
|
|
down_revision = "20260723_phase4_registration_lifecycle"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade():
|
|
op.create_table("credential_vault_entries",
|
|
sa.Column("id", sa.Integer(), primary_key=True), sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False), sa.Column("branch_id", sa.Integer(), sa.ForeignKey("branches.id", ondelete="SET NULL")), sa.Column("client_id", sa.Integer(), sa.ForeignKey("clients.id", ondelete="CASCADE")), sa.Column("registration_id", sa.Integer(), sa.ForeignKey("client_registrations.id", ondelete="SET NULL")),
|
|
sa.Column("title", sa.String(200), nullable=False), sa.Column("category", sa.String(60), nullable=False, server_default="government_portal"), sa.Column("portal_url", sa.String(800)), sa.Column("reference_number", sa.String(160)), sa.Column("username_encrypted", sa.Text()), sa.Column("secret_encrypted", sa.Text(), nullable=False), sa.Column("additional_secret_encrypted", sa.Text()), sa.Column("notes_encrypted", sa.Text()), sa.Column("encryption_version", sa.Integer(), nullable=False, server_default="1"),
|
|
sa.Column("sensitivity", sa.String(30), nullable=False, server_default="high"), sa.Column("status", sa.String(30), nullable=False, server_default="active"), sa.Column("expires_on", sa.Date()), sa.Column("rotation_due_on", sa.Date()), sa.Column("owner_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), sa.Column("allowed_user_ids_csv", sa.Text()), sa.Column("reveal_requires_reason", sa.Boolean(), nullable=False, server_default=sa.true()), sa.Column("last_rotated_at_utc", sa.DateTime(timezone=True)), sa.Column("archived_at_utc", sa.DateTime(timezone=True)), sa.Column("created_by_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), sa.Column("updated_by_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False), sa.Column("updated_at_utc", sa.DateTime(timezone=True), nullable=False))
|
|
for col in ["tenant_id","branch_id","client_id","registration_id","title","category","reference_number","status","expires_on","rotation_due_on","owner_user_id"]: op.create_index(f"ix_credential_vault_entries_{col}", "credential_vault_entries", [col])
|
|
op.create_table("credential_vault_versions", sa.Column("id", sa.Integer(), primary_key=True), sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False), sa.Column("entry_id", sa.Integer(), sa.ForeignKey("credential_vault_entries.id", ondelete="CASCADE"), nullable=False), sa.Column("version_number", sa.Integer(), nullable=False), sa.Column("username_encrypted", sa.Text()), sa.Column("secret_encrypted", sa.Text(), nullable=False), sa.Column("additional_secret_encrypted", sa.Text()), sa.Column("notes_encrypted", sa.Text()), sa.Column("change_reason", sa.Text(), nullable=False), sa.Column("changed_by_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), sa.Column("changed_at_utc", sa.DateTime(timezone=True), nullable=False))
|
|
op.create_index("ix_credential_vault_versions_tenant_id", "credential_vault_versions", ["tenant_id"]); op.create_index("ix_credential_vault_versions_entry_id", "credential_vault_versions", ["entry_id"])
|
|
op.create_table("credential_vault_access_logs", sa.Column("id", sa.Integer(), primary_key=True), sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False), sa.Column("branch_id", sa.Integer(), sa.ForeignKey("branches.id", ondelete="SET NULL")), sa.Column("entry_id", sa.Integer(), sa.ForeignKey("credential_vault_entries.id", ondelete="SET NULL")), sa.Column("actor_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), sa.Column("action", sa.String(40), nullable=False), sa.Column("reason", sa.Text()), sa.Column("fields_accessed_csv", sa.String(300)), sa.Column("success", sa.Boolean(), nullable=False, server_default=sa.true()), sa.Column("ip_address", sa.String(100)), sa.Column("user_agent", sa.String(500)), sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False))
|
|
for col in ["tenant_id","branch_id","entry_id","actor_user_id","action","created_at_utc"]: op.create_index(f"ix_credential_vault_access_logs_{col}", "credential_vault_access_logs", [col])
|
|
|
|
|
|
def downgrade():
|
|
op.drop_table("credential_vault_access_logs"); op.drop_table("credential_vault_versions"); op.drop_table("credential_vault_entries")
|