Merge branch 'main' into fix/docs-explorer-auth-note

2026-08-29 04:26:20 +00:00 · 2026-08-18 19:00:47 +05:30
parent 5109c6fab2 a1194a155d
commit 063f447202
22 changed files with 983 additions and 119 deletions
@@ -176,6 +176,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 ### Security

+- **Tarball restore path traversal, latent SQL injection, DNS-rebinding TOCTOU in the shared SSRF guard, stored XSS in report generation, and unvalidated SPARQL object IRIs in AnzoStore** (#1079) by @KaifAhmad1
+  - `semantica backup restore`'s tar extraction (`cli.py`) stripped only the literal `semantica-backup/` prefix and called `tar.extract()` with no path-containment check, no symlink/hardlink validation, and (on Python <3.12) no extraction filter — a crafted archive member (`../../<file>`, or a symlink pointing outside the restore root) could write arbitrary files above the restore directory. Every member is now validated for resolved-path containment before extraction, symlink/hardlink targets are rejected both lexically (absolute path, `..` segments) and by resolution, and `filter="data"` is applied on Python ≥3.12
+  - `DataExporter.export_table_data()` (`db_ingestor.py`) was missing the `text` import from `sqlalchemy` — a `NameError` that made the method non-functional, but latently: the query it built from raw f-string interpolation of `table_name`/`schema`/`where`/`order_by` was already injectable, so fixing the import alone (without also fixing the injection) would have silently armed it. Both are fixed together: the import is restored, `table_name`/`schema` are now validated against a strict identifier allowlist, and `where`/`order_by` are checked against a blocklist (statement separators, comments, UNION, DDL/DML keywords, time-based blind-injection primitives, schema-enumeration terms). This is a blocklist, not a grammar — it closes the concrete UNION-exfiltration path and common injection primitives, but a boolean-blind subquery using none of the blocked keywords could still get through; `where`/`order_by` must be treated as trusted/operator input, not exposed to untrusted end users, and the docstrings now say so explicitly
+  - `request_with_ssrf_guard()` (`ssrf.py`) validated a hostname's resolved IPs, then let the underlying HTTP client re-resolve the same hostname independently at connect time — a low-TTL or DNS-rebinding answer could differ between the two lookups, so a hostname that validated as public could still connect to a private/internal address. Ported the IP-pinning pattern already used by `explorer/routes/ontology.py`'s `_make_pinned_session` into the shared ingest guard: the one resolution that decides accept/reject is now also the one the connection is pinned to, via a custom `HTTPAdapter` that presents the real hostname over TLS SNI / Host header while connecting only to the validated IPs. Also closes the RFC 6598 Carrier-Grade NAT gap noted as a known limitation in #905/#868: `100.64.0.0/10` is now in `BLOCKED_NETWORKS`
+  - `ReportGenerator._generate_html()` (`export/report_generator.py`) f-string-interpolated report title/summary/metrics into HTML with no escaping — an ingested entity or document whose content flowed into a report (e.g. `<img src=x onerror=...>`) executed as stored XSS when the report was opened. All interpolated values are now `html.escape()`d
+  - `AnzoStore._format_object_for_sparql()` (`triplet_store/anzo_store.py`) validated the subject/predicate of a triplet via `sparql_escaping.validate_uri()` before interpolating them into a SPARQL `INSERT DATA` clause, but delegated the **object** position to a separate formatter that wrapped it as `<{obj}>` without the same validation — an object value containing `>`/`}`/`{`/`"` could close the intended `<...>` token early and inject additional SPARQL Update operations. The Blazegraph/RDF4J backends were hardened for the equivalent gap previously; Anzo's object position now goes through the same `validate_uri()` check
+  - Also hardened in the same pass: Apache AGE's `create_index()` `index_type` parameter is now allowlisted (was interpolated raw into a `USING` clause); Neo4j's `limit` is now explicitly validated (raises `ValidationError` for non-integer input instead of falling through to a generic `ProcessingError`); the `ffprobe` metadata-extraction subprocess call is guarded against a filename starting with `-` being parsed as an option; the MCP server no longer echoes raw exception text to JSON-RPC clients, logging full details server-side and returning a generic message plus the exception class name instead
+  - **Fixed during review** (@KaifAhmad1): the SSRF IP-pinning change introduced a connection-pool leak of its own — `requests.Session.mount()` silently drops whatever adapter it replaces without closing it, so a multi-hop redirect chain on a reused session leaked one pooled connection per hop. Pinned adapters are now tagged and explicitly closed before being replaced, both per-hop and on final restore
+  - **Fixed during review** (@KaifAhmad1): mounting a pinned adapter and setting a Host header on a caller-supplied `Session` is not inherently thread-safe — two guarded calls sharing the same session from different threads could interleave their mount/restore cycles. Added a per-session lock (`_get_session_lock`) so concurrent guarded calls on the same session now serialize instead of racing; verified with a two-thread test showing correct serialization and zero cross-contamination of per-request Host headers
+  - **Fixed during automated PR review** (Qodo): `export_table_data()`'s new identifier/fragment validation raised `ValidationError` from inside a `try` whose blanket `except Exception` re-wrapped it as `ProcessingError`, masking the distinction between "bad input" and "the export itself failed" that callers rely on elsewhere in this module. Added the `except ValidationError: raise` guard already used by its sibling methods
+  - **Fixed during automated PR review** (Qodo): on a hop where IP pinning doesn't apply (`allow_private_ips=True`), `_apply_connection_pin()` unconditionally popped the session's `Host` header instead of restoring whatever it was before pinning touched it — a caller-supplied session carrying its own legitimate `Host` override (e.g. fronting a private endpoint under a different name) had that override silently dropped for the in-flight request, only reappearing afterward via the outer `finally` restore. It now restores the session's own pre-call header state (set back if present, popped only if it was truly absent) instead of always popping
+  - **Fixed during automated PR review** (Qodo): the `where`/`order_by` blocklist matched keywords/punctuation inside properly quoted string literals and identifiers too, so legitimate data like `status = 'union'` or `name = 'a--b'` was rejected as if it were SQL syntax. The blocklist now runs against a copy with quoted-literal contents masked out (`_mask_sql_literals`) — a malformed/unterminated quote sequence doesn't match the masking pattern and is left fully exposed to the blocklist, so this closes false positives without opening a masking-based bypass; the fragment actually used in the query is unchanged
+  - Re-ran each finding's proof-of-concept (or an equivalent adversarial test) against the fix and confirmed it is blocked: tar path/symlink traversal (both lexical and resolved-path forms), SQL UNION exfiltration and identifier breakout, DNS-rebinding TOCTOU (including under a configured `HTTP_PROXY`, which the pinning adapter also rejects outright since a proxy would resolve DNS itself), stored XSS, and the AnzoStore SPARQL injection
+  - `pytest tests/ingest/`: 266 passed, 2 skipped (10 pre-existing failures unrelated to this change — identical failure set confirmed on unmodified `main`); full regression sweep across `graph_store`, `export`, `triplet_store`, `parse`, and backup/restore: 313 passed
+
 - **`Authorization`/`Proxy-Authorization` credentials could leak to a different origin across HTTP redirects, and several ingest paths bypassed the shared SSRF/redirect guard entirely** (#1067, closes #947) by @Sameer6305, reviewed by @KaifAhmad1
  - `request_with_ssrf_guard()` previously only stripped sensitive headers from per-request `kwargs["headers"]` on a cross-origin redirect; session-level `Authorization`/`Proxy-Authorization` headers, `session.auth`, and `session.trust_env` (`.netrc` lookup) could all still resurrect credentials on the hop to a foreign origin. All five credential sources are now stripped case-insensitively, kept stripped for the remainder of a multi-hop redirect chain (no resurrection even if a later hop returns to the original host), and unconditionally restored via `finally` — including on exceptions and redirect-limit errors
  - `MCPClient._send_request_http()` and `PublicAPIIngestor.detect_public_api()`/`ingest_public_api()` called `httpx.post()`/`requests.post()`/`session.request()` directly, bypassing `request_with_ssrf_guard()` entirely. Both now route through the shared guard, including when `validate_no_auth=False`
@@ -2,7 +2,15 @@

 <img src="Semantica Logo.png" alt="Semantica" width="420"/>

-<a href="https://trendshift.io/repositories/18986?utm_source=repository-badge&amp;utm_medium=badge&amp;utm_campaign=badge-repository-18986" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/repositories/18986" alt="semantica-agi%2Fsemantica | Trendshift" width="250" height="55"/></a>
+<div style="display:flex; gap:10px; align-items:center; flex-wrap:wrap;">
+  <a href="https://trendshift.io/repositories/18986?utm_source=repository-badge&amp;utm_medium=badge&amp;utm_campaign=badge-repository-18986" target="_blank" rel="noopener noreferrer">
+    <img src="https://trendshift.io/api/badge/repositories/18986" alt="semantica-agi/semantica | Trendshift" width="250" height="55"/>
+  </a>
+
+  <a href="https://trendshift.io/repositories/18986?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-18986" target="_blank" rel="noopener noreferrer">
+    <img src="https://trendshift.io/api/badge/trendshift/repositories/18986/weekly?language=Python" alt="semantica-agi/semantica | Trendshift" width="250" height="55"/>
+  </a>
+</div>

 ### Graph-Native Infrastructure for Context and Accountable AI Systems

@@ -295,17 +303,10 @@ graph.add_causal_relationship(d1, d2, relationship_type="CAUSED")
 prov.track_entity("patient_P4821", source="ehr/medication_orders_2024.json",
                  metadata={"extractor": "NamedEntityRecognizer"})

-# Export W3C PROV-O for regulator submission - RDFExporter expects
-# {"entities": [...], "relationships": [...]}, so map ContextGraph.to_dict()'s
-# {"nodes": [...], "edges": [...]} shape onto it first
-graph_dict = graph.to_dict()
-kg = {
-    "entities": [{"id": n["id"], "type": n["type"], "text": n["content"]} for n in graph_dict["nodes"]],
-    "relationships": [
-        {"source_id": e["source"], "target_id": e["target"], "type": e["type"]}
-        for e in graph_dict["edges"]
-    ],
-}
+# Export W3C PROV-O for regulator submission - to_kg_dict() is the official
+# adapter that emits the {"entities": [...], "relationships": [...]} /
+# source_id shape RDFExporter expects, so no manual field mapping is needed
+kg = graph.to_kg_dict()
 RDFExporter().export(kg, "audit_trail.ttl", format="turtle")
 ```

@@ -879,20 +880,14 @@ fact = BiTemporalFact(
    recorded_at=datetime(2024, 3, 5),
 )

-# Query facts valid within a time window - query_time_range() expects
-# {"relationships": [...]} with source_id/target_id keys, which differs from
-# ContextGraph.to_dict()'s {"nodes", "edges"} shape, so map it first
-graph_dict = graph.to_dict()
-kg_relationships = {
-    "relationships": [
-        {**e, "source_id": e["source"], "target_id": e["target"]}
-        for e in graph_dict["edges"]
-    ]
-}
+# Query facts valid within a time window - to_kg_dict() is the official
+# adapter that emits {"entities", "relationships"} with source_id/target_id
+# keys, the shape query_time_range() expects (no manual mapping required)
+kg = graph.to_kg_dict()

 tq = TemporalGraphQuery()
 facts_in_window = tq.query_time_range(
-    kg_relationships, query="valid_facts", start_time="2024-01-01", end_time="2024-12-31"
+    kg, query="valid_facts", start_time="2024-01-01", end_time="2024-12-31"
 )

 # Normalize natural language temporal expressions - returns a (start, end) range
@@ -93,7 +93,14 @@ def _handle_tools_call(req_id: Any, params: dict) -> dict:
        result = tool["_handler"](args)
    except Exception as exc:
        log.exception("Tool %s raised an exception", name)
-        return _err(req_id, _INTERNAL_ERROR, str(exc))
+        # The exception's class name (e.g. "ValidationError", "TimeoutError")
+        # is safe to surface — unlike str(exc), it never carries paths,
+        # connection strings, or other internal detail — and lets the
+        # client distinguish failure kinds without a full message.
+        return _err(
+            req_id, _INTERNAL_ERROR,
+            f"Tool '{name}' failed ({type(exc).__name__}). See server logs for details.",
+        )

    # MCP spec: content must be a list of content items
    return _ok(req_id, {
@@ -171,7 +178,10 @@ class SemanticaMCPServer:
            log.exception("Unhandled error in method %s", method)
            if req_id is None:
                return None
-            return _err(req_id, _INTERNAL_ERROR, str(exc))
+            return _err(
+                req_id, _INTERNAL_ERROR,
+                f"Method '{method}' failed ({type(exc).__name__}). See server logs for details.",
+            )

    # ------------------------------------------------------------------
    def run(self) -> None:
@@ -20,7 +20,7 @@ if sys.platform == "win32":
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")

 from dataclasses import asdict, dataclass, field, is_dataclass
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Sequence, Tuple

 import yaml
@@ -3933,15 +3933,68 @@ def backup_restore(cli_ctx: CLIContext, source: str, local_dry: bool) -> None:

        try:
            if _tf.is_tarfile(str(work_path)):
-                restore_root = Path.cwd()
+                restore_root = Path.cwd().resolve()
                with _tf.open(str(work_path), "r:*") as tar:
                    # Dry-run listing was already handled above; extract now
                    for member in tar.getmembers():
                        # Strip the leading "semantica-backup/" prefix
                        member.name = member.name.replace("semantica-backup/", "", 1)
-                        if member.name:
-                            tar.extract(member, path=str(restore_root))
-                            console.print(f"  restored: {member.name}")
+                        if not member.name:
+                            continue
+
+                        # Reject members whose resolved path escapes the
+                        # restore root (path traversal / absolute paths),
+                        # regardless of the "semantica-backup/" prefix.
+                        member_path = (restore_root / member.name).resolve()
+                        try:
+                            member_path.relative_to(restore_root)
+                        except ValueError:
+                            raise click.ClickException(
+                                f"Refusing to restore '{member.name}': "
+                                "path escapes the restore directory."
+                            )
+
+                        # Reject symlink/hardlink members whose target
+                        # escapes the restore root. Checked two ways:
+                        # lexically (linkname itself, so an absolute path or
+                        # a literal ".." segment is rejected outright, with
+                        # no dependence on what else does or doesn't already
+                        # exist on disk) and by resolution (catches any
+                        # remaining traversal the lexical check misses).
+                        if member.issym() or member.islnk():
+                            linkname = member.linkname or ""
+                            linkname_parts = PurePosixPath(
+                                linkname.replace("\\", "/")
+                            ).parts
+                            if (
+                                not linkname
+                                or os.path.isabs(linkname)
+                                or PureWindowsPath(linkname).is_absolute()
+                                or ".." in linkname_parts
+                            ):
+                                raise click.ClickException(
+                                    f"Refusing to restore '{member.name}': "
+                                    "link target is absolute or traverses "
+                                    "out of the archive."
+                                )
+                            link_target = (
+                                member_path.parent / linkname
+                            ).resolve()
+                            try:
+                                link_target.relative_to(restore_root)
+                            except ValueError:
+                                raise click.ClickException(
+                                    f"Refusing to restore '{member.name}': "
+                                    "link target escapes the restore directory."
+                                )
+
+                        extract_kwargs: Dict[str, Any] = {"path": str(restore_root)}
+                        if hasattr(_tf, "data_filter"):
+                            # Python >=3.12: also reject device files, and
+                            # further harden the traversal/ownership checks.
+                            extract_kwargs["filter"] = "data"
+                        tar.extract(member, **extract_kwargs)
+                        console.print(f"  restored: {member.name}")
            elif src.is_dir():
                restore_root = Path.cwd()
                for f in src.rglob("*"):
@@ -2449,6 +2449,97 @@ class ContextGraph:
                },
            }

+    def to_kg_dict(self, entities_only: bool = False) -> Dict[str, Any]:
+        """Export graph in the canonical knowledge-graph shape.
+
+        This is the official adapter that converts the ContextGraph's internal
+        ``{"nodes", "edges"}`` / ``source`` representation into the
+        ``{"entities", "relationships"}`` / ``source_id`` shape expected by
+        downstream consumers such as
+        :class:`~semantica.export.rdf_exporter.RDFExporter` and
+        :meth:`~semantica.kg.temporal_query.TemporalGraphQuery.query_time_range`.
+
+        Users no longer need to hand-map field names between APIs.
+
+        Args:
+            entities_only: If True, only nodes whose ``node_type`` is
+                ``"entity"`` are exported as entities. When False (default),
+                every node is exported. Relationships whose endpoints are not
+                in the exported entity set are dropped to avoid dangling
+                references in downstream consumers.
+
+        Returns:
+            dict: A knowledge-graph dictionary with:
+                - ``entities``: list of ``{"id", "text", "type", "properties",
+                  "metadata"}`` (plus ``valid_from`` / ``valid_until`` when set)
+                - ``relationships``: list of ``{"source_id", "target_id",
+                  "type", "weight", "id", "familyId"}`` (plus ``metadata`` and
+                  ``valid_from`` / ``valid_until`` when set)
+                - ``statistics``: ``{"entity_count", "relationship_count"}``
+        """
+        with self._lock:
+            entities_out = []
+            for n in self.nodes.values():
+                if entities_only and n.node_type != "entity":
+                    continue
+                # Normalize the entity id to ``str`` so it matches ContextEdge,
+                # which coerces its endpoints to ``str`` in ``__post_init__``.
+                # Without this, non-string node ids (e.g. numeric ids loaded via
+                # ``from_dict``) would fail the ``valid_ids`` membership check
+                # below and silently drop otherwise-valid relationships.
+                entity_id = str(n.node_id)
+                entity: Dict[str, Any] = {
+                    "id": entity_id,
+                    "text": n.content,
+                    "type": n.node_type,
+                    # ``properties`` / ``metadata`` may be ``None`` when a node
+                    # was loaded from JSON containing an explicit ``null``;
+                    # guard with ``or {}`` so ``dict(...)`` never raises.
+                    "properties": dict(n.properties or {}),
+                    "metadata": dict(n.metadata or {}),
+                }
+                if n.valid_from is not None:
+                    entity["valid_from"] = n.valid_from
+                if n.valid_until is not None:
+                    entity["valid_until"] = n.valid_until
+                entities_out.append(entity)
+
+            # When only entity nodes are exported, drop relationships whose
+            # endpoints were filtered out so downstream consumers never see a
+            # source_id/target_id that is absent from ``entities``.
+            valid_ids = {e["id"] for e in entities_out} if entities_only else None
+
+            relationships_out = []
+            for e in self.edges:
+                if valid_ids is not None and (
+                    e.source_id not in valid_ids or e.target_id not in valid_ids
+                ):
+                    continue
+                rel: Dict[str, Any] = {
+                    "id": e.edge_id,
+                    "familyId": e.family_id or e.edge_id,
+                    "source_id": e.source_id,
+                    "target_id": e.target_id,
+                    "type": e.edge_type,
+                    "weight": e.weight,
+                }
+                if e.metadata:
+                    rel["metadata"] = dict(e.metadata)
+                if e.valid_from is not None:
+                    rel["valid_from"] = e.valid_from
+                if e.valid_until is not None:
+                    rel["valid_until"] = e.valid_until
+                relationships_out.append(rel)
+
+            return {
+                "entities": entities_out,
+                "relationships": relationships_out,
+                "statistics": {
+                    "entity_count": len(entities_out),
+                    "relationship_count": len(relationships_out),
+                },
+            }
+
    def from_dict(self, graph_dict: Dict[str, Any]) -> None:
        """Load graph from dictionary format."""
        # Clear existing graph
@@ -23,6 +23,7 @@ Author: Semantica Contributors
 License: MIT
 """

+import html
 import json
 from datetime import datetime
 from pathlib import Path
@@ -362,7 +363,7 @@ class ReportGenerator:
            '  <meta name="viewport" content="width=device-width, initial-scale=1.0">'
        )
        title = data.get("title", "Report")
-        lines.append(f"  <title>{title}</title>")
+        lines.append(f"  <title>{html.escape(str(title))}</title>")
        lines.append("  <style>")
        lines.append("    body { font-family: Arial, sans-serif; margin: 20px; }")
        lines.append("    h1 { color: #333; }")
@@ -380,11 +381,14 @@ class ReportGenerator:

        # Title
        title = data.get("title", "Report")
-        lines.append(f"  <h1>{title}</h1>")
+        lines.append(f"  <h1>{html.escape(str(title))}</h1>")

        # Generated at
        if "generated_at" in data:
-            lines.append(f'  <p><strong>Generated:</strong> {data["generated_at"]}</p>')
+            lines.append(
+                f'  <p><strong>Generated:</strong> '
+                f'{html.escape(str(data["generated_at"]))}</p>'
+            )

        # Summary
        if "summary" in data:
@@ -393,10 +397,13 @@ class ReportGenerator:
            if isinstance(summary, dict):
                lines.append("  <ul>")
                for key, value in summary.items():
-                    lines.append(f"    <li><strong>{key}:</strong> {value}</li>")
+                    lines.append(
+                        f"    <li><strong>{html.escape(str(key))}:</strong> "
+                        f"{html.escape(str(value))}</li>"
+                    )
                lines.append("  </ul>")
            else:
-                lines.append(f"  <p>{summary}</p>")
+                lines.append(f"  <p>{html.escape(str(summary))}</p>")

        # Metrics
        if "metrics" in data:
@@ -483,7 +490,10 @@ class ReportGenerator:
            else:
                value_str = str(value)

-            lines.append(f"    <tr><td>{key}</td><td>{value_str}</td></tr>")
+            lines.append(
+                f"    <tr><td>{html.escape(str(key))}</td>"
+                f"<td>{html.escape(value_str)}</td></tr>"
+            )

        lines.append("  </table>")

@@ -66,6 +66,12 @@ except (ImportError, OSError):
 # Helpers
 # ---------------------------------------------------------------------------

+# create_index's index_type reaches a raw SQL keyword position (`USING
+# {index_type}`) that can't be bound as a query parameter; only the
+# documented, PostgreSQL-recognized types are allowed through.
+_ALLOWED_INDEX_TYPES = frozenset({"btree", "gin", "hash", "gist", "brin"})
+
+
 def _sanitize_label(label: str) -> str:
    """
    Sanitize a Cypher label to prevent injection.
@@ -1214,6 +1220,11 @@ class ApacheAgeStore:
            safe_label = _sanitize_label(label)
            if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", property_name):
                raise ValidationError(f"Invalid property name: '{property_name}'")
+            if index_type not in _ALLOWED_INDEX_TYPES:
+                raise ValidationError(
+                    f"Invalid index_type: {index_type!r}. "
+                    f"Allowed: {sorted(_ALLOWED_INDEX_TYPES)}"
+                )

            index_name = options.get(
                "index_name", f"idx_{self.graph_name}_{safe_label}_{property_name}"
@@ -503,6 +503,16 @@ class Neo4jStore:
        Returns:
            List of matching nodes
        """
+        # LIMIT can't be bound as a query parameter in a way Neo4j accepts
+        # here, so it's interpolated directly; validate explicitly rather
+        # than trust the `limit: int` type hint, which Python doesn't
+        # enforce at runtime. Done outside the try/except below so a bad
+        # limit raises ValidationError, not a generic ProcessingError.
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid limit: {limit!r}") from exc
+
        try:
            # Build query
            if labels:
@@ -698,6 +708,15 @@ class Neo4jStore:
        Returns:
            List of matching relationships
        """
+        # See get_nodes: LIMIT is interpolated directly, so validate
+        # explicitly rather than trust the unenforced `limit: int` hint,
+        # outside the try/except below so a bad limit raises
+        # ValidationError, not a generic ProcessingError.
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid limit: {limit!r}") from exc
+
        try:
            type_filter = f":{sanitize_identifier(rel_type, 'relationship type')}" if rel_type else ""

@@ -29,6 +29,7 @@ License: MIT
 """

 import json
+import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -38,6 +39,96 @@ from ..utils.exceptions import ProcessingError, ValidationError
 from ..utils.logging import get_logger
 from ..utils.progress_tracker import get_progress_tracker

+_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+# Fragments that turn a filter/order clause into a second statement, a
+# data-exfiltration UNION, a time-based blind-injection oracle, or schema
+# enumeration, rather than a boolean/ordering expression.
+_SQL_FRAGMENT_BLOCKLIST_RE = re.compile(
+    r";|--|/\*|\*/|\bunion\b|\binsert\b|\bupdate\b|\bdelete\b|\bdrop\b|"
+    r"\balter\b|\bcreate\b|\bexec\b|\bexecute\b|\bgrant\b|\brevoke\b|"
+    r"\battach\b|\bpragma\b|\bxp_\w+|\bsp_\w+|\binto\s+outfile\b|\bload_file\b|"
+    r"\bsleep\s*\(|\bbenchmark\s*\(|\bpg_sleep\s*\(|\bwaitfor\b|"
+    r"\bdbms_\w+|\butl_\w+|\binformation_schema\b|\bpg_catalog\b",
+    re.IGNORECASE,
+)
+
+# SQL single-quoted string literals (''  is the standard escaped-quote) and
+# double-quoted identifiers (""  likewise) — matched only when properly
+# closed, so a malformed/unterminated quote sequence is left alone and
+# still hits the blocklist above rather than being treated as "inside a
+# literal" and skipped.
+_SQL_STRING_LITERAL_RE = re.compile(r"'(?:[^']|'')*'")
+_SQL_QUOTED_IDENTIFIER_RE = re.compile(r'"(?:[^"]|"")*"')
+
+
+def _mask_sql_literals(fragment: str) -> str:
+    """Blank the contents of quoted literals so they can't trip the blocklist.
+
+    A legitimate value or quoted identifier that happens to contain a
+    blocked word or character as *data* — e.g. ``status = 'union'`` or
+    ``"my--column" = 1`` — is not SQL syntax and shouldn't be rejected as
+    if it were. Only the quoted span's interior is replaced (with `?`,
+    keeping the surrounding quotes and the fragment's length/positions
+    intact for the error message); text outside any properly closed quote
+    is passed through unchanged and still fully scrutinized.
+    """
+    fragment = _SQL_STRING_LITERAL_RE.sub(
+        lambda m: "'" + "?" * (len(m.group(0)) - 2) + "'", fragment
+    )
+    fragment = _SQL_QUOTED_IDENTIFIER_RE.sub(
+        lambda m: '"' + "?" * (len(m.group(0)) - 2) + '"', fragment
+    )
+    return fragment
+
+
+def _validate_sql_identifier(name: str, kind: str) -> str:
+    """Validate a table/schema name used as a raw SQL identifier.
+
+    ``export_table_data`` interpolates *name* directly into the query text
+    (SQLAlchemy has no bind-parameter syntax for identifiers), so anything
+    outside a plain alphanumeric/underscore identifier is a potential
+    breakout of the surrounding ``"..."`` quoting.
+    """
+    if not isinstance(name, str) or not _IDENTIFIER_RE.match(name):
+        raise ValidationError(
+            f"Invalid {kind}: {name!r}. Must start with a letter or "
+            "underscore and contain only alphanumeric characters and "
+            "underscores."
+        )
+    return name
+
+
+def _validate_sql_fragment(fragment: str, kind: str) -> str:
+    """Reject WHERE/ORDER BY fragments that smuggle a second statement.
+
+    These clauses can't be bound as query parameters (they're arbitrary
+    boolean/ordering expressions, not values), so this blocks the concrete
+    injection primitives (statement separators, comments, UNION, DML/DDL
+    keywords, time-based blind oracles, schema enumeration) rather than
+    parameterizing.
+
+    This is a blocklist, not a grammar: it cannot exhaustively prove
+    *fragment* is safe, only reject known-dangerous constructs, so a
+    boolean-blind subquery expressed with none of the blocked keywords
+    (e.g. ``id = (SELECT 1 FROM t WHERE ...)``) still passes. ``where``/
+    ``order_by`` are a raw-SQL-fragment API by design (see
+    ``export_table_data``'s docstring); treat them as trusted/operator
+    input, not something to expose directly to untrusted end users.
+    """
+    if not isinstance(fragment, str):
+        raise ValidationError(f"Invalid {kind}: must be a string")
+    # Check the blocklist against literal-masked text so a blocked word
+    # appearing only as quoted data (not as SQL syntax) doesn't false-
+    # positive; the original, unmodified fragment is still what's returned
+    # and used in the query.
+    if _SQL_FRAGMENT_BLOCKLIST_RE.search(_mask_sql_literals(fragment)):
+        raise ValidationError(
+            f"Invalid {kind}: {fragment!r} contains disallowed SQL "
+            "keywords or statement-boundary characters"
+        )
+    return fragment
+

@dataclass
 class TableData:
@@ -253,8 +344,13 @@ class DataExporter:
            schema: Schema name (for databases with schema support, optional)
            limit: Maximum number of rows to export (optional)
            offset: Row offset for pagination (optional)
-            where: WHERE clause for filtering (optional, e.g., "age > 18")
-            order_by: ORDER BY clause for sorting (optional, e.g., "name ASC")
+            where: WHERE clause for filtering (optional, e.g., "age > 18").
+                Raw SQL, checked against a keyword/character blocklist (see
+                ``_validate_sql_fragment``) but not fully sanitized — treat
+                as trusted/operator input, never pass untrusted end-user
+                text here directly.
+            order_by: ORDER BY clause for sorting (optional, e.g., "name ASC").
+                Same trust requirement as ``where``.
            **options: Additional export options (unused)

        Returns:
@@ -269,7 +365,16 @@ class DataExporter:
            ProcessingError: If table export fails
        """
        try:
-            from sqlalchemy import inspect
+            from sqlalchemy import inspect, text
+
+            _validate_sql_identifier(table_name, "table_name")
+            if schema:
+                _validate_sql_identifier(schema, "schema")
+            if where:
+                _validate_sql_fragment(where, "where")
+            if order_by:
+                _validate_sql_fragment(order_by, "order_by")
+
            inspector = inspect(connection)

            # Get column information
@@ -344,6 +449,8 @@ class DataExporter:
                schema=schema,
            )

+        except ValidationError:
+            raise
        except Exception as e:
            self.logger.error(f"Failed to export table {table_name}: {e}")
            raise ProcessingError(f"Failed to export table: {e}") from e
@@ -760,8 +867,11 @@ class DBIngestor:
            schema: Schema name (for databases with schema support, optional)
            limit: Maximum number of rows to export (optional)
            offset: Row offset for pagination (optional)
-            where: WHERE clause for filtering (optional, e.g., "status = 'active'")
-            order_by: ORDER BY clause for sorting (optional, e.g., "created_at DESC")
+            where: WHERE clause for filtering (optional, e.g., "status = 'active'").
+                Raw SQL passed through to ``export_table_data`` — same trust
+                requirement documented there: not for untrusted end-user text.
+            order_by: ORDER BY clause for sorting (optional, e.g., "created_at DESC").
+                Same trust requirement as ``where``.
            transform: Whether to apply data transformations (default: False)
            **filters: Additional filtering options (merged with above parameters)

@@ -11,7 +11,7 @@ import concurrent.futures
 import ipaddress
 import socket
 import threading
-from typing import Any, Iterable, Optional
+from typing import Any, Iterable, List, Optional
 from urllib.parse import urljoin, urlparse

 import requests
@@ -138,6 +138,7 @@ def _get_dns_executor() -> concurrent.futures.ThreadPoolExecutor:
 BLOCKED_NETWORKS = (
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("100.64.0.0/10"),  # CGNAT (RFC 6598) — routable inside carrier/cloud NAT
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # link-local / cloud metadata
    ipaddress.ip_network("172.16.0.0/12"),
@@ -262,6 +263,218 @@ def validate_url_for_request(
        )


+def _resolve_pinned_ips(
+    url: str, *, allow_private_ips: bool
+) -> Optional[List[str]]:
+    """Validate *url* and return every resolved IP for connection pinning.
+
+    ``validate_url_for_request`` and the subsequent connection used to
+    resolve the same hostname independently, which reopens a DNS-rebinding
+    TOCTOU window: a low-TTL or rebinding DNS answer can differ between the
+    validation lookup and the connect-time lookup, so a hostname that
+    validated as public can still connect to a private/internal address.
+    This performs the one resolution that is actually used for both the
+    accept/reject decision *and* the connection (see
+    ``_make_pinned_adapter``), closing that window the same way
+    ``explorer/routes/ontology.py``'s ``_validate_fetch_url`` /
+    ``_make_pinned_session`` pair already does.
+
+    Returns ``None`` when ``allow_private_ips`` is True (the caller
+    explicitly trusts this host, e.g. an operator-configured internal
+    endpoint that may rely on live DNS/service discovery — pinning is
+    skipped so it keeps resolving normally) or when the URL has no host.
+    Otherwise returns the deduplicated, resolution-ordered list of
+    validated IP addresses.
+    """
+    validate_url_for_request(url, allow_private_ips=allow_private_ips)
+    if allow_private_ips:
+        return None
+
+    host = urlparse(url).hostname
+    if not host:
+        return None
+
+    try:
+        literal_ip = ipaddress.ip_address(host)
+    except ValueError:
+        literal_ip = None
+    if literal_ip is not None:
+        return [str(literal_ip)]
+
+    executor = _get_dns_executor()
+    owned_executor = False
+    try:
+        try:
+            future = executor.submit(socket.getaddrinfo, host, None)
+        except RuntimeError:
+            executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+            owned_executor = True
+            future = executor.submit(socket.getaddrinfo, host, None)
+        resolved: Iterable = future.result(timeout=_DNS_RESOLVE_TIMEOUT_SECONDS)
+    except (socket.gaierror, concurrent.futures.TimeoutError, OSError) as exc:
+        raise ValidationError(
+            f"URL host '{host}' could not be resolved safely "
+            "(DNS error or timeout); request blocked"
+        ) from exc
+    finally:
+        if owned_executor:
+            _shutdown_executor(executor)
+
+    pinned_ips: List[str] = []
+    for info in resolved:
+        addr = ipaddress.ip_address(info[4][0])
+        if _ip_is_blocked(addr):
+            raise ValidationError(
+                f"URL host '{host}' resolves to a blocked (private/loopback/"
+                "link-local) address"
+            )
+        addr_str = str(addr)
+        if addr_str not in pinned_ips:
+            pinned_ips.append(addr_str)
+    if not pinned_ips:
+        raise ValidationError(
+            f"URL host '{host}' could not be resolved to a usable address"
+        )
+    return pinned_ips
+
+
+def _make_pinned_adapter(pinned_ips: List[str], hostname: str) -> "requests.adapters.HTTPAdapter":
+    """Build an HTTPAdapter that connects only to *pinned_ips*.
+
+    Falls back across every pinned address in order (a hostname can have
+    multiple A/AAAA records) while presenting *hostname* as the TLS SNI /
+    certificate identity and outgoing Host header, so DNS resolution is
+    bypassed entirely for the actual connection — mirroring
+    ``explorer/routes/ontology.py``'s ``_make_pinned_session``.
+    """
+    import urllib3.util.connection as _u3_connection
+    from urllib3.exceptions import NewConnectionError
+
+    class _MultiIPConnectionMixin:
+        def _new_conn(self):
+            last_exc: Optional[BaseException] = None
+            for ip in pinned_ips:
+                try:
+                    return _u3_connection.create_connection(
+                        (ip, self.port),
+                        self.timeout,
+                        source_address=self.source_address,
+                        socket_options=self.socket_options,
+                    )
+                except OSError as exc:
+                    last_exc = exc
+                    continue
+            raise NewConnectionError(
+                self,
+                f"Failed to establish a connection to any of {pinned_ips}: {last_exc}",
+            )
+
+    class _PinnedIPHTTPAdapter(requests.adapters.HTTPAdapter):
+        def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
+            # A proxy performs its own DNS resolution outside this
+            # process's control, which would silently reopen the exact
+            # rebinding race pinning exists to close. Fail closed instead.
+            if requests.utils.select_proxy(request.url, proxies):
+                raise ValidationError(
+                    "Proxied requests are not supported through the "
+                    "SSRF-guarded request path (a proxy would resolve the "
+                    "host itself and bypass IP pinning)."
+                )
+            host_params, pool_kwargs = self.build_connection_pool_key_attributes(
+                request, verify, cert
+            )
+            if host_params.get("scheme") == "https":
+                pool_kwargs.setdefault("assert_hostname", hostname)
+                pool_kwargs.setdefault("server_hostname", hostname)
+            host_params["host"] = pinned_ips[0]
+            pool = self.poolmanager.connection_from_host(
+                **host_params, pool_kwargs=pool_kwargs
+            )
+            base_connection_cls = pool.ConnectionCls
+            if not issubclass(base_connection_cls, _MultiIPConnectionMixin):
+                pool.ConnectionCls = type(
+                    "_PinnedConnection",
+                    (_MultiIPConnectionMixin, base_connection_cls),
+                    {},
+                )
+            return pool
+
+    return _PinnedIPHTTPAdapter()
+
+
+def _apply_connection_pin(
+    active_session: "requests.Session",
+    url: str,
+    pinned_ips: Optional[List[str]],
+    orig_http_adapter: "requests.adapters.HTTPAdapter",
+    orig_https_adapter: "requests.adapters.HTTPAdapter",
+    had_host_header: bool,
+    orig_host_header: Optional[str],
+) -> None:
+    """Mount (or remove) IP pinning on *active_session* for the next hop."""
+    # Session.mount() silently drops whatever adapter it replaces without
+    # closing it. Across a multi-hop redirect chain, each hop gets its own
+    # fresh pinned adapter (a new pool), so failing to close the one from
+    # the previous hop would leak its pooled connection.
+    _current = active_session.adapters.get("http://")
+    if getattr(_current, "_semantica_pinned", False):
+        _current.close()
+
+    parsed = urlparse(url)
+    if pinned_ips:
+        port = parsed.port
+        default_port = _DEFAULT_PORTS.get(parsed.scheme, 80)
+        host_header = (
+            parsed.hostname
+            if port in (None, default_port)
+            else f"{parsed.hostname}:{port}"
+        )
+        adapter = _make_pinned_adapter(pinned_ips, parsed.hostname or "")
+        adapter._semantica_pinned = True
+        active_session.mount("http://", adapter)
+        active_session.mount("https://", adapter)
+        active_session.headers["Host"] = host_header
+    else:
+        active_session.mount("http://", orig_http_adapter)
+        active_session.mount("https://", orig_https_adapter)
+        # Restore the session's own pre-call Host header state rather than
+        # unconditionally clearing it — a caller-supplied session may carry
+        # a legitimate Host override (e.g. a private/internal endpoint
+        # fronted by a name that differs from the connection host), which
+        # a hop that happens not to need pinning must not silently drop.
+        if had_host_header:
+            active_session.headers["Host"] = orig_host_header
+        else:
+            active_session.headers.pop("Host", None)
+
+
+_SESSION_LOCK_ATTR = "_semantica_ssrf_lock"
+_session_lock_registry_lock = threading.Lock()
+
+
+def _get_session_lock(session: "requests.Session") -> threading.Lock:
+    """Return a lock private to *session*, creating one on first use.
+
+    request_with_ssrf_guard mutates a caller-supplied session's adapters
+    and Host header for the duration of one guarded call (including every
+    redirect hop). Without serializing on the session itself, two guarded
+    calls sharing the same session from different threads could interleave
+    their mount()/restore cycles — one call's request could go out pinned
+    to (or carrying the Host header for) a completely different call's
+    target host. Double-checked locking so concurrent first-use doesn't
+    attach two different locks to the same session.
+    """
+    lock = getattr(session, _SESSION_LOCK_ATTR, None)
+    if lock is not None:
+        return lock
+    with _session_lock_registry_lock:
+        lock = getattr(session, _SESSION_LOCK_ATTR, None)
+        if lock is None:
+            lock = threading.Lock()
+            setattr(session, _SESSION_LOCK_ATTR, lock)
+        return lock
+
+
 def request_with_ssrf_guard(
    method: str,
    url: str,
@@ -317,10 +530,12 @@ def request_with_ssrf_guard(

    Session state that was removed is unconditionally restored in a ``finally``
    block so the session is left in its original state after this call returns,
-    regardless of how it exits (normal return, exception, redirect cap).  The
-    loop is sequential and single-threaded within one call, so the mutation is
-    safe as long as the caller does not share the session across concurrent
-    threads (the standard Semantica pattern: one session per ingestor instance).
+    regardless of how it exits (normal return, exception, redirect cap). A
+    caller-supplied session is also serialized on internally (see
+    ``_get_session_lock``): two guarded calls sharing the same session from
+    different threads block on each other for the call's duration rather than
+    interleaving their mutations, so concurrent use of a shared session is
+    safe, if not concurrent.

    Once credentials have been stripped for a cross-origin hop they are NOT
    re-added for subsequent hops in the same chain, even if a later hop
@@ -337,13 +552,38 @@ def request_with_ssrf_guard(
    )
    _original_host = (urlparse(url).hostname or "").lower()

-    validate_url_for_request(url, allow_private_ips=allow_private_ips)
+    current_pinned_ips = _resolve_pinned_ips(url, allow_private_ips=allow_private_ips)

-    requester = session.request if session is not None else requests.request
+    _owns_session = session is None
+    active_session = session if session is not None else requests.Session()
+    requester = active_session.request
    current_url = url
    current_method = method.upper()
    redirects_followed = 0

+    # A caller-supplied session is mutated (adapters + Host header, and
+    # potentially auth/trust_env below) for the duration of this call,
+    # including every redirect hop; serialize on the session itself so a
+    # second guarded call sharing it from another thread can't interleave
+    # its own mount()/restore cycle into the middle of this one. An owned
+    # session is private to this call, so no lock is needed. Released in
+    # the outermost `finally` below, alongside the state it protects.
+    _session_lock = None if _owns_session else _get_session_lock(active_session)
+    if _session_lock is not None:
+        _session_lock.acquire()
+
+    # Snapshot the session's pre-existing adapters/Host header so pinning
+    # (mounted per-hop below) can be fully undone when this call returns —
+    # required for a caller-supplied session, which outlives this call.
+    _orig_http_adapter = (
+        active_session.adapters.get("http://") or requests.adapters.HTTPAdapter()
+    )
+    _orig_https_adapter = (
+        active_session.adapters.get("https://") or requests.adapters.HTTPAdapter()
+    )
+    _had_host_header = "Host" in active_session.headers
+    _orig_host_header = active_session.headers.get("Host")
+
    # -- issue #947: snapshot every session-level credential source so we can
    #    restore them unconditionally when this call exits.
    _SENSITIVE = ("Authorization", "Proxy-Authorization")
@@ -369,6 +609,15 @@ def request_with_ssrf_guard(

    try:
        while True:
+            _apply_connection_pin(
+                active_session,
+                current_url,
+                current_pinned_ips,
+                _orig_http_adapter,
+                _orig_https_adapter,
+                _had_host_header,
+                _orig_host_header,
+            )
            response = requester(
                current_method,
                current_url,
@@ -407,7 +656,9 @@ def request_with_ssrf_guard(
                if next_host and next_host == _original_host
                else redirect_allow_private_ips
            )
-            validate_url_for_request(next_url, allow_private_ips=hop_allow_private_ips)
+            current_pinned_ips = _resolve_pinned_ips(
+                next_url, allow_private_ips=hop_allow_private_ips
+            )

            # Do not leak sensitive headers or auth handlers to a different
            # origin on redirects.  All four credential sources are cleared:
@@ -472,13 +723,31 @@ def request_with_ssrf_guard(
            redirects_followed += 1

    finally:
-        # Unconditionally restore every session credential source we touched,
-        # so the session is in its original state after this call returns or raises.
-        if session is not None:
+        if _owns_session:
+            # No caller holds a reference to this session; just release it.
+            active_session.close()
+        else:
+            # Unconditionally restore every session credential source and
+            # pinning artifact we touched, so the session is in its
+            # original state after this call returns or raises.
            if _session_auth_backup:
                for _h, _v in _session_auth_backup.items():
-                    session.headers[_h] = _v
+                    active_session.headers[_h] = _v
            # Restore session.auth to whatever it was before this call.
-            session.auth = _session_auth_handler_backup
+            active_session.auth = _session_auth_handler_backup
            # Restore session.trust_env (.netrc / env-proxy lookup flag).
-            session.trust_env = _session_trust_env_backup
+            active_session.trust_env = _session_trust_env_backup
+            # Undo any IP-pinning adapter/Host header mounted for a hop,
+            # closing the last pinned adapter so its pooled connection
+            # isn't leaked (see _apply_connection_pin).
+            _current = active_session.adapters.get("http://")
+            if getattr(_current, "_semantica_pinned", False):
+                _current.close()
+            active_session.mount("http://", _orig_http_adapter)
+            active_session.mount("https://", _orig_https_adapter)
+            if _had_host_header:
+                active_session.headers["Host"] = _orig_host_header
+            else:
+                active_session.headers.pop("Host", None)
+        if _session_lock is not None:
+            _session_lock.release()
@@ -22,7 +22,7 @@ License: MIT

 from typing import Any, Dict, List, Optional

-from ..deduplication.duplicate_detector import DuplicateDetector
+from ..deduplication.duplicate_detector import DuplicateDetector, DuplicateGroup
 from ..deduplication.entity_merger import EntityMerger
 from ..utils.logging import get_logger
 from ..utils.progress_tracker import get_progress_tracker
@@ -138,9 +138,7 @@ class EntityResolver:
            self.logger.debug(
                f"Detecting duplicate groups with threshold {self.similarity_threshold}"
            )
-            duplicate_groups = self.duplicate_detector.detect_duplicate_groups(
-                entities, threshold=self.similarity_threshold
-            )
+            duplicate_groups = self._detect_duplicate_groups(entities)

            self.logger.debug(f"Found {len(duplicate_groups)} duplicate group(s)")

@@ -150,6 +148,7 @@ class EntityResolver:
            # Step 2: Merge duplicates in each group
            merged_entities = []
            processed_entity_ids = set()  # Track which entities have been merged
+            processed_entity_objects = set()

            for group in duplicate_groups:
                # Skip groups with less than 2 entities (not duplicates)
@@ -157,9 +156,16 @@ class EntityResolver:
                    continue

                # Merge the duplicate group into a single canonical entity
-                merge_operations = self.entity_merger.merge_duplicates(
-                    group.entities, **self.config
-                )
+                if self.resolution_strategy == "exact":
+                    merge_operations = [
+                        self.entity_merger.merge_entity_group(
+                            group.entities, **self.config
+                        )
+                    ]
+                else:
+                    merge_operations = self.entity_merger.merge_duplicates(
+                        group.entities, **self.config
+                    )

                # Process each merge operation
                for operation in merge_operations:
@@ -168,30 +174,26 @@ class EntityResolver:

                    # Mark all source entities as processed
                    for source_entity in operation.source_entities:
-                        entity_id = (
-                            source_entity.get("id")
-                            if isinstance(source_entity, dict)
-                            else getattr(source_entity, "id", None)
-                        ) or (
-                            source_entity.get("entity_id")
-                            if isinstance(source_entity, dict)
-                            else getattr(source_entity, "entity_id", None)
-                        )
-                        if entity_id:
+                        entity_id = self._get_entity_id(source_entity)
+                        if entity_id is None:
+                            processed_entity_objects.add(id(source_entity))
+                            continue
+                        try:
                            processed_entity_ids.add(entity_id)
+                        except TypeError:
+                            processed_entity_objects.add(id(source_entity))

            # Step 3: Add non-duplicate entities (entities not in any duplicate group)
            for entity in entities:
-                entity_id = (
-                    entity.get("id")
-                    if isinstance(entity, dict)
-                    else getattr(entity, "id", None)
-                ) or (
-                    entity.get("entity_id")
-                    if isinstance(entity, dict)
-                    else getattr(entity, "entity_id", None)
-                )
-                if entity_id and entity_id not in processed_entity_ids:
+                entity_id = self._get_entity_id(entity)
+                if entity_id is None:
+                    is_unprocessed = id(entity) not in processed_entity_objects
+                else:
+                    try:
+                        is_unprocessed = entity_id not in processed_entity_ids
+                    except TypeError:
+                        is_unprocessed = id(entity) not in processed_entity_objects
+                if is_unprocessed:
                    # This entity was not merged, add it as-is
                    merged_entities.append(entity)

@@ -213,6 +215,50 @@ class EntityResolver:
            )
            raise

+    def _detect_duplicate_groups(
+        self, entities: List[Dict[str, Any]]
+    ) -> List[DuplicateGroup]:
+        """Detect duplicate groups according to the configured strategy."""
+        if self.resolution_strategy != "exact":
+            return self.duplicate_detector.detect_duplicate_groups(
+                entities, threshold=self.similarity_threshold
+            )
+
+        groups = {}
+        for entity in entities:
+            name = self._get_entity_name(entity)
+            normalized = str(name).strip() if name is not None else ""
+            if normalized:
+                groups.setdefault(normalized.casefold(), []).append(entity)
+
+        return [
+            DuplicateGroup(entities=group, confidence=1.0)
+            for group in groups.values()
+            if len(group) > 1
+        ]
+
+    @staticmethod
+    def _get_entity_id(entity: Any) -> Any:
+        """Return an entity ID while supporting dictionary and object inputs."""
+        if isinstance(entity, dict):
+            return entity.get("id") or entity.get("entity_id")
+        return getattr(entity, "id", None) or getattr(entity, "entity_id", None)
+
+    @staticmethod
+    def _get_entity_name(entity: Any) -> Optional[str]:
+        """Return an entity name, falling back to text-based entity input."""
+        if isinstance(entity, dict):
+            name = entity.get("name")
+            return (
+                name if name is not None and str(name).strip() else entity.get("text")
+            )
+        name = getattr(entity, "name", None)
+        return (
+            name
+            if name is not None and str(name).strip()
+            else getattr(entity, "text", None)
+        )
+
    def merge_duplicates(self, entities: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        Merge duplicate entities.
@@ -185,8 +185,25 @@ class GraphValidator:

        # 3. Relationship Validation
        for i, rel in enumerate(relationships):
-            # Check required fields
-            missing = self.required_rel_fields - set(rel.keys())
+            # Endpoints may use either the legacy ``source``/``target`` keys or
+            # the canonical ``source_id``/``target_id`` keys emitted by
+            # ``ContextGraph.to_kg_dict()``. Accept either variant so both
+            # representations validate consistently.
+            src = rel.get("source")
+            if src is None:
+                src = rel.get("source_id")
+            tgt = rel.get("target")
+            if tgt is None:
+                tgt = rel.get("target_id")
+
+            # Check required fields: ``type`` plus a resolvable source/target.
+            missing = set()
+            if "type" not in rel:
+                missing.add("type")
+            if src is None:
+                missing.add("source")
+            if tgt is None:
+                missing.add("target")
            if missing:
                issues.append(ValidationIssue(
                    code="MISSING_FIELD",
@@ -196,9 +213,6 @@ class GraphValidator:
                    details={"index": i}
                ))
                continue
-
-            src = rel.get("source")
-            tgt = rel.get("target")
            
            # Check Dangling Edges
            def is_valid_id(node_id):
@@ -535,7 +535,8 @@ class TemporalGraphQuery:
            relationships = [
                rel
                for rel in relationships
-                if rel.get("source") == entity or rel.get("target") == entity
+                if (rel.get("source") or rel.get("source_id")) == entity
+                or (rel.get("target") or rel.get("target_id")) == entity
            ]

        if relationship:
@@ -642,8 +643,14 @@ class TemporalGraphQuery:
        parsed_end_time = self._parse_time(end_time) if end_time else None

        for rel in relationships:
+            # Accept both the legacy ``source``/``target`` keys and the
+            # canonical ``source_id``/``target_id`` keys from ``to_kg_dict()``.
            s = rel.get("source")
+            if s is None:
+                s = rel.get("source_id")
            t = rel.get("target")
+            if t is None:
+                t = rel.get("target_id")

            # Check temporal validity
            if start_time or end_time:
@@ -243,6 +243,14 @@ class MediaParser:
        try:
            import subprocess

+            # ffprobe's own argument parser doesn't reliably honor a bare
+            # "--" end-of-options marker, so a filename starting with "-"
+            # could otherwise be parsed as an option; neutralize that by
+            # forcing a relative-path prefix ffprobe can't mistake for a flag.
+            ffprobe_path = str(file_path)
+            if ffprobe_path.startswith("-"):
+                ffprobe_path = f"./{ffprobe_path}"
+
            result = subprocess.run(
                [
                    "ffprobe",
@@ -252,7 +260,7 @@ class MediaParser:
                    "json",
                    "-show_format",
                    "-show_streams",
-                    str(file_path),
+                    ffprobe_path,
                ],
                capture_output=True,
                text=True,
@@ -383,9 +383,9 @@ class AnzoStore:
        if self._is_uri_value(obj):
            if obj.startswith("<") and obj.endswith(">"):
                inner = obj[1:-1]
-                if " " in inner or ">" in inner:
-                    raise ValueError(f"IRI contains invalid characters: {obj!r}")
+                sparql_escaping.validate_uri(inner)
                return obj
+            sparql_escaping.validate_uri(obj)
            return f"<{obj}>"

        escaped = sparql_escaping.escape_literal(obj)
@@ -0,0 +1,150 @@
+"""Tests for ContextGraph.to_kg_dict() — the official KG-shape adapter.
+
+These tests lock in the contract that to_kg_dict() emits the
+``{"entities", "relationships"}`` / ``source_id`` shape expected by
+downstream consumers (RDFExporter, TemporalGraphQuery.query_time_range),
+so users never need to hand-map field names.
+"""
+
+from semantica.context.context_graph import ContextEdge, ContextGraph, ContextNode
+
+
+def _build_graph():
+    g = ContextGraph()
+    g._add_internal_node(ContextNode(node_id="e1", node_type="entity", content="Alice"))
+    g._add_internal_node(ContextNode(node_id="e2", node_type="entity", content="Bob"))
+    g._add_internal_node(
+        ContextNode(node_id="c1", node_type="conversation", content="chat log")
+    )
+    g._add_internal_edge(
+        ContextEdge(
+            source_id="e1",
+            target_id="e2",
+            edge_type="knows",
+            valid_from="2024-01-01",
+            valid_until="2024-12-31",
+        )
+    )
+    # Edge touching a non-entity node — used to test entities_only filtering.
+    g._add_internal_edge(
+        ContextEdge(source_id="c1", target_id="e1", edge_type="mentions")
+    )
+    return g
+
+
+def test_basic_shape():
+    kg = _build_graph().to_kg_dict()
+    assert set(kg.keys()) == {"entities", "relationships", "statistics"}
+    # Entity shape uses id/text/type (not id/content).
+    entity = next(e for e in kg["entities"] if e["id"] == "e1")
+    assert entity["text"] == "Alice"
+    assert entity["type"] == "entity"
+
+
+def test_relationship_uses_source_id_target_id():
+    kg = _build_graph().to_kg_dict()
+    rel = next(r for r in kg["relationships"] if r["type"] == "knows")
+    assert rel["source_id"] == "e1"
+    assert rel["target_id"] == "e2"
+    # "source"/"target" (the internal names) must NOT leak through.
+    assert "source" not in rel
+    assert "target" not in rel
+
+
+def test_temporal_fields_passthrough():
+    kg = _build_graph().to_kg_dict()
+    rel = next(r for r in kg["relationships"] if r["type"] == "knows")
+    assert rel["valid_from"] == "2024-01-01"
+    assert rel["valid_until"] == "2024-12-31"
+
+
+def test_statistics_counts():
+    kg = _build_graph().to_kg_dict()
+    assert kg["statistics"]["entity_count"] == len(kg["entities"])
+    assert kg["statistics"]["relationship_count"] == len(kg["relationships"])
+
+
+def test_entities_only_filters_nodes():
+    kg = _build_graph().to_kg_dict(entities_only=True)
+    types = {e["type"] for e in kg["entities"]}
+    assert types == {"entity"}
+    assert len(kg["entities"]) == 2
+
+
+def test_entities_only_drops_dangling_relationships():
+    # The "mentions" edge points from a conversation node (filtered out under
+    # entities_only) and must not appear as a dangling relationship.
+    kg = _build_graph().to_kg_dict(entities_only=True)
+    rel_types = {r["type"] for r in kg["relationships"]}
+    assert "mentions" not in rel_types
+    assert rel_types == {"knows"}
+
+
+def test_returned_dicts_are_isolated_from_internal_state():
+    g = _build_graph()
+    kg = g.to_kg_dict()
+    entity = next(e for e in kg["entities"] if e["id"] == "e1")
+    # Mutating the returned dict must not corrupt internal node properties.
+    entity["properties"]["injected"] = True
+    assert "injected" not in g.nodes["e1"].properties
+
+
+def test_null_properties_and_metadata_do_not_crash():
+    """Nodes loaded from JSON ``null`` keep None props/metadata; to_kg_dict
+    must normalize them instead of raising TypeError (Qodo bug 1)."""
+    g = ContextGraph()
+    n = ContextNode(node_id="e1", node_type="entity", content="Alice")
+    n.properties = None
+    n.metadata = None
+    g._add_internal_node(n)
+
+    kg = g.to_kg_dict()
+    entity = kg["entities"][0]
+    assert entity["properties"] == {}
+    assert entity["metadata"] == {}
+
+
+def test_non_string_node_id_is_normalized_and_keeps_edges():
+    """ContextEdge coerces endpoints to str; entity ids must be coerced too
+    so entities_only filtering does not drop valid edges (Qodo bug 3)."""
+    g = ContextGraph()
+    g._add_internal_node(ContextNode(node_id=1, node_type="entity", content="one"))
+    g._add_internal_node(ContextNode(node_id=2, node_type="entity", content="two"))
+    g._add_internal_edge(ContextEdge(source_id=1, target_id=2, edge_type="links"))
+
+    kg = g.to_kg_dict(entities_only=True)
+    ids = {e["id"] for e in kg["entities"]}
+    assert ids == {"1", "2"}
+    assert all(isinstance(e["id"], str) for e in kg["entities"])
+    # The edge must survive filtering despite the int-vs-str origin.
+    assert {r["type"] for r in kg["relationships"]} == {"links"}
+
+
+def test_output_is_consumable_by_kg_utilities():
+    """to_kg_dict output must validate and be traversable by KG utilities that
+    historically read ``source``/``target`` (Qodo bug 2, consumer side)."""
+    from semantica.kg.graph_validator import GraphValidator, ValidationSeverity
+    from semantica.kg.temporal_query import TemporalGraphQuery
+
+    g = ContextGraph()
+    g._add_internal_node(ContextNode(node_id="e1", node_type="entity", content="Alice"))
+    g._add_internal_node(ContextNode(node_id="e2", node_type="entity", content="Bob"))
+    g._add_internal_edge(ContextEdge(source_id="e1", target_id="e2", edge_type="knows"))
+    kg = g.to_kg_dict()
+
+    # Validator requires entity ``name``; add it so only endpoint compat is tested.
+    for e in kg["entities"]:
+        e["name"] = e["text"]
+
+    result = GraphValidator().validate(kg)
+    endpoint_errors = [
+        i for i in result.issues
+        if i.code in {"MISSING_FIELD", "DANGLING_EDGE"}
+        and i.element_type == "relationship"
+    ]
+    assert endpoint_errors == [], endpoint_errors
+
+    # TemporalGraphQuery.analyze_evolution must see the relationship for "e1".
+    tq = TemporalGraphQuery()
+    filtered = tq.analyze_evolution(kg, entity="e1")
+    assert filtered is not None
@@ -254,7 +254,7 @@ class TestAuthHandlerStripping:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -285,7 +285,7 @@ class TestAuthHandlerStripping:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -416,7 +416,7 @@ class TestAuthHandlerStripping:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[hop1, hop2, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -520,7 +520,7 @@ class TestCredentialResurrection:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[hop1, hop2, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -578,7 +578,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -600,7 +600,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -619,7 +619,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -638,7 +638,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -657,7 +657,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -676,7 +676,7 @@ class TestRedirectTypesAndOriginChanges:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -701,7 +701,7 @@ class TestAllowPrivateIpsOnRedirect:
        redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=redirect,
        ):
            with pytest.raises(ValidationError, match="blocked"):
@@ -718,7 +718,7 @@ class TestAllowPrivateIpsOnRedirect:
        final = _mock_final()

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            request_with_ssrf_guard(
@@ -736,7 +736,7 @@ class TestAllowPrivateIpsOnRedirect:
        redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, _mock_final()],
        ) as mock_req:
            # allow_private_ips=True with no override: redirect target validation
@@ -784,7 +784,7 @@ class TestMCPClientAuthRedirect:
        )

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -806,7 +806,7 @@ class TestMCPClientAuthRedirect:
        )

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -821,7 +821,7 @@ class TestMCPClientAuthRedirect:
        client = MCPClient(url="http://localhost:8000/mcp")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=final,
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -834,7 +834,7 @@ class TestMCPClientAuthRedirect:
        client = MCPClient(url="http://127.0.0.1:9000/mcp")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=final,
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -849,7 +849,7 @@ class TestMCPClientAuthRedirect:
        client = MCPClient(url="http://localhost:8000/mcp")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -869,7 +869,7 @@ class TestMCPClientAuthRedirect:
        client = MCPClient(url="https://mcp.example.com/mcp")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=redirect,
        ):
            with pytest.raises(ValidationError, match="blocked"):
@@ -888,7 +888,7 @@ class TestMCPClientAuthRedirect:
        )

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            side_effect=[redirect, final],
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -904,7 +904,7 @@ class TestMCPClientAuthRedirect:
        client = MCPClient(url="https://mcp.example.com/mcp")

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=hop,
        ):
            with pytest.raises((ValidationError, Exception), match="[Rr]edirect|[Ee]xceeded"):
@@ -920,7 +920,7 @@ class TestMCPClientAuthRedirect:
        )

        with patch(
-            "semantica.ingest.ssrf.requests.request",
+            "requests.Session.request",
            return_value=final,
        ) as mock_req:
            client._send_request_http({"jsonrpc": "2.0", "method": "ping"})
@@ -13,7 +13,7 @@ class TestCookbookIntegration:
        # MCPClient._send_request_http now routes through request_with_ssrf_guard,
        # which calls requests.request (not httpx.post / requests.post directly).
        # Patch at the point where the guard issues the actual HTTP call.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:

            def side_effect(method, url, json=None, **kwargs):
                if not json:
@@ -222,7 +222,7 @@ def test_discover_feeds_empty() -> None:
        "semantica.ingest.ssrf.socket.getaddrinfo",
        return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
    ):
-        with patch("requests.request", side_effect=fake_request):
+        with patch("requests.Session.request", side_effect=fake_request):
            feeds = ingestor.discover_feeds("http://site.com")

    assert len(feeds) == 0
@@ -254,7 +254,7 @@ def test_discover_feeds_found() -> None:
        "semantica.ingest.ssrf.socket.getaddrinfo",
        return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
    ):
-        with patch("requests.request", side_effect=fake_request):
+        with patch("requests.Session.request", side_effect=fake_request):
            feeds = ingestor.discover_feeds("http://site.com")

    assert "http://site.com/rss.xml" in feeds
@@ -447,7 +447,7 @@ class TestSitemapCrawlerSSRF:
        redirect.close = MagicMock()

        with patch(
-            "semantica.ingest.ssrf.requests.request", return_value=redirect
+            "requests.Session.request", return_value=redirect
        ), patch(
            "semantica.ingest.ssrf.socket.getaddrinfo",
            return_value=[(None, None, None, None, ("93.184.216.34", 0))],
@@ -198,7 +198,7 @@ class TestMCPClient:
        # MCPClient._send_request_http now routes through request_with_ssrf_guard,
        # which calls requests.request (not requests.post) with allow_redirects=False.
        # Patch the requests.request call inside ssrf.py.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:
            mock_response = MagicMock()
            mock_response.status_code = 200

@@ -226,7 +226,7 @@ class TestMCPClient:

    def test_call_tool_mock_check(self):
        # Redo with the corrected patch target.
-        with patch("semantica.ingest.ssrf.requests.request") as mock_request:
+        with patch("requests.Session.request") as mock_request:
            mock_response = MagicMock()
            mock_response.status_code = 200

@@ -0,0 +1,56 @@
+from semantica.kg.entity_resolver import EntityResolver
+
+
+def test_exact_resolution_does_not_merge_similar_names():
+    entities = [
+        {"id": "1", "name": "Alice"},
+        {"id": "2", "name": "Alicia"},
+    ]
+
+    resolved = EntityResolver(strategy="exact").resolve_entities(entities)
+
+    assert {entity["id"] for entity in resolved} == {"1", "2"}
+
+
+def test_exact_resolution_merges_case_and_whitespace_variants():
+    entities = [
+        {"id": "1", "name": " Alice "},
+        {"id": "2", "name": "alice"},
+    ]
+
+    resolved = EntityResolver(strategy="exact").resolve_entities(entities)
+
+    assert len(resolved) == 1
+
+
+def test_resolution_preserves_non_duplicate_entities_without_ids():
+    entities = [
+        {"name": "Alice"},
+        {"name": "Bob"},
+    ]
+
+    resolved = EntityResolver(strategy="exact").resolve_entities(entities)
+
+    assert resolved == entities
+
+
+def test_exact_resolution_does_not_merge_whitespace_only_names():
+    entities = [
+        {"id": "1", "name": "   "},
+        {"id": "2", "name": "\t"},
+    ]
+
+    resolved = EntityResolver(strategy="exact").resolve_entities(entities)
+
+    assert {entity["id"] for entity in resolved} == {"1", "2"}
+
+
+def test_exact_resolution_falls_back_to_text_when_name_is_blank():
+    entities = [
+        {"id": "1", "name": " ", "text": "Alice"},
+        {"id": "2", "name": "Alice"},
+    ]
+
+    resolved = EntityResolver(strategy="exact").resolve_entities(entities)
+
+    assert len(resolved) == 1