gsadmin/haproxy-openmanager
1
0
Fork 0
mirror of https://github.com/taylanbakircioglu/haproxy-openmanager.git synced 2026-09-12 05:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
haproxy-openmanager/backend/database/migrations.py
T
taylanbakircioglu 4e2d936c27 fix(requestlog): stop the table size from scaling with fleet size 
The row rate of `request_logs` was a function of how many nodes are installed,
not of what anyone did. Counted from the agent loop in linux_install.sh, each
agent's 30s cycle issues three logged calls - config, pending-requests,
upgrade-status (the heartbeat is already on the default exclude list) - plus
keepalived-config and keepalived-status every fifth cycle. That is ~9 800
rows/day per agent, essentially all of them 200s meaning "nothing changed".

Measured on PostgreSQL 15 against the real DDL and all nine indexes, at 2 424
bytes/row:

     20 agents     ~196k rows/day    453 MB/day   row cap reached in 2.5 days
    200 agents     ~2.0M rows/day    4.4 GB/day   row cap reached in  6 hours
    500 agents     ~4.9M rows/day     11 GB/day   row cap reached in  2 hours

The cap holds, so nothing runs away - but it holds by deleting, and what it
deletes is everything else. The shipped policy says 7 days of successes and 30
days of failures; on a 200-node fleet it delivers about six HOURS of both. The
forensic record the feature exists for is evicted by polling noise, and the
larger the installation the less history it keeps.

`requestlog.capture_agent_success`, default FALSE: a SUCCESSFUL inbound call
from an agent is not recorded. Failures always are, whatever the flag says -
they are what an operator needs and they are rare, so they cost nothing. With
this the table's size follows operator activity, and adding nodes does not
shorten anyone's retention.

Agent traffic is identified by header only, no database round-trip on the hot
path: the installed agent sends `X-API-Key` and never `Authorization`, the UI
sends a JWT and never an agent key. `generate-install-script`, the one endpoint
that accepts either, classifies correctly under the same rule - an operator
generating a script sends Authorization, a self-upgrading agent sends only the
key. The result is stored in the existing `target` column, which already means
"who was on the other end" for outbound rows and now means the same for inbound
ones, so no schema change and the existing target index applies.

Second half, and the reason this is one commit: `operator` holds
`requestlog.read` because, per the migration that grants it, "operators debug
failing applies and ACME orders". They could not. An apply fails on the NODE,
and the node reports that over its own API key, so the row carrying the
diagnosis has `user_id IS NULL` - and own-rows-only scoping hid it from exactly
the role the grant was written for. Scoping now admits agent rows alongside the
caller's own. Deliberately keyed on `target = 'agent'` rather than `user_id IS
NULL`: anonymous traffic is not agent traffic, so failed logins and their
usernames, and unauthenticated probes, stay admin-only.

Verified end to end through the real middleware: a successful agent poll is
dropped, a 422 from config-validation-failed is kept, operator and anonymous
calls are unaffected, and flipping the setting on restores the old behaviour.
2026-08-15 11:04:30 +03:00

4429 lines
208 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import logging
import json
import os
import hashlib
import secrets
from datetime import datetime, timedelta
from database.connection import get_database_connection, close_database_connection
logger = logging.getLogger(__name__)
async def run_init_sql():
"""Run the initial database setup from init.sql"""
conn = None
try:
conn = await get_database_connection()
# Check if basic tables already exist
tables_exist = await conn.fetchval("""
SELECT COUNT(*) FROM information_schema.tables
WHERE table_schema = 'public' AND table_name IN ('haproxy_clusters', 'users', 'config_versions')
""")
# init.sql is deprecated - all schema creation handled by migration system
logger.info("init.sql is deprecated - using migration system for all schema creation")
return
except Exception as e:
logger.error(f"Error running init.sql: {e}")
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_agents_table():
"""Ensures that all necessary tables, columns, and types exist in the database."""
conn = None
try:
conn = await get_database_connection()
# Ensure config_status enum type exists
enum_exists = await conn.fetchval("""
SELECT 1 FROM pg_type WHERE typname = 'config_status'
""")
if not enum_exists:
logger.info("Creating config_status enum type...")
await conn.execute("CREATE TYPE config_status AS ENUM ('PENDING', 'APPLIED');")
logger.info("Successfully created config_status enum type.")
await conn.execute("ALTER TYPE config_status ADD VALUE IF NOT EXISTS 'REJECTED';")
await conn.execute("ALTER TYPE config_status ADD VALUE IF NOT EXISTS 'DELETION';")
logger.info("Ensured REJECTED and DELETION values exist in config_status enum.")
# First, create essential tables if they don't exist.
#
# Rolling-restart resilience: create_essential_tables() runs idempotent
# CREATE ... IF NOT EXISTS statements on every startup. Its CREATE INDEX
# block needs a SHARE lock that conflicts with concurrent writes (e.g.
# agent heartbeats updating backend_servers/agents). During a redeploy a
# writer can hold that lock, so the DDL blocked for the full 60s
# command_timeout -> TimeoutError -> startup crash -> crash-loop.
#
# Fix: fail fast on locks (lock_timeout), retry briefly, and on
# persistent contention SKIP the idempotent bootstrap and continue — on
# an established DB the objects already exist; a fresh DB has no writers
# so the first attempt always succeeds. lock_timeout is scoped to this
# call and RESET afterwards, so every other migration below keeps its
# original (wait-indefinitely) behavior. Non-lock errors still propagate
# (genuine schema problems must NOT be masked).
import asyncio as _asyncio
_lock_excs = (_asyncio.TimeoutError,)
try:
import asyncpg as _asyncpg
_lock_excs = _lock_excs + (
_asyncpg.exceptions.LockNotAvailableError,
_asyncpg.exceptions.QueryCanceledError,
)
except Exception:
pass
try:
await conn.execute("SET lock_timeout = '10s'")
except Exception:
pass
try:
for _attempt in range(1, 4):
try:
await create_essential_tables(conn)
break
except _lock_excs as _lock_err:
if _attempt < 3:
logger.warning(
f"create_essential_tables: lock contention "
f"(attempt {_attempt}/3), retrying in 3s "
f"({type(_lock_err).__name__})"
)
await _asyncio.sleep(3)
else:
logger.warning(
"create_essential_tables: persistent lock contention; "
"skipping idempotent schema bootstrap and continuing "
"startup (objects already exist on an established DB) "
f"({type(_lock_err).__name__})"
)
finally:
try:
await conn.execute("RESET lock_timeout")
except Exception:
pass
# Ensure status column exists in config_versions table
status_column_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name='config_versions' AND column_name='status'
""")
if not status_column_exists:
logger.info("Adding status column to config_versions table...")
await conn.execute("ALTER TABLE config_versions ADD COLUMN status config_status DEFAULT 'APPLIED';")
logger.info("Successfully added status column to config_versions.")
# Update existing NULL status values to 'APPLIED'
await conn.execute("UPDATE config_versions SET status = 'APPLIED' WHERE status IS NULL;")
logger.info("Updated existing NULL status values to 'APPLIED'.")
# Ensure waf_rules.config JSONB column exists (for consolidated WAF rule configs)
waf_config_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name='waf_rules' AND column_name='config'
""")
if not waf_config_exists:
logger.info("Adding config JSONB column to waf_rules table...")
await conn.execute("ALTER TABLE waf_rules ADD COLUMN config JSONB NOT NULL DEFAULT '{}'::jsonb;")
logger.info("Successfully added waf_rules.config column.")
# Add cluster_id column to waf_rules for cluster-specific WAF rules
waf_cluster_id_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name='waf_rules' AND column_name='cluster_id'
""")
if not waf_cluster_id_exists:
logger.info("Adding cluster_id column to waf_rules table...")
await conn.execute("ALTER TABLE waf_rules ADD COLUMN cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE;")
logger.info("Successfully added cluster_id column to waf_rules table.")
# Update unique constraint to be cluster-specific
try:
# Drop existing unique constraint on name
await conn.execute("ALTER TABLE waf_rules DROP CONSTRAINT IF EXISTS waf_rules_name_key;")
# Add new unique constraint on (name, cluster_id)
await conn.execute("ALTER TABLE waf_rules ADD CONSTRAINT waf_rules_name_cluster_unique UNIQUE (name, cluster_id);")
logger.info("Updated waf_rules unique constraint to be cluster-specific (name, cluster_id).")
except Exception as constraint_error:
logger.warning(f"Could not update waf_rules unique constraint: {constraint_error}")
# Ensure haproxy_user and haproxy_group columns in haproxy_clusters
columns_to_add = {
'haproxy_user': "ALTER TABLE haproxy_clusters ADD COLUMN haproxy_user VARCHAR(255) DEFAULT 'haproxy';",
'haproxy_group': "ALTER TABLE haproxy_clusters ADD COLUMN haproxy_group VARCHAR(255) DEFAULT 'haproxy';"
}
for col, query in columns_to_add.items():
column_exists = await conn.fetchval(f"""
SELECT 1 FROM information_schema.columns
WHERE table_name='haproxy_clusters' AND column_name='{col}'
""")
if not column_exists:
logger.info(f"Column '{col}' not found in 'haproxy_clusters', adding it...")
await conn.execute(query)
logger.info(f"Successfully added column '{col}' to 'haproxy_clusters'.")
# Ensure frontend columns exist (for comprehensive frontend config support)
frontend_columns = {
'ssl_cert': "ALTER TABLE frontends ADD COLUMN ssl_cert TEXT;",
# PR-2 (R11.B): default flipped from 'optional' to NULL.
# Pre-PR-2 every newly INSERTED frontend row carried
# ``ssl_verify='optional'``, which the HAProxy config
# generator then rendered as ``bind ... ssl ... verify
# optional`` — but without a client-CA bundle (no
# ``ssl_client_ca_certificate_id`` column exists yet),
# HAProxy emitted the fatal ALERT
# ``verify is enabled but no CA file specified``. The
# explicit data cleanup migration further down (`pr2_…`)
# also flips existing ``DEFAULT 'optional'`` definitions
# on already-deployed databases via
# ``ALTER COLUMN ... DROP DEFAULT``. The Python-side
# safeguard in services/haproxy_config.py
# (``_apply_bind_ssl_verify``) is the runtime backstop;
# this is the data-side fix.
'ssl_verify': "ALTER TABLE frontends ADD COLUMN ssl_verify VARCHAR(50);",
'timeout_client': "ALTER TABLE frontends ADD COLUMN timeout_client INTEGER;",
'timeout_http_request': "ALTER TABLE frontends ADD COLUMN timeout_http_request INTEGER;",
'rate_limit': "ALTER TABLE frontends ADD COLUMN rate_limit INTEGER;",
'compression': "ALTER TABLE frontends ADD COLUMN compression BOOLEAN DEFAULT FALSE;",
'log_separate': "ALTER TABLE frontends ADD COLUMN log_separate BOOLEAN DEFAULT FALSE;",
'monitor_uri': "ALTER TABLE frontends ADD COLUMN monitor_uri VARCHAR(255);",
'use_backend_rules': "ALTER TABLE frontends ADD COLUMN use_backend_rules JSONB DEFAULT '[]'::jsonb;",
'request_headers': "ALTER TABLE frontends ADD COLUMN request_headers TEXT;",
'response_headers': "ALTER TABLE frontends ADD COLUMN response_headers TEXT;",
'maxconn': "ALTER TABLE frontends ADD COLUMN maxconn INTEGER;",
# Issue #38: SPOE filter directives (e.g. Coraza WAF) and frontend
# log-format were silently dropped on bulk-import / manual edit
# because the parser recognised only a fixed set of directives.
# These nullable TEXT columns persist them verbatim (multi-line for
# `filters`), mirroring the request_headers/options passthrough.
'log_format': "ALTER TABLE frontends ADD COLUMN log_format TEXT;",
'filters': "ALTER TABLE frontends ADD COLUMN filters TEXT;"
}
for col, query in frontend_columns.items():
column_exists = await conn.fetchval(f"""
SELECT 1 FROM information_schema.columns
WHERE table_name='frontends' AND column_name='{col}'
""")
if not column_exists:
logger.info(f"Column '{col}' not found in 'frontends', adding it...")
await conn.execute(query)
logger.info(f"Successfully added column '{col}' to 'frontends'.")
# ─────────────────────────────────────────────────────────────────
# PR-2 R11.B: ssl_verify default flip + invalid value cleanup.
# Already-deployed databases (created before PR-2) still have the
# column DEFAULT set to 'optional'. Drop the default in-place so
# all subsequent INSERTs leave the column NULL when no value is
# supplied. Existing rows are NOT mass-rewritten — operators may
# have legitimately enabled mTLS, and the runtime safeguard in
# services/haproxy_config.py handles them. We only clean up rows
# where the value is OUTSIDE the canonical Literal set, which is
# always-incorrect data (legacy 'true'/'false'/'1' artefacts that
# the unified Pydantic Literal would now reject).
# ─────────────────────────────────────────────────────────────────
try:
ssl_verify_default = await conn.fetchval("""
SELECT column_default
FROM information_schema.columns
WHERE table_name='frontends' AND column_name='ssl_verify'
""")
if ssl_verify_default and "'optional'" in str(ssl_verify_default):
logger.info(
"PR-2 R11.B: dropping legacy DEFAULT 'optional' from "
"frontends.ssl_verify (new INSERTs will leave the "
"column NULL → no `verify` directive emitted by the "
"HAProxy config generator until a client-CA bundle "
"is configured)."
)
await conn.execute(
"ALTER TABLE frontends ALTER COLUMN ssl_verify DROP DEFAULT;"
)
logger.info("PR-2 R11.B: ssl_verify DEFAULT dropped successfully.")
# Cleanup any rows whose ssl_verify is outside the canonical
# Literal set ({'none','optional','required'}). NULL and
# canonical values are preserved as-is. The cleanup is
# idempotent and bounded — only invalid values get rewritten.
#
# R11-audit-3 (FIX-3): the pre-fix block used `fetchval` with
# a CTE that returned multi-row `RETURNING f.id`, which
# silently kept only the first row's id and logged a
# misleading "cleaned up row id=<single id>" message even
# when N>1 rows were actually rewritten. Switched to
# `execute()` so we can parse the asyncpg status string
# (`'UPDATE N'`) and report the true row count.
cleanup_status = await conn.execute("""
UPDATE frontends
SET ssl_verify = NULL
WHERE ssl_verify IS NOT NULL
AND ssl_verify NOT IN ('none', 'optional', 'required')
""")
cleanup_n = 0
try:
# asyncpg returns a status tag like 'UPDATE 0' / 'UPDATE 7'
cleanup_n = int(str(cleanup_status).split()[-1])
except (ValueError, IndexError, AttributeError):
cleanup_n = 0
if cleanup_n > 0:
logger.info(
f"PR-2 R11.B: cleaned up {cleanup_n} frontends row(s) "
f"with ssl_verify outside the canonical Literal set"
)
except Exception as e:
# Idempotency guard: any failure here is non-fatal (the
# runtime safeguard still prevents the fatal HAProxy ALERT).
logger.warning(
f"PR-2 R11.B ssl_verify cleanup migration encountered "
f"a non-fatal error (continuing): {e}"
)
# Entity config status enum and per-entity status columns
entity_status_enum_exists = await conn.fetchval("""
SELECT 1 FROM pg_type WHERE typname = 'config_entity_status'
""")
if not entity_status_enum_exists:
logger.info("Creating config_entity_status enum type...")
await conn.execute("CREATE TYPE config_entity_status AS ENUM ('PENDING','APPLIED','REJECTED');")
logger.info("Successfully created config_entity_status enum type.")
# Add last_config_status to entities if missing
entity_tables = ['waf_rules', 'frontends', 'backends', 'ssl_certificates']
for table in entity_tables:
col_exists = await conn.fetchval(f"""
SELECT 1 FROM information_schema.columns
WHERE table_name='{table}' AND column_name='last_config_status'
""")
if not col_exists:
logger.info(f"Adding last_config_status to {table}...")
await conn.execute(f"ALTER TABLE {table} ADD COLUMN last_config_status config_entity_status DEFAULT 'APPLIED';")
logger.info(f"Added last_config_status to {table}.")
# Add missing columns to existing tables
missing_columns = {
'backends': {
'backend_id': "ALTER TABLE backends ADD COLUMN backend_id VARCHAR(255);",
'cluster_id': "ALTER TABLE backends ADD COLUMN cluster_id INTEGER;"
},
'users': {
'role': "ALTER TABLE users ADD COLUMN role VARCHAR(50) DEFAULT 'user';",
'last_login_at': "ALTER TABLE users ADD COLUMN last_login_at TIMESTAMP;"
},
'ssl_certificates': {
'expires_at': "ALTER TABLE ssl_certificates ADD COLUMN expires_at TIMESTAMP;",
'auto_renew': "ALTER TABLE ssl_certificates ADD COLUMN auto_renew BOOLEAN DEFAULT FALSE;",
'cluster_id': "ALTER TABLE ssl_certificates ADD COLUMN cluster_id INTEGER;",
'usage_type': "ALTER TABLE ssl_certificates ADD COLUMN usage_type VARCHAR(50) DEFAULT 'frontend';"
},
'agents': {
'architecture': "ALTER TABLE agents ADD COLUMN architecture VARCHAR(50) DEFAULT 'amd64';",
'version': "ALTER TABLE agents ADD COLUMN version VARCHAR(50) DEFAULT '1.0.0';",
'enabled': "ALTER TABLE agents ADD COLUMN enabled BOOLEAN DEFAULT TRUE;",
'haproxy_status': "ALTER TABLE agents ADD COLUMN haproxy_status VARCHAR(50) DEFAULT 'unknown';",
'haproxy_version': "ALTER TABLE agents ADD COLUMN haproxy_version VARCHAR(50);",
'config_version': "ALTER TABLE agents ADD COLUMN config_version VARCHAR(100);",
'applied_config_version': "ALTER TABLE agents ADD COLUMN applied_config_version VARCHAR(100);",
'last_validation_error': "ALTER TABLE agents ADD COLUMN last_validation_error TEXT;",
'last_validation_error_at': "ALTER TABLE agents ADD COLUMN last_validation_error_at TIMESTAMP;",
'keepalive_state': "ALTER TABLE agents ADD COLUMN keepalive_state VARCHAR(20) DEFAULT NULL;",
'keepalive_ip': "ALTER TABLE agents ADD COLUMN keepalive_ip VARCHAR(45) DEFAULT NULL;"
},
'haproxy_cluster_pools': {
'location': "ALTER TABLE haproxy_cluster_pools ADD COLUMN location VARCHAR(255);",
'is_active': "ALTER TABLE haproxy_cluster_pools ADD COLUMN is_active BOOLEAN DEFAULT TRUE;"
},
'haproxy_clusters': {
'pool_id': "ALTER TABLE haproxy_clusters ADD COLUMN pool_id INTEGER;",
'haproxy_config_path': "ALTER TABLE haproxy_clusters ADD COLUMN haproxy_config_path VARCHAR(500) DEFAULT '/etc/haproxy/haproxy.cfg';"
},
'user_activity_logs': {
'created_at': "ALTER TABLE user_activity_logs ADD COLUMN created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP;"
},
'config_versions': {
'cluster_id': "ALTER TABLE config_versions ADD COLUMN cluster_id INTEGER;",
'validation_error': "ALTER TABLE config_versions ADD COLUMN validation_error TEXT;",
'validation_error_reported_at': "ALTER TABLE config_versions ADD COLUMN validation_error_reported_at TIMESTAMP;"
}
}
for table_name, columns in missing_columns.items():
for col_name, alter_query in columns.items():
try:
# Check if table exists first
table_exists = await conn.fetchval(f"""
SELECT 1 FROM information_schema.tables
WHERE table_name='{table_name}'
""")
if not table_exists:
logger.warning(f"Table '{table_name}' does not exist yet, skipping column '{col_name}'")
continue
column_exists = await conn.fetchval(f"""
SELECT 1 FROM information_schema.columns
WHERE table_name='{table_name}' AND column_name='{col_name}'
""")
if not column_exists:
logger.info(f"Adding missing column '{col_name}' to table '{table_name}'...")
await conn.execute(alter_query)
logger.info(f"✅ Successfully added column '{col_name}' to '{table_name}'.")
else:
logger.debug(f"Column '{col_name}' already exists in '{table_name}'")
except Exception as col_error:
# Log error but don't crash - some columns may be optional
logger.error(f"❌ Could not add column '{col_name}' to '{table_name}': {col_error}")
# For critical columns like usage_type, we should still try to continue
# The error will be visible in logs for manual intervention
# Fix agents table ID issue - recreate if needed
try:
# Check if agents table has proper SERIAL PRIMARY KEY
id_check = await conn.fetchval("""
SELECT column_default FROM information_schema.columns
WHERE table_name='agents' AND column_name='id'
""")
if not id_check or 'nextval' not in str(id_check):
logger.info("Agents table ID column not properly configured, recreating table...")
# Backup existing agents data if any
existing_agents = await conn.fetch("SELECT * FROM agents")
logger.info(f"Found {len(existing_agents)} existing agents, backing up...")
# Drop and recreate agents table with proper ID
await conn.execute("DROP TABLE IF EXISTS agents CASCADE")
await conn.execute("""
CREATE TABLE agents (
id SERIAL PRIMARY KEY,
name VARCHAR(255) UNIQUE NOT NULL,
pool_id INTEGER,
platform VARCHAR(50) DEFAULT 'linux',
architecture VARCHAR(50) DEFAULT 'amd64',
version VARCHAR(50) DEFAULT '1.0.0',
hostname VARCHAR(255),
ip_address INET,
operating_system VARCHAR(255),
kernel_version VARCHAR(255),
uptime INTEGER,
cpu_count INTEGER,
memory_total INTEGER,
disk_space INTEGER,
network_interfaces TEXT[],
capabilities TEXT[],
status VARCHAR(50) DEFAULT 'offline',
last_seen TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
logger.info("Agents table recreated with proper SERIAL PRIMARY KEY")
except Exception as agent_fix_error:
logger.warning(f"Could not fix agents table: {agent_fix_error}")
# Ensure core tables exist
tables_to_create = [
"""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(255) UNIQUE NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
full_name VARCHAR(255),
phone VARCHAR(50),
role VARCHAR(50) DEFAULT 'user',
is_active BOOLEAN DEFAULT TRUE,
is_verified BOOLEAN DEFAULT FALSE,
last_login TIMESTAMP,
last_login_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS roles (
id SERIAL PRIMARY KEY,
name VARCHAR(50) UNIQUE NOT NULL,
display_name VARCHAR(100) NOT NULL,
description TEXT,
permissions JSONB NOT NULL DEFAULT '{}',
cluster_ids JSONB DEFAULT NULL,
is_active BOOLEAN DEFAULT TRUE,
is_system BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS user_activity_logs (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
action VARCHAR(100) NOT NULL,
resource_type VARCHAR(50),
resource_id VARCHAR(100),
details JSONB,
ip_address INET,
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS haproxy_cluster_pools (
id SERIAL PRIMARY KEY,
name VARCHAR(255) UNIQUE NOT NULL,
description TEXT,
environment VARCHAR(50) DEFAULT 'development',
location VARCHAR(255),
default_config JSONB,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS haproxy_clusters (
id SERIAL PRIMARY KEY,
name VARCHAR(255) UNIQUE NOT NULL,
description TEXT,
connection_type VARCHAR(50) DEFAULT 'agent',
stats_socket_path VARCHAR(500) DEFAULT '/run/haproxy/admin.sock',
haproxy_config_path VARCHAR(500) DEFAULT '/etc/haproxy/haproxy.cfg',
keepalived_config_path VARCHAR(500) DEFAULT '/etc/keepalived/keepalived.conf',
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE SET NULL,
haproxy_user VARCHAR(255) DEFAULT 'haproxy',
haproxy_group VARCHAR(255) DEFAULT 'haproxy',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS agents (
id SERIAL PRIMARY KEY,
name VARCHAR(255) UNIQUE NOT NULL,
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE CASCADE,
platform VARCHAR(50) DEFAULT 'linux',
architecture VARCHAR(50) DEFAULT 'amd64',
version VARCHAR(50) DEFAULT '1.0.0',
hostname VARCHAR(255),
ip_address INET,
operating_system VARCHAR(255),
kernel_version VARCHAR(255),
uptime INTEGER,
cpu_count INTEGER,
memory_total INTEGER,
disk_space INTEGER,
network_interfaces TEXT[],
capabilities TEXT[],
status VARCHAR(50) DEFAULT 'offline',
last_seen TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS ssl_certificates (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL UNIQUE,
domain VARCHAR(255) NOT NULL,
certificate_content TEXT NOT NULL,
private_key_content TEXT,
chain_content TEXT,
expires_at TIMESTAMP,
expiry_date DATE,
usage_type VARCHAR(50) DEFAULT 'frontend',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS backends (
id SERIAL PRIMARY KEY,
name VARCHAR(100) UNIQUE NOT NULL,
backend_id VARCHAR(255),
balance_method VARCHAR(50) DEFAULT 'roundrobin',
mode VARCHAR(20) DEFAULT 'http',
health_check_uri VARCHAR(255),
cluster_id INTEGER,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS backend_servers (
id SERIAL PRIMARY KEY,
backend_name VARCHAR(100) NOT NULL,
server_name VARCHAR(100) NOT NULL,
server_address VARCHAR(255) NOT NULL,
server_port INTEGER NOT NULL,
weight INTEGER DEFAULT 100,
check_enabled BOOLEAN DEFAULT TRUE,
backup_server BOOLEAN DEFAULT FALSE,
ssl_enabled BOOLEAN DEFAULT FALSE,
cluster_id INTEGER,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS waf_rules (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL UNIQUE,
rule_type VARCHAR(50) NOT NULL,
action VARCHAR(20) NOT NULL DEFAULT 'block',
priority INTEGER NOT NULL DEFAULT 100,
is_active BOOLEAN DEFAULT TRUE,
description TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS config_versions (
id SERIAL PRIMARY KEY,
cluster_id INTEGER,
version_number INTEGER NOT NULL,
config_content TEXT NOT NULL,
checksum VARCHAR(64),
status VARCHAR(20) DEFAULT 'APPLIED',
created_by INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
applied_at TIMESTAMP
);
""",
"""
CREATE TABLE IF NOT EXISTS frontend_waf_rules (
id SERIAL PRIMARY KEY,
frontend_id INTEGER NOT NULL,
waf_rule_id INTEGER NOT NULL,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (frontend_id) REFERENCES frontends(id) ON DELETE CASCADE,
FOREIGN KEY (waf_rule_id) REFERENCES waf_rules(id) ON DELETE CASCADE
);
"""
]
for table_sql in tables_to_create:
try:
await conn.execute(table_sql)
logger.info(f"Table created/verified successfully")
except Exception as table_error:
logger.warning(f"Table creation failed: {table_error}")
# Specifically ensure frontend_waf_rules table exists (critical for WAF functionality)
try:
# First, drop the table if it exists but is malformed
await conn.execute("DROP TABLE IF EXISTS frontend_waf_rules CASCADE")
logger.info("Dropped existing frontend_waf_rules table if it existed")
# Create the table fresh
logger.info("Creating frontend_waf_rules table...")
await conn.execute("""
CREATE TABLE frontend_waf_rules (
id SERIAL PRIMARY KEY,
frontend_id INTEGER NOT NULL,
waf_rule_id INTEGER NOT NULL,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
logger.info("frontend_waf_rules table created successfully")
# Verify table was created properly
verify = await conn.fetchval("""
SELECT COUNT(*) FROM information_schema.columns
WHERE table_name = 'frontend_waf_rules' AND column_name = 'frontend_id'
""")
if verify:
logger.info("frontend_waf_rules table verified - frontend_id column exists")
else:
logger.error("frontend_waf_rules table creation failed - frontend_id column missing")
except Exception as waf_table_error:
logger.error(f"Failed to create frontend_waf_rules table: {waf_table_error}")
# Insert initial data if not exists
try:
# Check if admin user exists
admin_exists = await conn.fetchval("SELECT 1 FROM users WHERE username = 'admin'")
if not admin_exists:
logger.info("Creating admin user...")
await conn.execute("""
INSERT INTO users (username, email, password_hash, is_admin, is_verified)
VALUES ('admin', 'admin@haproxy-openmanager.local', '$2b$12$yubNpwPopBGooz/kGVyLSuQIY3u32nA4ROXveHFVvgYWMzc/K1ymS', TRUE, TRUE)
""")
logger.info("Admin user created successfully")
# Fix host column constraint issue - make it nullable since it's not used in agent-based architecture
try:
await conn.execute("ALTER TABLE haproxy_clusters ALTER COLUMN host DROP NOT NULL")
logger.info("Made host column nullable in haproxy_clusters")
except Exception as host_error:
logger.warning(f"Could not modify host column: {host_error}")
# Add haproxy_bin_path column if it doesn't exist
try:
column_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name = 'haproxy_clusters' AND column_name = 'haproxy_bin_path'
""")
if not column_exists:
await conn.execute("ALTER TABLE haproxy_clusters ADD COLUMN haproxy_bin_path VARCHAR(500) DEFAULT '/usr/sbin/haproxy'")
logger.info("Added haproxy_bin_path column to haproxy_clusters")
except Exception as bin_path_error:
logger.warning(f"Could not add haproxy_bin_path column: {bin_path_error}")
# Add server status tracking columns to backend_servers
try:
# Add haproxy_status column for real-time server status from agents
haproxy_status_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers' AND column_name = 'haproxy_status'
""")
if not haproxy_status_exists:
await conn.execute("ALTER TABLE backend_servers ADD COLUMN haproxy_status VARCHAR(20) DEFAULT NULL")
logger.info("Added haproxy_status column to backend_servers table")
# Add haproxy_status_updated_at column for tracking status freshness
status_updated_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers' AND column_name = 'haproxy_status_updated_at'
""")
if not status_updated_exists:
await conn.execute("ALTER TABLE backend_servers ADD COLUMN haproxy_status_updated_at TIMESTAMP DEFAULT NULL")
logger.info("Added haproxy_status_updated_at column to backend_servers table")
# Create index for faster status lookups
index_exists = await conn.fetchval("""
SELECT 1 FROM pg_indexes
WHERE indexname = 'idx_backend_servers_status_lookup'
""")
if not index_exists:
await conn.execute("""
CREATE INDEX idx_backend_servers_status_lookup
ON backend_servers(backend_name, cluster_id, haproxy_status_updated_at)
""")
logger.info("Created index idx_backend_servers_status_lookup on backend_servers")
except Exception as status_error:
logger.warning(f"Could not add server status tracking columns: {status_error}")
# Default records removed - users will create their own pools and clusters as needed
except Exception as data_error:
logger.warning(f"Could not insert initial data: {data_error}")
# Create initial system roles and users if they don't exist
await create_initial_system_data(conn)
logger.info("Database schema check/update completed successfully.")
except Exception as e:
logger.error(f"Database schema check/update failed: {e}", exc_info=True)
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_config_versions_metadata_column():
"""Add metadata column to config_versions table for storing pre-apply snapshots."""
conn = None
try:
conn = await get_database_connection()
# Check if metadata column exists
metadata_column_exists = await conn.fetchval("""
SELECT 1 FROM information_schema.columns
WHERE table_name='config_versions' AND column_name='metadata'
""")
if not metadata_column_exists:
logger.info("Adding metadata column to config_versions table...")
await conn.execute("ALTER TABLE config_versions ADD COLUMN metadata JSONB;")
logger.info("Successfully added metadata column to config_versions.")
else:
logger.info("Metadata column already exists in config_versions table.")
except Exception as e:
logger.error(f"Error ensuring config_versions metadata column: {e}")
raise e
finally:
if conn:
await close_database_connection(conn)
async def ensure_user_pool_access_table():
"""Ensure user_pool_access table exists for multi-cluster access control"""
conn = await get_database_connection()
try:
# Check if table exists
table_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'user_pool_access'
)
""")
if not table_exists:
logger.info("Creating user_pool_access table...")
await conn.execute("""
CREATE TABLE user_pool_access (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE CASCADE,
access_level VARCHAR(20) DEFAULT 'read_write',
granted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
granted_by INTEGER REFERENCES users(id),
is_active BOOLEAN DEFAULT TRUE,
expires_at TIMESTAMP NULL,
UNIQUE(user_id, pool_id)
);
""")
# Create indexes
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_pool_access_user ON user_pool_access(user_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_pool_access_pool ON user_pool_access(pool_id);
""")
# Grant admin user access to all existing pools
await conn.execute("""
INSERT INTO user_pool_access (user_id, pool_id, access_level, granted_by)
SELECT u.id, p.id, 'admin', u.id
FROM users u, haproxy_cluster_pools p
WHERE u.username = 'admin' AND u.is_admin = TRUE
ON CONFLICT (user_id, pool_id) DO NOTHING;
""")
logger.info("✅ user_pool_access table created successfully with admin access")
else:
logger.info("user_pool_access table already exists")
except Exception as e:
logger.error(f"Failed to create user_pool_access table: {e}")
raise
finally:
await close_database_connection(conn)
async def ensure_backend_servers_last_config_status():
"""Add last_config_status column to backend_servers table"""
conn = await get_database_connection()
try:
# Check if column exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers'
AND column_name = 'last_config_status'
);
""")
if not column_exists:
logger.info("Adding last_config_status column to backend_servers table...")
await conn.execute("""
ALTER TABLE backend_servers
ADD COLUMN last_config_status VARCHAR(20) DEFAULT 'APPLIED'
""")
# Update existing servers to APPLIED status
await conn.execute("""
UPDATE backend_servers
SET last_config_status = 'APPLIED'
WHERE last_config_status IS NULL
""")
logger.info("Successfully added last_config_status column to backend_servers table")
else:
logger.info("last_config_status column already exists in backend_servers table")
except Exception as e:
logger.error(f"Error adding last_config_status column to backend_servers: {e}")
raise
finally:
await close_database_connection(conn)
async def ensure_backend_servers_haproxy_status():
"""Add haproxy_status and haproxy_status_updated_at columns to backend_servers table"""
conn = await get_database_connection()
try:
# Check if columns exist
haproxy_status_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers'
AND column_name = 'haproxy_status'
);
""")
haproxy_status_updated_at_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers'
AND column_name = 'haproxy_status_updated_at'
);
""")
if not haproxy_status_exists:
logger.info("Adding haproxy_status column to backend_servers table...")
await conn.execute("""
ALTER TABLE backend_servers
ADD COLUMN haproxy_status VARCHAR(20) DEFAULT 'UNKNOWN'
""")
logger.info("Successfully added haproxy_status column to backend_servers table")
else:
logger.info("haproxy_status column already exists in backend_servers table")
if not haproxy_status_updated_at_exists:
logger.info("Adding haproxy_status_updated_at column to backend_servers table...")
await conn.execute("""
ALTER TABLE backend_servers
ADD COLUMN haproxy_status_updated_at TIMESTAMP
""")
logger.info("Successfully added haproxy_status_updated_at column to backend_servers table")
else:
logger.info("haproxy_status_updated_at column already exists in backend_servers table")
except Exception as e:
logger.error(f"Error adding haproxy_status columns to backend_servers: {e}")
raise
finally:
await close_database_connection(conn)
async def ensure_ssl_certificates_new_columns():
"""Ensure ssl_certificates table has new schema columns"""
conn = None
try:
conn = await get_database_connection()
# Check if ssl_certificates table exists
table_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'ssl_certificates'
)
""")
if not table_exists:
logger.info("ssl_certificates table does not exist, skipping migration")
return
# CRITICAL: Make private_key_content nullable for server SSL support
# Check if private_key_content exists and is NOT NULL
private_key_not_null = await conn.fetchval("""
SELECT is_nullable = 'NO'
FROM information_schema.columns
WHERE table_name = 'ssl_certificates' AND column_name = 'private_key_content'
""")
if private_key_not_null:
logger.info("🔧 Altering private_key_content to allow NULL (for server SSL)...")
await conn.execute("""
ALTER TABLE ssl_certificates
ALTER COLUMN private_key_content DROP NOT NULL
""")
logger.info("✅ private_key_content is now nullable (server SSL support)")
# Add missing columns if they don't exist
columns_to_add = [
("primary_domain", "VARCHAR(255)"),
("issuer", "VARCHAR(255)"),
("fingerprint", "VARCHAR(128)"),
("status", "VARCHAR(20) DEFAULT 'valid'"),
("days_until_expiry", "INTEGER DEFAULT 0"),
("all_domains", "JSONB DEFAULT '[]'"),
("usage_type", "VARCHAR(50) DEFAULT 'frontend'")
]
for column_name, column_def in columns_to_add:
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'ssl_certificates' AND column_name = $1
)
""", column_name)
if not column_exists:
await conn.execute(f"""
ALTER TABLE ssl_certificates
ADD COLUMN {column_name} {column_def}
""")
logger.info(f"Added column '{column_name}' to ssl_certificates table")
# Backfill primary_domain from legacy domain column for upgraded installations
has_domain_col = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'ssl_certificates' AND column_name = 'domain'
)
""")
if has_domain_col:
await conn.execute("""
UPDATE ssl_certificates
SET primary_domain = domain
WHERE primary_domain IS NULL AND domain IS NOT NULL
""")
logger.info("Backfilled primary_domain from legacy domain column")
# Make legacy domain column nullable (new code uses primary_domain)
if has_domain_col:
domain_not_null = await conn.fetchval("""
SELECT is_nullable = 'NO'
FROM information_schema.columns
WHERE table_name = 'ssl_certificates' AND column_name = 'domain'
""")
if domain_not_null:
await conn.execute("""
ALTER TABLE ssl_certificates
ALTER COLUMN domain DROP NOT NULL
""")
logger.info("Made legacy domain column nullable")
# Update existing certificates to have valid status
await conn.execute("""
UPDATE ssl_certificates
SET status = 'valid'
WHERE status IS NULL
""")
logger.info("SSL certificates table schema updated successfully")
except Exception as e:
logger.error(f"Failed to update ssl_certificates schema: {e}")
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_ssl_cluster_junction_table():
"""Create SSL-Cluster many-to-many junction table"""
try:
conn = await get_database_connection()
# Create ssl_certificate_clusters junction table
await conn.execute("""
CREATE TABLE IF NOT EXISTS ssl_certificate_clusters (
id SERIAL PRIMARY KEY,
ssl_certificate_id INTEGER NOT NULL REFERENCES ssl_certificates(id) ON DELETE CASCADE,
cluster_id INTEGER NOT NULL REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(ssl_certificate_id, cluster_id)
)
""")
# Migrate existing ssl_certificates.cluster_id data to junction table
existing_mappings = await conn.fetch("""
SELECT id, cluster_id FROM ssl_certificates
WHERE cluster_id IS NOT NULL
""")
for mapping in existing_mappings:
# Insert into junction table if not exists
await conn.execute("""
INSERT INTO ssl_certificate_clusters (ssl_certificate_id, cluster_id)
VALUES ($1, $2)
ON CONFLICT (ssl_certificate_id, cluster_id) DO NOTHING
""", mapping['id'], mapping['cluster_id'])
await close_database_connection(conn)
logger.info("✅ SSL-Cluster junction table migration completed")
except Exception as e:
logger.error(f"❌ SSL-Cluster junction table migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def fix_backend_unique_constraint():
"""Fix backend name unique constraint to be cluster-specific"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting backend unique constraint migration...")
# Check if global unique constraint exists
constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'backends'
AND constraint_name = 'backends_name_key'
AND constraint_type = 'UNIQUE'
)
""")
if constraint_exists:
logger.info("📝 Dropping global unique constraint on backends.name...")
await conn.execute("ALTER TABLE backends DROP CONSTRAINT IF EXISTS backends_name_key")
# Check if cluster-specific unique constraint already exists
cluster_constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'backends'
AND constraint_name = 'backends_name_cluster_unique'
AND constraint_type = 'UNIQUE'
)
""")
if not cluster_constraint_exists:
logger.info("➕ Adding cluster-specific unique constraint on backends (name, cluster_id)...")
await conn.execute("""
ALTER TABLE backends
ADD CONSTRAINT backends_name_cluster_unique
UNIQUE (name, cluster_id)
""")
logger.info("✅ Backend unique constraint fixed successfully")
else:
logger.info("ℹ️ Cluster-specific unique constraint already exists")
else:
logger.info("ℹ️ Global unique constraint on backends.name does not exist")
await close_database_connection(conn)
logger.info("✅ Backend unique constraint migration completed")
except Exception as e:
logger.error(f"❌ Backend unique constraint migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def fix_frontend_unique_constraint():
"""Fix frontend name unique constraint to be cluster-specific"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting frontend unique constraint migration...")
# Check if global unique constraint exists
constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'frontends'
AND constraint_name = 'frontends_name_key'
AND constraint_type = 'UNIQUE'
)
""")
if constraint_exists:
logger.info("📝 Dropping global unique constraint on frontends.name...")
await conn.execute("ALTER TABLE frontends DROP CONSTRAINT IF EXISTS frontends_name_key")
# Check if cluster-specific unique constraint already exists
cluster_constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'frontends'
AND constraint_name = 'frontends_name_cluster_unique'
AND constraint_type = 'UNIQUE'
)
""")
if not cluster_constraint_exists:
logger.info("➕ Adding cluster-specific unique constraint on frontends (name, cluster_id)...")
await conn.execute("""
ALTER TABLE frontends
ADD CONSTRAINT frontends_name_cluster_unique
UNIQUE (name, cluster_id)
""")
logger.info("✅ Frontend unique constraint fixed successfully")
else:
logger.info("ℹ️ Cluster-specific unique constraint already exists")
else:
logger.info("ℹ️ Global unique constraint on frontends.name does not exist")
await close_database_connection(conn)
logger.info("✅ Frontend unique constraint migration completed")
except Exception as e:
logger.error(f"❌ Frontend unique constraint migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def fix_waf_unique_constraint():
"""Fix WAF rules name unique constraint to be cluster-specific"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting WAF rules unique constraint migration...")
# Check if global unique constraint exists
constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'waf_rules'
AND constraint_name = 'waf_rules_name_key'
AND constraint_type = 'UNIQUE'
)
""")
if constraint_exists:
logger.info("📝 Dropping global unique constraint on waf_rules.name...")
await conn.execute("ALTER TABLE waf_rules DROP CONSTRAINT IF EXISTS waf_rules_name_key")
# Check if cluster-specific unique constraint already exists
cluster_constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'waf_rules'
AND constraint_name = 'waf_rules_name_cluster_unique'
AND constraint_type = 'UNIQUE'
)
""")
if not cluster_constraint_exists:
logger.info("➕ Adding cluster-specific unique constraint on waf_rules (name, cluster_id)...")
await conn.execute("""
ALTER TABLE waf_rules
ADD CONSTRAINT waf_rules_name_cluster_unique
UNIQUE (name, cluster_id)
""")
logger.info("✅ WAF rules unique constraint fixed successfully")
else:
logger.info("ℹ️ Cluster-specific unique constraint already exists")
else:
logger.info("ℹ️ Global unique constraint on waf_rules.name does not exist")
await close_database_connection(conn)
logger.info("✅ WAF rules unique constraint migration completed")
except Exception as e:
logger.error(f"❌ WAF rules unique constraint migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def fix_backend_servers_unique_constraint():
"""Fix backend servers unique constraint to be cluster-specific"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting backend servers unique constraint migration...")
# Check if cluster-specific unique constraint already exists
constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.table_constraints
WHERE table_name = 'backend_servers'
AND constraint_name = 'backend_servers_backend_name_server_name_cluster_id_key'
AND constraint_type = 'UNIQUE'
)
""")
if not constraint_exists:
logger.info("➕ Adding cluster-specific unique constraint on backend_servers (backend_name, server_name, cluster_id)...")
await conn.execute("""
ALTER TABLE backend_servers
ADD CONSTRAINT backend_servers_backend_name_server_name_cluster_id_key
UNIQUE (backend_name, server_name, cluster_id)
""")
logger.info("✅ Backend servers unique constraint added successfully")
else:
logger.info("ℹ️ Cluster-specific unique constraint already exists")
await close_database_connection(conn)
logger.info("✅ Backend servers unique constraint migration completed")
except Exception as e:
logger.error(f"❌ Backend servers unique constraint migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def add_frontend_ssl_port_column():
"""Add ssl_port column to frontends table"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting frontend ssl_port column migration...")
# Check if ssl_port column exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends'
AND column_name = 'ssl_port'
)
""")
if not column_exists:
logger.info("➕ Adding ssl_port column to frontends table...")
await conn.execute("ALTER TABLE frontends ADD COLUMN ssl_port INTEGER")
logger.info("✅ ssl_port column added successfully")
else:
logger.info("ℹ️ ssl_port column already exists")
await close_database_connection(conn)
logger.info("✅ Frontend ssl_port column migration completed")
except Exception as e:
logger.error(f"❌ Frontend ssl_port column migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def add_config_versions_updated_at_column():
"""Add updated_at column to config_versions table"""
conn = None
try:
conn = await get_database_connection()
logger.info("🔧 Starting config_versions updated_at column migration...")
# Check if updated_at column exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'config_versions'
AND column_name = 'updated_at'
)
""")
if not column_exists:
logger.info("➕ Adding updated_at column to config_versions table...")
await conn.execute("""
ALTER TABLE config_versions
ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
""")
logger.info("✅ updated_at column added successfully")
else:
logger.info("ℹ️ updated_at column already exists")
await close_database_connection(conn)
logger.info("✅ Config versions updated_at column migration completed")
except Exception as e:
logger.error(f"❌ Config versions updated_at column migration failed: {e}")
if conn:
await close_database_connection(conn)
raise
async def ensure_roles_cluster_ids_column():
"""Add cluster_ids JSONB column to roles table for cluster-specific roles"""
conn = None
try:
conn = await get_database_connection()
# Check if cluster_ids column exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'roles' AND column_name = 'cluster_ids'
)
""")
if not column_exists:
logger.info("Adding cluster_ids column to roles table...")
await conn.execute("""
ALTER TABLE roles
ADD COLUMN cluster_ids JSONB DEFAULT NULL
""")
logger.info("Successfully added cluster_ids column to roles table")
await close_database_connection(conn)
except Exception as e:
logger.error(f"Failed to add cluster_ids column to roles: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
async def update_system_roles_to_enterprise_rbac():
"""Update system roles to new enterprise RBAC permissions structure"""
conn = None
try:
conn = await get_database_connection()
# Define enterprise permissions for each role
enterprise_roles = {
'super_admin': {
'display_name': 'Super Administrator',
'description': 'Full system access with all permissions',
'permissions': [
# All permissions - full access
'dashboard.read', 'dashboard.statistics', 'dashboard.metrics',
'frontends.read', 'frontends.create', 'frontends.update', 'frontends.delete', 'frontends.toggle', 'frontends.history',
'backends.read', 'backends.create', 'backends.update', 'backends.delete', 'backends.toggle', 'backends.servers', 'backends.history',
'waf.read', 'waf.create', 'waf.update', 'waf.delete', 'waf.toggle', 'waf.history',
'ssl.read', 'ssl.create', 'ssl.update', 'ssl.delete', 'ssl.download', 'ssl.history',
'apply.read', 'apply.execute', 'apply.reject', 'apply.history', 'apply.bulk', 'apply.emergency',
'agents.read', 'agents.create', 'agents.update', 'agents.delete', 'agents.script', 'agents.toggle', 'agents.upgrade', 'agents.version', 'agents.logs',
'clusters.read', 'clusters.create', 'clusters.update', 'clusters.delete', 'clusters.switch', 'clusters.config',
'vip.read', 'vip.create', 'vip.update', 'vip.delete', 'vip.apply',
'config.read', 'config.update', 'config.download', 'config.upload', 'config.backup', 'config.restore', 'config.history', 'config.bulk_import', 'config.view_request', 'config.download_request',
'users.read', 'users.create', 'users.update', 'users.delete', 'users.password', 'users.roles',
'roles.read', 'roles.create', 'roles.update', 'roles.delete', 'roles.permissions',
'statistics.read', 'statistics.performance', 'statistics.agents', 'statistics.health', 'statistics.export',
'activity.read', 'activity.all', 'activity.export',
# v1.11.0 — request/response log. `read` browses the log,
# `manage` edits retention/capture settings and triggers a
# manual purge.
'requestlog.read', 'requestlog.manage',
'settings.read', 'settings.update', 'settings.system', 'settings.security',
'system.restart', 'system.logs', 'system.database', 'system.services', 'system.emergency'
]
},
'operator': {
'display_name': 'Operator',
'description': 'Daily operational access for managing HAProxy configurations and applying changes',
'permissions': [
'dashboard.read', 'dashboard.statistics', 'dashboard.metrics',
'frontends.read', 'frontends.create', 'frontends.update', 'frontends.delete', 'frontends.toggle', 'frontends.history',
'backends.read', 'backends.create', 'backends.update', 'backends.delete', 'backends.toggle', 'backends.servers', 'backends.history',
'waf.read', 'waf.create', 'waf.update', 'waf.delete', 'waf.toggle', 'waf.history',
'ssl.read', 'ssl.create', 'ssl.update', 'ssl.delete', 'ssl.download', 'ssl.history',
'apply.read', 'apply.execute', 'apply.reject', 'apply.history', 'apply.bulk',
'agents.read', 'agents.update', 'agents.toggle', 'agents.upgrade', 'agents.version', 'agents.logs',
'clusters.read', 'clusters.switch', 'clusters.config',
'vip.read', 'vip.create', 'vip.update', 'vip.delete', 'vip.apply',
'config.read', 'config.update', 'config.download', 'config.history', 'config.bulk_import', 'config.view_request', 'config.download_request',
'statistics.read', 'statistics.performance', 'statistics.agents', 'statistics.health',
'activity.read',
# v1.11.0 — operators debug failing applies and ACME orders,
# so they get read access to the request log; retention and
# purge stay with the admins.
'requestlog.read'
]
},
'security_admin': {
'display_name': 'Security Administrator',
'description': 'Security-focused access for WAF rules, SSL certificates, and security monitoring',
'permissions': [
'dashboard.read', 'dashboard.statistics', 'dashboard.metrics',
'frontends.read', 'frontends.history',
'backends.read', 'backends.history',
'waf.read', 'waf.create', 'waf.update', 'waf.delete', 'waf.toggle', 'waf.history',
'ssl.read', 'ssl.create', 'ssl.update', 'ssl.delete', 'ssl.download', 'ssl.history',
'apply.read', 'apply.execute', 'apply.reject', 'apply.history',
'agents.read', 'agents.version', 'agents.logs',
'clusters.read', 'clusters.switch',
'vip.read',
'config.read', 'config.history', 'config.view_request', 'config.download_request',
'statistics.read', 'statistics.performance', 'statistics.agents', 'statistics.health',
'activity.read', 'activity.all', 'activity.export',
# v1.11.0 — the request log is a security-forensics surface,
# so the security admin gets both read and retention control.
'requestlog.read', 'requestlog.manage',
'settings.read', 'settings.security'
]
},
# NOTE (v1.11.0): `viewer` deliberately gets NEITHER requestlog
# permission. Even redacted, captured request/response bodies are a
# far broader disclosure surface than the read-only configuration
# views a viewer is meant to have.
'viewer': {
'display_name': 'Viewer',
'description': 'Read-only access to view configurations, statistics, and monitor system status',
'permissions': [
'dashboard.read', 'dashboard.statistics', 'dashboard.metrics',
'frontends.read', 'frontends.history',
'backends.read', 'backends.history',
'waf.read', 'waf.history',
'ssl.read', 'ssl.history',
'apply.read', 'apply.history',
'agents.read',
'clusters.read', 'clusters.switch',
'vip.read',
'config.read', 'config.history', 'config.view_request',
'statistics.read', 'statistics.performance', 'statistics.agents', 'statistics.health',
'activity.read',
'settings.read'
]
}
}
# Update each system role
for role_name, role_data in enterprise_roles.items():
try:
# Check if role exists
role_exists = await conn.fetchval("""
SELECT id FROM roles WHERE name = $1
""", role_name)
if role_exists:
# Update existing role
await conn.execute("""
UPDATE roles SET
display_name = $1,
description = $2,
permissions = $3,
updated_at = CURRENT_TIMESTAMP
WHERE name = $4
""",
role_data['display_name'],
role_data['description'],
json.dumps(role_data['permissions']),
role_name
)
logger.info(f"✅ Updated system role: {role_name} with {len(role_data['permissions'])} permissions")
else:
# Create new role
await conn.execute("""
INSERT INTO roles (name, display_name, description, permissions, is_active, is_system)
VALUES ($1, $2, $3, $4, $5, $6)
""",
role_name,
role_data['display_name'],
role_data['description'],
json.dumps(role_data['permissions']),
True,
True
)
logger.info(f"✅ Created system role: {role_name} with {len(role_data['permissions'])} permissions")
except Exception as role_error:
logger.warning(f"Failed to update role {role_name}: {role_error}")
# Ensure default users exist with correct roles
default_users = {
'admin': {
'email': 'admin@haproxy-openmanager.local',
'full_name': 'System Administrator',
'phone': '+1-555-0001',
'role': 'super_admin',
'is_admin': True
},
'operator1': {
'email': 'operator@haproxy-openmanager.local',
'full_name': 'Daily Operator',
'phone': '+1-555-0101',
'role': 'operator',
'is_admin': False
},
'security1': {
'email': 'security@haproxy-openmanager.local',
'full_name': 'Security Analyst',
'phone': '+1-555-0102',
'role': 'security_admin',
'is_admin': False
},
'viewer1': {
'email': 'viewer@haproxy-openmanager.local',
'full_name': 'Read Only User',
'phone': '+1-555-0103',
'role': 'viewer',
'is_admin': False
}
}
# Default password hash for all users: admin123
password_hash = '$2b$12$yubNpwPopBGooz/kGVyLSuQIY3u32nA4ROXveHFVvgYWMzc/K1ymS'
for username, user_data in default_users.items():
try:
# Check if user exists
user_exists = await conn.fetchval("""
SELECT id FROM users WHERE username = $1
""", username)
if not user_exists:
# Create user
user_id = await conn.fetchval("""
INSERT INTO users (username, email, password_hash, full_name, phone, is_active, is_admin, is_verified)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
RETURNING id
""",
username,
user_data['email'],
password_hash,
user_data['full_name'],
user_data['phone'],
True, # is_active
user_data['is_admin'],
True # is_verified
)
logger.info(f"✅ Created default user: {username}")
else:
user_id = user_exists
logger.info(f"✅ Default user already exists: {username}")
# Assign role to user
role_id = await conn.fetchval("""
SELECT id FROM roles WHERE name = $1
""", user_data['role'])
if role_id:
# Check if role assignment exists
assignment_exists = await conn.fetchval("""
SELECT id FROM user_roles WHERE user_id = $1 AND role_id = $2
""", user_id, role_id)
if not assignment_exists:
await conn.execute("""
INSERT INTO user_roles (user_id, role_id, assigned_by)
VALUES ($1, $2, $3)
""", user_id, role_id, 1) # Assigned by admin (id=1)
logger.info(f"✅ Assigned role {user_data['role']} to user {username}")
except Exception as user_error:
logger.warning(f"Failed to create/update user {username}: {user_error}")
await close_database_connection(conn)
logger.info("✅ Enterprise RBAC system roles and users migration completed")
except Exception as e:
logger.error(f"Failed to update system roles to enterprise RBAC: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
async def ensure_user_activity_logs_table():
"""Ensure user_activity_logs table exists with proper schema"""
conn = None
try:
conn = await get_database_connection()
# Check if table exists
table_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'user_activity_logs'
)
""")
if not table_exists:
logger.info("Creating user_activity_logs table...")
await conn.execute("""
CREATE TABLE user_activity_logs (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
action VARCHAR(100) NOT NULL,
resource_type VARCHAR(50),
resource_id VARCHAR(100),
details JSONB,
ip_address INET,
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create indexes for better performance
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_user_id ON user_activity_logs(user_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_created_at ON user_activity_logs(created_at DESC);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_resource ON user_activity_logs(resource_type, resource_id);
""")
logger.info("Successfully created user_activity_logs table with indexes")
else:
logger.info("user_activity_logs table already exists")
# Ensure all required columns exist
required_columns = {
'created_at': 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP',
'ip_address': 'INET',
'user_agent': 'TEXT',
'details': 'JSONB'
}
for column_name, column_type in required_columns.items():
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'user_activity_logs' AND column_name = $1
)
""", column_name)
if not column_exists:
logger.info(f"Adding missing column {column_name} to user_activity_logs table...")
await conn.execute(f"""
ALTER TABLE user_activity_logs
ADD COLUMN {column_name} {column_type}
""")
await close_database_connection(conn)
logger.info("✅ user_activity_logs table migration completed")
except Exception as e:
logger.error(f"Failed to ensure user_activity_logs table: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
async def ensure_agent_activity_logs_table():
"""Create agent_activity_logs table for tracking meaningful agent actions"""
conn = None
try:
conn = await get_database_connection()
# Check if table exists
table_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = 'agent_activity_logs'
)
""")
if not table_exists:
logger.info("Creating agent_activity_logs table...")
await conn.execute("""
CREATE TABLE agent_activity_logs (
id SERIAL PRIMARY KEY,
agent_id INTEGER REFERENCES agents(id) ON DELETE CASCADE,
agent_name VARCHAR(255) NOT NULL,
action_type VARCHAR(50) NOT NULL,
action_details JSONB,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create indexes for performance
await conn.execute("""
CREATE INDEX idx_agent_activity_logs_agent_id ON agent_activity_logs(agent_id);
CREATE INDEX idx_agent_activity_logs_agent_name ON agent_activity_logs(agent_name);
CREATE INDEX idx_agent_activity_logs_timestamp ON agent_activity_logs(timestamp DESC);
CREATE INDEX idx_agent_activity_logs_action_type ON agent_activity_logs(action_type);
""")
logger.info("✅ agent_activity_logs table created successfully")
else:
logger.info("agent_activity_logs table already exists")
# Add last_action_time column to agents table
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT FROM information_schema.columns
WHERE table_name = 'agents' AND column_name = 'last_action_time'
)
""")
if not column_exists:
logger.info("Adding last_action_time column to agents table...")
await conn.execute("""
ALTER TABLE agents ADD COLUMN last_action_time TIMESTAMP;
""")
logger.info("✅ last_action_time column added to agents table")
await close_database_connection(conn)
except Exception as e:
logger.error(f"Failed to ensure agent_activity_logs table: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
# Schema-version gate for the migration runner.
#
# >>> BUMP THIS whenever you add/modify ANY step in _run_all_migrations_inner()
# >>> that changes the schema (table/column/index/constraint) OR seeded/role data
# >>> (e.g. update_system_roles_to_enterprise_rbac). Otherwise the new step will
# >>> NOT run on databases already marked at the current version.
#
# When the DB already records >= this version, run_all_migrations() skips the
# whole (lock-heavy) idempotent sequence, so redeploys/scale-ups issue NO DDL and
# a concurrently-serving replica's traffic cannot block ALTER / CREATE INDEX (the
# rolling-deploy startup crash that motivated this gate).
#
# Backward compatibility (the product runs at many versions across companies):
# - First start on this code: no marker -> applied_version is NULL -> the FULL
# sequence runs (upgrades any prior version), THEN the marker is written. So
# upgrading from any older version is unaffected.
# - The marker is written ONLY after _run_all_migrations_inner() completes with
# no exception, so an interrupted/failed migration never marks an incomplete
# schema as done — the next start retries.
# - Behavior change vs the historical "re-run every idempotent ensure_* on every
# start": once marked, same-version restarts no longer re-run (and therefore no
# longer auto-repair manual drift). To force a re-run, bump SCHEMA_VERSION or
# delete the schema_migrations row.
#
# v1.7.0 (Issue #27 — HA/VIP Keepalived management): bumped 1 -> 2 so the new
# additive ensure_vip_tables() step (two brand-new tables) actually runs on
# databases already marked at version 1. The whole re-run is idempotent.
# v1.7.0 self-review: bumped 2 -> 3 so the additive `applied_snapshot` column on
# vip_instances (enables VIP reject/restore-to-previous) lands on DBs marked at 2.
# v1.7.0 self-review: bumped 3 -> 4 for the additive `keepalived_config_path` column on
# haproxy_clusters (cluster-driven keepalived.conf path, like haproxy_config_path).
# v1.7.0 self-review: bumped 4 -> 5 to drop the table-level UNIQUE on vip_instances.name
# and replace it with a partial unique index (active rows only), so a soft-deleted VIP's
# name is reusable — consistent with the address/VRID partial indexes. Idempotent re-run.
# v1.7.2: bumped 5 -> 6 for the additive `purge_on_teardown` column on vip_instances
# (opt-in "also uninstall the keepalived package on delete"; default FALSE keeps the safe
# graceful-teardown behaviour). Additive + idempotent.
# v1.7.2: bumped 6 -> 7 for the additive `pending_delete` column on vip_instances
# (approval-gated VIP deletion: a delete is staged for Apply Management and the VIP keeps
# running until APPROVED, so an agent never tears down without explicit human approval).
# v1.8.0 (Issue #35 — ACME DNS-01 challenge support): bumped 7 -> 8 for additive DNS-01
# columns on letsencrypt_accounts/letsencrypt_orders/acme_challenges and the brand-new
# letsencrypt_account_dns_credentials table (ensure_letsencrypt_dns_credentials step).
# All additive + idempotent; default challenge_type 'http-01' keeps existing flows byte-identical.
# v1.8.8 (Issue #38 — SPOE filter + frontend log-format): bumped 8 -> 9 for the additive
# `log_format` + `filters` TEXT columns on `frontends` (frontend_columns loop). Without this
# bump, already-deployed databases (version >= 8) skip the whole migration run and never gain
# the columns, so the frontends SELECT/INSERT would fail. Additive + idempotent + nullable;
# existing rows stay NULL and render byte-identical.
# v1.9.0 (CSR creation): bumped 9 -> 10 for the brand-new `ssl_csrs` table
# (ensure_ssl_csrs_table step). Holds a locally generated private key + CSR PEM
# until the operator imports the CA-signed certificate; the import creates a
# normal ssl_certificates row and NULLs the key copy here. Additive + idempotent;
# no existing table is altered, agents never read this table.
# v1.10.4 (VIP adoption): bumped 10 -> 11 for the new `vip_discoveries` table plus two
# additive columns (`vip_instances.adopted_at`, `vip_members.takeover_expected_hash`).
# Holds the keepalived.conf an agent found already on a node so an existing VIP can be
# adopted instead of retyped. Additive + idempotent; no existing table is altered and no
# existing row changes. NOTE for the upgrade notes: a SCHEMA_VERSION bump re-seeds the four
# built-in roles to their defaults, so role customizations are lost on this upgrade.
# v1.11.0 (unified request/response log): bumped 11 -> 12 for the brand-new
# `request_logs` table (ensure_request_logs_table), its retention-settings seed
# (ensure_request_log_settings), and the new `requestlog.read` /
# `requestlog.manage` permissions added to the built-in roles in
# update_system_roles_to_enterprise_rbac().
#
# 12, NOT 11. The feature branch was cut when this constant was still 10 and
# proposed 11, but 11 was taken in the meantime by v1.10.4 (vip_discoveries)
# above. Landing it as 11 would be silently inert: run_all_migrations() returns
# early on `applied_version >= SCHEMA_VERSION`, so every database already at 11
# would skip the whole sequence and get neither the table nor the permissions,
# while a fresh install would get both. Same class of bug the bump exists to
# prevent, one number later.
#
# Additive + idempotent; no existing table is altered, agents never read this
# table. Same caveat as v1.10.4: this bump re-seeds the four built-in roles to
# their defaults, so export role customizations before upgrading.
SCHEMA_VERSION = 12
async def run_all_migrations():
"""Run all database migrations.
Hardened for multiple backend replicas / rolling deploys:
- A session-level advisory lock serializes the run so only one pod migrates
at a time (others wait, then hit the version gate and skip). It is
session-scoped, so it auto-releases if a pod dies mid-migration.
- A schema-version marker (schema_migrations) gates the run: when the DB is
already at SCHEMA_VERSION the whole idempotent sequence is skipped, so no
DDL is issued and a serving replica's traffic can't block it.
If the advisory lock or marker can't be used, we fall back to running the
(idempotent) migrations rather than crashing startup.
"""
logger.info("Starting database migrations...")
MIGRATION_ADVISORY_LOCK_KEY = 1836016242 # single-key advisory space ("migr"); distinct from the (ns,id) locks used elsewhere
lock_conn = None
lock_acquired = False
try:
lock_conn = await get_database_connection()
try:
await lock_conn.execute("SELECT pg_advisory_lock($1)", MIGRATION_ADVISORY_LOCK_KEY)
lock_acquired = True
logger.info("Acquired migration advisory lock (migrations serialized across pods)")
except Exception as _lock_e:
logger.warning(f"Could not acquire migration advisory lock; proceeding (migrations are idempotent): {_lock_e}")
# Schema-version gate: skip the lock-heavy sequence if the DB is current.
applied_version = None
try:
await lock_conn.execute("""
CREATE TABLE IF NOT EXISTS schema_migrations (
id INTEGER PRIMARY KEY DEFAULT 1,
version INTEGER NOT NULL,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT schema_migrations_singleton CHECK (id = 1)
)
""")
applied_version = await lock_conn.fetchval("SELECT version FROM schema_migrations WHERE id = 1")
except Exception as _mk_e:
logger.warning(f"schema_migrations marker unavailable; running full migrations: {_mk_e}")
applied_version = None
if applied_version is not None and applied_version >= SCHEMA_VERSION:
logger.info(f"Schema already at version {applied_version} (>= {SCHEMA_VERSION}); skipping migration run.")
return
await _run_all_migrations_inner()
try:
await lock_conn.execute("""
INSERT INTO schema_migrations (id, version, applied_at)
VALUES (1, $1, CURRENT_TIMESTAMP)
ON CONFLICT (id) DO UPDATE SET version = EXCLUDED.version, applied_at = EXCLUDED.applied_at
""", SCHEMA_VERSION)
logger.info(f"Recorded schema version {SCHEMA_VERSION} in schema_migrations.")
except Exception as _wr_e:
logger.warning(f"Could not record schema version marker (migrations still applied): {_wr_e}")
finally:
if lock_acquired and lock_conn is not None:
try:
await lock_conn.execute("SELECT pg_advisory_unlock($1)", MIGRATION_ADVISORY_LOCK_KEY)
except Exception:
pass
if lock_conn is not None:
await close_database_connection(lock_conn)
async def _run_all_migrations_inner():
"""The full idempotent migration sequence. Runs under the migration advisory
lock and is gated by the schema-version marker in run_all_migrations()."""
# First, ensure basic database schema exists
await run_init_sql()
# Then run additional migrations
await ensure_agents_table()
await ensure_config_versions_metadata_column()
await ensure_user_pool_access_table()
await ensure_backend_servers_last_config_status()
await ensure_backend_servers_haproxy_status()
await ensure_ssl_certificates_new_columns()
await ensure_ssl_cluster_junction_table()
await fix_backend_unique_constraint()
await fix_frontend_unique_constraint()
await fix_waf_unique_constraint()
await fix_backend_servers_unique_constraint()
# Enum migration removed - REJECTED value already in init.sql for fresh databases
await add_frontend_ssl_port_column()
await add_config_versions_updated_at_column()
await ensure_roles_cluster_ids_column()
await update_system_roles_to_enterprise_rbac()
await ensure_user_activity_logs_table()
await ensure_agent_api_key_columns()
await ensure_frontends_ssl_columns()
await ensure_validation_error_columns()
await remove_haproxy_user_group_columns()
await ensure_agent_versions_table()
await ensure_agent_script_templates_source_hash()
await ensure_agent_script_templates_table()
await ensure_agent_activity_logs_table()
await ensure_agent_config_management_tables()
await add_cluster_id_to_agent_config_requests()
await fix_users_unique_constraints_for_soft_delete()
await add_ssl_certificate_id_to_backend_servers()
await add_options_to_backends()
await add_options_to_frontends()
await add_ssl_advanced_options_to_frontends()
await add_ssl_advanced_options_to_servers()
await add_preserved_listen_blocks_to_agents()
await ensure_system_settings_table()
await ensure_acme_tables()
await ensure_acme_columns_on_existing_tables()
# Issue #35 (v1.8.0 — ACME DNS-01): per-account encrypted DNS provider credentials.
# MUST run after ensure_acme_tables() (FK references letsencrypt_accounts).
await ensure_letsencrypt_dns_credentials()
# Issue #11 cleanup: must run AFTER acme_tables/columns to ensure FK refs exist
await cleanup_orphan_acme_challenge_backend()
# v1.5.0 Feature A (ACME diagnostics) + Feature B (site wizard)
# Order matters: letsencrypt_orders column additions BEFORE acme_order_events
# FK setup; both BEFORE wizard_drafts (user FK uses pre-existing users table).
await ensure_letsencrypt_orders_post_completion_actions_column()
await ensure_letsencrypt_orders_wizard_staged_until_column()
await ensure_letsencrypt_orders_pending_apply_version_name_column()
await ensure_letsencrypt_orders_created_by_column()
await ensure_acme_order_events_table()
await ensure_wizard_drafts_table()
await ensure_user_activity_logs_user_action_time_index()
# R18c round 3 #1 (KRITIK concurrency): partial unique on
# (cluster_id, bind_address, bind_port) WHERE is_active.
await ensure_frontends_bind_unique_constraint()
# Issue #18 — TOTP MFA (v1.6.0): additive columns + 3 new tables
await ensure_mfa_columns()
# Issue #27 — HA/VIP Keepalived management (v1.7.0): two brand-new tables.
# MUST run after its FK targets (haproxy_cluster_pools/agents/users), all created above.
await ensure_vip_tables()
# v1.9.0 — CSR creation: brand-new ssl_csrs table. FK-references
# ssl_certificates/users, both created above.
await ensure_ssl_csrs_table()
# v1.11.0 — unified request/response log: brand-new request_logs table
# (no FK targets) plus the seed for its operator-tunable retention
# settings. Both are additive and idempotent.
await ensure_request_logs_table()
await ensure_request_log_settings()
logger.info("Database migrations completed successfully.")
async def ensure_ssl_csrs_table():
"""v1.9.0 — CSR (Certificate Signing Request) creation. Additive only:
one brand-new table (ssl_csrs) + indexes. No ALTER of any existing table,
so the entire current fleet is byte-identical. Fully idempotent
(CREATE TABLE/INDEX IF NOT EXISTS). FK targets (ssl_certificates, users)
are created earlier in the sequence.
A CSR row holds a locally generated private key + CSR PEM until the
operator imports the CA-signed certificate. The import creates a normal
ssl_certificates row (source='csr', last_config_status='PENDING') and
NULLs the private_key_pem copy here — the key then lives only on the
certificate row, like every other key in the system. Agents never read
this table: the agent SSL delivery endpoint selects from
ssl_certificates only, so a pending CSR can never leak to an agent.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
CREATE TABLE IF NOT EXISTS ssl_csrs (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
common_name VARCHAR(253) NOT NULL,
subject JSONB NOT NULL DEFAULT '{}'::jsonb,
sans JSONB NOT NULL DEFAULT '[]'::jsonb,
key_algorithm VARCHAR(20) NOT NULL DEFAULT 'rsa-2048',
csr_pem TEXT NOT NULL,
private_key_pem TEXT,
status VARCHAR(20) NOT NULL DEFAULT 'pending',
ssl_certificate_id INTEGER REFERENCES ssl_certificates(id) ON DELETE SET NULL,
completed_at TIMESTAMP,
created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT ssl_csrs_status_check CHECK (status IN ('pending', 'completed'))
);
""")
# Only PENDING CSRs reserve their name: the name becomes the
# ssl_certificates.name (and thus /etc/ssl/haproxy/{name}.pem on every
# agent) at import time, so two open CSRs must not target the same
# cert name. Completed CSRs are history and may share a name across
# reissues — mirrors the uq_vip_name_active partial-index rationale.
await conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_ssl_csrs_name_pending ON ssl_csrs(name) WHERE status = 'pending';"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_ssl_csrs_status ON ssl_csrs(status);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_ssl_csrs_cert ON ssl_csrs(ssl_certificate_id);"
)
logger.info("ssl_csrs table ensured (v1.9.0 CSR creation)")
except Exception as e:
logger.error(f"Error ensuring ssl_csrs table: {e}")
# Re-raise (ensure_ssl_cluster_junction_table precedent): this step is
# part of the SCHEMA_VERSION=10 bump, and run_all_migrations() records
# the marker only after the inner sequence completes cleanly. Swallowing
# a failure here would stamp version 10 with no ssl_csrs table, and the
# version gate would then skip every future retry — permanently.
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_request_logs_table():
"""v1.11.0 — unified inbound/outbound request/response log.
Additive only: one brand-new table (request_logs) + indexes. No ALTER of
any existing table; agents never read this table.
Deliberately has NO foreign key on user_id. This is the highest-volume
table in the system — one row per API call — and per-insert FK validation
is not worth it here; `username` is a denormalized snapshot so a row stays
readable after the user who made the request is deleted. That is also the
correct audit semantics: the record should outlive the account.
Fully idempotent (CREATE TABLE/INDEX IF NOT EXISTS). Uses only PostgreSQL
9.5+ features (BIGSERIAL, JSONB, partial indexes, varchar_pattern_ops) so
there is no server-version floor beyond what the rest of the schema needs.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
CREATE TABLE IF NOT EXISTS request_logs (
id BIGSERIAL PRIMARY KEY,
request_id VARCHAR(64) NOT NULL,
direction VARCHAR(8) NOT NULL,
target VARCHAR(32),
method VARCHAR(10) NOT NULL,
url TEXT NOT NULL,
path VARCHAR(512),
query_params JSONB,
status_code INTEGER,
status_class SMALLINT NOT NULL DEFAULT 0,
duration_ms INTEGER NOT NULL DEFAULT 0,
user_id INTEGER,
username VARCHAR(50),
client_ip INET,
user_agent TEXT,
request_headers JSONB,
request_body JSONB,
request_body_bytes INTEGER NOT NULL DEFAULT 0,
response_headers JSONB,
response_body JSONB,
response_body_bytes INTEGER NOT NULL DEFAULT 0,
error TEXT,
truncated BOOLEAN NOT NULL DEFAULT FALSE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT request_logs_direction_check
CHECK (direction IN ('inbound', 'outbound'))
);
""")
# Indexes run UNCONDITIONALLY on every startup, not only on first
# creation (the R16-2 rule established for acme_order_events): an older
# deploy that raced ahead of an index would otherwise be stuck doing
# sequential scans forever. All are IF NOT EXISTS, so re-running is free.
# --- read paths: the filters the log viewer actually issues ---
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_created_at "
"ON request_logs(created_at DESC);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_dir_created "
"ON request_logs(direction, created_at DESC);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_status_created "
"ON request_logs(status_class, created_at DESC);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_user_created "
"ON request_logs(user_id, created_at DESC) WHERE user_id IS NOT NULL;"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_target_created "
"ON request_logs(target, created_at DESC) WHERE target IS NOT NULL;"
)
# Correlates one inbound row with the outbound calls it caused — this is
# what makes "which request went where" readable as a single trace.
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_request_id "
"ON request_logs(request_id);"
)
# Prefix search on path (LIKE 'x%') needs pattern_ops to be usable under
# a non-C collation.
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_path_prefix "
"ON request_logs(path varchar_pattern_ops);"
)
# --- prune paths: the TTL delete is split by outcome, so a plain
# (status_class, created_at) index would still range-scan the half it
# is not interested in.
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_prune_ok "
"ON request_logs(created_at) WHERE status_class BETWEEN 1 AND 3;"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_request_logs_prune_err "
"ON request_logs(created_at) WHERE status_class = 0 OR status_class >= 4;"
)
logger.info("request_logs table ensured (v1.11.0 request/response log)")
except Exception as e:
logger.error(f"Error ensuring request_logs table: {e}")
# Re-raise (ssl_csrs precedent): this step is part of the
# SCHEMA_VERSION=11 bump and the version marker is written only after
# the inner sequence completes cleanly. Swallowing here would stamp
# version 11 with no request_logs table, and the version gate would
# then skip every future retry — permanently.
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_request_log_settings():
"""v1.11.0 — seed the request/response-log retention defaults.
Runs UNCONDITIONALLY rather than inside an `if not table_exists:` branch,
so an install that already has `system_settings` picks the rows up too.
ON CONFLICT DO NOTHING means an operator's tuning is never overwritten by a
later upgrade.
Defaults are mirrored in utils/request_log_settings.py; the pair is pinned
by backend/tests/test_request_log_settings.py so they cannot drift apart.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
INSERT INTO system_settings (key, value, category, description) VALUES
('requestlog.enabled', 'true', 'requestlog', 'Master switch for the request/response log'),
('requestlog.capture_inbound', 'true', 'requestlog', 'Log inbound API calls'),
('requestlog.capture_outbound', 'true', 'requestlog', 'Log outbound HTTP calls made by the backend'),
('requestlog.capture_bodies', 'true', 'requestlog', 'Capture redacted, size-capped request/response bodies'),
('requestlog.capture_get', 'true', 'requestlog', 'Log inbound GET requests'),
('requestlog.capture_agent_success', 'false', 'requestlog', 'Log SUCCESSFUL agent polls too (failures are always logged); off by default because the row rate scales with fleet size, not operator activity'),
('requestlog.max_body_bytes', '8192', 'requestlog', 'Per-body capture cap in bytes'),
('requestlog.sample_rate', '1.0', 'requestlog', 'Sampling rate for successful inbound requests (errors always 1.0)'),
('requestlog.exclude_paths', '["/api/request-logs","/api/health","/api/docs","/api/redoc","/api/openapi.json","/.well-known/acme-challenge","/api/agents/heartbeat","/static","/favicon.ico"]', 'requestlog', 'Path prefixes that are never logged'),
('requestlog.success_retention_days', '7', 'requestlog', 'Retention for 1xx/2xx/3xx rows, in days'),
('requestlog.error_retention_days', '30', 'requestlog', 'Retention for 4xx/5xx/transport-error rows, in days'),
('requestlog.max_rows', '500000', 'requestlog', 'Hard row cap; oldest rows are pruned beyond this'),
('requestlog.prune_interval_minutes', '60', 'requestlog', 'Minimum interval between retention prune passes')
ON CONFLICT (key) DO NOTHING
""")
logger.info("request_log retention settings seeded (v1.11.0)")
except Exception as e:
logger.error(f"Error seeding request_log settings: {e}")
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_mfa_columns():
"""Issue #18 — TOTP MFA (v1.6.0): additive columns on users + 3 new tables.
All operations are idempotent (ADD COLUMN IF NOT EXISTS, CREATE TABLE IF NOT EXISTS).
Default behavior preserved: every existing user gets mfa_enabled=FALSE, so login
flow is byte-identical for accounts that don't opt in.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
ALTER TABLE users
ADD COLUMN IF NOT EXISTS mfa_enabled BOOLEAN DEFAULT FALSE NOT NULL,
ADD COLUMN IF NOT EXISTS mfa_method VARCHAR(20),
ADD COLUMN IF NOT EXISTS mfa_secret_encrypted TEXT,
ADD COLUMN IF NOT EXISTS mfa_enrolled_at TIMESTAMP,
ADD COLUMN IF NOT EXISTS mfa_last_used_at TIMESTAMP,
ADD COLUMN IF NOT EXISTS mfa_last_used_totp_step BIGINT;
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS mfa_backup_codes (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
code_hash VARCHAR(255) NOT NULL,
used_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_mfa_backup_codes_user
ON mfa_backup_codes(user_id);
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS mfa_pending_logins (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
challenge_token VARCHAR(64) UNIQUE NOT NULL,
attempts INTEGER DEFAULT 0 NOT NULL,
expires_at TIMESTAMP NOT NULL,
used_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ip_address INET
);
""")
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_mfa_pending_token ON mfa_pending_logins(challenge_token);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_mfa_pending_expires ON mfa_pending_logins(expires_at);"
)
await conn.execute("""
CREATE TABLE IF NOT EXISTS mfa_pending_enrollments (
user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
secret_encrypted TEXT NOT NULL,
attempts INTEGER DEFAULT 0 NOT NULL,
expires_at TIMESTAMP NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_mfa_pending_enroll_expires ON mfa_pending_enrollments(expires_at);"
)
logger.info("✅ MFA migration completed (Issue #18 — Phase 1)")
except Exception as e:
logger.error(f"Failed to ensure MFA columns: {e}")
# Don't raise — follow the same defensive pattern as ensure_user_activity_logs_table
finally:
if conn:
await close_database_connection(conn)
async def ensure_vip_tables():
"""Issue #27 — HA/VIP (Keepalived) management (v1.7.0). Additive only:
two brand-new tables (vip_instances, vip_members) + indexes. No ALTER of any
existing table, so the entire current fleet is byte-identical. Fully idempotent
(CREATE TABLE/INDEX IF NOT EXISTS). FK targets (haproxy_cluster_pools, agents,
users) are created earlier in the sequence — this function is registered LAST.
Backward-compat: a cluster/agent with no VIP row is unaffected; the agent
delivery endpoint returns 'not_configured' for every node without a membership.
"""
conn = None
try:
conn = await get_database_connection()
# Cluster-driven keepalived.conf path (mirrors haproxy_config_path): additive +
# idempotent, with a universal default so operators need set nothing. The agent
# pulls this from its cluster, exactly like the HAProxy paths.
await conn.execute(
"ALTER TABLE haproxy_clusters ADD COLUMN IF NOT EXISTS keepalived_config_path "
"VARCHAR(500) DEFAULT '/etc/keepalived/keepalived.conf';")
# VIP instance: one row per virtual IP (one VRRP group), anchored to a pool.
await conn.execute("""
CREATE TABLE IF NOT EXISTS vip_instances (
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT,
pool_id INTEGER NOT NULL REFERENCES haproxy_cluster_pools(id) ON DELETE CASCADE,
virtual_ip VARCHAR(45) NOT NULL,
prefix_length INTEGER NOT NULL DEFAULT 24,
virtual_router_id INTEGER NOT NULL,
advert_int INTEGER NOT NULL DEFAULT 1,
auth_pass_encrypted TEXT,
use_unicast BOOLEAN NOT NULL DEFAULT TRUE,
track_haproxy BOOLEAN NOT NULL DEFAULT TRUE,
is_active BOOLEAN NOT NULL DEFAULT TRUE,
last_config_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
applied_snapshot JSONB,
created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT vip_vrid_range CHECK (virtual_router_id BETWEEN 1 AND 255)
);
""")
# Additive (idempotent) for DBs that created vip_instances before applied_snapshot
# existed (v1.7.0 self-review): holds the field-level state as of the last Apply so
# a pending edit can be rejected and fully reverted to the previous applied state.
await conn.execute("ALTER TABLE vip_instances ADD COLUMN IF NOT EXISTS applied_snapshot JSONB;")
# Opt-in package removal (v1.7.2): when an operator deletes a VIP and explicitly ticks
# "also uninstall keepalived from the node(s)", we set this flag so the teardown
# delivery tells the agent to purge the OS package. Default FALSE = the safe enterprise
# default (stop+disable+remove our config, but KEEP the package). Additive + idempotent;
# a node we never managed stays untouched regardless.
await conn.execute("ALTER TABLE vip_instances ADD COLUMN IF NOT EXISTS purge_on_teardown BOOLEAN NOT NULL DEFAULT FALSE;")
# Approval-gated deletion (v1.7.2): deleting a RUNNING VIP from the UI does NOT take
# effect immediately — it sets pending_delete=TRUE and stages a vip-*-delete version
# for Apply Management. The VIP stays is_active=TRUE (agents keep serving it, NOTHING
# is torn down) until the operator APPROVES; only then does apply flip is_active=FALSE
# and the agents tear down. Reject clears the flag and the VIP keeps running untouched.
# This guarantees an agent never tears a VIP down without an explicit human approval —
# protecting production. Additive + idempotent.
await conn.execute("ALTER TABLE vip_instances ADD COLUMN IF NOT EXISTS pending_delete BOOLEAN NOT NULL DEFAULT FALSE;")
# Uniqueness as PARTIAL indexes on active rows so a soft-deleted VIP frees its
# name/address/VRID for immediate reuse (a table-level UNIQUE would keep blocking it).
# NAME (v1.7.0 self-review): the original CREATE used a table-level UNIQUE on name,
# which left a soft-deleted VIP's name blocked (you couldn't re-create a VIP with the
# same name) — inconsistent with addr/VRID. Drop that constraint and use a partial
# index instead. Idempotent: no-op on a fresh table (no inline UNIQUE) and on re-run.
await conn.execute("ALTER TABLE vip_instances DROP CONSTRAINT IF EXISTS vip_instances_name_key;")
await conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_vip_name_active ON vip_instances(name) WHERE is_active=TRUE;"
)
await conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_vip_addr_active ON vip_instances(virtual_ip) WHERE is_active=TRUE;"
)
await conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_vip_vrid_active ON vip_instances(pool_id, virtual_router_id) WHERE is_active=TRUE;"
)
# Per-node membership: which agents participate + their VRRP role/priority,
# the applied (delivered) config snapshot, and the agent's deploy ack.
await conn.execute("""
CREATE TABLE IF NOT EXISTS vip_members (
id SERIAL PRIMARY KEY,
vip_id INTEGER NOT NULL REFERENCES vip_instances(id) ON DELETE CASCADE,
agent_id INTEGER NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
network_interface VARCHAR(64) NOT NULL,
role VARCHAR(10) NOT NULL DEFAULT 'BACKUP',
priority INTEGER NOT NULL DEFAULT 100,
applied_config_content TEXT,
applied_config_hash VARCHAR(64),
last_deploy_state VARCHAR(24),
last_deploy_message TEXT,
last_deploy_hash VARCHAR(64),
last_deploy_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT vip_member_role CHECK (role IN ('MASTER','BACKUP')),
CONSTRAINT vip_member_priority_range CHECK (priority BETWEEN 1 AND 254),
CONSTRAINT vip_member_unique UNIQUE (vip_id, agent_id)
);
""")
# Last line of defense against split-brain: at most one MASTER per VIP.
await conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_vip_one_master ON vip_members(vip_id) WHERE role='MASTER';"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_vip_members_vip ON vip_members(vip_id);"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_vip_members_agent ON vip_members(agent_id);"
)
# ── v1.10.4 — VIP adoption: what the agent found already on the node ──────────
# A node with a hand-maintained keepalived.conf reports it here so an existing VIP can
# be adopted instead of retyped. One row per agent (the file is per-node); the agent
# only reports a config it does NOT own, and only when the content changed.
#
# SECRETS: `raw_config` is stored MASKED (auth_pass replaced) because it is served to
# the UI. The real VRRP password is Fernet-encrypted in auth_pass_encrypted, mirroring
# vip_instances, so adoption can carry it into the managed VIP without it ever being
# readable through the API or a DB dump. `analysis` is the parser output with auth_pass
# stripped out.
await conn.execute("""
CREATE TABLE IF NOT EXISTS vip_discoveries (
id SERIAL PRIMARY KEY,
agent_id INTEGER NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
config_path VARCHAR(500) NOT NULL,
config_hash VARCHAR(64) NOT NULL,
is_managed BOOLEAN NOT NULL DEFAULT FALSE,
raw_config_masked TEXT,
auth_pass_encrypted TEXT,
analysis JSONB,
parse_error TEXT,
adopted_vip_id INTEGER REFERENCES vip_instances(id) ON DELETE SET NULL,
reported_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT vip_discovery_agent_unique UNIQUE (agent_id)
);
""")
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_vip_discoveries_agent ON vip_discoveries(agent_id);"
)
# Adoption provenance + the one-shot takeover authorisation. The agent refuses to
# overwrite a keepalived.conf that lacks our ownership marker, which is exactly the
# guard adoption has to pass. Rather than weaken it, an adopted VIP carries the hash of
# the file we analysed: the agent takes over ONLY if the file on disk still hashes to
# that value, so a config that changed after adoption is never clobbered.
await conn.execute(
"ALTER TABLE vip_instances ADD COLUMN IF NOT EXISTS adopted_at TIMESTAMP;")
await conn.execute(
"ALTER TABLE vip_members ADD COLUMN IF NOT EXISTS takeover_expected_hash VARCHAR(64);")
logger.info("✅ VIP tables ensured (Issue #27 — HA/VIP Keepalived management)")
except Exception as e:
logger.error(f"Failed to ensure VIP tables: {e}")
# Don't raise — follow the same defensive pattern as ensure_mfa_columns
finally:
if conn:
await close_database_connection(conn)
async def add_ssl_certificate_id_to_backend_servers():
"""Add ssl_certificate_id column to backend_servers table for SSL certificate management"""
conn = None
try:
conn = await get_database_connection()
# Check if ssl_certificate_id column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers'
AND column_name = 'ssl_certificate_id'
)
""")
if not column_exists:
# Add ssl_certificate_id column
await conn.execute("""
ALTER TABLE backend_servers
ADD COLUMN ssl_certificate_id INTEGER,
ADD CONSTRAINT fk_backend_server_ssl_certificate
FOREIGN KEY (ssl_certificate_id)
REFERENCES ssl_certificates(id)
ON DELETE SET NULL
""")
logger.info("Added ssl_certificate_id column to backend_servers table with FK constraint")
else:
logger.info("ssl_certificate_id column already exists in backend_servers table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error adding ssl_certificate_id to backend_servers: {e}")
# Don't raise - we'll try to proceed
async def remove_haproxy_user_group_columns():
"""Remove haproxy_user and haproxy_group columns from haproxy_clusters table"""
conn = None
try:
conn = await get_database_connection()
# Check if columns exist before trying to drop them
user_column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'haproxy_clusters'
AND column_name = 'haproxy_user'
)
""")
group_column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'haproxy_clusters'
AND column_name = 'haproxy_group'
)
""")
if user_column_exists:
await conn.execute("ALTER TABLE haproxy_clusters DROP COLUMN haproxy_user")
logger.info("✅ Dropped haproxy_user column from haproxy_clusters")
else:
logger.info("ℹ️ haproxy_user column already removed")
if group_column_exists:
await conn.execute("ALTER TABLE haproxy_clusters DROP COLUMN haproxy_group")
logger.info("✅ Dropped haproxy_group column from haproxy_clusters")
else:
logger.info("ℹ️ haproxy_group column already removed")
logger.info("✅ HAProxy user/group columns removal completed")
except Exception as e:
logger.error(f"❌ Failed to remove HAProxy user/group columns: {e}")
raise
finally:
if conn:
await close_database_connection(conn)
async def ensure_agent_api_key_columns():
"""Add API key columns to agents table for secure agent authentication"""
conn = None
try:
conn = await get_database_connection()
# Check if api_key column exists
api_key_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'agents' AND column_name = 'api_key'
)
""")
if not api_key_exists:
logger.info("Adding api_key column to agents table...")
await conn.execute("""
ALTER TABLE agents
ADD COLUMN api_key VARCHAR(128),
ADD COLUMN api_key_name VARCHAR(100),
ADD COLUMN api_key_created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
ADD COLUMN api_key_expires_at TIMESTAMP,
ADD COLUMN api_key_last_used TIMESTAMP,
ADD COLUMN api_key_created_by INTEGER REFERENCES users(id)
""")
# Create index for fast API key lookups
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agents_api_key ON agents(api_key) WHERE api_key IS NOT NULL;
""")
logger.info("Successfully added API key columns to agents table")
else:
logger.info("API key columns already exist in agents table")
await close_database_connection(conn)
logger.info("✅ Agent API key columns migration completed")
except Exception as e:
logger.error(f"Failed to add agent API key columns: {e}")
if conn:
await close_database_connection(conn)
async def create_initial_system_data(conn):
"""Create initial system roles and users"""
try:
# Create system roles with basic permissions (cluster_ids will be added later)
system_roles = [
{
'name': 'super_admin',
'display_name': 'Super Administrator',
'description': 'Full system access with all permissions',
'permissions': ["dashboard.read","dashboard.statistics","frontends.read","frontends.create","frontends.update","frontends.delete","backends.read","backends.create","backends.update","backends.delete","waf.read","waf.create","waf.update","waf.delete","ssl.read","ssl.create","ssl.update","ssl.delete","apply.read","apply.execute","agents.read","agents.create","agents.update","agents.delete","clusters.read","clusters.create","clusters.update","clusters.delete","config.read","config.update","config.bulk_import","config.view_request","config.download_request","users.read","users.create","users.update","users.delete","roles.read","roles.create","roles.update","roles.delete","requestlog.read","requestlog.manage"]
},
{
'name': 'operator',
'display_name': 'Operator',
'description': 'Daily operational access for managing HAProxy configurations',
'permissions': ["dashboard.read","dashboard.statistics","frontends.read","frontends.create","frontends.update","backends.read","backends.create","backends.update","waf.read","waf.create","waf.update","ssl.read","ssl.create","ssl.update","apply.read","apply.execute","agents.read","clusters.read","config.read","config.update","config.bulk_import","config.view_request","config.download_request","requestlog.read"]
},
{
'name': 'security_admin',
'display_name': 'Security Administrator',
'description': 'Security-focused access for WAF rules and SSL certificates',
'permissions': ["dashboard.read","frontends.read","backends.read","waf.read","waf.create","waf.update","waf.delete","ssl.read","ssl.create","ssl.update","ssl.delete","apply.read","apply.execute","agents.read","clusters.read","config.read","config.view_request","config.download_request","requestlog.read","requestlog.manage"]
},
{
'name': 'viewer',
'display_name': 'Viewer',
'description': 'Read-only access to view configurations and monitor system status',
'permissions': ["dashboard.read","frontends.read","backends.read","waf.read","ssl.read","apply.read","agents.read","clusters.read","config.read","config.view_request"]
}
]
for role_data in system_roles:
# Check if role exists first (safer than ON CONFLICT)
role_exists = await conn.fetchval("SELECT id FROM roles WHERE name = $1", role_data['name'])
if not role_exists:
await conn.execute("""
INSERT INTO roles (name, display_name, description, permissions, is_active, is_system)
VALUES ($1, $2, $3, $4, $5, $6)
""",
role_data['name'],
role_data['display_name'],
role_data['description'],
json.dumps(role_data['permissions']), # FIXED: Convert list to JSON string for JSONB column
True,
True
)
logger.info("✅ System roles created")
# Create default admin user
import bcrypt
# Check if admin user already exists first (safer than ON CONFLICT)
admin_id = await conn.fetchval("SELECT id FROM users WHERE username = 'admin'")
if not admin_id:
password_hash = bcrypt.hashpw('admin123'.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
admin_id = await conn.fetchval("""
INSERT INTO users (username, email, password_hash, full_name, is_active, is_admin, is_verified)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id
""", 'admin', 'admin@haproxy-openmanager.local', password_hash, 'System Administrator', True, True, True)
if admin_id:
# Assign super_admin role to admin user
super_admin_role = await conn.fetchval("SELECT id FROM roles WHERE name = 'super_admin'")
if super_admin_role:
# Check if role assignment exists
role_assigned = await conn.fetchval("""
SELECT id FROM user_roles
WHERE user_id = $1 AND role_id = $2
""", admin_id, super_admin_role)
if not role_assigned:
await conn.execute("""
INSERT INTO user_roles (user_id, role_id, assigned_by)
VALUES ($1, $2, $3)
""", admin_id, super_admin_role, admin_id)
logger.info("✅ Default admin user created")
else:
logger.info("✅ Default admin user already exists")
except Exception as e:
logger.error(f"Failed to create initial system data: {e}")
raise
async def create_essential_tables(conn):
"""Create all essential tables if they don't exist"""
try:
# Create enum types first
await conn.execute("""
DO $$ BEGIN
CREATE TYPE config_status AS ENUM ('PENDING', 'APPLIED', 'REJECTED');
EXCEPTION
WHEN duplicate_object THEN null;
END $$;
""")
# Create users table
await conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE,
password_hash VARCHAR(255) NOT NULL,
full_name VARCHAR(100),
phone VARCHAR(20),
role VARCHAR(50) DEFAULT 'admin',
is_active BOOLEAN DEFAULT TRUE,
is_admin BOOLEAN DEFAULT FALSE,
is_verified BOOLEAN DEFAULT FALSE,
last_login TIMESTAMP,
last_login_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create roles table
await conn.execute("""
CREATE TABLE IF NOT EXISTS roles (
id SERIAL PRIMARY KEY,
name VARCHAR(50) UNIQUE NOT NULL,
display_name VARCHAR(100) NOT NULL,
description TEXT,
permissions JSONB DEFAULT '[]'::jsonb,
cluster_ids JSONB DEFAULT NULL,
is_active BOOLEAN DEFAULT TRUE,
is_system BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create user_roles junction table
await conn.execute("""
CREATE TABLE IF NOT EXISTS user_roles (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
role_id INTEGER REFERENCES roles(id) ON DELETE CASCADE,
assigned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
assigned_by INTEGER REFERENCES users(id),
is_active BOOLEAN DEFAULT TRUE,
UNIQUE(user_id, role_id)
);
""")
# Create haproxy_cluster_pools table
await conn.execute("""
CREATE TABLE IF NOT EXISTS haproxy_cluster_pools (
id SERIAL PRIMARY KEY,
name VARCHAR(100) UNIQUE NOT NULL,
description TEXT,
environment VARCHAR(50) DEFAULT 'development',
location VARCHAR(255),
default_config JSONB,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create haproxy_clusters table
await conn.execute("""
CREATE TABLE IF NOT EXISTS haproxy_clusters (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
description TEXT,
environment VARCHAR(50),
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE SET NULL,
haproxy_config_path VARCHAR(255) DEFAULT '/etc/haproxy/haproxy.cfg',
haproxy_bin_path VARCHAR(255) DEFAULT '/usr/sbin/haproxy',
stats_socket_path VARCHAR(255) DEFAULT '/var/run/haproxy/admin.sock',
haproxy_user VARCHAR(255) DEFAULT 'haproxy',
haproxy_group VARCHAR(255) DEFAULT 'haproxy',
installation_type VARCHAR(50) DEFAULT 'agent',
deployment_type VARCHAR(50) DEFAULT 'standalone',
host VARCHAR(255),
port INTEGER,
connection_type VARCHAR(50) DEFAULT 'agent',
is_active BOOLEAN DEFAULT TRUE,
is_default BOOLEAN DEFAULT FALSE,
last_connected_at TIMESTAMP,
connection_status VARCHAR(50) DEFAULT 'unknown',
connection_error TEXT,
haproxy_version VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(name, pool_id)
);
""")
# Create config_versions table
await conn.execute("""
CREATE TABLE IF NOT EXISTS config_versions (
id SERIAL PRIMARY KEY,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
version_name VARCHAR(100) NOT NULL,
description TEXT,
config_content TEXT NOT NULL,
checksum VARCHAR(64),
file_size INTEGER,
status config_status DEFAULT 'APPLIED',
metadata JSONB DEFAULT '{}'::jsonb,
is_active BOOLEAN DEFAULT TRUE,
created_by INTEGER REFERENCES users(id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(cluster_id, version_name)
);
""")
# Create agents table
await conn.execute("""
CREATE TABLE IF NOT EXISTS agents (
id SERIAL PRIMARY KEY,
name VARCHAR(100) UNIQUE NOT NULL,
hostname VARCHAR(255),
ip_address INET,
platform VARCHAR(50),
architecture VARCHAR(50),
version VARCHAR(50),
operating_system VARCHAR(100),
kernel_version VARCHAR(100),
uptime BIGINT,
cpu_count INTEGER,
memory_total BIGINT,
disk_space BIGINT,
network_interfaces JSONB DEFAULT '[]'::jsonb,
capabilities JSONB DEFAULT '[]'::jsonb,
status VARCHAR(20) DEFAULT 'offline',
haproxy_status VARCHAR(20) DEFAULT 'unknown',
haproxy_version VARCHAR(50),
keepalive_state VARCHAR(20) DEFAULT NULL,
keepalive_ip VARCHAR(45) DEFAULT NULL,
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE SET NULL,
enabled BOOLEAN DEFAULT TRUE,
is_active BOOLEAN DEFAULT TRUE,
last_seen TIMESTAMP,
applied_config_version VARCHAR(100),
config_version VARCHAR(100),
api_key VARCHAR(128),
api_key_name VARCHAR(100),
api_key_created_at TIMESTAMP,
api_key_expires_at TIMESTAMP,
api_key_last_used TIMESTAMP,
api_key_created_by INTEGER REFERENCES users(id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create backends table
await conn.execute("""
CREATE TABLE IF NOT EXISTS backends (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
balance_method VARCHAR(50) DEFAULT 'roundrobin',
mode VARCHAR(10) DEFAULT 'http',
health_check_uri VARCHAR(255),
health_check_interval INTEGER DEFAULT 2000,
health_check_expected_status INTEGER DEFAULT 200,
fullconn INTEGER,
timeout_connect INTEGER DEFAULT 10000,
timeout_server INTEGER DEFAULT 60000,
timeout_queue INTEGER DEFAULT 60000,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
maxconn INTEGER,
last_config_status config_status DEFAULT 'APPLIED',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(name, cluster_id)
);
""")
# Add new columns to existing backends table if they don't exist
await conn.execute("""
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='health_check_expected_status') THEN
ALTER TABLE backends ADD COLUMN health_check_expected_status INTEGER DEFAULT 200;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='fullconn') THEN
ALTER TABLE backends ADD COLUMN fullconn INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='cookie_name') THEN
ALTER TABLE backends ADD COLUMN cookie_name VARCHAR(100);
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='cookie_options') THEN
ALTER TABLE backends ADD COLUMN cookie_options TEXT;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='default_server_inter') THEN
ALTER TABLE backends ADD COLUMN default_server_inter INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='default_server_fall') THEN
ALTER TABLE backends ADD COLUMN default_server_fall INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='default_server_rise') THEN
ALTER TABLE backends ADD COLUMN default_server_rise INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='request_headers') THEN
ALTER TABLE backends ADD COLUMN request_headers TEXT;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backends' AND column_name='response_headers') THEN
ALTER TABLE backends ADD COLUMN response_headers TEXT;
END IF;
END $$;
""")
# Create backend_servers table
await conn.execute("""
CREATE TABLE IF NOT EXISTS backend_servers (
id SERIAL PRIMARY KEY,
backend_id INTEGER REFERENCES backends(id) ON DELETE CASCADE,
backend_name VARCHAR(100) NOT NULL,
server_name VARCHAR(100) NOT NULL,
server_address VARCHAR(255) NOT NULL,
server_port INTEGER NOT NULL,
weight INTEGER DEFAULT 100,
maxconn INTEGER,
check_enabled BOOLEAN DEFAULT TRUE,
check_port INTEGER,
backup_server BOOLEAN DEFAULT FALSE,
ssl_enabled BOOLEAN DEFAULT FALSE,
ssl_verify VARCHAR(20),
cookie_value VARCHAR(100),
inter INTEGER,
fall INTEGER,
rise INTEGER,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
is_active BOOLEAN DEFAULT TRUE,
last_config_status config_status DEFAULT 'APPLIED',
haproxy_status VARCHAR(20) DEFAULT 'unknown',
haproxy_status_updated_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(backend_id, server_name)
);
""")
# Add new columns to existing backend_servers table if they don't exist
await conn.execute("""
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='check_port') THEN
ALTER TABLE backend_servers ADD COLUMN check_port INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='ssl_verify') THEN
ALTER TABLE backend_servers ADD COLUMN ssl_verify VARCHAR(20);
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='cookie_value') THEN
ALTER TABLE backend_servers ADD COLUMN cookie_value VARCHAR(100);
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='inter') THEN
ALTER TABLE backend_servers ADD COLUMN inter INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='fall') THEN
ALTER TABLE backend_servers ADD COLUMN fall INTEGER;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='backend_servers' AND column_name='rise') THEN
ALTER TABLE backend_servers ADD COLUMN rise INTEGER;
END IF;
END $$;
""")
# Create frontends table
await conn.execute("""
CREATE TABLE IF NOT EXISTS frontends (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
bind_address VARCHAR(255) DEFAULT '*',
bind_port INTEGER NOT NULL,
default_backend VARCHAR(100),
mode VARCHAR(10) DEFAULT 'http',
ssl_enabled BOOLEAN DEFAULT FALSE,
ssl_certificate_id INTEGER,
ssl_port INTEGER,
ssl_cert_path VARCHAR(255),
ssl_cert TEXT,
ssl_verify VARCHAR(20), -- PR-2 (R11.B): no DEFAULT; NULL means "omit verify directive"
acl_rules JSONB DEFAULT '[]'::jsonb,
redirect_rules JSONB DEFAULT '[]'::jsonb,
use_backend_rules JSONB DEFAULT '[]'::jsonb,
request_headers TEXT,
response_headers TEXT,
tcp_request_rules TEXT,
timeout_client INTEGER,
timeout_http_request INTEGER,
rate_limit INTEGER,
compression BOOLEAN DEFAULT FALSE,
log_separate BOOLEAN DEFAULT FALSE,
monitor_uri VARCHAR(255),
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
maxconn INTEGER,
last_config_status config_status DEFAULT 'APPLIED',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(name, cluster_id)
);
""")
# Add new column to existing frontends table if it doesn't exist
await conn.execute("""
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name='frontends' AND column_name='tcp_request_rules') THEN
ALTER TABLE frontends ADD COLUMN tcp_request_rules TEXT;
END IF;
END $$;
""")
# Create ssl_certificates table
await conn.execute("""
CREATE TABLE IF NOT EXISTS ssl_certificates (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
primary_domain VARCHAR(255),
certificate_content TEXT NOT NULL,
private_key_content TEXT,
chain_content TEXT,
expiry_date TIMESTAMP,
issuer TEXT,
status VARCHAR(20) DEFAULT 'valid',
fingerprint VARCHAR(128),
days_until_expiry INTEGER DEFAULT 0,
all_domains JSONB DEFAULT '[]'::jsonb,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE SET NULL,
last_config_status config_status DEFAULT 'APPLIED',
usage_type VARCHAR(50) DEFAULT 'frontend',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create waf_rules table
await conn.execute("""
CREATE TABLE IF NOT EXISTS waf_rules (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
rule_type VARCHAR(50) NOT NULL,
config JSONB NOT NULL DEFAULT '{}'::jsonb,
action VARCHAR(20) DEFAULT 'deny',
priority INTEGER DEFAULT 100,
description TEXT,
enabled BOOLEAN DEFAULT TRUE,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
last_config_status config_status DEFAULT 'APPLIED',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(name, cluster_id)
);
""")
# Create user_activity_logs table
await conn.execute("""
CREATE TABLE IF NOT EXISTS user_activity_logs (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
action VARCHAR(100) NOT NULL,
resource_type VARCHAR(50),
resource_id VARCHAR(100),
details JSONB,
ip_address INET,
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Create user_pool_access table
await conn.execute("""
CREATE TABLE IF NOT EXISTS user_pool_access (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
pool_id INTEGER REFERENCES haproxy_cluster_pools(id) ON DELETE CASCADE,
access_level VARCHAR(20) DEFAULT 'read_write',
granted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
granted_by INTEGER REFERENCES users(id),
is_active BOOLEAN DEFAULT TRUE,
UNIQUE(user_id, pool_id)
);
""")
# Create frontend_waf_rules junction table
await conn.execute("""
CREATE TABLE IF NOT EXISTS frontend_waf_rules (
id SERIAL PRIMARY KEY,
frontend_id INTEGER REFERENCES frontends(id) ON DELETE CASCADE,
waf_rule_id INTEGER REFERENCES waf_rules(id) ON DELETE CASCADE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(frontend_id, waf_rule_id)
);
""")
# Create ssl_certificate_clusters junction table
await conn.execute("""
CREATE TABLE IF NOT EXISTS ssl_certificate_clusters (
id SERIAL PRIMARY KEY,
ssl_certificate_id INTEGER REFERENCES ssl_certificates(id) ON DELETE CASCADE,
cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(ssl_certificate_id, cluster_id)
);
""")
# Create indexes for performance
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agents_api_key ON agents(api_key) WHERE api_key IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_user_id ON user_activity_logs(user_id);
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_created_at ON user_activity_logs(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_resource ON user_activity_logs(resource_type, resource_id);
CREATE INDEX IF NOT EXISTS idx_config_versions_cluster_status ON config_versions(cluster_id, status);
CREATE INDEX IF NOT EXISTS idx_config_versions_created_at ON config_versions(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_agents_pool_status ON agents(pool_id, status);
CREATE INDEX IF NOT EXISTS idx_agents_last_seen ON agents(last_seen DESC);
CREATE INDEX IF NOT EXISTS idx_backend_servers_backend_cluster ON backend_servers(backend_id, cluster_id);
CREATE INDEX IF NOT EXISTS idx_frontend_waf_rules_frontend ON frontend_waf_rules(frontend_id);
CREATE INDEX IF NOT EXISTS idx_frontend_waf_rules_waf ON frontend_waf_rules(waf_rule_id);
CREATE INDEX IF NOT EXISTS idx_ssl_certificate_clusters_ssl ON ssl_certificate_clusters(ssl_certificate_id);
CREATE INDEX IF NOT EXISTS idx_ssl_certificate_clusters_cluster ON ssl_certificate_clusters(cluster_id);
""")
logger.info("✅ Essential tables and indexes created/verified")
except Exception as e:
logger.error(f"Failed to create essential tables: {e}")
raise
# REMOVED: Duplicate function that was overriding the correct ensure_agents_table() at line 33
# The detailed version (line 33) includes missing_columns loop with usage_type migration
# This basic version only called create_essential_tables() without column additions
# Keeping it removed to ensure proper migration execution
async def ensure_frontends_ssl_columns():
"""Add missing SSL columns to frontends table for proper SSL certificate support"""
conn = None
try:
conn = await get_database_connection()
# Check if ssl_certificate_id column exists
ssl_cert_id_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'ssl_certificate_id'
)
""")
# Check if ssl_port column exists
ssl_port_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'ssl_port'
)
""")
# Add missing columns
if not ssl_cert_id_exists:
logger.info("Adding ssl_certificate_id column to frontends table...")
await conn.execute("""
ALTER TABLE frontends
ADD COLUMN ssl_certificate_id INTEGER
""")
if not ssl_port_exists:
logger.info("Adding ssl_port column to frontends table...")
await conn.execute("""
ALTER TABLE frontends
ADD COLUMN ssl_port INTEGER DEFAULT 443
""")
# CRITICAL NEW FEATURE: Check and add ssl_certificate_ids column for multiple SSL certificates
# HAProxy supports multiple certificates on single bind: bind :443 ssl crt file1.pem crt file2.pem
ssl_cert_ids_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'ssl_certificate_ids'
)
""")
if not ssl_cert_ids_exists:
logger.info("Adding ssl_certificate_ids JSONB column for multiple SSL certificates...")
await conn.execute("""
ALTER TABLE frontends
ADD COLUMN ssl_certificate_ids JSONB DEFAULT '[]'::jsonb
""")
logger.info("✅ ssl_certificate_ids column added successfully")
# Migrate existing ssl_certificate_id to ssl_certificate_ids array
logger.info("Migrating existing ssl_certificate_id values to ssl_certificate_ids array...")
await conn.execute("""
UPDATE frontends
SET ssl_certificate_ids = jsonb_build_array(ssl_certificate_id)
WHERE ssl_certificate_id IS NOT NULL AND ssl_certificate_id > 0
""")
logger.info("✅ SSL certificate migration completed - single IDs converted to arrays")
# AGENT UPGRADE TRACKING: Check and add upgrade status columns
# These columns track agent self-upgrade process initiated from backend
upgrade_status_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'agents' AND column_name = 'upgrade_status'
)
""")
if not upgrade_status_exists:
logger.info("Adding agent upgrade tracking columns...")
await conn.execute("""
ALTER TABLE agents
ADD COLUMN upgrade_status VARCHAR(50),
ADD COLUMN upgrade_target_version VARCHAR(50),
ADD COLUMN upgraded_at TIMESTAMP
""")
logger.info("✅ Agent upgrade tracking columns added successfully")
logger.info(" - upgrade_status: Tracks upgrade process state (upgrading, completed, failed)")
logger.info(" - upgrade_target_version: Version being upgraded to")
logger.info(" - upgraded_at: Timestamp of last upgrade status change")
# Update acl_rules, redirect_rules, and use_backend_rules to JSONB if they're still TEXT[] or TEXT
acl_rules_type = await conn.fetchval("""
SELECT data_type FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'acl_rules'
""")
if acl_rules_type == 'ARRAY':
logger.info("Converting acl_rules from TEXT[] to JSONB...")
await conn.execute("""
ALTER TABLE frontends
ALTER COLUMN acl_rules TYPE JSONB USING array_to_json(acl_rules)::JSONB
""")
redirect_rules_type = await conn.fetchval("""
SELECT data_type FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'redirect_rules'
""")
if redirect_rules_type == 'ARRAY':
logger.info("Converting redirect_rules from TEXT[] to JSONB...")
await conn.execute("""
ALTER TABLE frontends
ALTER COLUMN redirect_rules TYPE JSONB USING array_to_json(redirect_rules)::JSONB
""")
# CRITICAL FIX: Convert use_backend_rules to JSONB for consistency with acl_rules and redirect_rules
use_backend_rules_type = await conn.fetchval("""
SELECT data_type FROM information_schema.columns
WHERE table_name = 'frontends' AND column_name = 'use_backend_rules'
""")
if use_backend_rules_type in ('text', 'character varying'):
logger.info("Converting use_backend_rules from TEXT to JSONB...")
await conn.execute("""
ALTER TABLE frontends
ALTER COLUMN use_backend_rules TYPE JSONB USING
CASE
WHEN use_backend_rules IS NULL THEN '[]'::jsonb
WHEN use_backend_rules = '' THEN '[]'::jsonb
ELSE use_backend_rules::jsonb
END
""")
logger.info("use_backend_rules converted to JSONB successfully")
await close_database_connection(conn)
logger.info("✅ Frontends SSL columns migration completed")
except Exception as e:
logger.error(f"Failed to ensure frontends SSL columns: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
async def ensure_validation_error_columns():
"""
Add validation_error columns to config_versions and agents tables
These columns store HAProxy validation errors reported by agents
"""
conn = None
try:
conn = await get_database_connection()
# VALIDATION ERROR TRACKING: Add columns to config_versions table
validation_error_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'config_versions' AND column_name = 'validation_error'
)
""")
if not validation_error_exists:
logger.info("Adding validation error columns to config_versions table...")
await conn.execute("""
ALTER TABLE config_versions
ADD COLUMN validation_error TEXT,
ADD COLUMN validation_error_reported_at TIMESTAMP
""")
logger.info("✅ Validation error columns added to config_versions")
logger.info(" - validation_error: Stores HAProxy validation error message from agent")
logger.info(" - validation_error_reported_at: Timestamp when error was reported")
else:
logger.info("ℹ️ Validation error columns already exist in config_versions table")
# VALIDATION ERROR TRACKING: Add columns to agents table
agent_validation_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'agents' AND column_name = 'last_validation_error'
)
""")
if not agent_validation_exists:
logger.info("Adding validation error columns to agents table...")
await conn.execute("""
ALTER TABLE agents
ADD COLUMN last_validation_error TEXT,
ADD COLUMN last_validation_error_at TIMESTAMP
""")
logger.info("✅ Validation error columns added to agents")
logger.info(" - last_validation_error: Last HAProxy validation error encountered by this agent")
logger.info(" - last_validation_error_at: Timestamp of last validation error")
else:
logger.info("ℹ️ Validation error columns already exist in agents table")
logger.info("✅ Validation error columns migration completed")
except Exception as e:
logger.error(f"Failed to ensure validation error columns: {e}")
if conn:
await close_database_connection(conn)
# Don't raise - this is not critical for system operation
async def ensure_agent_versions_table():
"""Create agent_versions table for managing agent script versions"""
conn = None
try:
conn = await get_database_connection()
# Create agent_versions table
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_versions (
id SERIAL PRIMARY KEY,
platform VARCHAR(50) NOT NULL,
version VARCHAR(20) NOT NULL,
release_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
changelog TEXT[],
is_active BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(platform, version)
)
""")
# Insert default versions for each platform
await conn.execute("""
INSERT INTO agent_versions (platform, version, changelog)
VALUES
('macos', '1.0.0', ARRAY[
'Initial agent version with heartbeat sync fix',
'Complete HAProxy configuration management',
'SSL certificate deployment support',
'Zero-downtime configuration reload',
'Production-ready agent foundation'
]),
('linux', '1.0.0', ARRAY[
'Initial agent version with heartbeat sync fix',
'Complete HAProxy configuration management',
'SSL certificate deployment support',
'Zero-downtime configuration reload',
'Production-ready agent foundation'
])
ON CONFLICT (platform, version) DO NOTHING
""")
await close_database_connection(conn)
logger.info("✅ Agent versions table created successfully")
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error creating agent_versions table: {e}")
# Don't raise - this is not critical for system operation
async def ensure_agent_script_templates_table():
"""Create agent_script_templates table for storing editable script templates"""
conn = None
try:
conn = await get_database_connection()
# Create agent_script_templates table
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_script_templates (
id SERIAL PRIMARY KEY,
platform VARCHAR(50) NOT NULL,
version VARCHAR(20) NOT NULL,
script_content TEXT NOT NULL,
source_file_hash VARCHAR(64),
is_active BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(platform, version)
)
""")
# Load initial script templates from files and sync with existing versions
import os
script_dir = os.path.join(os.path.dirname(__file__), '..', 'utils', 'agent_scripts')
# Get current agent versions from database
macos_version_row = await conn.fetchrow("SELECT version FROM agent_versions WHERE platform = 'macos' AND is_active = true ORDER BY created_at DESC LIMIT 1")
linux_version_row = await conn.fetchrow("SELECT version FROM agent_versions WHERE platform = 'linux' AND is_active = true ORDER BY created_at DESC LIMIT 1")
macos_current_version = macos_version_row['version'] if macos_version_row else '1.0.0'
linux_current_version = linux_version_row['version'] if linux_version_row else '1.0.0'
# Load macOS script with current version
macos_script_path = os.path.join(script_dir, 'macos_install.sh')
if os.path.exists(macos_script_path):
with open(macos_script_path, 'r') as f:
macos_script = f.read()
macos_hash = hashlib.sha256(macos_script.encode()).hexdigest()
await conn.execute("""
INSERT INTO agent_script_templates (platform, version, script_content, source_file_hash)
VALUES ('macos', $1, $2, $3)
ON CONFLICT (platform, version) DO NOTHING
""", macos_current_version, macos_script, macos_hash)
logger.info(f"✅ Loaded macOS script template version {macos_current_version}")
# Load Linux script with current version
linux_script_path = os.path.join(script_dir, 'linux_install.sh')
if os.path.exists(linux_script_path):
with open(linux_script_path, 'r') as f:
linux_script = f.read()
linux_hash = hashlib.sha256(linux_script.encode()).hexdigest()
await conn.execute("""
INSERT INTO agent_script_templates (platform, version, script_content, source_file_hash)
VALUES ('linux', $1, $2, $3)
ON CONFLICT (platform, version) DO NOTHING
""", linux_current_version, linux_script, linux_hash)
logger.info(f"✅ Loaded Linux script template version {linux_current_version}")
await close_database_connection(conn)
logger.info("✅ Agent script templates table created successfully")
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error creating agent_script_templates table: {e}")
# Don't raise - this is not critical for system operation
async def ensure_agent_script_templates_source_hash():
"""Add source_file_hash column to agent_script_templates for update detection"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
ALTER TABLE agent_script_templates
ADD COLUMN IF NOT EXISTS source_file_hash VARCHAR(64)
""")
await close_database_connection(conn)
logger.info("Agent script templates source_file_hash column ensured")
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error adding source_file_hash column: {e}")
async def ensure_agent_config_management_tables():
"""Create tables for Configuration Management feature"""
conn = None
try:
conn = await get_database_connection()
# Create agent_config_requests table
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_config_requests (
id SERIAL PRIMARY KEY,
agent_id INTEGER REFERENCES agents(id) ON DELETE CASCADE,
agent_name VARCHAR(255) NOT NULL,
request_type VARCHAR(50) NOT NULL,
status VARCHAR(50) DEFAULT 'pending',
requested_by INTEGER REFERENCES users(id),
requested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP DEFAULT (CURRENT_TIMESTAMP + INTERVAL '5 minutes'),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create agent_config_responses table
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_config_responses (
id SERIAL PRIMARY KEY,
request_id INTEGER REFERENCES agent_config_requests(id) ON DELETE CASCADE,
agent_name VARCHAR(255) NOT NULL,
config_content TEXT NOT NULL,
config_path VARCHAR(500),
file_size BIGINT,
response_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP DEFAULT (CURRENT_TIMESTAMP + INTERVAL '10 minutes'),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create indexes for performance
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_requests_agent_name
ON agent_config_requests(agent_name);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_requests_status
ON agent_config_requests(status);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_requests_expires_at
ON agent_config_requests(expires_at);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_responses_request_id
ON agent_config_responses(request_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_responses_expires_at
ON agent_config_responses(expires_at);
""")
await close_database_connection(conn)
logger.info("✅ Agent configuration management tables created successfully")
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error creating agent configuration management tables: {e}")
# Don't raise - this is not critical for system operation
async def add_cluster_id_to_agent_config_requests():
"""Add cluster_id column to agent_config_requests table
This fixes the issue where config requests don't know which cluster they belong to,
causing wrong cluster to be selected when multiple clusters share the same pool.
"""
conn = None
try:
conn = await get_database_connection()
# Check if cluster_id column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'agent_config_requests'
AND column_name = 'cluster_id'
)
""")
if not column_exists:
# Add cluster_id column
await conn.execute("""
ALTER TABLE agent_config_requests
ADD COLUMN cluster_id INTEGER REFERENCES haproxy_clusters(id) ON DELETE CASCADE
""")
# Create index for performance
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_config_requests_cluster_id
ON agent_config_requests(cluster_id)
""")
logger.info("✅ Added cluster_id column to agent_config_requests table")
else:
logger.info("ℹ️ cluster_id column already exists in agent_config_requests table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error adding cluster_id to agent_config_requests: {e}")
# Don't raise - not critical for system operation
async def fix_users_unique_constraints_for_soft_delete():
"""Fix users table unique constraints to support soft delete
Replace table-level UNIQUE constraints with partial unique indexes
that only apply to active users (is_active = TRUE).
This allows soft-deleted users to be recreated with the same username/email.
"""
conn = None
try:
conn = await get_database_connection()
# Check if the old constraints exist
username_constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'users_username_key'
)
""")
email_constraint_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'users_email_key'
)
""")
# Check if partial indexes already exist
username_index_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM pg_indexes
WHERE indexname = 'users_username_active_key'
)
""")
email_index_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM pg_indexes
WHERE indexname = 'users_email_active_key'
)
""")
# Drop old username constraint if exists
if username_constraint_exists:
logger.info("Dropping users_username_key constraint...")
await conn.execute("ALTER TABLE users DROP CONSTRAINT users_username_key")
logger.info("✅ Dropped users_username_key constraint")
# Drop old email constraint if exists
if email_constraint_exists:
logger.info("Dropping users_email_key constraint...")
await conn.execute("ALTER TABLE users DROP CONSTRAINT users_email_key")
logger.info("✅ Dropped users_email_key constraint")
# Create partial unique index for username (only active users)
if not username_index_exists:
logger.info("Creating partial unique index for username...")
await conn.execute("""
CREATE UNIQUE INDEX users_username_active_key
ON users(username)
WHERE is_active = TRUE
""")
logger.info("✅ Created users_username_active_key partial index")
else:
logger.info("ℹ️ users_username_active_key index already exists")
# Create partial unique index for email (only active users)
if not email_index_exists:
logger.info("Creating partial unique index for email...")
await conn.execute("""
CREATE UNIQUE INDEX users_email_active_key
ON users(email)
WHERE is_active = TRUE
""")
logger.info("✅ Created users_email_active_key partial index")
else:
logger.info("ℹ️ users_email_active_key index already exists")
logger.info("✅ Users table unique constraints fixed for soft delete support")
except Exception as e:
logger.error(f"❌ Failed to fix users unique constraints: {e}")
raise
finally:
if conn:
await close_database_connection(conn)
async def add_options_to_backends():
"""Add options column to backends table for HAProxy backend options (option http-keep-alive, etc.)"""
conn = None
try:
conn = await get_database_connection()
# Check if options column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backends'
AND column_name = 'options'
)
""")
if not column_exists:
# Add options column as TEXT (multi-line HAProxy options)
await conn.execute("""
ALTER TABLE backends
ADD COLUMN options TEXT
""")
logger.info("✅ Added options column to backends table for HAProxy backend options")
else:
logger.info("ℹ️ options column already exists in backends table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"❌ Error adding options to backends: {e}")
# Don't raise - we'll try to proceed
async def add_options_to_frontends():
"""Add options column to frontends table for HAProxy frontend options (option httplog, option forwardfor, etc.)"""
conn = None
try:
conn = await get_database_connection()
# Check if options column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends'
AND column_name = 'options'
)
""")
if not column_exists:
# Add options column as TEXT (multi-line HAProxy options)
await conn.execute("""
ALTER TABLE frontends
ADD COLUMN options TEXT
""")
logger.info("✅ Added options column to frontends table for HAProxy frontend options")
else:
logger.info("ℹ️ options column already exists in frontends table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"❌ Error adding options to frontends: {e}")
# Don't raise - we'll try to proceed
async def add_ssl_advanced_options_to_frontends():
"""Add SSL advanced options columns to frontends table for bind SSL parameters"""
conn = None
try:
conn = await get_database_connection()
# List of SSL parameters to add
ssl_columns = [
('ssl_alpn', 'TEXT'), # Application-Layer Protocol Negotiation (e.g., "h2,http/1.1")
('ssl_npn', 'TEXT'), # Next Protocol Negotiation (legacy, before ALPN)
('ssl_ciphers', 'TEXT'), # Cipher suite list (e.g., "ECDHE-RSA-AES128-GCM-SHA256:...")
('ssl_ciphersuites', 'TEXT'), # TLS 1.3 cipher suites
('ssl_min_ver', 'VARCHAR(20)'), # Minimum TLS version (TLSv1.2, TLSv1.3)
('ssl_max_ver', 'VARCHAR(20)'), # Maximum TLS version
('ssl_strict_sni', 'BOOLEAN DEFAULT FALSE'), # Strict SNI requirement
]
for column_name, column_type in ssl_columns:
# Check if column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'frontends'
AND column_name = $1
)
""", column_name)
if not column_exists:
# Add column
await conn.execute(f"""
ALTER TABLE frontends
ADD COLUMN {column_name} {column_type}
""")
logger.info(f"✅ Added {column_name} column to frontends table")
else:
logger.info(f"ℹ️ {column_name} column already exists in frontends table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"❌ Error adding SSL advanced options to frontends: {e}")
# Don't raise - we'll try to proceed
async def add_ssl_advanced_options_to_servers():
"""Add SSL advanced options columns to backend_servers table for server SSL parameters"""
conn = None
try:
conn = await get_database_connection()
# List of SSL parameters to add
ssl_columns = [
('ssl_sni', 'VARCHAR(255)'), # SNI hostname for backend SSL connections
('ssl_min_ver', 'VARCHAR(20)'), # Minimum TLS version (TLSv1.2, TLSv1.3)
('ssl_max_ver', 'VARCHAR(20)'), # Maximum TLS version
('ssl_ciphers', 'TEXT'), # Cipher suite list for backend connections
]
for column_name, column_type in ssl_columns:
# Check if column already exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'backend_servers'
AND column_name = $1
)
""", column_name)
if not column_exists:
# Add column
await conn.execute(f"""
ALTER TABLE backend_servers
ADD COLUMN {column_name} {column_type}
""")
logger.info(f"✅ Added {column_name} column to backend_servers table")
else:
logger.info(f"ℹ️ {column_name} column already exists in backend_servers table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"❌ Error adding SSL advanced options to servers: {e}")
# Don't raise - we'll try to proceed
async def add_preserved_listen_blocks_to_agents():
"""
Add preserved_listen_blocks column to agents table.
This stores the listen block names from agent's local haproxy.cfg
so backend can detect potential naming collisions before entity creation.
COLLISION PREVENTION:
- Agent sends its current config via config-sync
- Backend parses listen blocks and stores their names here
- When creating frontend/backend, check against this list for collisions
- Prevents HAProxy "proxy has same name" errors at runtime
"""
conn = None
try:
conn = await get_database_connection()
# Check if preserved_listen_blocks column exists
column_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'agents'
AND column_name = 'preserved_listen_blocks'
)
""")
if not column_exists:
await conn.execute("""
ALTER TABLE agents
ADD COLUMN preserved_listen_blocks JSONB DEFAULT '[]'::jsonb
""")
logger.info("✅ Added preserved_listen_blocks column to agents table")
logger.info(" This column stores listen block names from agent's local config")
logger.info(" Used for collision detection when creating new entities")
else:
logger.info("ℹ️ preserved_listen_blocks column already exists in agents table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"❌ Error adding preserved_listen_blocks column: {e}")
# Don't raise - not critical for system operation
async def ensure_system_settings_table():
"""Create system_settings table for centralized application configuration."""
conn = None
try:
conn = await get_database_connection()
table_exists = await conn.fetchval("""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'system_settings'
)
""")
if not table_exists:
await conn.execute("""
CREATE TABLE system_settings (
key VARCHAR(100) PRIMARY KEY,
value JSONB NOT NULL DEFAULT '{}',
category VARCHAR(50) DEFAULT 'general',
description TEXT,
updated_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL
)
""")
logger.info("Created system_settings table")
await conn.execute("""
INSERT INTO system_settings (key, value, category, description) VALUES
('acme.provider', '"letsencrypt"', 'acme', 'ACME CA provider (letsencrypt, zerossl, google, custom)'),
('acme.directory_url', '"https://acme-v02.api.letsencrypt.org/directory"', 'acme', 'ACME directory URL'),
('acme.staging_mode', 'false', 'acme', 'Use staging/test environment'),
('acme.contact_email', '""', 'acme', 'Contact email for ACME account registration'),
('acme.auto_renew_enabled', 'false', 'acme', 'Enable automatic certificate renewal'),
('acme.renew_before_days', '30', 'acme', 'Days before expiry to trigger renewal'),
('acme.tos_accepted', 'false', 'acme', 'Terms of Service accepted'),
('acme.eab_kid', '""', 'acme', 'External Account Binding Key ID'),
('acme.eab_hmac_key', '""', 'acme', 'External Account Binding HMAC Key'),
('acme.challenge_backend_url', '""', 'acme', 'Override URL for ACME challenge backend (leave empty for auto-detect)')
ON CONFLICT (key) DO NOTHING
""")
logger.info("Seeded default ACME settings")
else:
logger.info("system_settings table already exists")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error creating system_settings table: {e}")
async def ensure_acme_tables():
"""Create ACME-related tables for Let's Encrypt / ACME certificate automation."""
conn = None
try:
conn = await get_database_connection()
# letsencrypt_accounts
exists = await conn.fetchval("""
SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'letsencrypt_accounts')
""")
if not exists:
await conn.execute("""
CREATE TABLE letsencrypt_accounts (
id SERIAL PRIMARY KEY,
email VARCHAR(255) NOT NULL,
directory_url VARCHAR(500) NOT NULL,
account_url VARCHAR(500),
jwk_private_key TEXT NOT NULL,
status VARCHAR(30) DEFAULT 'pending',
tos_agreed BOOLEAN DEFAULT FALSE,
eab_kid VARCHAR(255),
created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
UNIQUE(email, directory_url)
)
""")
logger.info("Created letsencrypt_accounts table")
# letsencrypt_orders
exists = await conn.fetchval("""
SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'letsencrypt_orders')
""")
if not exists:
await conn.execute("""
CREATE TABLE letsencrypt_orders (
id SERIAL PRIMARY KEY,
account_id INTEGER NOT NULL REFERENCES letsencrypt_accounts(id) ON DELETE CASCADE,
order_url VARCHAR(500),
status VARCHAR(30) DEFAULT 'pending',
domains JSONB NOT NULL DEFAULT '[]',
certificate_url VARCHAR(500),
finalize_url VARCHAR(500),
cert_private_key TEXT,
expires_at TIMESTAMPTZ,
error_detail TEXT,
ssl_certificate_id INTEGER REFERENCES ssl_certificates(id) ON DELETE SET NULL,
cluster_ids JSONB DEFAULT '[]',
created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
)
""")
logger.info("Created letsencrypt_orders table")
else:
try:
await conn.execute(
"ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS cert_private_key TEXT"
)
except Exception:
pass
# acme_challenges
exists = await conn.fetchval("""
SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'acme_challenges')
""")
if not exists:
await conn.execute("""
CREATE TABLE acme_challenges (
id SERIAL PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES letsencrypt_orders(id) ON DELETE CASCADE,
domain VARCHAR(255) NOT NULL,
token VARCHAR(500) NOT NULL,
key_authorization TEXT NOT NULL,
challenge_url VARCHAR(500),
status VARCHAR(30) DEFAULT 'pending',
validated_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
)
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_acme_challenges_token ON acme_challenges(token)
""")
logger.info("Created acme_challenges table with token index")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error creating ACME tables: {e}")
async def ensure_acme_columns_on_existing_tables():
"""Add ACME-related columns to existing ssl_certificates and haproxy_clusters tables."""
conn = None
try:
conn = await get_database_connection()
for col, sql in [
('source', "ALTER TABLE ssl_certificates ADD COLUMN IF NOT EXISTS source VARCHAR(20) DEFAULT 'manual'"),
('letsencrypt_order_id', "ALTER TABLE ssl_certificates ADD COLUMN IF NOT EXISTS letsencrypt_order_id INTEGER REFERENCES letsencrypt_orders(id) ON DELETE SET NULL"),
('auto_renew', "ALTER TABLE ssl_certificates ADD COLUMN IF NOT EXISTS auto_renew BOOLEAN DEFAULT FALSE"),
('acme_enabled', "ALTER TABLE haproxy_clusters ADD COLUMN IF NOT EXISTS acme_enabled BOOLEAN DEFAULT FALSE"),
('acme_backend_url', "ALTER TABLE haproxy_clusters ADD COLUMN IF NOT EXISTS acme_backend_url VARCHAR(500)"),
# Issue #12 / Commit 5a: track challenge response attempts for rate-limit + retry policy
('attempts', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS attempts INTEGER DEFAULT 0"),
('last_attempt_at', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS last_attempt_at TIMESTAMPTZ"),
# Commit 3a: track auto-completion task lock/poll timestamps for atomic claim across replicas
('orders_updated_at_idx', "CREATE INDEX IF NOT EXISTS idx_letsencrypt_orders_status_updated ON letsencrypt_orders(status, updated_at) WHERE status = 'valid' AND ssl_certificate_id IS NULL"),
# Issue #35 (v1.8.0 — ACME DNS-01): per-account challenge method + DNS provider selection
('acct_challenge_type', "ALTER TABLE letsencrypt_accounts ADD COLUMN IF NOT EXISTS challenge_type VARCHAR(20) DEFAULT 'http-01'"),
('acct_dns_provider', "ALTER TABLE letsencrypt_accounts ADD COLUMN IF NOT EXISTS dns_provider VARCHAR(50)"),
# per-order challenge method + bounded DNS-01 retry chain (dns01_parent_order_id is a PLAIN INTEGER, not a FK,
# to avoid a self-referential cascade interacting with account/order bulk DELETEs)
('order_challenge_type', "ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS challenge_type VARCHAR(20) DEFAULT 'http-01'"),
('order_dns01_attempts', "ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS dns01_attempts INTEGER DEFAULT 0"),
('order_dns01_last_attempt_at', "ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS dns01_last_attempt_at TIMESTAMPTZ"),
('order_dns01_parent_order_id', "ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS dns01_parent_order_id INTEGER"),
('order_dns01_retry_claimed', "ALTER TABLE letsencrypt_orders ADD COLUMN IF NOT EXISTS dns01_retry_claimed BOOLEAN DEFAULT FALSE"),
# per-challenge DNS-01 lifecycle state
('chal_challenge_type', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS challenge_type VARCHAR(20) DEFAULT 'http-01'"),
('chal_dns_txt_value', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS dns_txt_value TEXT"),
('chal_dns_record_published', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS dns_record_published BOOLEAN DEFAULT FALSE"),
('chal_dns_record_cleaned', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS dns_record_cleaned BOOLEAN DEFAULT FALSE"),
('chal_dns_published_at', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS dns_published_at TIMESTAMPTZ"),
('chal_manual_confirm_deadline', "ALTER TABLE acme_challenges ADD COLUMN IF NOT EXISTS manual_confirm_deadline TIMESTAMPTZ"),
]:
try:
await conn.execute(sql)
logger.info(f"Ensured column/index exists: {col}")
except Exception as col_err:
logger.warning(f"Column/index {col} migration note: {col_err}")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error adding ACME columns: {e}")
async def ensure_letsencrypt_dns_credentials():
"""Issue #35 (v1.8.0 — ACME DNS-01): per-account encrypted DNS provider credentials.
Idempotent (CREATE TABLE IF NOT EXISTS). FK to letsencrypt_accounts (created earlier by
ensure_acme_tables). Credentials are Fernet-encrypted at rest (backend/utils/dns_credentials.py);
only the provider name + timestamps are ever surfaced to the API.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute("""
CREATE TABLE IF NOT EXISTS letsencrypt_account_dns_credentials (
id SERIAL PRIMARY KEY,
account_id INTEGER NOT NULL UNIQUE REFERENCES letsencrypt_accounts(id) ON DELETE CASCADE,
dns_provider VARCHAR(50) NOT NULL,
credentials_encrypted TEXT NOT NULL,
created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
)
""")
logger.info("Ensured letsencrypt_account_dns_credentials table")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error ensuring letsencrypt_account_dns_credentials: {e}")
async def cleanup_orphan_acme_challenge_backend():
"""
Issue #11: One-time cleanup of orphan `_acme_challenge_backend` rows that may
have been persisted by previous versions where agent sync did not filter
auto-managed backends. Idempotent (NO-OP if zero rows).
backend_servers.backend_id has ON DELETE CASCADE, so deleting parent backends
will cascade-delete dependent server rows. We also explicitly delete by
backend_name first to clean up any orphan rows where backend_id may be NULL
or stale (string-based references).
"""
conn = None
try:
conn = await get_database_connection()
cnt = await conn.fetchval(
"SELECT COUNT(*) FROM backends WHERE name = '_acme_challenge_backend'"
)
if cnt and cnt > 0:
logger.warning(
f"CLEANUP MIGRATION: Found {cnt} orphan '_acme_challenge_backend' "
f"row(s). Removing (Issue #11 cleanup)."
)
async with conn.transaction():
# Defensive: clean up any backend_servers rows by name first
# (catches orphans where backend_id is NULL or stale)
bs_cnt = await conn.execute(
"DELETE FROM backend_servers WHERE backend_name = '_acme_challenge_backend'"
)
logger.info(f"CLEANUP MIGRATION: Removed backend_servers rows: {bs_cnt}")
# Delete parent backends - FK CASCADE removes any remaining backend_servers
be_cnt = await conn.execute(
"DELETE FROM backends WHERE name = '_acme_challenge_backend'"
)
logger.info(f"CLEANUP MIGRATION: Removed backends rows: {be_cnt}")
else:
logger.info("CLEANUP MIGRATION: No orphan '_acme_challenge_backend' rows found (clean state)")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in cleanup_orphan_acme_challenge_backend: {e}")
# =============================================================================
# v1.5.0 migrations: ACME diagnostic panel (Feature A) + site wizard (B)
# =============================================================================
async def ensure_letsencrypt_orders_post_completion_actions_column():
"""v1.5.0: add post_completion_actions JSONB column on letsencrypt_orders.
Carries the deferred HTTPS frontend create payload (and any future
post-completion actions) for wizard-staged ACME orders. Idempotent.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute(
"""
ALTER TABLE letsencrypt_orders
ADD COLUMN IF NOT EXISTS post_completion_actions JSONB DEFAULT '[]'::jsonb
"""
)
logger.info("Ensured letsencrypt_orders.post_completion_actions column")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in ensure_letsencrypt_orders_post_completion_actions_column: {e}")
async def ensure_letsencrypt_orders_wizard_staged_until_column():
"""v1.5.0: add wizard_staged_until TIMESTAMPTZ column on letsencrypt_orders.
Used by complete_pending_acme_orders to abandon stale wizard_staged orders
after 24h (M25). NULL for non-wizard orders.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute(
"""
ALTER TABLE letsencrypt_orders
ADD COLUMN IF NOT EXISTS wizard_staged_until TIMESTAMPTZ
"""
)
logger.info("Ensured letsencrypt_orders.wizard_staged_until column")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in ensure_letsencrypt_orders_wizard_staged_until_column: {e}")
async def ensure_letsencrypt_orders_pending_apply_version_name_column():
"""v1.5.0: add pending_apply_version_name VARCHAR + partial index for fast
`wizard_staged` lookups by config version name.
The wizard records the bulk-site-create-{ts} version name (legacy
naming pre-rename: bulk-proxied-host-create-{ts}) into
this column when it stages the order; the background task uses
string-equality vs agents.applied_config_version to gate LE API calls.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute(
"""
ALTER TABLE letsencrypt_orders
ADD COLUMN IF NOT EXISTS pending_apply_version_name VARCHAR(255)
"""
)
await conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_letsencrypt_orders_wizard_staged
ON letsencrypt_orders (pending_apply_version_name)
WHERE status = 'wizard_staged'
"""
)
logger.info(
"Ensured letsencrypt_orders.pending_apply_version_name column + partial index"
)
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(
f"Error in ensure_letsencrypt_orders_pending_apply_version_name_column: {e}"
)
async def ensure_letsencrypt_orders_created_by_column():
"""v1.5.0: add created_by INTEGER on letsencrypt_orders (R31/M23).
Carries the requesting user_id so post_completion_actions auto-apply
can attribute the apply to the original wizard caller. ON DELETE
SET NULL so deleting the user does not break orphan orders.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute(
"""
ALTER TABLE letsencrypt_orders
ADD COLUMN IF NOT EXISTS created_by INTEGER
REFERENCES users(id) ON DELETE SET NULL
"""
)
logger.info("Ensured letsencrypt_orders.created_by column (FK ON DELETE SET NULL)")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in ensure_letsencrypt_orders_created_by_column: {e}")
async def ensure_acme_order_events_table():
"""v1.5.0 Feature A: detailed ACME event log table.
Used by record_event() for diagnostic timeline display. CASCADE on order
delete so deleting a letsencrypt_order also cleans up its event trail.
Daily-watermarked TTL prune (90d) lives in main.py background task.
"""
conn = None
try:
conn = await get_database_connection()
exists = await conn.fetchval(
"""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'acme_order_events'
)
"""
)
if not exists:
await conn.execute(
"""
CREATE TABLE acme_order_events (
id BIGSERIAL PRIMARY KEY,
order_id INTEGER NOT NULL
REFERENCES letsencrypt_orders(id) ON DELETE CASCADE,
event_type VARCHAR(64) NOT NULL,
severity VARCHAR(16) NOT NULL DEFAULT 'INFO',
message TEXT,
details JSONB DEFAULT '{}'::jsonb,
correlation_id VARCHAR(64),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
"""
)
logger.info("Created acme_order_events table")
# R16 hardening (#R16-2): indexes must run UNCONDITIONALLY on every
# startup, not just on first table creation. An older deploy that
# raced ahead of these indexes (or where the table was created by a
# previous v1.5.0 push before the daily-watermarked retention task
# existed) would otherwise be stuck doing sequential scans for the
# 90-day prune query. Both `CREATE INDEX IF NOT EXISTS` calls are
# idempotent so re-running is safe.
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_acme_order_events_order_id "
"ON acme_order_events(order_id, created_at DESC)"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_acme_order_events_created_at "
"ON acme_order_events(created_at)"
)
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in ensure_acme_order_events_table: {e}")
async def ensure_wizard_drafts_table():
"""v1.5.0 Feature B: persisted wizard drafts.
expires_at defaults to NOW() + 30d; daily-watermarked prune in main.py.
user_id ON DELETE CASCADE so deleting a user removes their drafts.
"""
conn = None
try:
conn = await get_database_connection()
exists = await conn.fetchval(
"""
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_name = 'wizard_drafts'
)
"""
)
if not exists:
await conn.execute(
"""
CREATE TABLE wizard_drafts (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL
REFERENCES users(id) ON DELETE CASCADE,
wizard_type VARCHAR(64) NOT NULL DEFAULT 'site',
title VARCHAR(255),
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
expires_at TIMESTAMPTZ NOT NULL DEFAULT (NOW() + INTERVAL '30 days'),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
"""
)
logger.info("Created wizard_drafts table")
# R16 hardening (#R16-2): see acme_order_events fix above. Indexes
# MUST run unconditionally so existing v1.5.0 first-deploy tables
# also get the prune-supporting expires_at index.
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_wizard_drafts_user_type "
"ON wizard_drafts(user_id, wizard_type, updated_at DESC)"
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_wizard_drafts_expires_at "
"ON wizard_drafts(expires_at)"
)
# Phase I: Site rebrand — flip the schema-level DEFAULT for the
# `wizard_type` column from the legacy 'proxied_host' value to
# the post-rebrand 'site' value so all NEW rows (when callers
# rely on the column default) land with the canonical naming.
# This is purely an `ALTER TABLE … ALTER COLUMN … SET DEFAULT`
# — idempotent, takes a SHARE UPDATE EXCLUSIVE-equivalent
# metadata lock that does NOT block readers/writers, and does
# not rewrite existing rows. Pre-rename rows still carry
# `wizard_type='proxied_host'`; the application code path
# accepts BOTH values via dual-filter (`IN ('site',
# 'proxied_host')`) on every read/delete query, so older
# drafts remain visible to their owner and remain rejectable
# via the cluster cleanup.
await conn.execute(
"ALTER TABLE wizard_drafts ALTER COLUMN wizard_type SET DEFAULT 'site'"
)
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(f"Error in ensure_wizard_drafts_table: {e}")
async def ensure_user_activity_logs_user_action_time_index():
"""v1.5.0 (M33/R50): composite index on user_activity_logs for the new
per-user-per-minute rate-limit COUNT(*) query used by ACME diagnostics
and wizard preflight rate-limits.
"""
conn = None
try:
conn = await get_database_connection()
await conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_user_activity_logs_user_action_time
ON user_activity_logs (user_id, action, created_at DESC)
"""
)
logger.info("Ensured user_activity_logs (user_id, action, created_at) composite index")
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
logger.error(
f"Error in ensure_user_activity_logs_user_action_time_index: {e}"
)
async def ensure_frontends_bind_unique_constraint():
"""v1.5.0 (R18c round 3 #1 — KRITIK concurrency): partial unique
constraint on (cluster_id, bind_address, bind_port) WHERE
is_active.
Pre-fix: `services/frontend_service.check_bind_port_collision`
ran a plain `SELECT` outside the wizard transaction with no
`FOR UPDATE`, and the schema had NO uniqueness on
(cluster_id, bind_address, bind_port). Two concurrent wizards
targeting the same cluster + bind could both pass the check
and both INSERT, producing TWO active frontends bound to the
same port — HAProxy then refused to reload (port already in
use) and the cluster was wedged until manual cleanup.
Adding a partial UNIQUE INDEX serializes the race at the
database level: the second INSERT raises UniqueViolationError,
which the wizard router (R18b round 3 #11) already maps to a
clean 409.
NOT auto-deduplicating: `CREATE UNIQUE INDEX IF NOT EXISTS`
only skips when the index NAME already exists. If a deployment
already has duplicate active rows (a pre-fix race that landed),
the migration FAILS with "could not create unique index" and is
logged as non-fatal — runtime continues without the index, which
means the database-level race protection is OFF until an operator
manually consolidates the conflicting rows. Operationally:
# find conflicting active rows
SELECT cluster_id, bind_address, bind_port, COUNT(*)
FROM frontends
WHERE is_active = TRUE
GROUP BY 1,2,3 HAVING COUNT(*) > 1;
Even without the index, the wizard router still maps the
happy-path race outcome to 409 via UniqueViolationError when
the index DOES exist, so this migration is the belt-and-
suspenders layer rather than the only protection.
Partial WHERE is_active is intentional — soft-deleted
frontends (is_active=false) are kept for audit and would
otherwise prevent re-creating a binding after deactivation.
"""
conn = None
try:
conn = await get_database_connection()
# Use a unique INDEX (not constraint) because PostgreSQL
# only allows partial uniqueness via an INDEX, not a table
# CONSTRAINT. Functionally equivalent for asyncpg's
# UniqueViolationError path.
await conn.execute(
"""
CREATE UNIQUE INDEX IF NOT EXISTS
idx_frontends_active_bind_unique
ON frontends (cluster_id, bind_address, bind_port)
WHERE is_active = TRUE
"""
)
logger.info(
"Ensured frontends partial UNIQUE on "
"(cluster_id, bind_address, bind_port) WHERE is_active=TRUE"
)
await close_database_connection(conn)
except Exception as e:
if conn:
await close_database_connection(conn)
# Existing duplicates would surface here as
# 'could not create unique index'. Log loudly so operators
# see the conflicting rows in their migration logs but do
# NOT abort startup — the wizard 409-mapping path already
# covers the steady-state race; the constraint is the
# belt-and-suspenders.
logger.error(
f"Error in ensure_frontends_bind_unique_constraint: {e} "
"(non-fatal — wizard router still maps UniqueViolation "
"to 409 even without the index)"
)
Reference in New Issue View Git Blame Copy Permalink