fix(security): confine an LLC work product's storage_path before opening it (#17302, #17300 AC3)

autobot-backend/llc/api/work_items.py

@@ -1315,9 +1315,9 @@ async def upload_attachment(
             uploaded_by_agent_id=uploaded_by_agent_id,
             uploaded_by_user_id=uploaded_by_user_id,
         )
-    except AttachmentTooLarge as exc:
-        logger.error("Exception in API handler: %s", exc, exc_info=True)
-        raise HTTPException(status_code=413, detail="Internal server error")
+    except (AttachmentTooLarge, ValueError) as exc:  # #17302: ValueError was an uncaught 500
+        logger.warning("Attachment upload refused for %s: %s", work_item_id, exc)
+        raise HTTPException(413 if isinstance(exc, AttachmentTooLarge) else 400, "Attachment rejected")
     return _attachment_to_dict(row)
 
 

autobot-backend/llc/kb/artifact_ingestor.py

@@ -23,7 +23,10 @@
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from autobot_shared.security.path_validator import validate_path
+
 from ..models.work_product import LLCWorkProduct
+from ..storage_root import llc_local_storage_root
 from .write_guard import assert_not_writing_to_ancestor_kb
 
 logger = logging.getLogger(__name__)
@@ -61,10 +64,49 @@
 
 
 def _is_text_path(path: str) -> bool:
+    """Whether this is a shape we can parse. NOT a containment check (#17302).
+
+    An extension allowlist answers "can we read this", and for a while it was
+    standing in for "are we allowed to read this". The two questions have no
+    overlap: `/etc/anything.yaml` passes this and is not ours to read. Use
+    `_contained_storage_path` for the second question; this one stays exactly
+    as narrow as its name.
+    """
     ext = os.path.splitext(path)[1].lower()
     return ext in _TEXT_EXTENSIONS
 
 
+def _contained_storage_path(raw: str, product_id: object) -> Optional[str]:
+    """*raw* resolved inside the LLC storage root, or None if it escapes (#17302).
+
+    `storage_path` is a free-form string on the agent request model
+    (`llc/api/agent_api.py`), stored unvalidated and opened later. Nothing in
+    this codebase produces one -- the field is only ever set from a caller's
+    body -- so an authenticated agent could name any path on the host with an
+    allowlisted extension and have it read into the knowledge index, where it
+    becomes retrievable. `.yaml`/`.toml` config, `.json` credential files,
+    `.py` source and `.sql` dumps are all in that allowlist.
+
+    Returns the **validated** string, which is the string the caller must open:
+    validating one path and opening something rebuilt from the original input
+    is the bypass this shape invites (THREAT_MODEL.md section 1).
+
+    Fails closed. A path that cannot be proven inside the root is skipped with
+    a warning rather than read, because "we could not establish this is ours"
+    and "this is ours" must not produce the same outcome.
+    """
+    root = llc_local_storage_root()
+    try:
+        return str(validate_path(raw, allowed_roots=(str(root.resolve()),)))
+    except (ValueError, OSError) as exc:
+        logger.warning(
+            "ArtifactIngestor: refusing storage_path outside the LLC storage root " "for product %s: %s (#17302)",
+            product_id,
+            exc,
+        )
+        return None
+
+
 def _split_text(text: str) -> List[str]:
     """Split text into overlapping chunks of ~_CHUNK_SIZE characters."""
     if len(text) <= _CHUNK_SIZE:
@@ -237,20 +279,48 @@ async def _resolve_text(self, product: LLCWorkProduct) -> Optional[str]:
 
         if product.storage_path:
             if not _is_text_path(product.storage_path):
+                # The product id and nothing derived from the path (#17302).
+                # `storage_path` is agent-supplied, and CLAUDE.md forbids
+                # internal filesystem paths in logs -- doubly so for one an
+                # attacker chooses.
+                #
+                # An earlier version logged `splitext(path)[1][:16]`, reasoning
+                # that a 16-char extension is bounded. Two things were wrong
+                # with that. A bound is not a sanitiser: 16 characters of
+                # attacker-chosen text is still attacker-chosen text, including
+                # newlines and control characters that a log reader renders.
+                # And CodeQL was right to keep flagging it -- taint propagates
+                # through `splitext` and survives a slice, so the sink genuinely
+                # still received data derived from the untrusted string.
+                #
+                # An allowlist does not rescue it here: this branch runs
+                # precisely when the extension is NOT in `_TEXT_EXTENSIONS`, so
+                # "log it if allowlisted else a constant" logs the constant
+                # every time. The message already states the reason, and the
+                # product id is the join key to the row that holds the path.
                 logger.info(
-                    "ArtifactIngestor: skipping binary storage_path %s for product %s",
-                    product.storage_path,
+                    "ArtifactIngestor: skipping product %s -- storage_path is not an indexable text type",
                     product.id,
                 )
                 return None
+            contained = _contained_storage_path(product.storage_path, product.id)
+            if contained is None:
+                return None
             try:
-                with open(product.storage_path, encoding="utf-8", errors="replace") as fh:
+                # `contained`, never `product.storage_path`: the validated
+                # string is the string used (#17302).
+                with open(contained, encoding="utf-8", errors="replace") as fh:
                     return fh.read()
-            except Exception:
-                logger.exception(
-                    "ArtifactIngestor: could not read storage_path %s for product %s",
-                    product.storage_path,
+            except Exception as exc:
+                # Type, not the exception object: an OSError stringifies with
+                # the filename, so `logger.exception` would put the
+                # agent-supplied path back into the log through the traceback
+                # after it was removed from the format string. The cost is the
+                # traceback; the product id recovers the path from the row.
+                logger.warning(
+                    "ArtifactIngestor: unreadable storage_path for product %s (%s)",
                     product.id,
+                    type(exc).__name__,
                 )
                 return None
 

autobot-backend/llc/kb/artifact_ingestor_path_containment_17302_test.py

@@ -0,0 +1,242 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""A work product's `storage_path` cannot name a file outside the LLC root (#17302).
+
+`storage_path` is a free-form optional string on the agent request model
+(`llc/api/agent_api.py`), stored unvalidated by `WorkProductService.create` and
+opened later by the ingestor. Nothing in this codebase ever produces one -- the
+field is only ever set from a caller's request body -- so an authenticated agent
+could name any path on the host and have its contents read into the knowledge
+index, where they become retrievable.
+
+The only gate between the two was `_is_text_path`, an extension allowlist. That
+is the near-miss worth naming: validation existed, and it was not the validation
+that was needed. `.yaml`, `.toml`, `.ini`, `.cfg`, `.py`, `.sh` and `.sql` are all
+allowlisted, so "is this a shape we can parse" was answering "are we allowed to
+read this".
+The two questions do not overlap at all.
+
+Why these assert on `open` rather than on a return value: a refusal that
+returns None after reading the file is indistinguishable from one that never
+read it, if you only check the result. The property is that the bytes are never
+touched, so the tests check that the bytes are never touched.
+"""
+
+from __future__ import annotations
+
+import builtins
+import pathlib
+import uuid
+from typing import List, Optional
+
+import pytest
+
+from llc.kb import artifact_ingestor
+from llc.kb.artifact_ingestor import ArtifactIngestor, _contained_storage_path
+
+
+def _plant(path: pathlib.Path, text: str) -> pathlib.Path:
+    """Create a file for a test to point at.
+
+    A plain sync function on purpose: these writes are arrangement, not the
+    behaviour under test, and #7444's guard correctly refuses blocking I/O
+    inside an `async def`. Keeping the setup out of the coroutine satisfies it
+    without a `# noqa`, which would have claimed something untrue -- the calls
+    do not run in a thread.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text, encoding="utf-8")
+    return path
+
+
+class _Product:
+    """The two fields `_resolve_text` reads, and an id for the log line."""
+
+    def __init__(self, storage_path: Optional[str]) -> None:
+        self.id = uuid.uuid4()
+        self.content_text = None
+        self.storage_path = storage_path
+
+
+@pytest.fixture()
+def llc_root(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
+    root = tmp_path / "llc-storage"
+    root.mkdir()
+    monkeypatch.setenv("LLC_STORAGE_PATH", str(root))
+    return root
+
+
+@pytest.fixture()
+def opened(monkeypatch: pytest.MonkeyPatch) -> List[str]:
+    """Every path handed to `open`, so "was it read" is answerable directly."""
+    seen: List[str] = []
+    real_open = builtins.open
+
+    def spy(file, *args, **kwargs):
+        seen.append(str(file))
+        return real_open(file, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "open", spy)
+    return seen
+
+
+@pytest.mark.asyncio
+async def test_an_absolute_path_outside_the_root_is_never_opened(
+    llc_root: pathlib.Path, tmp_path: pathlib.Path, opened: List[str]
+) -> None:
+    """The defect proper, with an allowlisted extension so only containment can refuse it."""
+    secret = _plant(tmp_path / "elsewhere" / "credentials.yaml", "token: hunter2\n")
+
+    text = await ArtifactIngestor()._resolve_text(_Product(str(secret)))
+
+    assert text is None, "the file outside the storage root must not be ingested"
+    assert str(secret) not in opened, (
+        "the file was opened before being refused -- a refusal after the read still leaks it "
+        "into memory and into any log that echoes content"
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_traversal_out_of_the_root_is_never_opened(
+    llc_root: pathlib.Path, tmp_path: pathlib.Path, opened: List[str]
+) -> None:
+    """The relative form, which an absolute-path check alone would miss."""
+    secret = _plant(tmp_path / "elsewhere" / "config.toml", "key = 'value'\n")
+    traversal = str(llc_root / ".." / "elsewhere" / "config.toml")
+
+    text = await ArtifactIngestor()._resolve_text(_Product(traversal))
+
+    assert text is None
+    assert not any("elsewhere" in path for path in opened)
+
+
+@pytest.mark.asyncio
+async def test_an_in_root_product_still_ingests(llc_root: pathlib.Path) -> None:
+    """The contrast, without which "refuses everything" passes every test above."""
+    product_file = _plant(llc_root / "tenant" / "artifact.md", "# real content\n")
+
+    text = await ArtifactIngestor()._resolve_text(_Product(str(product_file)))
+
+    assert text == "# real content\n"
+
+
+@pytest.mark.asyncio
+async def test_a_row_stored_before_the_fix_still_ingests_when_it_is_in_root(
+    llc_root: pathlib.Path,
+) -> None:
+    """#17302 AC5. The fix must not orphan rows that were always legitimate.
+
+    Containment is decided by where the path points, not by when the row was
+    written, so an existing in-root row is indistinguishable from a new one --
+    which is the property that makes this safe to deploy without a migration.
+    """
+    legacy = _plant(llc_root / "legacy-company" / "legacy-item" / "old.txt", "written before #17302\n")
+
+    assert await ArtifactIngestor()._resolve_text(_Product(str(legacy))) == "written before #17302\n"
+
+
+def test_the_returned_string_is_the_validated_one(llc_root: pathlib.Path) -> None:
+    """THREAT_MODEL section 1: the validated string is the string used.
+
+    Pinned because the bypass this shape invites is validating one path and then
+    opening something rebuilt from the original input. The helper returns the
+    resolved path so the caller cannot reach for the raw one, and this asserts
+    the redundant components are actually gone from what it hands back.
+    """
+    target = _plant(llc_root / "sub" / "file.txt", "x\n")
+    noisy = str(llc_root / "sub" / "." / "file.txt")
+
+    contained = _contained_storage_path(noisy, "product-1")
+
+    assert contained == str(target.resolve())
+    assert "/./" not in contained
+
+
+def test_an_extension_check_is_not_a_containment_check(llc_root: pathlib.Path) -> None:
+    """The two gates answer different questions, and one used to stand in for the other.
+
+    `_is_text_path` says yes to a path it has no business reading; that is not a
+    bug in `_is_text_path`, it is the reason it cannot be the only gate.
+    """
+    from llc.kb.artifact_ingestor import _is_text_path
+
+    outsider = "/etc/anything.yaml"
+    assert _is_text_path(outsider) is True, "premise: the allowlist admits it"
+    assert _contained_storage_path(outsider, "product-1") is None, "and containment refuses it"
+
+
+@pytest.mark.asyncio
+async def test_the_path_opened_is_the_decoded_one_not_the_raw_string(llc_root: pathlib.Path) -> None:
+    """THREAT_MODEL section 1, asserted at the CALL SITE rather than on the helper.
+
+    Written because a mutation found the gap: reverting the `open()` back to
+    `product.storage_path` left every other test in this file passing. They all
+    used paths whose raw and validated forms are identical, so "which string is
+    opened" was never observable -- the helper was tested, the wiring was not.
+
+    A percent-encoded in-root path makes the two forms differ. `_canonicalize`
+    decodes `%73ub` to `sub` before resolving (decode-before-resolve is the
+    house rule), so the validated path exists and the raw string does not. Open
+    the wrong one and this returns None instead of the content.
+    """
+    target = _plant(llc_root / "sub" / "file.txt", "real content\n")
+    encoded = str(llc_root / "%73ub" / "file.txt")
+
+    assert not pathlib.Path(encoded).exists(), "premise: the raw string names nothing on disk"
+
+    assert await ArtifactIngestor()._resolve_text(_Product(encoded)) == "real content\n"
+
+
+@pytest.mark.asyncio
+async def test_a_refusal_is_a_clean_skip_not_a_swallowed_exception(
+    llc_root: pathlib.Path, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The explicit `if contained is None` guard, which a mutation showed was unpinned.
+
+    Without it, a refused path falls through to `open(None)`, raises TypeError,
+    and is swallowed by the broad `except Exception` -- same return value, so
+    every assertion on the result still passed. The difference is entirely in
+    what an operator sees: a warning naming containment and #17302, versus an
+    exception traceback saying the file could not be read, which reads like a
+    missing file rather than a rejected one and sends the reader somewhere else.
+
+    The negative assertion names `unreadable storage_path`, the literal the
+    read-failure branch logs. That coupling is deliberate and was nearly lost:
+    when that message was reworded, this assertion went on checking for the OLD
+    text, which no longer appears anywhere -- passing unconditionally, for a
+    string that could never be present. A negative assertion against a literal
+    is only as good as the literal still existing, so `test_the_read_failure_log_literal_still_exists`
+    below pins it.
+    """
+    import logging
+
+    # `.ini`, not `.json`: #17302's text names `.json` and `.csv` as admitted
+    # extensions and neither is in `_TEXT_EXTENSIONS`. The real list still
+    # carries plenty -- `.ini`, `.cfg`, `.yaml`, `.toml`, `.sh`, `.py`, `.sql` --
+    # so the threat stands; only the issue's examples were off.
+    secret = _plant(tmp_path / "elsewhere" / "service.ini", "[auth]\ntoken = hunter2\n")
+
+    with caplog.at_level(logging.DEBUG):
+        assert await ArtifactIngestor()._resolve_text(_Product(str(secret))) is None
+
+    assert "#17302" in caplog.text, "the refusal must say why it refused"
+    assert (
+        "unreadable storage_path" not in caplog.text
+    ), "a refused path was reported as an unreadable file -- the operator is told the wrong thing"
+
+
+def test_the_read_failure_log_literal_still_exists() -> None:
+    """The negative assertion above is worthless if its literal is renamed.
+
+    `assert "x" not in caplog.text` passes for two different reasons: the
+    branch did not fire, or `x` is not what the branch says any more. Only the
+    first is a result. This pins the second so a rename fails here, loudly,
+    instead of quietly making the other test unconditional.
+    """
+    source = pathlib.Path(artifact_ingestor.__file__).read_text(encoding="utf-8")
+    assert "unreadable storage_path" in source, (
+        "the read-failure log message was renamed; update the negative assertion in "
+        "test_a_refusal_is_a_clean_skip_not_a_swallowed_exception to match, or it now asserts nothing"
+    )