test(fuzz): Atheris + Hypothesis fuzzing for untrusted-input surfaces

backend/app/ddl/export.py

@@ -6,6 +6,21 @@
 DdlDialect = Literal["postgresql", "snowflake"]
 
 
+def _rows(snapshot: dict, key: str) -> list[dict]:
+    """Return ``snapshot[key]`` as a list of dict rows, tolerating junk.
+
+    Snapshot payloads are weakly-typed JSON; a malformed key (missing, or not a
+    list, or holding non-dict entries) must degrade to "no rows" rather than
+    raise. Mirrors the defensive contract already used by the spec/lint
+    generators (``app.spec.reversing._rows`` etc.).
+    """
+
+    value = snapshot.get(key)
+    if not isinstance(value, list):
+        return []
+    return [row for row in value if isinstance(row, dict)]
+
+
 def _normalize_dialect(dialect: str) -> DdlDialect:
     normalized = dialect.lower().replace("_", "-")
     if normalized in ("postgres", "postgresql", "pg"):
@@ -65,7 +80,12 @@ def _normalize_type_text(data_type: str) -> str:
 def _postgres_type_to_snowflake(column: dict) -> str:
     base_type = column.get("domain_base_type")
     if isinstance(base_type, str):
-        return _postgres_type_to_snowflake({**column, "data_type": base_type})
+        # Re-map using the domain's base type, but clear domain_base_type so the
+        # recursion terminates (otherwise a column carrying a string
+        # domain_base_type recurses forever -> RecursionError / DoS).
+        return _postgres_type_to_snowflake(
+            {**column, "data_type": base_type, "domain_base_type": None}
+        )
 
     data_type = column.get("data_type")
     if not isinstance(data_type, str):
@@ -237,11 +257,10 @@ def _column_default_clause(default_expr: object, target: DdlDialect) -> str | No
 
 
 def _snapshot_tables(snapshot: dict) -> list[dict]:
-    relations = snapshot.get("relations", [])
     return [
         r
-        for r in relations
-        if isinstance(r, dict) and r.get("relation_kind") in ("r", "p")
+        for r in _rows(snapshot, "relations")
+        if r.get("relation_kind") in ("r", "p")
     ]
 
 
@@ -455,13 +474,12 @@ def _snapshot_json_to_postgresql_sql(snapshot: dict) -> str:
     Limitations (intentional, MVP): partitioning clauses and some table options are not reconstructed.
     """
 
-    relations = snapshot.get("relations", [])
-    columns = snapshot.get("columns", [])
-    constraints = snapshot.get("constraints", [])
-    indexes = snapshot.get("indexes", [])
+    columns = _rows(snapshot, "columns")
+    constraints = _rows(snapshot, "constraints")
+    indexes = _rows(snapshot, "indexes")
     source_dialect = _snapshot_source_dialect(snapshot)
 
-    tables = [r for r in relations if r.get("relation_kind") in ("r", "p")]
+    tables = _snapshot_tables(snapshot)
 
     cols_by_oid = _group_by_relation(columns)
     constraints_by_oid = _group_by_relation(constraints)
@@ -563,9 +581,9 @@ def _snapshot_json_to_snowflake_sql(snapshot: dict) -> str:
     """Generate Snowflake DDL from a captured PostgreSQL/Snowflake snapshot."""
 
     source_dialect = _snapshot_source_dialect(snapshot)
-    columns = snapshot.get("columns", [])
-    constraints = snapshot.get("constraints", [])
-    indexes = snapshot.get("indexes", [])
+    columns = _rows(snapshot, "columns")
+    constraints = _rows(snapshot, "constraints")
+    indexes = _rows(snapshot, "indexes")
 
     tables = _snapshot_tables(snapshot)
     cols_by_oid = _group_by_relation(columns)

backend/app/dsn_redaction.py

@@ -16,25 +16,58 @@
 )
 
 
+def _split_dsn_best_effort(dsn: str) -> tuple[str, str]:
+    """Extract (netloc, query) from a DSN without ``urlsplit``.
+
+    ``urllib.parse.urlsplit`` raises ``ValueError`` (e.g. "Invalid IPv6 URL")
+    on malformed authorities such as an unbalanced ``[``. Redaction must never
+    crash on hostile input, otherwise the raw, un-redacted error message could
+    still reach a client. This fallback recovers the credential-bearing parts
+    with plain string slicing so embedded secrets are still stripped.
+    """
+
+    remainder = dsn
+    scheme_sep = remainder.find("://")
+    if scheme_sep != -1:
+        remainder = remainder[scheme_sep + 3 :]
+    remainder = remainder.split("#", 1)[0]
+    if "?" in remainder:
+        remainder, query = remainder.split("?", 1)
+    else:
+        query = ""
+    netloc = remainder.split("/", 1)[0]
+    return netloc, query
+
+
 def _password_candidates_from_dsn(dsn: str) -> set[str]:
     candidates: set[str] = set()
-    parsed = urlsplit(dsn)
-    if "://" in dsn and not parsed.netloc:
-        # ponytail: keep urlsplit; only swap the non-RFC scheme so userinfo parses.
-        parsed = urlsplit("http://" + dsn.split("://", 1)[1])
 
-    if parsed.password:
-        candidates.add(parsed.password)
-        candidates.add(quote(parsed.password, safe=""))
+    password: str | None = None
+    try:
+        parsed = urlsplit(dsn)
+        if "://" in dsn and not parsed.netloc:
+            # ponytail: keep urlsplit; only swap the non-RFC scheme so userinfo parses.
+            parsed = urlsplit("http://" + dsn.split("://", 1)[1])
+        netloc = parsed.netloc
+        password = parsed.password
+        query = parsed.query
+    except ValueError:
+        # Malformed DSN (e.g. invalid IPv6 literal). Fall back to best-effort
+        # parsing so any embedded credentials are still redacted.
+        netloc, query = _split_dsn_best_effort(dsn)
+
+    if password:
+        candidates.add(password)
+        candidates.add(quote(password, safe=""))
 
-    if "@" in parsed.netloc:
-        userinfo = parsed.netloc.rsplit("@", 1)[0]
+    if "@" in netloc:
+        userinfo = netloc.rsplit("@", 1)[0]
         if ":" in userinfo:
             raw_password = userinfo.split(":", 1)[1]
             candidates.add(raw_password)
             candidates.add(unquote(raw_password))
 
-    for part in parsed.query.split("&"):
+    for part in query.split("&"):
         key, sep, raw_value = part.partition("=")
         if not sep:
             continue

backend/app/spec/naming_lint.py

@@ -47,6 +47,20 @@
 _VALID_UNQUOTED = re.compile(r"^[a-z_][a-z0-9_$]*$")
 
 
+def _rows(snapshot: dict[str, Any], key: str) -> list[dict[str, Any]]:
+    """Return ``snapshot[key]`` as dict rows, tolerating malformed JSON.
+
+    A key that is missing, not a list, or holds non-dict entries degrades to
+    "no rows" rather than raising -- the same defensive contract used by the
+    other snapshot generators (``app.spec.reversing._rows`` etc.).
+    """
+
+    value = snapshot.get(key)
+    if not isinstance(value, list):
+        return []
+    return [row for row in value if isinstance(row, dict)]
+
+
 def _case_style(name: str) -> str | None:
     """Classify identifier case style, or None if it can't be classified."""
     if re.fullmatch(r"[a-z][a-z0-9]*(_[a-z0-9]+)*", name):
@@ -65,8 +79,8 @@ def _item(category: str, severity: str, target: str, detail: str) -> dict[str, A
 def lint_naming(snapshot: dict[str, Any] | None) -> dict[str, Any]:
     """Return naming-convention findings + a summary, breaking issues first."""
     snapshot = snapshot or {}
-    relations = snapshot.get("relations") or []
-    columns = snapshot.get("columns") or []
+    relations = _rows(snapshot, "relations")
+    columns = _rows(snapshot, "columns")
     rel_by_oid = {r.get("relation_oid"): r for r in relations}
 
     # (label, name) for every identifier: tables and columns.

backend/app/spec/wide_tables.py

@@ -18,15 +18,29 @@
 INFO = "info"
 
 
+def _rows(snapshot: dict[str, Any], key: str) -> list[dict[str, Any]]:
+    """Return ``snapshot[key]`` as dict rows, tolerating malformed JSON.
+
+    Missing / non-list / non-dict-entry payloads degrade to "no rows" rather
+    than raising -- matching the defensive contract of the other snapshot
+    generators (``app.spec.reversing._rows`` etc.).
+    """
+
+    value = snapshot.get(key)
+    if not isinstance(value, list):
+        return []
+    return [row for row in value if isinstance(row, dict)]
+
+
 def detect_wide_tables(
     snapshot: dict[str, Any] | None,
     warn_threshold: int = 40,
     info_threshold: int = 25,
 ) -> dict[str, Any]:
     """Return tables exceeding the column-count thresholds, widest first."""
     snapshot = snapshot or {}
-    relations = snapshot.get("relations") or []
-    columns = snapshot.get("columns") or []
+    relations = _rows(snapshot, "relations")
+    columns = _rows(snapshot, "columns")
 
     # Only ordinary/partitioned tables count (views legitimately project many cols).
     table_oids = {

backend/fuzz/README.md

@@ -0,0 +1,67 @@
+# Fuzzing
+
+Coverage-guided and property-based fuzzing for pg-erd-cloud's untrusted-input
+surfaces. All targets are **pure, deterministic, network-free** functions that
+consume attacker-influenced data (connection strings, driver error messages, and
+DB-introspection *snapshots* that may originate from a hostile database).
+
+## Tools & licenses (permissive only)
+
+| Tool | License | Role |
+| --- | --- | --- |
+| [Atheris](https://github.com/google/atheris) | Apache-2.0 | coverage-guided (libFuzzer) harnesses in `backend/fuzz/` |
+| [Hypothesis](https://github.com/HypothesisWorks/hypothesis) | MPL-2.0 | property tests in `backend/tests/test_fuzz_properties.py` |
+
+Neither is GPL/AGPL. Hypothesis is **not** in the pinned runtime/dev lock, so
+`test_fuzz_properties.py` is `importorskip`-guarded and is skipped by the normal
+hash-locked `pytest` job unless Hypothesis is installed locally.
+
+## Targets (surfaced with CodeGraph)
+
+Discovered via `codegraph init` + `codegraph explore` on the fresh clone:
+
+| Harness | Target | Untrusted input | Invariant |
+| --- | --- | --- | --- |
+| `fuzz_dsn_redaction.py` | `app.dsn_redaction.redact_dsn_error_message` | driver error text + DSN | no crash; **DSN password never leaks verbatim** |
+| `fuzz_dsn_guard.py` | `app.pg_introspect.dsn_guard` parse helpers | DSN query/host/port/IP | only `DsnTargetError` raised; IP parsing total |
+| `fuzz_ddl_export.py` | `app.ddl.export.snapshot_json_to_sql` | schema snapshot + dialect | no crash (bad dialect → `ValueError` only); deterministic |
+| `fuzz_spec_generators.py` | `app.spec.reversing` / `index_design` / `naming_lint` / `wide_tables` | schema snapshot | no crash; deterministic; report summary counts self-consistent |
+
+`_snapshot.py` turns fuzzer bytes into snapshot-shaped dicts (the keys/value
+types the generators branch on) so the fuzzer spends its budget on structure.
+
+## Run locally
+
+Atheris ships wheels for CPython 3.10–3.12 (Linux/macOS):
+
+```bash
+cd backend
+python -m pip install atheris hypothesis
+export PYTHONPATH=.
+
+# Coverage-guided, time-bounded:
+python fuzz/fuzz_dsn_redaction.py   -max_total_time=60 fuzz/corpus/dsn_redaction
+python fuzz/fuzz_dsn_guard.py       -max_total_time=60 fuzz/corpus/dsn_guard
+python fuzz/fuzz_ddl_export.py      -max_total_time=60 fuzz/corpus/ddl_export
+python fuzz/fuzz_spec_generators.py -max_total_time=60 fuzz/corpus/reversing_spec
+
+# Property-based (portable, no native build):
+pytest tests/test_fuzz_properties.py -q
+```
+
+A crash writes a `crash-*` reproducer; re-run the harness with that file as the
+sole argument to replay it.
+
+## CI
+
+The existing backend CI installs from `requirements-dev.lock` with
+`--require-hashes` and runs `pytest -q`. The Hypothesis suite is skipped there
+unless Hypothesis is added to that hash-locked dev set; the Atheris harnesses are
+manual, bounded checks for security/robustness investigations.
+
+## Findings
+
+The initial harness run hardened four latent robustness bugs (see the PR):
+unbounded recursion on a domain-typed column in the Snowflake export, and three
+snapshot generators crashing on a non-list top-level key. Regression coverage
+now lives in these fuzz targets.