fix(ci): reject non-regular scanner reports without blocking

Signed-off-by: Adrien Langou <alangou@nvidia.com>

Author: alangou

tasks/scripts/codex_security_diagnostics.py

@@ -10,6 +10,7 @@
 import json
 import os
 import shutil
+import stat
 import subprocess
 import time
 from pathlib import Path
@@ -18,9 +19,15 @@
 def read_document(path: Path) -> tuple[dict, str]:
     # Never echo parser errors, paths, or arbitrary scanner-generated text.
     try:
-        if path.is_symlink():
+        if not stat.S_ISREG(path.lstat().st_mode):
             return {}, "invalid"
-        with path.open("rb") as stream:
+        # Check the opened file too: the path may change after lstat().
+        flags = (
+            os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0)
+        )
+        with os.fdopen(os.open(path, flags), "rb") as stream:
+            if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
+                return {}, "invalid"
             data = stream.read(4 * 1024 * 1024 + 1)
         if len(data) > 4 * 1024 * 1024:
             return {}, "oversized"

tasks/scripts/codex_security_diagnostics_test.py

@@ -128,6 +128,43 @@ def test_missing_invalid_and_symlinked_reports_do_not_echo_contents(tmp_path):
     assert coverage_summary(tmp_path)["coverage_document"] == "invalid"
 
 
+@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="requires named pipes")
+@pytest.mark.parametrize(
+    ("filename", "field"),
+    [
+        ("coverage.json", "coverage_document"),
+        ("scan-manifest.json", "manifest_document"),
+    ],
+)
+def test_fifo_report_does_not_block_diagnostics(tmp_path, filename, field):
+    scan = tmp_path / "scan"
+    scan.mkdir()
+    os.mkfifo(scan / filename)
+    output = tmp_path / "diagnostics.json"
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(SCRIPT),
+            "--scan-dir",
+            str(scan),
+            "--output",
+            str(output),
+            "--",
+            sys.executable,
+            "-c",
+            "raise SystemExit(2)",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=5,
+    )
+    assert result.returncode == 2
+    report = json.loads(output.read_text())
+    assert report["exit_code"] == 2
+    assert report[field] == "invalid"
+
+
 def test_scan_output_directory_stays_empty_until_scanner_writes(tmp_path):
     scan = tmp_path / "scan"
     output = tmp_path / "diagnostics.json"