fix: prevent FIFO descriptors from blocking discovery

src/lumi_eggcracker/procfd.py

@@ -306,7 +306,16 @@ def open_process_fd(
     """
     if isinstance(number, bool) or not isinstance(number, int) or number < 0:
         raise JsonInputError("target descriptor number is invalid")
-    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
+    # A target can deliberately expose a FIFO, device, or another descriptor
+    # whose open would wait for external activity.  Discovery must never let
+    # an untrusted workload block the supervisor before the descriptor has
+    # been validated as a regular file.  O_NONBLOCK is inert for ordinary
+    # files and lets the caller reject non-regular descriptors after fstat.
+    flags = (
+        os.O_RDONLY
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NONBLOCK", 0)
+    )
     path = proc / str(value.pid) / "fd" / str(number)
     try:
         descriptor = os.open(path, flags)

tests/test_procfd.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from lumi_eggcracker.discovery import identity
+from lumi_eggcracker.jsonio import JsonInputError
+from lumi_eggcracker.procfd import open_process_fd
+
+
+@unittest.skipUnless(os.name == "posix" and hasattr(os, "mkfifo"), "requires POSIX FIFOs")
+class ProcessDescriptorTests(unittest.TestCase):
+    def test_fifo_descriptor_reopen_cannot_block_discovery(self) -> None:
+        """A hostile FIFO descriptor must not stop autonomous scan progress."""
+        with tempfile.TemporaryDirectory() as raw:
+            fifo = Path(raw) / "hostile-fd"
+            os.mkfifo(fifo, mode=0o600)
+            keeper = os.open(fifo, os.O_RDWR | os.O_NONBLOCK | os.O_CLOEXEC)
+            target_fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK | os.O_CLOEXEC)
+            os.close(keeper)
+            process = identity(os.getpid())
+            self.assertIsNotNone(process)
+            assert process is not None
+
+            outcome: dict[str, object] = {}
+
+            def probe() -> None:
+                try:
+                    outcome["descriptor"] = open_process_fd(process, target_fd)[0]
+                except (JsonInputError, OSError) as error:
+                    outcome["error"] = error
+
+            worker = threading.Thread(target=probe, daemon=True)
+            writer = -1
+            worker.start()
+            worker.join(timeout=0.5)
+            blocked = worker.is_alive()
+            if blocked:
+                # Unblock an unfixed implementation so this regression cannot
+                # strand the test process after reporting the failure.
+                writer = os.open(fifo, os.O_WRONLY | os.O_NONBLOCK | os.O_CLOEXEC)
+                worker.join(timeout=1.0)
+            try:
+                self.assertFalse(blocked, "reopening a FIFO descriptor blocked discovery")
+                self.assertNotIn("error", outcome)
+            finally:
+                descriptor = outcome.get("descriptor")
+                if isinstance(descriptor, int):
+                    os.close(descriptor)
+                if writer >= 0:
+                    os.close(writer)
+                os.close(target_fd)
+
+
+if __name__ == "__main__":
+    unittest.main()