diff --git a/backend/app/api/connections.py b/backend/app/api/connections.py index 5689ce62..084dc7fc 100644 --- a/backend/app/api/connections.py +++ b/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)) diff --git a/backend/app/db_introspect.py b/backend/app/db_introspect.py index 934a5cb5..81b89d83 100644 --- a/backend/app/db_introspect.py +++ b/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 diff --git a/backend/app/pg_introspect/introspect.py b/backend/app/pg_introspect/introspect.py index 4fb1ebc4..0ebc55c8 100644 --- a/backend/app/pg_introspect/introspect.py +++ b/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.""" diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 0359799c..e6ae3d12 100644 --- a/backend/app/schemas.py +++ b/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.""" 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)