feat: add trusted local PostgreSQL snapshot CLI

.github/workflows/ci.yml

@@ -37,11 +37,11 @@ jobs:
           PYTHONPATH: .
         run: mypy app
 
-      - name: Tests (pytest)
+      - name: Tests and coverage (pytest)
         working-directory: backend
         env:
           PYTHONPATH: .
-        run: pytest -q
+        run: pytest -q --cov --cov-report=term-missing --cov-fail-under=100
 
   frontend:
     runs-on: ubuntu-latest

CHANGELOG.md

@@ -1,8 +1,13 @@
 # Changelog
 
 ## Unreleased
+
+- [BE] 🔐 **신뢰된 로컬 PostgreSQL 스냅샷 CLI**: 웹 API의 SSRF 허용 정책을 완화하거나 비밀번호가 포함된 DSN을 노출하지 않고, 명시적으로 검증된 Unix-domain socket을 통해 로컬 데이터베이스 스키마를 JSON으로 캡처할 수 있는 `pg-erd-snapshot` 명령을 추가했습니다. 공용 수집기를 웹·CLI 경로에서 재사용하고 Citus 유무·카탈로그 호환성·출력 및 오류 redaction을 회귀 테스트로 검증하며, 백엔드 CI에서 선택된 핵심 모듈의 100% coverage evidence를 강제합니다.
+- [BE] 🧭 **로컬 스냅샷 연결 시간 초과 처리**: Unix socket 연결이 시간 초과되면 내부 드라이버 메시지를 노출하지 않고 안정적인 실패 코드와 오류 유형만 반환합니다. 연결이 준비되면 같은 명령을 다시 실행하십시오.
+
 - [BE] 🔒 **Cryptography 50+ 보안 경계 갱신**: `pyproject.toml`과 두 hash-locked 요구사항 파일을 동일한 Cryptography 50+ 해석으로 정합화하여 PKCS#7 오류·타이밍 구분으로 인한 CVE-2026-69247 완화를 실제 설치·검증 경로에 반영했습니다.
 - [FE] ⚡ **검색 노드 참조 안정화 및 순차 스냅샷 폴링**: 같은 정규화 검색어와 원본 테이블 데이터에는 장식된 `node.data` 참조를 재사용하여 드래그 중 불필요한 하위 렌더링과 할당을 줄입니다. 스냅샷 폴링은 이전 요청이 끝난 뒤에만 다음 요청을 예약하며, 선택 변경·언마운트 후 도착한 오래된 성공 또는 실패 응답을 무시합니다.
+
 - [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 반영).

README.md

@@ -135,6 +135,22 @@ hypercorn --config python:app.hypercorn_config app.main:app \
   --access-logfile - --error-logfile -
 ```
 
+로컬 마이그레이션 DB를 웹 API의 SSRF 허용 목록에 넣지 않고 스냅샷하려면
+Unix-domain socket 전용 CLI를 사용합니다. 비밀번호가 포함된 DSN을 인자나 로그로
+전달하지 않으며 결과 JSON은 stdout으로만 출력합니다.
+
+```bash
+cd backend
+pg-erd-snapshot \
+  --host /tmp \
+  --database my_local_database \
+  --schema public \
+  --pretty > snapshot.json
+```
+
+원격 PostgreSQL은 기존 연결 API와 host allowlist를 계속 사용합니다. 이 CLI는 TCP
+호스트를 받지 않으므로 웹 API의 loopback/private-network 차단을 우회하지 않습니다.
+
 Snowflake 리버스 엔지니어링을 사용할 개발 환경에서는 백엔드 가상환경에서 선택 의존성을
 함께 설치합니다.
 

backend/app/local_snapshot_cli.py

@@ -0,0 +1,147 @@
+"""Trusted-local PostgreSQL snapshot CLI.
+
+The web API intentionally rejects loopback and private database targets. This
+separate operator CLI accepts only an absolute Unix-domain socket directory, so
+developers can reverse a local migration database without weakening the remote
+API's SSRF boundary or putting a password-bearing DSN in the process list.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Sequence
+
+import asyncpg
+
+from app.pg_introspect.snapshot_collect import collect_postgres_snapshot
+
+_SCHEMA_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_$]{0,62}$")
+
+
+def _socket_directory(value: str) -> str:
+    """Accept an existing absolute directory suitable for a Unix socket."""
+
+    path = Path(value)
+    if not path.is_absolute():
+        raise argparse.ArgumentTypeError(
+            "--host must be an absolute PostgreSQL Unix socket directory"
+        )
+    if not path.is_dir():
+        raise argparse.ArgumentTypeError("--host Unix socket directory does not exist")
+    return str(path)
+
+
+def _schema_name(value: str) -> str:
+    """Accept one unquoted PostgreSQL schema identifier."""
+
+    if not _SCHEMA_RE.fullmatch(value):
+        raise argparse.ArgumentTypeError("--schema is not a valid PostgreSQL identifier")
+    return value
+
+
+def _port(value: str) -> int:
+    """Parse a PostgreSQL port in the valid TCP and Unix-socket range."""
+
+    try:
+        port = int(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("--port must be an integer") from exc
+    if not 1 <= port <= 65535:
+        raise argparse.ArgumentTypeError("--port must be between 1 and 65535")
+    return port
+
+
+def build_parser() -> argparse.ArgumentParser:
+    """Build the argument parser for the trusted-local snapshot CLI."""
+
+    parser = argparse.ArgumentParser(
+        prog="pg-erd-snapshot",
+        description=(
+            "Write a pg-erd-cloud snapshot for a trusted local PostgreSQL Unix socket "
+            "to stdout."
+        ),
+    )
+    parser.add_argument(
+        "--database",
+        default=os.environ.get("PGDATABASE"),
+        required=os.environ.get("PGDATABASE") is None,
+        help="database name (defaults to PGDATABASE)",
+    )
+    host_default = os.environ.get("PGHOST")
+    parser.add_argument(
+        "--host",
+        type=_socket_directory,
+        default=host_default,
+        required=host_default is None,
+        help="absolute Unix socket directory (defaults to PGHOST; no implicit fallback)",
+    )
+    parser.add_argument(
+        "--port",
+        type=_port,
+        default=os.environ.get("PGPORT", "5432"),
+        metavar="1..65535",
+    )
+    parser.add_argument(
+        "--user",
+        default=os.environ.get("PGUSER"),
+        help="database user (defaults to PGUSER or the operating-system user)",
+    )
+    parser.add_argument("--schema", type=_schema_name, default=None)
+    parser.add_argument("--pretty", action="store_true")
+    return parser
+
+
+async def capture_local_snapshot(args: argparse.Namespace) -> dict:
+    """Connect over a validated Unix socket and collect one catalog snapshot."""
+
+    conn = await asyncpg.connect(
+        database=args.database,
+        host=args.host,
+        password="",
+        port=args.port,
+        user=args.user,
+        timeout=10,
+    )
+    try:
+        return await collect_postgres_snapshot(conn, args.schema)
+    finally:
+        await conn.close()
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    """Capture a trusted-local snapshot and write JSON to standard output."""
+
+    args = build_parser().parse_args(argv)
+    try:
+        snapshot = asyncio.run(capture_local_snapshot(args))
+    except (
+        OSError,
+        TimeoutError,
+        asyncio.TimeoutError,
+        asyncpg.PostgresError,
+    ) as exc:
+        print(
+            f"pg-erd-snapshot failed: {type(exc).__name__}",
+            file=sys.stderr,
+        )
+        return 1
+    json.dump(
+        snapshot,
+        sys.stdout,
+        ensure_ascii=False,
+        indent=2 if args.pretty else None,
+        sort_keys=True,
+        separators=None if args.pretty else (",", ":"),
+    )
+    sys.stdout.write("\n")
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())

backend/app/pg_introspect/introspect.py

@@ -1,16 +1,15 @@
+"""Guarded PostgreSQL connectivity and snapshot introspection helpers."""
+
 from __future__ import annotations
 
-import datetime as dt
 import ssl
 from urllib.parse import parse_qsl, urlparse
 
 import asyncpg
 
-from app.pg_introspect import queries
-from app.pg_introspect.column_examples import add_column_examples
 from app.pg_introspect.dsn_guard import validate_postgres_dsn_target
 from app.pg_introspect.forward_ddl import ForwardDdlBatch
-from app.sanitize import sanitize_for_storage
+from app.pg_introspect.snapshot_collect import collect_postgres_snapshot
 
 
 class _ServerHostnameSSLContext(ssl.SSLContext):
@@ -19,11 +18,13 @@ class _ServerHostnameSSLContext(ssl.SSLContext):
     _server_hostname: str
 
     def __new__(cls, server_hostname: str) -> "_ServerHostnameSSLContext":
+        """Create a client TLS context that retains the verified DSN hostname."""
         context = super().__new__(cls, ssl.PROTOCOL_TLS_CLIENT)
         context._server_hostname = server_hostname
         return context
 
     def __init__(self, server_hostname: str) -> None:
+        """Keep the SSL context initialized by ``__new__`` without resetting it."""
         return None
 
     def wrap_bio(
@@ -34,6 +35,7 @@ def wrap_bio(
         server_hostname: str | bytes | None = None,
         session: ssl.SSLSession | None = None,
     ) -> ssl.SSLObject:
+        """Wrap a TLS BIO while forcing certificate verification to the DSN host."""
         return super().wrap_bio(
             incoming,
             outgoing,
@@ -44,11 +46,13 @@ def wrap_bio(
 
 
 def _requires_verified_tls_hostname(dsn: str) -> bool:
+    """Return whether the DSN requests PostgreSQL ``verify-full`` TLS mode."""
     query = dict(parse_qsl(urlparse(dsn).query, keep_blank_values=True))
     return query.get("sslmode", "").lower() == "verify-full"
 
 
 def _verified_tls_context(dsn: str, server_hostname: str) -> ssl.SSLContext:
+    """Build a TLS context from DSN certificate options with hostname binding."""
     query = dict(parse_qsl(urlparse(dsn).query, keep_blank_values=True))
     context = _ServerHostnameSSLContext(server_hostname)
     if query.get("sslrootcert"):
@@ -63,6 +67,7 @@ def _verified_tls_context(dsn: str, server_hostname: str) -> ssl.SSLContext:
 async def _connect_guarded_postgres(
     dsn: str, *, timeout: float
 ) -> asyncpg.Connection:
+    """Validate a DSN and connect only to its resolved, permitted host targets."""
     target = await validate_postgres_dsn_target(dsn)
     connect_host: str | list[str] = (
         target.hosts[0] if len(target.hosts) == 1 else list(target.hosts)
@@ -136,49 +141,6 @@ async def introspect_postgres(dsn: str, schema_filter: str | None) -> dict:
     # Note: avoid logging DSN.
     conn = await _connect_guarded_postgres(dsn, timeout=10)
     try:
-        version = await conn.fetchval("SHOW server_version")
-        schema_name = schema_filter
-        include_system = False
-
-        schemas = await conn.fetch(queries.SCHEMAS_SQL, schema_name, include_system)
-        relations = await conn.fetch(queries.RELATIONS_SQL, schema_name, include_system)
-        columns = await conn.fetch(queries.COLUMNS_SQL, schema_name, include_system)
-        constraints = await conn.fetch(
-            queries.CONSTRAINTS_SQL, schema_name, include_system
-        )
-        indexes = await conn.fetch(queries.INDEXES_SQL, schema_name, include_system)
-        pk_columns = await conn.fetch(
-            queries.PK_COLUMNS_SQL, schema_name, include_system
-        )
-        fk_edges = await conn.fetch(queries.FK_EDGES_SQL, schema_name, include_system)
-        citus_distributed_tables = []
-        has_citus = await conn.fetchval(
-            "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_extension WHERE extname = 'citus')"
-        )
-        if has_citus:
-            try:
-                citus_distributed_tables = await conn.fetch(
-                    queries.CITUS_DISTRIBUTED_TABLES_SQL,
-                    schema_name,
-                    include_system,
-                )
-            except asyncpg.UndefinedTableError:
-                citus_distributed_tables = []
-
-        snapshot = {
-            "captured_at": dt.datetime.now(dt.timezone.utc).isoformat(),
-            "server_version": str(version),
-            "schema_filter": schema_filter,
-            "schemas": [dict(r) for r in schemas],
-            "relations": [dict(r) for r in relations],
-            "columns": add_column_examples([dict(r) for r in columns]),
-            "constraints": [dict(r) for r in constraints],
-            "indexes": [dict(r) for r in indexes],
-            "pk_columns": [dict(r) for r in pk_columns],
-            "fk_edges": [dict(r) for r in fk_edges],
-            "citus_distributed_tables": [dict(r) for r in citus_distributed_tables],
-        }
-
-        return sanitize_for_storage(snapshot)  # type: ignore[return-value]
+        return await collect_postgres_snapshot(conn, schema_filter)
     finally:
         await conn.close()

backend/app/pg_introspect/snapshot_collect.py

@@ -0,0 +1,71 @@
+"""Shared PostgreSQL catalog snapshot collector.
+
+This module deliberately has no application-settings import. Network trust and
+credential policy are established before callers hand it an open connection.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+
+import asyncpg
+
+from app.pg_introspect import queries
+from app.pg_introspect.column_examples import add_column_examples
+from app.sanitize import sanitize_for_storage
+
+
+async def collect_postgres_snapshot(
+    conn: asyncpg.Connection, schema_filter: str | None
+) -> dict:
+    """Collect the canonical snapshot from an authorized PostgreSQL connection."""
+
+    version = await conn.fetchval("SHOW server_version")
+    schema_name = schema_filter
+    include_system = False
+
+    # asyncpg rejects overlapping operations on one Connection. Keep these
+    # catalog reads sequential unless a future caller gives the collector a
+    # pool and an explicit multi-connection consistency policy.
+    schemas = await conn.fetch(queries.SCHEMAS_SQL, schema_name, include_system)
+    relations = await conn.fetch(queries.RELATIONS_SQL, schema_name, include_system)
+    columns = await conn.fetch(queries.COLUMNS_SQL, schema_name, include_system)
+    constraints = await conn.fetch(
+        queries.CONSTRAINTS_SQL, schema_name, include_system
+    )
+    indexes = await conn.fetch(queries.INDEXES_SQL, schema_name, include_system)
+    pk_columns = await conn.fetch(
+        queries.PK_COLUMNS_SQL, schema_name, include_system
+    )
+    fk_edges = await conn.fetch(queries.FK_EDGES_SQL, schema_name, include_system)
+    citus_distributed_tables = []
+    has_citus = await conn.fetchval(
+        "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_extension WHERE extname = 'citus')"
+    )
+    if has_citus:
+        try:
+            citus_distributed_tables = await conn.fetch(
+                queries.CITUS_DISTRIBUTED_TABLES_SQL,
+                schema_name,
+                include_system,
+            )
+        except asyncpg.UndefinedTableError:
+            citus_distributed_tables = []
+
+    snapshot = {
+        "captured_at": dt.datetime.now(dt.timezone.utc).isoformat(),
+        "server_version": str(version),
+        "schema_filter": schema_filter,
+        "schemas": [dict(row) for row in schemas],
+        "relations": [dict(row) for row in relations],
+        "columns": add_column_examples([dict(row) for row in columns]),
+        "constraints": [dict(row) for row in constraints],
+        "indexes": [dict(row) for row in indexes],
+        "pk_columns": [dict(row) for row in pk_columns],
+        "fk_edges": [dict(row) for row in fk_edges],
+        "citus_distributed_tables": [
+            dict(row) for row in citus_distributed_tables
+        ],
+    }
+
+    return sanitize_for_storage(snapshot)  # type: ignore[return-value]