security(api): reject non-text controls in multiline SQL

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## Unreleased
+- [BE] 🔒 **SQL 요청 제어문자·검증 응답 하드닝**: 멀티라인 DDL의 탭·LF·CR 및 Unicode는 보존하면서 NUL, 기타 비텍스트 C0 제어문자와 DEL을 요청 스키마에서 거부합니다. 모든 422 요청 검증 응답에서 원문 입력을 제거해 SQL·자격 증명 값이 반사되지 않도록 합니다.
 - [BE] 🔒 **공유 export 전 경로 redaction**: 공개 share의 SQL / index-design / reversing-spec export에서 코멘트·`example_value`를 제거합니다. 단위 테스트로 누출을 차단합니다.
 - [BE] 🛠️ **함수 인덱스 중복 오탐 수정**: `lower(email)` 등 expression index를 평문 컬럼 인덱스의 중복으로 잘못 판단하지 않도록 괄호 파서를 강화했습니다.
 - [Docs] README를 상용 기준 기능 설명으로 갱신 (MVP skeleton 표현 제거, share redaction·diff/export 반영).

backend/app/main.py

@@ -5,8 +5,11 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 
 from app.api.annotations import router as annotations_router
 from app.api.api_keys import router as api_keys_router
@@ -65,6 +68,33 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
 
 app = FastAPI(title="pg-erd-cloud backend", lifespan=lifespan)
 
+
+@app.exception_handler(RequestValidationError)
+async def redact_request_validation_input(
+    _: Request, exc: RequestValidationError
+) -> JSONResponse:
+    """Return validation details without reflecting secret-bearing inputs.
+
+    Pydantic includes rejected values in its structured errors by default.
+    API payloads may contain SQL or credentials, so responses retain the
+    location, type, and safe validator message while removing the raw input.
+    """
+
+    errors = []
+    for error in exc.errors():
+        sanitized = {key: value for key, value in error.items() if key != "input"}
+        context = sanitized.get("ctx")
+        if isinstance(context, dict) and "error" in context:
+            sanitized["ctx"] = {
+                **context,
+                "error": str(context["error"]),
+            }
+        errors.append(sanitized)
+    return JSONResponse(
+        status_code=422,
+        content=jsonable_encoder({"detail": errors}),
+    )
+
 CORS_ALLOW_HEADERS = [
     "Authorization",
     "Content-Type",

backend/app/schemas.py

@@ -4,7 +4,7 @@
 import uuid
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 
 class ProjectCreateIn(BaseModel):
@@ -81,6 +81,25 @@ class ApplySqlIn(BaseModel):
     # Default to a rolled-back pre-flight; the caller must opt in to persist.
     dry_run: bool = True
 
+    @field_validator("sql")
+    @classmethod
+    def reject_non_text_controls(cls, value: str) -> str:
+        """Reject transport-unsafe controls while preserving multiline SQL.
+
+        Tab, line feed, and carriage return are valid SQL formatting
+        characters.  Other C0 controls and DEL can corrupt log, parser, or
+        driver boundaries and therefore fail before DDL authorization.  The
+        same applies to the C1 control block (U+007F through U+009F).
+        """
+
+        if any(
+            (codepoint < 0x20 and codepoint not in {0x09, 0x0A, 0x0D})
+            or 0x7F <= codepoint <= 0x9F
+            for codepoint in map(ord, value)
+        ):
+            raise ValueError("SQL contains a disallowed control character")
+        return value
+
 
 class ApplySqlOut(BaseModel):
     """Result of applying forward DDL (DSN-redacted on failure)."""

backend/tests/test_api_connections.py

@@ -7,15 +7,18 @@
 
 import pytest
 from fastapi import FastAPI, HTTPException
+from fastapi.exceptions import RequestValidationError
 from fastapi.testclient import TestClient
 
 from app.api.connections import router
 from app.auth import CurrentUser, get_current_user
 from app.db import get_read_session, get_session
 from app.models import DbConnection
+from app.main import redact_request_validation_input
 from app.security import EncryptedBlob
 
 app = FastAPI()
+app.add_exception_handler(RequestValidationError, redact_request_validation_input)
 app.include_router(router)
 
 
@@ -180,3 +183,23 @@ def test_create_connection_invalid_payload() -> None:
     )
 
     assert response.status_code == 422
+
+
+def test_validation_response_does_not_echo_secret_bearing_sql(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    client = TestClient(app)
+    redaction_marker = "-".join(("never", "log-me"))
+    secret_sql = (
+        "CREATE TABLE audit_record "
+        f"(credential text DEFAULT '{redaction_marker}')\x00;"
+    )
+
+    response = client.post(
+        f"/api/connections/{uuid.uuid4()}/apply-sql",
+        json={"sql": secret_sql, "dry_run": True},
+    )
+
+    assert response.status_code == 422
+    assert redaction_marker not in response.text
+    assert redaction_marker not in caplog.text

backend/tests/test_permissions.py

@@ -33,6 +33,15 @@ async def test_require_project_member_allows_sufficient_role() -> None:
     assert role == "editor"
 
 
+@pytest.mark.asyncio
+async def test_require_project_member_allows_membership_without_minimum_role() -> None:
+    role = await require_project_member(
+        FakeSession("viewer"), uuid.uuid4(), uuid.uuid4()
+    )
+
+    assert role == "viewer"
+
+
 @pytest.mark.asyncio
 async def test_require_project_member_rejects_insufficient_role() -> None:
     with pytest.raises(HTTPException) as exc_info:

backend/tests/test_pooler.py

@@ -36,6 +36,23 @@ def test_build_admin_console_dsn_strips_sqlalchemy_driver() -> None:
     assert ":dummy@" not in dsn
 
 
+def test_build_admin_console_dsn_preserves_plain_postgresql_driver() -> None:
+    username = "u"
+    password_marker = "".join(("dum", "my"))
+    hostname = "localhost"
+    port = 5432
+    database_name = "appdb"
+    source_url = (
+        f"postgresql://{username}:{password_marker}"
+        f"@{hostname}:{port}/{database_name}"
+    )
+
+    dsn, password = build_admin_console_dsn(source_url, "pgcat")
+
+    assert dsn.startswith("postgresql://")
+    assert dsn.endswith("/pgcat")
+    assert password == password_marker
+
 def test_should_route_reads_to_read_only() -> None:
     ro_url = "postgresql+asyncpg://u:p@localhost:5432/ro"
 

backend/tests/test_schema_validation.py

@@ -3,7 +3,12 @@
 import pytest
 from pydantic import ValidationError
 
-from app.schemas import ConnectionCreateIn, ProjectCreateIn, ProjectMemberAddIn
+from app.schemas import (
+    ApplySqlIn,
+    ConnectionCreateIn,
+    ProjectCreateIn,
+    ProjectMemberAddIn,
+)
 
 
 def test_project_name_length_is_bounded() -> None:
@@ -37,3 +42,35 @@ def test_conn_name_rejects_control_characters() -> None:
         ConnectionCreateIn(conn_name="my\x00conn", dsn="postgresql://localhost/db")
     with pytest.raises(ValidationError):
         ConnectionCreateIn(conn_name="my\nconn", dsn="postgresql://localhost/db")
+
+
+def test_apply_sql_preserves_multiline_unicode_transport_characters() -> None:
+    sql = 'CREATE TABLE "注文" (\n\t"識別子" text\r\n);'
+
+    assert ApplySqlIn(sql=sql).sql == sql
+
+
+@pytest.mark.parametrize(
+    "codepoint",
+    [
+        *range(0x00, 0x09),
+        0x0B,
+        0x0C,
+        *range(0x0E, 0x20),
+        *range(0x7F, 0xA0),
+    ],
+)
+@pytest.mark.parametrize("position", ["beginning", "middle", "end"])
+def test_apply_sql_rejects_non_text_controls_at_every_position(
+    codepoint: int, position: str
+) -> None:
+    control = chr(codepoint)
+    safe_sql = "CREATE TABLE audit_record (secret_token text);"
+    values = {
+        "beginning": control + safe_sql,
+        "middle": safe_sql[:20] + control + safe_sql[20:],
+        "end": safe_sql + control,
+    }
+
+    with pytest.raises(ValidationError, match="disallowed control character"):
+        ApplySqlIn(sql=values[position])

backend/tests/test_security_headers.py

@@ -4,6 +4,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.testclient import TestClient
 from starlette.requests import Request
+from starlette.responses import Response
 
 from app import security_headers
 from app.csrf import CSRF_HEADER_NAME
@@ -164,3 +165,26 @@ def test_csp_path_normalization_handles_double_slash() -> None:
     }
     request = Request(scope)
     assert security_headers._should_apply_csp(request) is False
+
+
+def test_existing_security_header_is_not_overwritten() -> None:
+    scope = {
+        "type": "http",
+        "asgi": {"version": "3.0"},
+        "http_version": "1.1",
+        "method": "GET",
+        "scheme": "http",
+        "path": "/api/ping",
+        "raw_path": b"/api/ping",
+        "query_string": b"",
+        "headers": [],
+        "client": ("127.0.0.1", 12345),
+        "server": ("testserver", 80),
+        "root_path": "",
+    }
+    request = Request(scope)
+    response = Response(headers={"X-Frame-Options": "SAMEORIGIN"})
+
+    security_headers.apply_security_headers(request, response)
+
+    assert response.headers["X-Frame-Options"] == "SAMEORIGIN"

docs/security/apply-sql-transport-validation.md

@@ -0,0 +1,48 @@
+# Apply SQL transport validation
+
+## Status and boundary
+
+Implemented. `ApplySqlIn` rejects U+0000–U+0008, U+000B, U+000C,
+U+000E–U+001F, and U+007F–U+009F before authorization or database access.
+Horizontal tab, line feed, and carriage return remain accepted so ordinary
+multiline SQL is not damaged. The existing 262,144-character request limit and
+downstream PostgreSQL DDL allowlist remain authoritative.
+
+This validation protects transport, parser, driver, and audit-log integrity. It
+is not SQL-injection prevention and does not expand the permitted DDL grammar.
+Printable Unicode and SQL metacharacters are preserved for the dialect-owned
+parser to classify.
+
+## Failure privacy
+
+FastAPI request-validation errors normally carry the rejected `input`. The
+application-wide request-validation handler removes that field before forming
+a 422 response. It retains the field location, error type, and safe diagnostic
+message. Middleware records request metadata and status only; it does not log
+the rejected body. This boundary also prevents connection strings and other
+secret-bearing request fields from being reflected by validation responses.
+
+## Acceptance evidence
+
+- Every rejected C0, DEL, and C1 code point is tested at the beginning, middle,
+  and end of a realistic DDL request.
+- Tab, LF, CR, quoted Unicode identifiers, and existing length validation are
+  preserved.
+- An HTTP-boundary regression proves a secret-bearing SQL literal appears in
+  neither the response nor captured logs.
+- The conservative SQL parser/allowlist remains a separate execution gate.
+
+## References
+
+Bray, T. (2017). *The JavaScript Object Notation (JSON) data interchange
+format* (RFC 8259). RFC Editor. https://www.rfc-editor.org/rfc/rfc8259.html
+This defines the JSON string transport rules whose rejected-input diagnostics must not reflect raw request values.
+
+PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation:
+Lexical structure*. https://www.postgresql.org/docs/18/sql-syntax-lexical.html
+This preserves PostgreSQL-owned lexical authority, including multiline formatting and quoted identifiers, after transport validation.
+
+Unicode Consortium. (2025). *The Unicode standard, version 17.0: Chapter 23—
+Special areas and format characters*.
+https://www.unicode.org/versions/Unicode17.0.0/core-spec/chapter-23/
+This supports distinguishing transport-unsafe control characters from printable Unicode that must remain available to the SQL parser.