fix: verify /proc/PID/cmdline in isRinging to prevent PID reuse false positives
This commit is contained in:
+6
-3
@@ -19,12 +19,15 @@ def is_wecker_ringing() -> bool:
|
||||
if not os.path.exists(PID_FILE):
|
||||
return False
|
||||
try:
|
||||
pid: int
|
||||
with open(PID_FILE) as f:
|
||||
pid = int(f.read().strip())
|
||||
os.kill(pid, 0)
|
||||
return True
|
||||
except (ValueError, ProcessLookupError, PermissionError, OSError):
|
||||
# Guard against PID reuse: verify the running process is actually wecker.py.
|
||||
cmdline_path = f"/proc/{pid}/cmdline"
|
||||
with open(cmdline_path, "rb") as f:
|
||||
cmdline = f.read().replace(b"\0", b" ").decode()
|
||||
return "wecker.py" in cmdline
|
||||
except (ValueError, ProcessLookupError, PermissionError, OSError, FileNotFoundError):
|
||||
return False
|
||||
|
||||
|
||||
|
||||
+28
-1
@@ -1,4 +1,4 @@
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import patch, MagicMock
|
||||
from fastapi.testclient import TestClient
|
||||
import tempfile
|
||||
import os
|
||||
@@ -15,6 +15,33 @@ from api.main import app # noqa: E402
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_is_ringing_false_on_pid_reuse():
|
||||
"""isRinging returns False when the PID file points to a non-wecker process."""
|
||||
from api.schema import is_wecker_ringing
|
||||
|
||||
fake_pid = 9999
|
||||
with patch("api.schema.PID_FILE", "/tmp/fake_reuse.pid"), \
|
||||
patch("api.schema.os.path.exists", return_value=True), \
|
||||
patch("api.schema.os.kill") as mock_kill, \
|
||||
patch("builtins.open") as mock_open:
|
||||
mock_kill.return_value = None
|
||||
|
||||
pid_file_mock = MagicMock()
|
||||
pid_file_mock.read.return_value = str(fake_pid)
|
||||
cmdline_mock = MagicMock()
|
||||
cmdline_mock.read.return_value = b"systemd-journald"
|
||||
|
||||
def open_side_effect(path, *args, **kwargs):
|
||||
if str(path).endswith("fake_reuse.pid"):
|
||||
return pid_file_mock
|
||||
if str(path) == f"/proc/{fake_pid}/cmdline":
|
||||
return cmdline_mock
|
||||
raise FileNotFoundError(path)
|
||||
|
||||
mock_open.side_effect = open_side_effect
|
||||
assert is_wecker_ringing() is False
|
||||
|
||||
|
||||
def test_is_ringing_returns_false_when_not_running():
|
||||
"""Test that isRinging returns False when no wecker process is running."""
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
|
||||
Reference in New Issue
Block a user