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,129 @@
+"""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:
+    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:
+    if not _SCHEMA_RE.fullmatch(value):
+        raise argparse.ArgumentTypeError("--schema is not a valid PostgreSQL identifier")
+    return value
+
+
+def _port(value: str) -> int:
+    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:
+    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)",
+    )
+    parser.add_argument(
+        "--host",
+        type=_socket_directory,
+        default=os.environ.get("PGHOST", "/tmp"),
+        help="absolute Unix socket directory (defaults to PGHOST or /tmp)",
+    )
+    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_args: dict[str, object] = {
+        "database": args.database,
+        "host": args.host,
+        "port": args.port,
+        "timeout": 10,
+    }
+    if args.user:
+        connect_args["user"] = args.user
+    conn = await asyncpg.connect(**connect_args)
+    try:
+        return await collect_postgres_snapshot(conn, args.schema)
+    finally:
+        await conn.close()
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    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,68 @@
+"""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
+
+    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",

backend/tests/test_local_snapshot_cli.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from app import local_snapshot_cli
+
+
+class FakeConnection:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+def test_socket_directory_requires_existing_absolute_directory(tmp_path: Path) -> None:
+    assert local_snapshot_cli._socket_directory(str(tmp_path)) == str(tmp_path)
+    with pytest.raises(argparse.ArgumentTypeError):
+        local_snapshot_cli._socket_directory("relative/socket")
+    with pytest.raises(argparse.ArgumentTypeError):
+        local_snapshot_cli._socket_directory(str(tmp_path / "missing"))
+
+
+def test_environment_host_default_cannot_bypass_socket_boundary(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setenv("PGDATABASE", "catalog")
+    monkeypatch.setenv("PGHOST", "db.example.com")
+
+    with pytest.raises(SystemExit) as exc_info:
+        local_snapshot_cli.build_parser().parse_args([])
+
+    assert exc_info.value.code == 2
+    assert "absolute PostgreSQL Unix socket directory" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("value", ["not-a-port", "0", "65536"])
+def test_port_rejects_invalid_values(value: str) -> None:
+    with pytest.raises(argparse.ArgumentTypeError):
+        local_snapshot_cli._port(value)
+
+
+def test_schema_name_rejects_sql_fragments() -> None:
+    assert local_snapshot_cli._schema_name("catalog_v2") == "catalog_v2"
+    with pytest.raises(argparse.ArgumentTypeError):
+        local_snapshot_cli._schema_name("public; DROP SCHEMA public")
+
+
+@pytest.mark.asyncio
+async def test_capture_uses_local_connection_without_password_or_dsn(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    captured: dict[str, Any] = {}
+    connection = FakeConnection()
+
+    async def fake_connect(**kwargs: object) -> FakeConnection:
+        captured.update(kwargs)
+        return connection
+
+    async def fake_collect(conn: FakeConnection, schema: str | None) -> dict:
+        assert conn is connection
+        return {"schema_filter": schema, "relations": []}
+
+    monkeypatch.setattr(local_snapshot_cli.asyncpg, "connect", fake_connect)
+    monkeypatch.setattr(
+        local_snapshot_cli,
+        "collect_postgres_snapshot",
+        fake_collect,
+    )
+    args = argparse.Namespace(
+        database="catalog",
+        host=str(tmp_path),
+        port=5432,
+        user="operator",
+        schema="public",
+        pretty=False,
+    )
+
+    snapshot = await local_snapshot_cli.capture_local_snapshot(args)
+
+    assert snapshot == {"schema_filter": "public", "relations": []}
+    assert captured == {
+        "database": "catalog",
+        "host": str(tmp_path),
+        "port": 5432,
+        "timeout": 10,
+        "user": "operator",
+    }
+    assert "dsn" not in captured
+    assert "password" not in captured
+    assert connection.closed is True