fix: close final strict linkage review gaps

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/research_signal_context_pipelines/latest_linkage.py

@@ -6,6 +6,7 @@
 import errno
 import os
 import re
+import stat
 from pathlib import Path
 from typing import Any
 
@@ -15,6 +16,7 @@
 
 _SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$")
 _OPEN_SUPPORTS_DIR_FD = os.open in getattr(os, "supports_dir_fd", set())
+MAX_JSON_ARTIFACT_BYTES = 4 * 1024 * 1024
 
 
 def validate_latest_signal(
@@ -93,14 +95,18 @@ def _resolve_source_path(source: str, base: Path | None) -> Path:
 
 
 def _read_declaration(path: Path, base: Path | None, declared_source: str) -> bytes:
-    if getattr(os, "O_NOFOLLOW", None) is None or getattr(os, "O_DIRECTORY", None) is None:
+    if (
+        getattr(os, "O_NOFOLLOW", None) is None
+        or getattr(os, "O_DIRECTORY", None) is None
+        or getattr(os, "O_NONBLOCK", None) is None
+    ):
         raise SignalValidationError("secure descriptor-based source reading is unavailable")
     if not _OPEN_SUPPORTS_DIR_FD:
         raise SignalValidationError("secure openat source reading is unavailable")
     fd = -1
     try:
         if base is None:
-            fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+            fd = os.open(Path(declared_source), os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0))
         else:
             raw_path = Path(declared_source)
             try:
@@ -117,11 +123,18 @@ def _read_declaration(path: Path, base: Path | None, declared_source: str) -> by
                 fd = next_fd
             final_fd = -1
             try:
-                final_fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=fd)
+                final_fd = os.open(
+                    parts[-1], os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0), dir_fd=fd
+                )
             finally:
                 parent_fd = fd
                 fd = final_fd
                 os.close(parent_fd)
+        metadata = os.fstat(fd)
+        if not stat.S_ISREG(metadata.st_mode):
+            raise SignalValidationError("declaration source must be a regular file")
+        if metadata.st_size > MAX_JSON_ARTIFACT_BYTES:
+            raise SignalValidationError("declaration source exceeds maximum size")
         with os.fdopen(fd, "rb") as stream:
             fd = -1
             return stream.read()
@@ -181,5 +194,5 @@ def _parse_datetime(value: Any, name: str) -> dt.datetime:
     except ValueError as exc:
         raise SignalValidationError(f"{name} must be an ISO datetime") from exc
     if parsed.tzinfo is None:
-        return parsed.replace(tzinfo=dt.timezone.utc)
+        raise SignalValidationError(f"{name} must be an ISO datetime with an explicit timezone")
     return parsed.astimezone(dt.timezone.utc)

tests/test_latest_linkage.py

@@ -107,6 +107,15 @@ def test_final_symlink_is_rejected_without_following_it(tmp_path: Path) -> None:
         validate_latest_signal(signal_payload("alias.json"), signal_base_dir=tmp_path)
 
 
+def test_absolute_declared_symlink_is_rejected_without_resolving_identity(tmp_path: Path) -> None:
+    write_snapshot(tmp_path)
+    alias = tmp_path / "absolute-alias.json"
+    alias.symlink_to(tmp_path / "theme_momentum_snapshot.json")
+
+    with pytest.raises(SignalValidationError, match="no-follow|symlink"):
+        validate_latest_signal(signal_payload(str(alias)))
+
+
 def test_descriptor_read_rejects_symlink_swap_after_bounds_check(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     path = write_snapshot(tmp_path)
     outside = tmp_path.parent / "outside.json"
@@ -154,6 +163,35 @@ def open_once(*args: object, **kwargs: object) -> int:
     assert calls == 2
 
 
+@pytest.mark.parametrize("generated_at", ["2026-06-27", "2026-06-27T00:00:00"])
+def test_generated_at_requires_time_and_explicit_timezone(tmp_path: Path, generated_at: str) -> None:
+    payload = signal_payload()
+    payload["generated_at"] = generated_at
+    write_snapshot(tmp_path)
+
+    with pytest.raises(SignalValidationError, match="ISO datetime"):
+        validate_latest_signal(payload, signal_base_dir=tmp_path)
+
+    payload = signal_payload()
+    write_snapshot(tmp_path, generated_at=generated_at)
+    with pytest.raises(SignalValidationError, match="ISO datetime"):
+        validate_latest_signal(payload, signal_base_dir=tmp_path)
+
+
+def test_non_regular_and_oversized_declarations_fail_closed(tmp_path: Path) -> None:
+    fifo = tmp_path / "theme_momentum_snapshot.json"
+    os.mkfifo(fifo)
+    with pytest.raises(SignalValidationError, match="regular file"):
+        validate_latest_signal(signal_payload(), signal_base_dir=tmp_path)
+
+    fifo.unlink()
+    fifo.write_bytes(b"{}")
+    with fifo.open("ab") as stream:
+        stream.truncate(linkage.MAX_JSON_ARTIFACT_BYTES + 1)
+    with pytest.raises(SignalValidationError, match="maximum size"):
+        validate_latest_signal(signal_payload(), signal_base_dir=tmp_path)
+
+
 def test_bad_hash_and_as_of_mismatch_are_validation_errors(tmp_path: Path) -> None:
     path = write_snapshot(tmp_path)
     payload = signal_payload()