Audit round two: four more ways the tool could be turned on its user

src/markproof/probes/document.py

@@ -243,10 +243,22 @@ def _manifest_for(
             manifest = fetch(client, "GET", url, stay_on_origin=True)
         except httpx.HTTPError as exc:
             raise ProbeError(f"{url}: the linked manifest could not be fetched — {exc}") from exc
-        if manifest.status_code >= 400:
+        if manifest.status_code != 200:
+            # Not `>= 400`: a 204 or a 304 carries no manifest either, and treating
+            # an empty body as "the manifest" would hand the checker nothing and
+            # call it a manifest.
             raise ProbeError(
                 f"{url}: the document advertises a manifest that answers HTTP "
                 f"{manifest.status_code}. A dangling provenance link is worse than none: "
                 "it reads as marked and verifies as nothing."
             )
+        if len(manifest.content) > self.config.max_bytes:
+            # The document has a size limit and the manifest did not, so the
+            # cheapest way to exhaust this process was to advertise one.
+            raise ProbeError(
+                f"{url}: the manifest is {len(manifest.content)} bytes, over max_bytes "
+                f"({self.config.max_bytes})"
+            )
+        if not manifest.content:
+            raise ProbeError(f"{url}: the advertised manifest is empty")
         return manifest.content, source

src/markproof/probes/http.py

@@ -34,7 +34,7 @@
 
 from markproof.probes.base import ProbeError
 
-__all__ = ["MAX_REDIRECTS", "fetch", "same_origin"]
+__all__ = ["MAX_REDIRECTS", "fetch", "is_internal_host", "same_origin"]
 
 #: Enough for the redirect chains real deployments have (canonical host, trailing
 #: slash, http→https), few enough that a loop ends as an error rather than a hang.
@@ -110,3 +110,35 @@ def fetch(
         current = target
 
     raise ProbeError(f"{url}: more than {MAX_REDIRECTS} redirects — refusing to follow further")
+
+
+def is_internal_host(url: str) -> bool:
+    """Whether a URL points at an address only the runner can reach.
+
+    Loopback, link-local, private and reserved ranges. The one that matters is
+    ``169.254.169.254``: on every major cloud that address answers with instance
+    credentials, and a probed endpoint choosing the URLs markproof fetches is a
+    server-side request forgery with a very good payload.
+
+    Hostnames that are not literal addresses are left alone. Resolving them here
+    would make the verdict depend on DNS at check time, and a resolver that
+    answers differently on two runs is exactly what the determinism claim rules
+    out — the defence for those belongs in the network the runner sits on.
+    """
+    import ipaddress
+
+    host = urlparse(url).hostname
+    if not host:
+        return False
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        return False
+    return (
+        address.is_loopback
+        or address.is_private
+        or address.is_link_local
+        or address.is_reserved
+        or address.is_multicast
+        or address.is_unspecified
+    )

src/markproof/probes/media.py

@@ -30,7 +30,7 @@
     Turn,
     sha256_hex,
 )
-from markproof.probes.http import fetch
+from markproof.probes.http import fetch, is_internal_host
 from markproof.rules.schema import ProbeKind
 
 __all__ = ["MediaProbe"]
@@ -178,8 +178,28 @@ def _from_base64(self, encoded: str, index: int) -> Artifact:
             media_type=self.config.expect_media_type or "image/png",
         )
 
+    def _refuse_internal(self, url: str) -> None:
+        """Refuse an asset URL the response body chose that points inward.
+
+        The endpoint under test supplies these URLs, so it decides what markproof
+        fetches. Pointed at 169.254.169.254 that is a request for cloud instance
+        credentials made by a process the operator trusts, from inside their
+        network.
+
+        Allowed when the probe's own target is internal too: pointing markproof at
+        a bot on localhost is the documented way to try it, and refusing the
+        assets that bot serves would break the demo the README sends people to.
+        """
+        if is_internal_host(url) and not is_internal_host(self.config.url):
+            raise ProbeError(
+                f"{self.config.url} returned an asset URL on an internal address ({url}). "
+                "Refusing to fetch it — the endpoint under test does not get to choose "
+                "what this process requests from inside your network."
+            )
+
     def _from_url(self, url: str, index: int, client: httpx.Client) -> Artifact:
         """Fetch an asset the way a browser would."""
+        self._refuse_internal(url)
         try:
             response = client.get(url)
         except httpx.HTTPError as exc:

src/markproof/report/sign.py

@@ -86,6 +86,13 @@ def generate_keypair(out_dir: Path) -> tuple[Path, Path]:
     )
 
     fd = os.open(private_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, stat.S_IRUSR | stat.S_IWUSR)
+    # The mode argument to os.open applies only when the file is *created*. A
+    # pre-existing key file — a previous run under a wider umask, a placeholder
+    # somebody touched — keeps its permissions through O_TRUNC, and the CLI then
+    # prints "(mode 600)" over a key anyone on the machine can read. fchmod on the
+    # open descriptor fixes that without a window in which the path could be
+    # swapped.
+    os.fchmod(fd, stat.S_IRUSR | stat.S_IWUSR)
     with os.fdopen(fd, "wb") as handle:
         handle.write(private_pem)
     public_path.write_bytes(public_pem)

src/markproof/rules/schema.py

@@ -212,6 +212,21 @@ class LabelScope(StrEnum):
     ANY_OUTPUT = "any_output"
 
 
+def _bare_data_filename(value: str, field: str) -> str:
+    """A packaged data file is named, not located.
+
+    The loader resolves these against the package's own directory, so anything
+    with a separator or a parent reference would read a file the rulepack author
+    chose rather than one that ships. Rejecting the shape is cheaper and more
+    obvious than sandboxing the read.
+    """
+    if value != Path(value).name or value in ("", ".", ".."):
+        raise ValueError(
+            f"{field} must be a bare filename of a packaged data file, not a path: {value!r}"
+        )
+    return value
+
+
 class Source(BaseModel):
     """One citable source behind a rulepack."""
 
@@ -234,6 +249,13 @@ class DisclosurePatternCheck(BaseModel):
 
     type: Literal["disclosure-pattern"]
     patterns_file: str = Field(min_length=1)
+    """Name of a packaged pattern file. A bare filename, never a path.
+
+    Validated rather than trusted: a rulepack is loaded from a path the operator
+    passes, so its contents are as trusted as that file is, and
+    ``../../../etc/shadow`` in this field would otherwise be opened and its
+    existence reported through the error message.
+    """
     position: Position = Position.ANYWHERE_IN_FIRST_RESPONSE
     min_matches: int = Field(default=1, ge=1)
     prompt_ids: list[str] | None = None
@@ -246,6 +268,11 @@ class DisclosurePatternCheck(BaseModel):
     which question exposed the problem.
     """
 
+    @field_validator("patterns_file")
+    @classmethod
+    def _bare_filename(cls, v: str) -> str:
+        return _bare_data_filename(v, "patterns_file")
+
     @field_validator("prompt_ids")
     @classmethod
     def _non_empty_prompt_ids(cls, v: list[str] | None) -> list[str] | None:
@@ -279,6 +306,11 @@ class LabelPresenceCheck(BaseModel):
     at the other's vocabulary.
     """
 
+    @field_validator("labels_file")
+    @classmethod
+    def _bare_filename(cls, v: str) -> str:
+        return _bare_data_filename(v, "labels_file")
+
     category: LabelCategory
     """Which duty this rule is about. Required, with no default: a rule that did
     not say would be satisfied by whichever notice happened to be on the page."""

tests/test_document_probe.py

@@ -258,3 +258,31 @@ def test_no_manifest_is_an_absence_not_an_unreadable_payload(self) -> None:
         result = self._verify("unsigned.html", None)
         assert result.outcome is C2paOutcome.MANIFEST_MISSING
         assert "cannot carry an embedded manifest" in (result.detail or "")
+
+
+class TestTheManifestFetchIsBounded:
+    """The document had a size limit and the manifest did not."""
+
+    @respx.mock
+    def test_an_oversized_manifest_is_refused(self) -> None:
+        respx.get(_URL).mock(
+            return_value=httpx.Response(
+                200,
+                content=_fixture("signed-valid.html"),
+                headers={"content-type": "text/html"},
+            )
+        )
+        respx.get(_MANIFEST_URL).mock(return_value=httpx.Response(200, content=b"x" * 5000))
+        with pytest.raises(ProbeError, match="over max_bytes"):
+            _probe(max_bytes=2048).collect()
+
+    @respx.mock
+    def test_an_empty_manifest_is_not_a_manifest(self) -> None:
+        respx.get(_URL).mock(
+            return_value=httpx.Response(
+                200, content=_fixture("signed-valid.html"), headers={"content-type": "text/html"}
+            )
+        )
+        respx.get(_MANIFEST_URL).mock(return_value=httpx.Response(204))
+        with pytest.raises(ProbeError, match="HTTP 204"):
+            _probe().collect()

tests/test_redirects.py

@@ -200,3 +200,78 @@ def test_a_manifest_may_not_be_redirected_off_origin(self) -> None:
         with pytest.raises(ProbeError, match="different origin"):
             probe.collect()
         assert not elsewhere.calls, "the foreign manifest was fetched anyway"
+
+
+class TestTheResponseBodyDoesNotChooseWhatIsFetched:
+    """The media probe fetches asset URLs the endpoint under test supplies.
+
+    Pointed at `169.254.169.254` — which on every major cloud answers with
+    instance credentials — that is a server-side request forgery made by a
+    process the operator trusts, from inside their network, with a very good
+    payload.
+    """
+
+    @pytest.mark.parametrize(
+        "target",
+        [
+            "http://169.254.169.254/latest/meta-data/",
+            "http://127.0.0.1:8080/admin",
+            "http://10.0.0.5/internal",
+            "http://[::1]/x",
+        ],
+    )
+    @respx.mock
+    def test_an_internal_asset_url_is_refused(self, target: str) -> None:
+        respx.post(f"{_A}/v1/images").mock(
+            return_value=httpx.Response(200, json={"data": [{"url": target}]})
+        )
+        reached = respx.get(target).mock(return_value=httpx.Response(200, content=b"secrets"))
+        probe = MediaProbe(
+            MediaProbeConfig.model_validate(
+                {"id": "images", "type": "media", "url": f"{_A}/v1/images"}
+            )
+        )
+        with pytest.raises(ProbeError, match="internal address"):
+            probe.collect()
+        assert not reached.calls, "the internal address was fetched anyway"
+
+    @respx.mock
+    def test_a_public_asset_url_is_still_fetched(self) -> None:
+        respx.post(f"{_A}/v1/images").mock(
+            return_value=httpx.Response(200, json={"data": [{"url": f"{_A}/a.png"}]})
+        )
+        respx.get(f"{_A}/a.png").mock(
+            return_value=httpx.Response(
+                200, content=b"\x89PNG\r\n\x1a\n", headers={"content-type": "image/png"}
+            )
+        )
+        probe = MediaProbe(
+            MediaProbeConfig.model_validate(
+                {"id": "images", "type": "media", "url": f"{_A}/v1/images"}
+            )
+        )
+        assert probe.collect().turns[0].artifacts
+
+    @respx.mock
+    def test_a_local_target_may_serve_local_assets(self) -> None:
+        """Pointing markproof at a bot on localhost is the documented way to try it.
+
+        Refusing the assets that bot serves would break the demo the README sends
+        people to, so the rule is relative: internal is refused only when the probe
+        target itself is not.
+        """
+        local = "http://127.0.0.1:8099"
+        respx.post(f"{local}/v1/images").mock(
+            return_value=httpx.Response(200, json={"data": [{"url": f"{local}/a.png"}]})
+        )
+        respx.get(f"{local}/a.png").mock(
+            return_value=httpx.Response(
+                200, content=b"\x89PNG\r\n\x1a\n", headers={"content-type": "image/png"}
+            )
+        )
+        probe = MediaProbe(
+            MediaProbeConfig.model_validate(
+                {"id": "images", "type": "media", "url": f"{local}/v1/images"}
+            )
+        )
+        assert probe.collect().turns[0].artifacts

tests/test_report.py

@@ -321,3 +321,31 @@ def test_both_renderers_state_the_same_limit(self) -> None:
 
         stripped = summary.MARKING_LIMB_NOTE.replace("**Article 50(2) has two limbs.** ", "")
         assert stripped == pdf_reportlab.MARKING_LIMB_NOTE
+
+
+class TestKeygenPermissionsSurviveAnExistingFile:
+    """`os.open`'s mode argument applies only on creation.
+
+    A key file left by an earlier run under a wider umask, or a placeholder
+    somebody touched, keeps its permissions through `O_TRUNC` — and the CLI then
+    prints "(mode 600)" over a private key anyone on the machine can read. A claim
+    printed next to a fact that contradicts it is the exact defect class this
+    project exists to find in other people's systems.
+    """
+
+    def test_a_pre_existing_world_readable_file_is_tightened(self, tmp_path: Path) -> None:
+        from markproof.report.sign import generate_keypair
+
+        target = tmp_path / "markproof-signing-key.pem"
+        target.write_text("placeholder", encoding="utf-8")
+        target.chmod(0o644)
+
+        private_path, _ = generate_keypair(tmp_path)
+        mode = private_path.stat().st_mode
+        assert not mode & (stat.S_IRWXG | stat.S_IRWXO), oct(mode & 0o777)
+
+    def test_a_fresh_file_is_owner_only(self, tmp_path: Path) -> None:
+        from markproof.report.sign import generate_keypair
+
+        private_path, _ = generate_keypair(tmp_path)
+        assert not private_path.stat().st_mode & (stat.S_IRWXG | stat.S_IRWXO)