From 7ebb1a35b7608d01d02b6a12deb81d9e4e2b8a4f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 05:55:11 +0900 Subject: [PATCH] feat(forward): apply DDL/migration to a target database (dry-run default) Forward engineering (model -> DB), the missing half of the round-trip. POST /api/connections/{uuid}/apply-sql runs DDL in ONE transaction; dry_run=True (default) rolls back as a pre-flight, dry_run=False commits. Editor role, IDOR-safe (non-member 404, member-without-editor 403), SSRF-guarded via validate_postgres_dsn_target + pinned IP (apply_postgres_sql mirrors probe_postgres), DSN-redacted errors, ok=false on failure. PostgreSQL only for now. +7 tests, backend suite 265. SECURITY REVIEW REQUIRED before merge (writes to a live DB). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01AxU2xaupAjp912oDNFuWyd --- backend/app/api/connections.py | 61 +++++++++++- backend/app/db_introspect.py | 23 ++++- backend/app/pg_introspect/introspect.py | 41 ++++++++ backend/app/schemas.py | 16 +++ backend/tests/test_api_apply_sql.py | 123 ++++++++++++++++++++++++ docs/valuation-gap-analysis.md | 5 + 6 files changed, 266 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_api_apply_sql.py diff --git a/backend/app/api/connections.py b/backend/app/api/connections.py index 6ef46654..9c70a13f 100644 --- a/backend/app/api/connections.py +++ b/backend/app/api/connections.py @@ -9,10 +9,16 @@ from app.auth import CurrentUser, get_current_user from app.db import get_read_session, get_session -from app.db_introspect import probe_database +from app.db_introspect import apply_database_sql, probe_database from app.models import DbConnection from app.permissions import require_project_member -from app.schemas import ConnectionCreateIn, ConnectionOut, ConnectionTestOut +from app.schemas import ( + ApplySqlIn, + ApplySqlOut, + ConnectionCreateIn, + ConnectionOut, + ConnectionTestOut, +) from app.security import decrypt_text, encrypt_text from app.sanitize import sanitize_for_storage @@ -104,3 +110,54 @@ async def test_connection( return ConnectionTestOut(ok=True, server_version=version, error=None) except Exception as exc: # noqa: BLE001 - message is already DSN-redacted return ConnectionTestOut(ok=False, server_version=None, error=str(exc)) + + +@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)) diff --git a/backend/app/db_introspect.py b/backend/app/db_introspect.py index 6b41776e..b2ee8798 100644 --- a/backend/app/db_introspect.py +++ b/backend/app/db_introspect.py @@ -4,7 +4,11 @@ from urllib.parse import urlparse from app.dsn_redaction import redact_dsn_error_message -from app.pg_introspect.introspect import introspect_postgres, probe_postgres +from app.pg_introspect.introspect import ( + apply_postgres_sql, + introspect_postgres, + probe_postgres, +) from app.snowflake_introspect import introspect_snowflake from app.snowflake_introspect.introspect import probe_snowflake @@ -50,3 +54,20 @@ async def probe_database(dsn: str) -> str: 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 diff --git a/backend/app/pg_introspect/introspect.py b/backend/app/pg_introspect/introspect.py index 02d9cb5d..6faaf1bd 100644 --- a/backend/app/pg_introspect/introspect.py +++ b/backend/app/pg_introspect/introspect.py @@ -33,6 +33,47 @@ async def probe_postgres(dsn: str) -> str: await conn.close() +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.""" diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 9540c560..1f32d53e 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -67,6 +67,22 @@ class ConnectionTestOut(BaseModel): error: str | None = None +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.""" diff --git a/backend/tests/test_api_apply_sql.py b/backend/tests/test_api_apply_sql.py new file mode 100644 index 00000000..bb2402ae --- /dev/null +++ b/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) diff --git a/docs/valuation-gap-analysis.md b/docs/valuation-gap-analysis.md index 537f84b0..7eeeba6b 100644 --- a/docs/valuation-gap-analysis.md +++ b/docs/valuation-gap-analysis.md @@ -40,6 +40,11 @@ now cross into **act** (migration SQL). The gaps below are ordered by value-per- - **Data dictionary export** — Markdown/HTML doc of schema + **annotations** (now that annotations exist) + PK/FK/index. High value for the "living documentation" buyer; pure-logic, testable. - **Relationship inference** — suggest implicit FKs by naming/type heuristics (`*_id` → `id`). Composes with existing cardinality logic. +### P0b — Forward engineering (closes the round-trip: model → database) +The product was reverse-only (DB → model). **Forward engineering** — materializing a designed schema / applying a migration to a real database — is the other half and a major value unlock: design → deploy in one tool. +- **Apply DDL/migration to a connection** _(backend delivered — SECURITY REVIEW REQUIRED)_: `POST /api/connections/{uuid}/apply-sql` runs DDL against the connection's DB inside one transaction, **dry-run by default** (rolls back = pre-flight validation), editor-only, IDOR-safe, SSRF-guarded (reuses `validate_postgres_dsn_target` + pinned IP), DSN-redacted errors. Because it writes to a live DB, it must pass human security review before merge and needs a careful confirmation UI. +- **Next**: down-migration generation (up+down = full change management), framework-formatted migrations (Alembic/Flyway), and a guarded "deploy schema" UX with explicit confirm + dry-run preview. + ### P2 — Scale, governance, monetization (justifies a valuation, not just a product) - **Robust async jobs** — `jobs` is DB-queue MVP; harden the valkey worker path (retries/backoff/idempotency) for large-schema introspection at scale. - **Audit log + org multi-tenancy + usage quotas** — table-stakes for enterprise/SOC2 and for billing.