Verify mimic listeners with one lsof pass and give port-forward RPCs time

app/Sources/SquirrelOpsHelper/RPCMethods.swift

@@ -371,16 +371,14 @@ func registerMethods(router: RPCRouter) {
                     "Sensor service account is unavailable or changed"
                 )
             }
-            for listener in backendListeners {
-                try requireSensorOwnedListener(
-                    listener,
-                    serviceUID: sensorUID
-                ) { executable, arguments in
-                    try runCommand(
-                        executable: executable,
-                        arguments: arguments
-                    )
-                }
+            try requireSensorOwnedListeners(
+                backendListeners,
+                serviceUID: sensorUID
+            ) { executable, arguments in
+                try runCommand(
+                    executable: executable,
+                    arguments: arguments
+                )
             }
         }
 
@@ -432,16 +430,14 @@ func registerMethods(router: RPCRouter) {
                         "Sensor service account disappeared or changed"
                     )
                 }
-                for listener in backendListeners {
-                    try requireSensorOwnedListener(
-                        listener,
-                        serviceUID: sensorUID
-                    ) { executable, arguments in
-                        try runCommand(
-                            executable: executable,
-                            arguments: arguments
-                        )
-                    }
+                try requireSensorOwnedListeners(
+                    backendListeners,
+                    serviceUID: sensorUID
+                ) { executable, arguments in
+                    try runCommand(
+                        executable: executable,
+                        arguments: arguments
+                    )
                 }
             } catch {
                 try quarantinePortForwardingAfterListenerRace(
@@ -1498,34 +1494,64 @@ func requireSensorOwnedListener(
     serviceUID: uid_t,
     run: (String, [String]) throws -> CommandResult
 ) throws {
-    let result = try run(
-        "/usr/sbin/lsof",
-        [
-            "-n",
-            "-P",
-            "-l",
-            "-a",
-            "-iTCP:\(listener.port)",
-            "-sTCP:LISTEN",
-            "-Fpun",
-        ]
-    )
-    guard result.status == 0,
+    try requireSensorOwnedListeners([listener], serviceUID: serviceUID, run: run)
+}
+
+/// Verify every backend listener with one lsof pass.
+///
+/// A fleet-wide anchor load names every mimic's listeners, and lsof walks every
+/// process on the machine per invocation. One invocation per listener, twice
+/// per load, is what let a 30-second RPC time out behind a long ARP scan and
+/// then reported the torn-down listener as a wrong-owner bind. One invocation
+/// covers all ports; the records are grouped by port and each listener is
+/// judged on its own, with the failure naming the endpoint and the reason.
+func requireSensorOwnedListeners(
+    _ listeners: [PFBackendListener],
+    serviceUID: uid_t,
+    run: (String, [String]) throws -> CommandResult
+) throws {
+    if listeners.isEmpty { return }
+    var arguments = ["-n", "-P", "-l", "-a"]
+    for port in Set(listeners.map(\.port)).sorted() {
+        arguments.append("-iTCP:\(port)")
+    }
+    arguments.append(contentsOf: ["-sTCP:LISTEN", "-Fpun"])
+    let result = try run("/usr/sbin/lsof", arguments)
+    // lsof exits 1 when a selection matches nothing; that is a missing
+    // listener, judged per endpoint below, not a failure to run lsof.
+    guard result.status == 0 || result.status == 1,
           result.stderr.utf8.count <= maximumLSOFOutputBytes else {
         throw RPCError.internalError(
             "Could not verify mimic backend listener ownership"
         )
     }
     let records = try parseLSOFListeners(result.stdout)
-    let expectedEndpoint = "\(listener.ip):\(listener.port)"
-    guard !records.isEmpty,
-          records.allSatisfy({
-              $0.uid == serviceUID && $0.endpoint == expectedEndpoint
-          }),
-          Set(records.map(\.pid)).count == 1 else {
-        throw RPCError.internalError(
-            "Mimic backend listener is not an exact sensor-owned VIP bind"
-        )
+
+    for listener in listeners {
+        let expectedEndpoint = "\(listener.ip):\(listener.port)"
+        let suffix = ":\(listener.port)"
+        let matching = records.filter { $0.endpoint.hasSuffix(suffix) }
+        if matching.isEmpty {
+            throw RPCError.internalError(
+                "Mimic backend listener \(expectedEndpoint) has no listening socket"
+            )
+        }
+        if let foreign = matching.first(where: { $0.uid != serviceUID }) {
+            throw RPCError.internalError(
+                "Mimic backend listener \(expectedEndpoint) is owned by uid \(foreign.uid), not the sensor"
+            )
+        }
+        if let misbound = matching.first(where: { $0.endpoint != expectedEndpoint }) {
+            throw RPCError.internalError(
+                "Mimic backend listener \(expectedEndpoint) is bound to \(misbound.endpoint), not the exact VIP"
+            )
+        }
+        let pids = Set(matching.map(\.pid))
+        if pids.count != 1 {
+            throw RPCError.internalError(
+                "Mimic backend listener \(expectedEndpoint) is held by \(pids.count) processes, not one"
+            )
+        }
     }
 }
 

app/Tests/SquirrelOpsHelperTests/PFIsolationTests.swift

@@ -2208,3 +2208,75 @@ struct PFIsolationTests {
         return cache
     }
 }
+
+@Suite("Batched listener ownership")
+struct BatchedListenerOwnershipTests {
+    private let listeners = [
+        PFBackendListener(ip: "192.168.1.214", port: 50402),
+        PFBackendListener(ip: "192.168.1.216", port: 58449),
+    ]
+
+    @Test("One lsof invocation covers every backend listener")
+    func oneInvocationForAllListeners() throws {
+        var calls = 0
+        try requireSensorOwnedListeners(listeners, serviceUID: 309) { executable, arguments in
+            calls += 1
+            #expect(executable == "/usr/sbin/lsof")
+            #expect(arguments.contains("-iTCP:50402") && arguments.contains("-iTCP:58449"))
+            #expect(arguments.contains("-sTCP:LISTEN"))
+            return CommandResult(
+                status: 0,
+                stdout: "p123\nu309\nf10\nn192.168.1.214:50402\nf11\nn192.168.1.216:58449\n",
+                stderr: ""
+            )
+        }
+        #expect(calls == 1, "a fleet-wide PF load must not scan every process once per listener")
+    }
+
+    @Test("A missing listener is reported as missing, by endpoint")
+    func missingListenerIsNamed() {
+        let text = ownershipFailureText(stdout: "p123\nu309\nf10\nn192.168.1.214:50402\n")
+        #expect(text.contains("192.168.1.216:58449"), "\(text)")
+        #expect(text.contains("no listening socket"), "\(text)")
+    }
+
+    @Test("A foreign owner, a wrong bind, and a second pid are each named")
+    func failureKindsAreDistinct() {
+        let cases: [(String, String)] = [
+            ("p123\nu0\nf10\nn192.168.1.214:50402\nf11\nn192.168.1.216:58449\n", "owned by uid 0"),
+            ("p123\nu309\nf10\nn*:50402\nf11\nn192.168.1.216:58449\n", "bound to *:50402"),
+            ("p123\nu309\nf10\nn192.168.1.214:50402\np124\nu309\nf11\nn192.168.1.214:50402\nf12\nn192.168.1.216:58449\n", "2 processes"),
+        ]
+        for (stdout, expected) in cases {
+            let text = ownershipFailureText(stdout: stdout)
+            #expect(text.contains("192.168.1.214:50402"), "\(text)")
+            #expect(text.contains(expected), "\(text)")
+        }
+    }
+
+    /// Runs the batched check against canned lsof output and returns the
+    /// RPCError text. Captured with do/catch rather than the value of
+    /// `#expect(throws:)`, which is Void on the Swift 6.0 toolchain CI pins.
+    private func ownershipFailureText(stdout: String) -> String {
+        do {
+            try requireSensorOwnedListeners(listeners, serviceUID: 309) { _, _ in
+                CommandResult(status: 0, stdout: stdout, stderr: "")
+            }
+            Issue.record("Expected the listener ownership check to fail")
+            return ""
+        } catch let error as RPCError {
+            return String(describing: error)
+        } catch {
+            Issue.record("Expected an RPCError, got \(type(of: error))")
+            return ""
+        }
+    }
+
+    @Test("The single-listener check is the batched check with one entry")
+    func singleListenerDelegates() throws {
+        try requireSensorOwnedListener(listeners[0], serviceUID: 309) { _, arguments in
+            #expect(arguments.filter { $0.hasPrefix("-iTCP:") } == ["-iTCP:50402"])
+            return CommandResult(status: 0, stdout: "p123\nu309\nf10\nn192.168.1.214:50402\n", stderr: "")
+        }
+    }
+}

sensor/src/squirrelops_home_sensor/privileged/xpc.py

@@ -40,6 +40,14 @@
 SERVICE_SCAN_MAX_PROBES = 4096
 
 
+# Port-forward RPCs are the helper's slowest calls: it verifies every backend
+# listener with lsof before and after loading the PF anchor, and it serves
+# requests one at a time, so these calls can queue behind a full ARP scan. The
+# generic 30 s budget timed out in the field and the sensor rolled a healthy
+# mimic back while the helper was still verifying it.
+PORT_FORWARD_RPC_TIMEOUT = 180.0
+
+
 class MacOSPrivilegedOps(PrivilegedOperations):
     """macOS operations, delegating only privileged work to the helper.
 
@@ -339,6 +347,7 @@ async def setup_port_forwards(
                     "protected_endpoints": protected_endpoints or [],
                     "interface": interface,
                 },
+                timeout=PORT_FORWARD_RPC_TIMEOUT,
             )
             count = result.get("rules_count", 0)
             logger.info("Port forwarding: %d pfctl rules loaded", count)
@@ -350,7 +359,7 @@ async def setup_port_forwards(
     async def clear_port_forwards(self) -> bool:
         """Delegate pfctl port forward cleanup to the helper."""
         try:
-            result = await self._call("clearPortForwards")
+            result = await self._call("clearPortForwards", timeout=PORT_FORWARD_RPC_TIMEOUT)
             return result.get("success", False)
         except Exception:
             logger.exception("Failed to clear port forwards via helper")

sensor/tests/unit/test_package_security.py

@@ -438,7 +438,7 @@ def test_helper_pf_records_live_rules_before_post_load_owner_check() -> None:
         source.index('router.handlers["clearPortForwards"]')
     ]
     assert setup.index("recordSuccessfulLiveMutation(") < setup.index(
-        "requireSensorOwnedListener("
+        "requireSensorOwnedListeners("
     )
 
 
@@ -462,7 +462,7 @@ def test_helper_pf_mutations_require_owned_vips_and_sensor_listeners() -> None:
     validate = setup.index("validatePortForwardRequest(")
     build = setup.index("buildPFRules(")
     revalidate = setup.index("revalidatePortForwardOwnership(")
-    listener = setup.index("requireSensorOwnedListener(")
+    listener = setup.index("requireSensorOwnedListeners(")
     network_recheck = setup.index("requireUnchangedPFNetworkContext(")
     final_address_check = setup.index(
         "guard let currentInterfaceAddresses = ipv4InterfaceAddresses()"
@@ -490,7 +490,12 @@ def test_helper_pf_mutations_require_owned_vips_and_sensor_listeners() -> None:
     assert "ownedIPs.isSubset(of: endpointIPs)" in source
     assert '"/usr/sbin/lsof"' in source
     assert '"-sTCP:LISTEN"' in source
-    assert "$0.uid == serviceUID && $0.endpoint == expectedEndpoint" in source
+    # The batched check judges each listener on three conditions, each with
+    # its own failure text: owner, exact endpoint, and a single process.
+    assert "$0.uid != serviceUID" in source
+    assert "$0.endpoint != expectedEndpoint" in source
+    assert "pids.count != 1" in source
+    assert "has no listening socket" in source
     assert "quarantinePortForwardingAfterListenerRace(" in setup
     assert "requireNoLocalIPv4AddressConflicts(" in source
     assert "cleanupPFStates(for: [ip]" in source

sensor/tests/unit/test_privileged.py

@@ -966,3 +966,34 @@ async def never_respond(reader, writer):
             await server.wait_closed()
             if os.path.exists(sock_path):
                 os.unlink(sock_path)
+
+
+class TestMacOSPortForwardTimeout:
+    """Port-forward RPCs are the helper's slowest calls and must outlast a scan.
+
+    The helper is single-threaded and verifies every backend listener with
+    lsof before and after loading the PF anchor, queued behind whatever ARP
+    scan is in flight. A 30 s default timed out in the field; the sensor then
+    rolled the mimic back while the helper was still verifying it.
+    """
+
+    @pytest.mark.asyncio
+    async def test_setup_and_clear_use_the_long_timeout(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from squirrelops_home_sensor.privileged.xpc import (
+            PORT_FORWARD_RPC_TIMEOUT,
+            MacOSPrivilegedOps,
+        )
+
+        ops = MacOSPrivilegedOps.__new__(MacOSPrivilegedOps)
+        ops._call = AsyncMock(return_value={"success": True, "rules_count": 3})  # type: ignore[method-assign]
+
+        assert await ops.setup_port_forwards(
+            [{"from_ip": "192.168.1.200", "from_port": 80, "to_ip": "192.168.1.200", "to_port": 50080}]
+        )
+        assert await ops.clear_port_forwards()
+
+        assert PORT_FORWARD_RPC_TIMEOUT >= 120.0
+        for call in ops._call.await_args_list:
+            assert call.kwargs.get("timeout") == PORT_FORWARD_RPC_TIMEOUT, call