feat(ingest): gate http behind allow_network run-flag + dispatch (I6)

src/portfolio_optimiser/ingest.py

@@ -470,7 +470,12 @@ def _update_index_lines(
 
 
 def materialize(
-    manifest_path: str | Path, bundle_dir: str | Path, *, ingested_at: str
+    manifest_path: str | Path,
+    bundle_dir: str | Path,
+    *,
+    ingested_at: str,
+    allow_network: bool = False,
+    http_get: HttpGet | None = None,
 ) -> list[Path]:
     """Materialize a manifest's extractions into an OKF bundle (§5): the I2 invocation
     surface (a ``python -m`` CLI is deliberately deferred).
@@ -478,10 +483,15 @@ def materialize(
     Three explicit inputs — manifest, target bundle dir, and ``ingested_at`` (REQUIRED
     keyword, NO wall-clock default, stamped verbatim; mirrors the promotion gate's
     timestamp rule). Deterministic and offline: zero model calls, zero network for the
-    ``file`` source type. All extractions execute and render IN MEMORY before the first
-    disk mutation (crash-window mitigation for the non-atomic §5 replace sequence;
+    ``file``/``sql`` source types. All extractions execute and render IN MEMORY before the
+    first disk mutation (crash-window mitigation for the non-atomic §5 replace sequence;
     recovery = idempotent re-run, §10). Source calls are logged per §8 (which source,
-    when = the ``ingested_at`` argument, row count) — never cell contents, never secrets."""
+    when = the ``ingested_at`` argument, row count) — never cell contents, never secrets.
+
+    ``allow_network`` (I6, §8) is the per-run network opt-in: an ``http`` source is refused
+    fail-fast unless it is set — the manifest itself cannot grant network access (local-only
+    default, no silent egress). ``http_get`` optionally injects the transport seam (default
+    ``_urllib_get``, the only socket path); both are ignored for ``file``/``sql`` sources."""
     if not _INGESTED_AT_RE.match(ingested_at):
         raise ValueError(
             f"ingested_at must be ISO-8601 UTC with a Z suffix "
@@ -490,10 +500,14 @@ def materialize(
     manifest_file = Path(manifest_path)
     manifest, stamp = load_manifest(manifest_file)
     source = manifest.source
-    if isinstance(source, HttpSource):
+    # §8 network gate (I6): refuse http fail-fast BEFORE any source access unless the per-run
+    # flag is set. The `and not allow_network` clause is the load-bearing seam — the manifest
+    # cannot grant itself network access (local-only default, no silent egress).
+    if isinstance(source, HttpSource) and not allow_network:
         raise IngestError(
-            f"source type {source.type!r} is a gated extension point (I6) with no connector "
-            "in this implementation — the schema validates it, execution defers"
+            f"source type {source.type!r} requires the per-run network opt-in "
+            "(materialize(..., allow_network=True)) — the manifest cannot grant itself network "
+            "access (ingest spec §8, local-only default, no silent egress)"
         )
     # Pinned decision (file): a relative root resolves against the manifest file's directory —
     # never the process cwd, or the extraction would not be reproducible. (sql resolves its
@@ -508,17 +522,32 @@ def materialize(
     for extraction in manifest.extractions:
         if isinstance(source, FileSource):
             header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
-        else:  # SqlSource — http already refused above
+            body = render_table(header, rows)
+            row_count = len(rows)
+        elif isinstance(source, HttpSource):
+            # The gate above guarantees allow_network here; _urllib_get is the only socket path.
+            get = http_get if http_get is not None else _urllib_get
+            text = read_http(
+                source.base_url,
+                extraction.query,
+                max_rows=extraction.max_rows,
+                credential_ref=source.credential_ref,
+                get=get,
+            )
+            body = _render_fenced_block(text)  # verbatim fenced block, NOT table-escaped
+            row_count = len(text.splitlines())
+        else:  # SqlSource
             header, rows = read_sql(
                 source.connection_ref, extraction.query, max_rows=extraction.max_rows
             )
+            body = render_table(header, rows)
+            row_count = len(rows)
         _LOGGER.info(
             "source call: source=%s ingested_at=%s rows=%d",
             source.id,
             ingested_at,
-            len(rows),
+            row_count,
         )
-        body = render_table(header, rows)
         content = _render_concept_file(
             manifest, extraction, body, ingested_at=ingested_at, stamp=stamp
         )