shorefront/backend/alembic/versions/0001_initial.py

"""initial schema and seed data

Revision ID: 0001
Revises:
Create Date: 2026-02-28
"""
from alembic import op
import sqlalchemy as sa

revision = "0001"
down_revision = None
branch_labels = None
depends_on = None

# Pre-computed bcrypt hash of "admin". Used only for seeding the initial admin
# row; hashed_password becomes nullable in migration 0002 (SSO transition) and
# is never validated at login.
_ADMIN_HASH = "$2b$12$E8Td3Igd66V.udnDuflWFOhhhZDYFRHwLpgIsw.cRQtMf4h102sj."


def upgrade() -> None:
    # --- Schema ---
    op.create_table(
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("username", sa.String(64), nullable=False, unique=True),
        sa.Column("email", sa.String(255), nullable=False, unique=True),
        sa.Column("hashed_password", sa.String(255), nullable=False),
        sa.Column("is_active", sa.Boolean, nullable=False, server_default="true"),
    )
    op.create_table(
        "configs",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.String(128), nullable=False),
        sa.Column("description", sa.Text, server_default="''"),
        sa.Column("is_active", sa.Boolean, server_default="true"),
        sa.Column("created_at", sa.DateTime, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime, server_default=sa.func.now()),
        sa.Column("owner_id", sa.Integer, sa.ForeignKey("users.id"), nullable=False),
    )
    op.create_table(
        "zones",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("config_id", sa.Integer, sa.ForeignKey("configs.id"), nullable=False),
        sa.Column("name", sa.String(32), nullable=False),
        sa.Column("type", sa.String(16), nullable=False),
        sa.Column("options", sa.Text, server_default="''"),
        sa.UniqueConstraint("config_id", "name", name="uq_zone_name_per_config"),
    )
    op.create_table(
        "interfaces",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("config_id", sa.Integer, sa.ForeignKey("configs.id"), nullable=False),
        sa.Column("name", sa.String(32), nullable=False),
        sa.Column("zone_id", sa.Integer, sa.ForeignKey("zones.id"), nullable=False),
        sa.Column("options", sa.Text, server_default="''"),
    )
    op.create_table(
        "policies",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("config_id", sa.Integer, sa.ForeignKey("configs.id"), nullable=False),
        sa.Column("src_zone_id", sa.Integer, sa.ForeignKey("zones.id"), nullable=False),
        sa.Column("dst_zone_id", sa.Integer, sa.ForeignKey("zones.id"), nullable=False),
        sa.Column("policy", sa.String(16), nullable=False),
        sa.Column("log_level", sa.String(16), server_default="''"),
        sa.Column("comment", sa.Text, server_default="''"),
        sa.Column("position", sa.Integer, server_default="0"),
    )
    op.create_table(
        "rules",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("config_id", sa.Integer, sa.ForeignKey("configs.id"), nullable=False),
        sa.Column("action", sa.String(32), nullable=False),
        sa.Column("src_zone_id", sa.Integer, sa.ForeignKey("zones.id"), nullable=True),
        sa.Column("dst_zone_id", sa.Integer, sa.ForeignKey("zones.id"), nullable=True),
        sa.Column("src_ip", sa.String(64), server_default="''"),
        sa.Column("dst_ip", sa.String(64), server_default="''"),
        sa.Column("proto", sa.String(16), server_default="''"),
        sa.Column("dport", sa.String(64), server_default="''"),
        sa.Column("sport", sa.String(64), server_default="''"),
        sa.Column("comment", sa.Text, server_default="''"),
        sa.Column("position", sa.Integer, server_default="0"),
    )
    op.create_table(
        "masq",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("config_id", sa.Integer, sa.ForeignKey("configs.id"), nullable=False),
        sa.Column("source_network", sa.String(64), nullable=False),
        sa.Column("out_interface", sa.String(32), nullable=False),
        sa.Column("to_address", sa.String(64), server_default="''"),
        sa.Column("comment", sa.Text, server_default="''"),
    )

    # --- Seed data ---
    conn = op.get_bind()

    # Admin user
    conn.execute(
        sa.text(
            "INSERT INTO users (username, email, hashed_password, is_active) "
            "VALUES (:u, :e, :p, true)"
        ),
        {"u": "admin", "e": "admin@localhost", "p": _ADMIN_HASH},
    )
    user_id = conn.execute(sa.text("SELECT id FROM users WHERE username='admin'")).scalar()

    # Sample config
    conn.execute(
        sa.text(
            "INSERT INTO configs (name, description, is_active, owner_id) "
            "VALUES (:n, :d, true, :o)"
        ),
        {"n": "homelab", "d": "Sample homelab Shorewall config", "o": user_id},
    )
    config_id = conn.execute(sa.text("SELECT id FROM configs WHERE name='homelab'")).scalar()

    # Zones
    for z_name, z_type in [("fw", "firewall"), ("net", "ipv4"), ("loc", "ipv4")]:
        conn.execute(
            sa.text("INSERT INTO zones (config_id, name, type, options) VALUES (:c, :n, :t, '')"),
            {"c": config_id, "n": z_name, "t": z_type},
        )
    fw_id = conn.execute(sa.text("SELECT id FROM zones WHERE config_id=:c AND name='fw'"), {"c": config_id}).scalar()
    net_id = conn.execute(sa.text("SELECT id FROM zones WHERE config_id=:c AND name='net'"), {"c": config_id}).scalar()
    loc_id = conn.execute(sa.text("SELECT id FROM zones WHERE config_id=:c AND name='loc'"), {"c": config_id}).scalar()

    # Interface
    conn.execute(
        sa.text("INSERT INTO interfaces (config_id, name, zone_id, options) VALUES (:c, :n, :z, '')"),
        {"c": config_id, "n": "eth0", "z": net_id},
    )

    # Policies
    policies = [
        (loc_id, net_id, "ACCEPT", "", "loc to net", 1),
        (net_id, fw_id, "DROP", "info", "net to fw", 2),
        (net_id, loc_id, "DROP", "info", "net to loc", 3),
        (fw_id, net_id, "ACCEPT", "", "fw to net", 4),
        (fw_id, loc_id, "ACCEPT", "", "fw to loc", 5),
    ]
    for src, dst, pol, log, comment, pos in policies:
        conn.execute(
            sa.text(
                "INSERT INTO policies (config_id, src_zone_id, dst_zone_id, policy, log_level, comment, position) "
                "VALUES (:c, :s, :d, :p, :l, :cm, :pos)"
            ),
            {"c": config_id, "s": src, "d": dst, "p": pol, "l": log, "cm": comment, "pos": pos},
        )

    # Masq
    conn.execute(
        sa.text("INSERT INTO masq (config_id, source_network, out_interface, to_address, comment) VALUES (:c, :s, :o, '', :cm)"),
        {"c": config_id, "s": "192.168.1.0/24", "o": "eth0", "cm": "LAN masquerade"},
    )


def downgrade() -> None:
    op.drop_table("masq")
    op.drop_table("rules")
    op.drop_table("policies")
    op.drop_table("interfaces")
    op.drop_table("zones")
    op.drop_table("configs")
    op.drop_table("users")