Spaces:
Running
Running
File size: 3,812 Bytes
2341488 88aac15 2341488 88aac15 2341488 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | from __future__ import annotations
import os
import tempfile
import threading
import unittest
from pathlib import Path
from lumi_eggcracker.artifacts import from_argv_paths
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_argv_path_cannot_block_discovery(self) -> None:
"""An untrusted absolute argv path must be opened nonblocking."""
with tempfile.TemporaryDirectory() as raw:
fifo = Path(raw) / "hostile-argv"
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)
snapshot = type("Snapshot", (), {"argv": (str(fifo),)})()
outcome: dict[str, object] = {}
def probe() -> None:
try:
outcome["evidence"] = from_argv_paths(snapshot)
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:
writer = os.open(fifo, os.O_WRONLY | os.O_NONBLOCK | os.O_CLOEXEC)
worker.join(timeout=1.0)
try:
self.assertFalse(blocked, "opening an argv FIFO blocked discovery")
self.assertEqual((), outcome.get("evidence"))
self.assertNotIn("error", outcome)
finally:
if writer >= 0:
os.close(writer)
os.close(target_fd)
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()
|