feat: add trusted local PostgreSQL snapshot CLI

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,141 @@
+"""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,
+        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, 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,13 @@
 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):
@@ -136,49 +133,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]

backend/pyproject.toml

@@ -32,6 +32,9 @@ dependencies = [
   "python-multipart>=0.0.32",
 ]
 
+[project.scripts]
+pg-erd-snapshot = "app.local_snapshot_cli:main"
+
 [project.optional-dependencies]
 snowflake = [
   "snowflake-connector-python>=4,<5",