Per-deployment sensor: run FIFO + atime baits together (#100, dual-plant 1/3)

agent/thumper_agent.sh

@@ -68,6 +68,29 @@ probe_fifo_mode() {  # AUTO policy: default to FIFO on macOS only (Linux default
     [ "$(platform)" = "darwin" ] || return 0
     mkfifo_works && FIFO_MODE=1
 }
+# effective_sensor <i>: which sensor governs THIS bait. Precedence:
+#   1. an explicit operator --sensor (fifo|atime) - an intentional override that
+#      must win over the server (#164 F2): the operator opted out of/into FIFOs;
+#   2. the deployment's OWN sensor when the server sent one (dual-plant pairs);
+#   3. the platform default.
+# Lets one agent run a FIFO bait and an atime bait side by side.
+effective_sensor() {
+    case "$SENSOR" in fifo|atime) printf '%s' "$SENSOR"; return 0 ;; esac
+    eval "_es=\${dep_sensor_$1:-}"
+    [ -n "$_es" ] && { printf '%s' "$_es"; return 0; }
+    if [ "$FIFO_MODE" = 1 ]; then printf 'fifo'
+    elif [ "$(platform)" = linux ] && command -v inotifywait >/dev/null 2>&1; then printf 'inotify'
+    else printf 'atime'; fi
+}
+has_explicit_sensors() {  # 0 if any deployment carries its own sensor (server is sending pairs)
+    _i=1
+    while [ "$_i" -le "$DEP_COUNT" ]; do
+        eval "_s=\${dep_sensor_$_i:-}"
+        [ -n "$_s" ] && return 0
+        _i=$((_i + 1))
+    done
+    return 1
+}
 cache_path() { printf '%s/%s' "$BAITCACHE" "$1"; }   # cache_path <deployment-id>
 TAB=$(printf '\t')
 
@@ -258,14 +281,15 @@ pull_deployments() {
     oldifs=$IFS
     IFS="$TAB"
     # `printf | while` would subshell the counters away; feed via a here-doc.
-    while IFS="$TAB" read -r id path secret content_url callback_url; do
+    while IFS="$TAB" read -r id path secret content_url callback_url sensor; do
         [ -n "$id" ] || continue
         DEP_COUNT=$((DEP_COUNT + 1))
         eval "dep_id_$DEP_COUNT=\$id"
         eval "dep_path_$DEP_COUNT=\$(expand_path \"\$path\")"
         eval "dep_secret_$DEP_COUNT=\$secret"
         eval "dep_content_$DEP_COUNT=\$content_url"
         eval "dep_callback_$DEP_COUNT=\$callback_url"
+        eval "dep_sensor_$DEP_COUNT=\${sensor:-}"   # per-deployment sensor (fifo|atime|inotify); empty = use global default
         eval "dep_last_$DEP_COUNT=0"
     done <<EOF
 $body
@@ -344,7 +368,7 @@ plant() {  # plant <i>
         return 1
     fi
 
-    if [ "$FIFO_MODE" = 1 ]; then
+    if [ "$(effective_sensor "$1")" = fifo ]; then
         mkdir -p "$BAITCACHE"
         chmod 700 "$BAITCACHE" 2>/dev/null || true
         cf=$(cache_path "$id")
@@ -570,30 +594,37 @@ arm_atime() {  # arm_atime <path>: set atime to the past so the next read bumps
 read_atime() {  # read_atime <path>: portable access-time epoch (GNU %X first, then BSD %a - never %a on Linux, that's free blocks: #28)
     stat -c %X "$1" 2>/dev/null || stat -f %a "$1" 2>/dev/null || echo 0
 }
-watch_atime() {
-    log "atime poll every ${POLL}s on regular-file bait (re-armable; detection only - no process/user)"
-    i=1
-    while [ "$i" -le "$DEP_COUNT" ]; do
+all_indices() {  # "1 2 ... DEP_COUNT" - every deployment
+    _ai=""; _i=1
+    while [ "$_i" -le "$DEP_COUNT" ]; do _ai="$_ai $_i"; _i=$((_i + 1)); done
+    printf '%s' "$_ai"
+}
+# atime_poll "<idx idx ...>": arm + re-armable-poll only the given deployments.
+# The index list lets the mixed watcher poll just the atime baits while FIFO
+# baits are served separately; watch_atime() polls all (homogeneous + fallback).
+atime_poll() {
+    log "atime poll every ${POLL}s on regular-file bait(s) (re-armable; detection only - no process/user)"
+    # shellcheck disable=SC2086  # $1 is a space-separated index list; splitting is intended
+    for i in $1; do
         eval "p=\$dep_path_$i"
         arm_atime "$p"                                  # arm so relatime bumps atime on a read
         eval "atime_$i=\$(read_atime \"\$p\")"
-        i=$((i + 1))
     done
     while true; do
         sleep "$POLL"
-        i=1
-        while [ "$i" -le "$DEP_COUNT" ]; do
+        # shellcheck disable=SC2086
+        for i in $1; do
             eval "p=\$dep_path_$i prev=\$atime_$i"
             cur=$(read_atime "$p")
             if [ "$cur" != "0" ] && [ "$cur" -gt "$prev" ] 2>/dev/null; then
                 fire "$i" "atime-change" "" "" "" "$p"
                 arm_atime "$p"                          # RE-ARM so the NEXT read is detectable too
                 eval "atime_$i=\$(read_atime \"\$p\")"
             fi
-            i=$((i + 1))
         done
     done
 }
+watch_atime() { atime_poll "$(all_indices)"; }   # poll every bait (homogeneous atime mode + degradation fallback)
 
 # ── live sync (re-pull + reconcile) ───────────────────────────────────────────
 # A running agent re-pulls its deployment set every --sync-interval and applies
@@ -681,9 +712,32 @@ watch_fifo() {  # supervisor: keep one serve_fifo alive per bait; restart any th
     done
 }
 
+# Dual-plant: each deployment runs under its OWN sensor. FIFO baits (canonical,
+# definitive pid) are served individually; atime/inotify baits (companion,
+# detection) are atime-polled as a group. Used whenever the server sends pairs.
+watch_mixed() {
+    log "watching $DEP_COUNT bait(s) with per-deployment sensors"
+    _atidx=""; i=1
+    while [ "$i" -le "$DEP_COUNT" ]; do
+        if [ "$(effective_sensor "$i")" = fifo ]; then
+            serve_fifo "$i" &
+        else
+            _atidx="$_atidx $i"                         # atime/inotify/unknown -> atime poll (detection)
+        fi
+        i=$((i + 1))
+    done
+    [ -n "$_atidx" ] && atime_poll "$_atidx" &
+    wait
+    [ -e "${WATCH_STOP_FLAG:-/nonexistent}" ] && return 0
+    err "mixed watcher exited unexpectedly - degrading to atime poll"
+    atime_poll "$(all_indices)"
+}
+
 start_watcher() {  # launch the right sensor in the background; set WATCH_PID
     rm -f "${WATCH_STOP_FLAG:-}" 2>/dev/null || true   # this start is not a stop
-    if [ "$SENSOR" = atime ]; then
+    if has_explicit_sensors; then
+        watch_mixed &                                   # per-deployment sensors (dual-plant pairs)
+    elif [ "$SENSOR" = atime ]; then
         watch_atime &                                   # forced atime sensor (any platform)
     elif [ "$FIFO_MODE" = 1 ]; then
         watch_fifo &
@@ -769,7 +823,7 @@ verify_planted() {
             # and never re-plant through it (curl -o would write the target); report
             # failed so the lost coverage is visible.
             report_plant "$vid" failed
-        elif [ "$FIFO_MODE" = 1 ] && [ -e "$p" ] && ! [ -p "$p" ]; then
+        elif [ "$(effective_sensor "$i")" = fifo ] && [ -e "$p" ] && ! [ -p "$p" ]; then
             # A regular file where our FIFO should be = tampering/replacement.
             # Recover like the "missing" branch below: plant() removes the impostor
             # (our own path) and re-creates the FIFO, then REPLANTED restarts the

tests/test_agent_mixed.py

@@ -0,0 +1,196 @@
+"""Per-deployment sensor (#100 dual-plant, increment 1): each deployment record
+carries a 6th `sensor` field (fifo|atime). The agent plants and watches EACH
+bait per its own sensor, so a FIFO bait (canonical, definitive pid) and an atime
+bait (companion, normal-file detection) run side by side from one agent.
+
+macOS-gated: the pair includes a FIFO bait. The atime 'read' is simulated by
+os.utime() (deterministic); the FIFO read is a real open()."""
+
+import http.server
+import subprocess
+import threading
+import os
+import stat
+import time
+import platform as _platform
+from pathlib import Path
+import pytest
+
+pytestmark = pytest.mark.skipif(
+    _platform.system() != "Darwin", reason="pair includes a FIFO bait (macOS)"
+)
+
+AGENT = Path(__file__).resolve().parents[1] / "agent" / "thumper_agent.sh"
+BAIT_BODY = "AKIA-BAIT\nsecret=shhh\n"
+ARMED_MAX = 1_000_000_000
+
+
+class Stub(http.server.BaseHTTPRequestHandler):
+    callbacks = []  # (callback_path, body)
+    deployments = []  # list of (id, path, sensor)
+
+    def log_message(self, *a):
+        pass
+
+    def _t(self, body=""):
+        self.send_response(200)
+        self.send_header("Content-Type", "text/plain")
+        self.end_headers()
+        self.wfile.write(body.encode())
+
+    def do_POST(self):
+        n = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(n).decode()
+        if self.path == "/api/enroll":
+            return self._t("agent_token=tok-1\nendpoint_id=ep_1\n")
+        if self.path.startswith("/cb/"):
+            Stub.callbacks.append((self.path, body))
+            return self._t("ok")
+        if self.path.endswith("/state"):
+            return self._t("ok")
+        return self._t("ok")
+
+    def do_GET(self):
+        if self.path == "/api/agent/deployments":
+            base = f"http://127.0.0.1:{self.server.server_port}"
+            lines = [
+                "\t".join(
+                    [
+                        did,
+                        path,
+                        "sekret",
+                        f"{base}/content/{did}",
+                        f"{base}/cb/{did}",
+                        sensor,
+                    ]
+                )
+                for did, path, sensor in Stub.deployments
+            ]
+            return self._t("\n".join(lines) + "\n")
+        if self.path.startswith("/content/"):
+            return self._t(BAIT_BODY)
+        return self._t("")
+
+
+@pytest.fixture
+def server():
+    httpd = http.server.HTTPServer(("127.0.0.1", 0), Stub)
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+    Stub.callbacks = []
+    yield httpd
+    httpd.shutdown()
+
+
+def _spawn(server, tmp_path):
+    port = server.server_port
+    state = tmp_path / "agent.json"
+    # NO --sensor: the per-deployment field must govern, overriding the auto-probe.
+    return subprocess.Popen(
+        [
+            "sh",
+            str(AGENT),
+            "run",
+            "--server",
+            f"http://127.0.0.1:{port}",
+            "--enroll-token",
+            "e",
+            "--tripwire",
+            "tw_1",
+            "--state-file",
+            str(state),
+            "--poll",
+            "1",
+            "--heartbeat",
+            "0",
+            "--sync-interval",
+            "0",
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+
+
+def _wait(cond, t=15.0):
+    end = time.time() + t
+    while time.time() < end:
+        if cond():
+            return True
+        time.sleep(0.05)
+    return False
+
+
+def _fired(cb_id):
+    return any(cb_id in p for p, _ in Stub.callbacks)
+
+
+def test_mixed_sensors_plant_and_fire_together(server, tmp_path):
+    fifo = tmp_path / "credentials"  # canonical -> FIFO (pid)
+    atin = tmp_path / "config"  # companion -> regular file (atime detect)
+    Stub.deployments = [
+        ("dep_fifo", str(fifo), "fifo"),
+        ("dep_atime", str(atin), "atime"),
+    ]
+    agent = _spawn(server, tmp_path)
+    try:
+        # planted per its OWN sensor, not the global auto-probe
+        assert _wait(lambda: fifo.exists() and stat.S_ISFIFO(fifo.stat().st_mode)), (
+            "fifo-sensor bait was not planted as a named pipe"
+        )
+        assert _wait(
+            lambda: (
+                atin.exists() and atin.is_file() and atin.stat().st_atime < ARMED_MAX
+            )
+        ), "atime-sensor bait was not planted as an armed regular file"
+        # read the FIFO bait (blocks until the agent serves it) -> fires with pid
+        threading.Thread(target=lambda: open(fifo).read(), daemon=True).start()
+        # 'read' the atime bait -> fires (detection)
+        os.utime(atin, (time.time(), os.stat(atin).st_mtime))
+        assert _wait(lambda: _fired("/cb/dep_fifo")), "FIFO bait did not fire"
+        assert _wait(lambda: _fired("/cb/dep_atime")), "atime bait did not fire"
+    finally:
+        agent.terminate()
+        agent.wait(timeout=5)
+
+
+def test_explicit_sensor_overrides_server_per_deployment(server, tmp_path):
+    # Roee #164 F2: an operator's explicit --sensor atime is an intentional opt-out
+    # of FIFOs; it MUST win over the server's sensor=fifo, so the bait is planted as
+    # a regular file, never a named pipe.
+    bait = tmp_path / "credentials"
+    Stub.deployments = [("dep_1", str(bait), "fifo")]  # server asks for FIFO...
+    port = server.server_port
+    agent = subprocess.Popen(
+        [
+            "sh",
+            str(AGENT),
+            "run",
+            "--server",
+            f"http://127.0.0.1:{port}",
+            "--enroll-token",
+            "e",
+            "--tripwire",
+            "tw_1",
+            "--state-file",
+            str(tmp_path / "agent.json"),
+            "--sensor",
+            "atime",  # ...operator overrides to atime
+            "--poll",
+            "1",
+            "--heartbeat",
+            "0",
+            "--sync-interval",
+            "0",
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        assert _wait(lambda: bait.exists()), "bait was never planted"
+        assert bait.is_file() and not stat.S_ISFIFO(bait.stat().st_mode), (
+            "--sensor atime did not override server sensor=fifo (a pipe was planted)"
+        )
+    finally:
+        agent.terminate()
+        agent.wait(timeout=5)