"""initial schema and seed data Revision ID: 0001 Revises: Create Date: 2026-02-28 """ from alembic import op import sqlalchemy as sa from passlib.context import CryptContext revision = "0001" down_revision = None branch_labels = None depends_on = None pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 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": pwd_context.hash("admin")}, ) 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")