feat(forward): apply DDL to a database — SECURITY REVIEW REQUIRED

backend/app/api/connections.py

@@ -3,16 +3,17 @@
 import datetime as dt
 import uuid
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.auth import CurrentUser, get_current_user
 from app.db import get_read_session, get_session
+from app.db_introspect import apply_database_sql
 from app.models import DbConnection
 from app.permissions import require_project_member
-from app.schemas import ConnectionCreateIn, ConnectionOut
-from app.security import encrypt_text
+from app.schemas import ApplySqlIn, ApplySqlOut, ConnectionCreateIn, ConnectionOut
+from app.security import decrypt_text, encrypt_text
 from app.sanitize import sanitize_for_storage
 
 router = APIRouter(prefix="/api/connections", tags=["connections"])
@@ -62,3 +63,54 @@ async def create_connection(
     session.add(c)
     await session.commit()
     return ConnectionOut(db_connection_uuid=c.db_connection_uuid, conn_name=c.conn_name)
+
+
+@router.post("/{db_connection_uuid}/apply-sql", response_model=ApplySqlOut)
+async def apply_sql(
+    db_connection_uuid: uuid.UUID,
+    body: ApplySqlIn,
+    user: CurrentUser = Depends(get_current_user),
+    session: AsyncSession = Depends(get_read_session),
+) -> ApplySqlOut:
+    """Forward engineering: apply DDL/SQL to a stored connection's database.
+
+    SECURITY-SENSITIVE (writes to a live database):
+    * Requires the **editor** role on the connection's project.
+    * IDOR-safe: non-members get a uniform 404 (no enumeration); members
+      lacking editor get 403.
+    * The DSN is decrypted only in memory; the connection reuses the
+      introspectors' SSRF guard (validated + pinned IP).
+    * Runs the whole batch in one transaction; ``dry_run`` (default True) rolls
+      back so nothing persists -- callers must opt in to persist. Failure
+      messages are DSN-redacted, and an execution error is ``ok=false`` rather
+      than an error status (a failed apply is a normal result).
+    """
+    project_space_uuid = await session.scalar(
+        select(DbConnection.project_space_uuid).where(
+            DbConnection.db_connection_uuid == db_connection_uuid
+        )
+    )
+    if project_space_uuid is None:
+        raise HTTPException(status_code=404, detail="connection not found")
+    # Membership first (mask non-members to 404), then require editor (403).
+    try:
+        await require_project_member(
+            session, project_space_uuid, user.user_account_uuid
+        )
+    except HTTPException as exc:
+        if exc.status_code == 403:
+            raise HTTPException(status_code=404, detail="connection not found")
+        raise
+    await require_project_member(
+        session, project_space_uuid, user.user_account_uuid, minimum_role="editor"
+    )
+
+    conn = await session.get(DbConnection, db_connection_uuid)
+    if conn is None:
+        raise HTTPException(status_code=404, detail="connection not found")
+    dsn = decrypt_text(conn.dsn_ciphertext, conn.dsn_nonce)
+    try:
+        await apply_database_sql(dsn, body.sql, dry_run=body.dry_run)
+        return ApplySqlOut(ok=True, dry_run=body.dry_run, error=None)
+    except Exception as exc:  # noqa: BLE001 - message is already DSN-redacted
+        return ApplySqlOut(ok=False, dry_run=body.dry_run, error=str(exc))

backend/app/db_introspect.py

@@ -4,7 +4,7 @@
 from urllib.parse import urlparse
 
 from app.dsn_redaction import redact_dsn_error_message
-from app.pg_introspect.introspect import introspect_postgres
+from app.pg_introspect.introspect import apply_postgres_sql, introspect_postgres
 from app.snowflake_introspect import introspect_snowflake
 
 DatabaseDialect = Literal["postgresql", "snowflake"]
@@ -32,3 +32,20 @@ async def introspect_database(dsn: str, schema_filter: str | None) -> dict:
     except Exception as exc:
         message = str(exc) or type(exc).__name__
         raise RuntimeError(redact_dsn_error_message(message, dsn)) from None
+
+
+async def apply_database_sql(dsn: str, sql: str, dry_run: bool = True) -> None:
+    """Forward engineering: apply DDL/SQL to a target database.
+
+    PostgreSQL only for now (Snowflake DDL apply differs and is connector-gated).
+    Errors are DSN-redacted so credentials never surface in an API response.
+    """
+
+    try:
+        dialect = detect_dsn_dialect(dsn)
+        if dialect != "postgresql":
+            raise ValueError("forward apply is only supported for PostgreSQL")
+        await apply_postgres_sql(dsn, sql, dry_run=dry_run)
+    except Exception as exc:
+        message = str(exc) or type(exc).__name__
+        raise RuntimeError(redact_dsn_error_message(message, dsn)) from None

backend/app/pg_introspect/introspect.py

@@ -10,6 +10,47 @@
 from app.sanitize import sanitize_for_storage
 
 
+async def apply_postgres_sql(dsn: str, sql: str, dry_run: bool = True) -> None:
+    """Execute DDL/SQL against a target Postgres DB inside one transaction.
+
+    Forward engineering: materialize a designed schema / apply a migration.
+    SSRF-guarded exactly like introspection (``validate_postgres_dsn_target`` +
+    pinned IP). The whole batch runs in a single transaction:
+
+    * ``dry_run=True`` (default) rolls back, so nothing persists -- a pre-flight
+      validation that the SQL executes cleanly against the real database.
+    * ``dry_run=False`` commits.
+
+    PostgreSQL DDL is transactional, so a mid-batch failure rolls everything
+    back. Statements that cannot run inside a transaction (e.g.
+    ``CREATE INDEX CONCURRENTLY``) will raise here -- that is expected.
+    """
+    target = await validate_postgres_dsn_target(dsn)
+    connect_host: str | list[str] = (
+        target.hosts[0] if len(target.hosts) == 1 else list(target.hosts)
+    )
+    if target.port is not None:
+        conn = await asyncpg.connect(
+            dsn, host=connect_host, port=target.port, timeout=15
+        )
+    else:
+        conn = await asyncpg.connect(dsn, host=connect_host, timeout=15)
+    try:
+        tx = conn.transaction()
+        await tx.start()
+        try:
+            await conn.execute(sql)
+        except BaseException:
+            await tx.rollback()
+            raise
+        if dry_run:
+            await tx.rollback()
+        else:
+            await tx.commit()
+    finally:
+        await conn.close()
+
+
 async def introspect_postgres(dsn: str, schema_filter: str | None) -> dict:
     """Introspect a PostgreSQL database and return a snapshot JSON."""
 

backend/app/schemas.py

@@ -58,6 +58,22 @@ class ConnectionOut(BaseModel):
     conn_name: str
 
 
+class ApplySqlIn(BaseModel):
+    """Request body for forward-engineering: apply DDL/SQL to a connection."""
+
+    sql: str = Field(min_length=1, max_length=262_144)
+    # Default to a rolled-back pre-flight; the caller must opt in to persist.
+    dry_run: bool = True
+
+
+class ApplySqlOut(BaseModel):
+    """Result of applying SQL (DSN-redacted on failure)."""
+
+    ok: bool
+    dry_run: bool
+    error: str | None = None
+
+
 class SnapshotCreateIn(BaseModel):
     """Request body for creating a schema snapshot."""
 

backend/tests/test_api_apply_sql.py

@@ -0,0 +1,123 @@
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from app.api.connections import apply_sql
+from app.auth import CurrentUser
+from app.db_introspect import apply_database_sql
+from app.schemas import ApplySqlIn
+
+
+def _user():
+    return CurrentUser(user_account_uuid=uuid.uuid4(), subject="t", display_name="T")
+
+
+def _conn():
+    return SimpleNamespace(dsn_ciphertext=b"c", dsn_nonce=b"n")
+
+
+def _body(dry_run=True):
+    return ApplySqlIn(sql="CREATE TABLE t (id bigint);", dry_run=dry_run)
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_returns_404_when_missing():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=None)
+    with pytest.raises(HTTPException) as e:
+        await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_masks_non_member_as_404():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    with patch(
+        "app.api.connections.require_project_member",
+        new_callable=AsyncMock,
+        side_effect=HTTPException(status_code=403, detail="project access denied"),
+    ):
+        with pytest.raises(HTTPException) as e:
+            await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_returns_403_when_member_lacks_editor():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    # first call (membership) passes, second call (editor) raises 403
+    with patch(
+        "app.api.connections.require_project_member",
+        new_callable=AsyncMock,
+        side_effect=[None, HTTPException(status_code=403, detail="insufficient project role")],
+    ):
+        with pytest.raises(HTTPException) as e:
+            await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_reports_ok_true_on_success():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    session.get = AsyncMock(return_value=_conn())
+    with patch(
+        "app.api.connections.require_project_member", new_callable=AsyncMock
+    ), patch(
+        "app.api.connections.decrypt_text", return_value="postgresql://u@db.example.com/x"
+    ), patch(
+        "app.api.connections.apply_database_sql", new_callable=AsyncMock
+    ) as apply_mock:
+        out = await apply_sql(
+            db_connection_uuid=uuid.uuid4(), body=_body(dry_run=True), user=_user(), session=session
+        )
+    assert out.ok is True and out.dry_run is True and out.error is None
+    apply_mock.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_reports_ok_false_on_error():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    session.get = AsyncMock(return_value=_conn())
+    with patch(
+        "app.api.connections.require_project_member", new_callable=AsyncMock
+    ), patch(
+        "app.api.connections.decrypt_text", return_value="postgresql://u@db.example.com/x"
+    ), patch(
+        "app.api.connections.apply_database_sql",
+        new_callable=AsyncMock,
+        side_effect=RuntimeError("syntax error at or near"),
+    ):
+        out = await apply_sql(
+            db_connection_uuid=uuid.uuid4(), body=_body(dry_run=False), user=_user(), session=session
+        )
+    assert out.ok is False and out.dry_run is False and "syntax error" in (out.error or "")
+
+
+@pytest.mark.asyncio
+async def test_apply_database_sql_rejects_non_postgres():
+    with pytest.raises(RuntimeError):
+        await apply_database_sql("snowflake://u@acct/db", "CREATE TABLE t (id int);")
+
+
+@pytest.mark.asyncio
+async def test_apply_database_sql_routes_postgres_and_redacts():
+    with patch("app.db_introspect.apply_postgres_sql", new_callable=AsyncMock) as m:
+        await apply_database_sql(
+            "postgresql://u@db.example.com/x", "CREATE TABLE t (id int);", dry_run=True
+        )
+    m.assert_awaited_once()
+    with patch(
+        "app.db_introspect.apply_postgres_sql",
+        new_callable=AsyncMock,
+        side_effect=OSError("connect to postgresql://user:s3cret@db"),
+    ):
+        with pytest.raises(RuntimeError) as e:
+            await apply_database_sql("postgresql://user:s3cret@db.example.com/x", "X")
+    assert "s3cret" not in str(e.value)