fix(security): probe argument-injection + same-owner alert routing

app/core/services/escalation_service.py

@@ -148,6 +148,9 @@ async def due_escalations(
                 if elapsed_minutes < step.delay_minutes:
                     break  # later steps have larger delays
                 channel = await self._session.get(NotificationChannel, step.channel_id)
+                # Defense-in-depth: never deliver to a channel of a different owner.
+                if channel is not None and channel.owner_id != problem.owner_id:
+                    channel = None
                 if channel is not None and channel.is_enabled:
                     # Auto-remediation must reach a machine endpoint, never an inbox.
                     if step.action_command is not None and channel.type != ChannelType.WEBHOOK:

app/tasks/notifications/dispatcher.py

@@ -9,6 +9,7 @@
 
 from app.core.config import get_settings
 from app.core.db.session import SessionLocal
+from app.core.models.host import Host
 from app.core.models.monitor import Monitor
 from app.core.models.notification_channel import (
     ChannelType,
@@ -150,9 +151,12 @@ async def _channels_for_monitor(monitor_id: uuid.UUID) -> list[NotificationChann
         stmt = (
             select(NotificationChannel)
             .join(monitor_channels, monitor_channels.c.channel_id == NotificationChannel.id)
+            .join(Monitor, Monitor.id == monitor_channels.c.monitor_id)
             .where(
                 monitor_channels.c.monitor_id == monitor_id,
                 NotificationChannel.is_enabled.is_(True),
+                # Defense-in-depth: only ever route to the monitor owner's channels.
+                NotificationChannel.owner_id == Monitor.owner_id,
             )
         )
         result = await session.execute(stmt)
@@ -164,9 +168,12 @@ async def _channels_for_host(host_id: uuid.UUID) -> list[NotificationChannel]:
         stmt = (
             select(NotificationChannel)
             .join(host_channels, host_channels.c.channel_id == NotificationChannel.id)
+            .join(Host, Host.id == host_channels.c.host_id)
             .where(
                 host_channels.c.host_id == host_id,
                 NotificationChannel.is_enabled.is_(True),
+                # Defense-in-depth: only ever route to the host owner's channels.
+                NotificationChannel.owner_id == Host.owner_id,
             )
         )
         result = await session.execute(stmt)

app/tasks/probes.py

@@ -174,6 +174,7 @@ async def _probe_ping(url: str) -> ProbeOutcome:
             "1",
             "-W",
             str(int(DEFAULT_TIMEOUT_SECONDS)),
+            "--",  # end of options: a target starting with '-' can't be read as a flag
             target,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,

tests/test_escalation.py

@@ -303,3 +303,42 @@ async def test_api_rejects_remediation_on_non_webhook(
     )
     assert resp.status_code == 422
     assert "webhook" in resp.text.lower()
+
+
+async def test_due_escalations_skip_foreign_owner_channel(session: Any, user: Any) -> None:
+    from app.core.models.user import AuthProvider, User
+    from app.core.security.passwords import hash_password
+
+    bob = User(
+        email="[email protected]",
+        full_name="Bob",
+        auth_provider=AuthProvider.LOCAL,
+        password_hash=hash_password("bob-secret-password"),
+        is_active=True,
+    )
+    session.add(bob)
+    await session.flush()
+    bob_channel = NotificationChannel(
+        name="bob-hook",
+        type=ChannelType.WEBHOOK,
+        config={"url": "https://hook.invalid/bob"},
+        owner_id=bob.id,
+        min_severity=Severity.INFO,
+    )
+    session.add(bob_channel)
+    await session.flush()
+
+    trig = await _trigger(session, user.id)
+    # Force a policy that targets another owner's channel (the API would reject this).
+    await EscalationService(session).create(
+        user.id,
+        EscalationPolicyCreate(
+            name="leaky",
+            steps=[EscalationStepCreate(step_order=1, delay_minutes=0, channel_id=bob_channel.id)],
+        ),
+    )
+    await _problem(session, user.id, trig.id, NOW - timedelta(minutes=1))
+    await session.commit()
+
+    # The engine must not route user A's problem through user B's channel.
+    assert await EscalationService(session).due_escalations(NOW) == []

tests/test_probes.py

@@ -291,3 +291,33 @@ async def test_probe_snmp_invalid_target() -> None:
     outcome = await _probe_snmp("")
     assert outcome.status == ProbeStatus.DOWN
     assert "invalid snmp target" in (outcome.error or "")
+
+
+async def test_ping_inserts_argv_separator_before_target(monkeypatch: pytest.MonkeyPatch) -> None:
+    import asyncio as _asyncio
+
+    from app.tasks import probes
+
+    captured: dict[str, tuple[str, ...]] = {}
+
+    class _FakeProc:
+        returncode = 0
+
+        async def communicate(self) -> tuple[bytes, bytes]:
+            return (b"64 bytes time=1.2 ms", b"")
+
+        def kill(self) -> None:  # pragma: no cover
+            pass
+
+    async def _fake_exec(*args: str, **_: object) -> _FakeProc:
+        captured["args"] = args
+        return _FakeProc()
+
+    monkeypatch.setattr(_asyncio, "create_subprocess_exec", _fake_exec)
+    outcome = await probes._probe_ping("8.8.8.8")
+    assert outcome.status is ProbeStatus.UP
+
+    args = captured["args"]
+    # "--" must precede the (user-controlled) target so a '-x' target can't be a flag.
+    assert "--" in args
+    assert args.index("--") < args.index("8.8.8.8")

tests/test_zk.py

@@ -18,7 +18,10 @@
     is_passphrase_token,
 )
 
-runner = CliRunner()
+# Clear the zk env-var fallbacks so an inherited GHOSTMON_ZK_* (which would make a
+# `--key` invocation also see a passphrase, tripping the mutual-exclusion check)
+# can't make these CLI tests order-dependent.
+runner = CliRunner(env={"GHOSTMON_ZK_KEY": None, "GHOSTMON_ZK_PASSPHRASE": None})
 
 
 def test_crypto_roundtrip_and_wrong_key() -> None: