fix(security): fix alembic.ini placeholder database URL (#44)

quantara/web_app/alembic.ini

@@ -60,7 +60,7 @@ version_path_separator = os  # Use os.pathsep. Default configuration used for ne
 # are written from script.py.mako
 # output_encoding = utf-8
 
-sqlalchemy.url = driver://user:pass@localhost/dbname
+sqlalchemy.url = REPLACE_ME
 
 
 [post_write_hooks]

quantara/web_app/alembic/env.py

@@ -1,84 +1,75 @@
 """
 Alembic migration environment configuration.
-
-This module sets up the environment for Alembic database migrations, 
-including database URL configuration, logging setup, and migration 
-execution modes (online and offline). It integrates with SQLAlchemy models 
-and provides the core functionality needed for database schema version control.
 """
 
+import os
 from logging.config import fileConfig
 
 from sqlalchemy import engine_from_config
 from sqlalchemy import pool
 
 from alembic import context
 
-from web_app.db.database import get_database_url
 from web_app.db.models import Base
 
-# this is the Alembic Config object, which provides
-# access to the values within the .ini file in use.
 config = context.config
-config.set_main_option("sqlalchemy.url", get_database_url())
 
-# Interpret the config file for Python logging.
-# This line sets up loggers basically.
 if config.config_file_name is not None:
     fileConfig(config.config_file_name)
 
-# add your model's MetaData object here
-# for 'autogenerate' support
-# from myapp import mymodel
-# target_metadata = mymodel.Base.metadata
 target_metadata = Base.metadata
 
-# other values from the config, defined by the needs of env.py,
-# can be acquired:
-# my_important_option = config.get_main_option("my_important_option")
-# ... etc.
 
+def _build_db_url() -> str:
+    """Construct PostgreSQL URL from environment variables.
 
-def run_migrations_offline() -> None:
-    """Run migrations in 'offline' mode.
+    Raises RuntimeError with a clear message if any required variable is absent,
+    so Alembic fails fast instead of silently connecting to a wrong host.
+    """
+    required = {
+        "DB_USER": os.getenv("DB_USER"),
+        "DB_PASSWORD": os.getenv("DB_PASSWORD"),
+        "DB_HOST": os.getenv("DB_HOST"),
+        "DB_NAME": os.getenv("DB_NAME"),
+    }
+    missing = [k for k, v in required.items() if not v]
+    if missing:
+        raise RuntimeError(
+            "Alembic cannot connect: missing required environment variables: "
+            + ", ".join(missing)
+        )
+    port = os.getenv("DB_PORT", "5432")
+    return (
+        f"postgresql://{required['DB_USER']}:{required['DB_PASSWORD']}"
+        f"@{required['DB_HOST']}:{port}/{required['DB_NAME']}"
+    )
 
-    This configures the context with just a URL
-    and not an Engine, though an Engine is acceptable
-    here as well.  By skipping the Engine creation
-    we don't even need a DBAPI to be available.
 
-    Calls to context.execute() here emit the given string to the
-    script output.
+config.set_main_option("sqlalchemy.url", _build_db_url())
 
-    """
+
+def run_migrations_offline() -> None:
+    """Run migrations in 'offline' mode."""
     url = config.get_main_option("sqlalchemy.url")
     context.configure(
         url=url,
         target_metadata=target_metadata,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
     )
-
     with context.begin_transaction():
         context.run_migrations()
 
 
 def run_migrations_online() -> None:
-    """Run migrations in 'online' mode.
-
-    In this scenario we need to create an Engine
-    and associate a connection with the context.
-
-    """
+    """Run migrations in 'online' mode."""
     connectable = engine_from_config(
         config.get_section(config.config_ini_section, {}),
         prefix="sqlalchemy.",
         poolclass=pool.NullPool,
     )
-
     with connectable.connect() as connection:
         context.configure(connection=connection, target_metadata=target_metadata)
-
         with context.begin_transaction():
             context.run_migrations()
 

quantara/web_app/tests/test_alembic_env.py

@@ -0,0 +1,69 @@
+"""Tests for alembic database URL construction and configuration."""
+
+import os
+
+import pytest
+
+
+def test_get_database_url_constructs_correctly(monkeypatch):
+    """URL is built from the five DB_* env vars."""
+    monkeypatch.setenv("DB_USER", "alice")
+    monkeypatch.setenv("DB_PASSWORD", "s3cr3t")
+    monkeypatch.setenv("DB_HOST", "db.local")
+    monkeypatch.setenv("DB_PORT", "5432")
+    monkeypatch.setenv("DB_NAME", "mydb")
+
+    import importlib
+    import web_app.db.database as db_mod
+    importlib.reload(db_mod)
+    url = db_mod.get_database_url()
+    assert url == "postgresql://alice:s3cr3t@db.local:5432/mydb"
+
+
+def test_get_database_url_default_port(monkeypatch):
+    """DB_PORT defaults to 5432 when not set."""
+    monkeypatch.setenv("DB_USER", "alice")
+    monkeypatch.setenv("DB_PASSWORD", "s3cr3t")
+    monkeypatch.setenv("DB_HOST", "db.local")
+    monkeypatch.delenv("DB_PORT", raising=False)
+    monkeypatch.setenv("DB_NAME", "mydb")
+
+    import importlib
+    import web_app.db.database as db_mod
+    importlib.reload(db_mod)
+    url = db_mod.get_database_url()
+    assert "5432" in url
+
+
+def test_alembic_ini_sentinel_value():
+    """alembic.ini must not contain the old placeholder URL."""
+    ini_path = os.path.join(
+        os.path.dirname(__file__), "..", "alembic.ini"
+    )
+    with open(ini_path) as fh:
+        content = fh.read()
+    assert "driver://user:pass@localhost/dbname" not in content, (
+        "alembic.ini still contains the insecure placeholder URL"
+    )
+    assert "REPLACE_ME" in content, (
+        "alembic.ini must use REPLACE_ME as the sqlalchemy.url sentinel"
+    )
+
+
+def test_get_database_url_includes_all_components(monkeypatch):
+    """All components appear in the final URL."""
+    monkeypatch.setenv("DB_USER", "myuser")
+    monkeypatch.setenv("DB_PASSWORD", "mypass")
+    monkeypatch.setenv("DB_HOST", "pghost")
+    monkeypatch.setenv("DB_PORT", "5433")
+    monkeypatch.setenv("DB_NAME", "testdb")
+
+    import importlib
+    import web_app.db.database as db_mod
+    importlib.reload(db_mod)
+    url = db_mod.get_database_url()
+    assert "myuser" in url
+    assert "mypass" in url
+    assert "pghost" in url
+    assert "5433" in url
+    assert "testdb" in url