feat: robust argv-based domain blocking

- Connect-time check: emit CONNECT_ATTEMPT for non-blocked-ips connects,
  userspace checks cmdline for blocked domains and kills before connect completes
- Domain rules checked first (before malicious_hashes, before LIN-*)
- Exempt alert dispatch (openclaw agent + ClawEDR Alert) to prevent cascade
- Reduce perf buffer poll timeout 50ms->5ms for faster event processing
- _blocked_domains for fast substring check

Made-with: Cursor

Author: leos565

deploy/linux/bpf_hooks.c

@@ -32,8 +32,8 @@ struct event_t {
   char comm[TASK_COMM_LEN];
   char filename[MAX_FILENAME_LEN];
   u8 action; // 0 = observed (enter), 1 = blocked (SIGKILL), 2 = post-exec
-             // (exit, for deny_rules)
-  u32 blocked_ip; // for connect events: blocked IP in host byte order
+             // (exit, for deny_rules), 3 = connect_attempt (userspace domain check)
+  u32 blocked_ip; // for connect events: IP in host byte order
 };
 
 /* --- Maps populated by monitor.py --- */
@@ -256,20 +256,25 @@ TRACEPOINT_PROBE(syscalls, sys_enter_connect) {
                             4); /* offset of sin_addr in sockaddr_in */
 
     u8 *is_blocked = blocked_ips.lookup(&ip);
-    if (is_blocked) {
-      struct event_t evt = {};
-      evt.pid = pid;
-      evt.ns_pid = get_ns_pid();
-      evt.uid = bpf_get_current_uid_gid() & 0xFFFFFFFF;
-      bpf_get_current_comm(&evt.comm, sizeof(evt.comm));
+    struct event_t evt = {};
+    evt.pid = pid;
+    evt.ns_pid = get_ns_pid();
+    evt.uid = bpf_get_current_uid_gid() & 0xFFFFFFFF;
+    bpf_get_current_comm(&evt.comm, sizeof(evt.comm));
+    evt.blocked_ip = ip;
 
+    if (is_blocked) {
       const char msg[] = "NETWORK_CONNECT";
       __builtin_memcpy((void *)evt.filename, msg, sizeof(msg));
-      evt.blocked_ip = ip;
-
       evt.action = 1;
       events.perf_submit(args, &evt, sizeof(evt));
       bpf_send_signal(SIGKILL);
+    } else {
+      /* Not in blocked_ips: emit for userspace domain check (argv-based) */
+      const char msg[] = "CONNECT_ATTEMPT";
+      __builtin_memcpy((void *)evt.filename, msg, sizeof(msg));
+      evt.action = 3;
+      events.perf_submit(args, &evt, sizeof(evt));
     }
   }
 

deploy/linux/monitor.py

@@ -62,6 +62,7 @@
 
 _bpf_instance = None
 _deny_rules: dict[str, dict] = {}
+_blocked_domains: dict[str, str] = {}  # rule_id -> domain, for fast connect-time check
 _malicious_hashes: dict[str, str] = {}
 _hash_to_rule_id: dict[int, str] = {}
 _ip_to_rule_id: dict[str, str] = {}
@@ -219,6 +220,11 @@ def _print_event(cpu, data, size):
                 pid=ns_pid,
                 comm=comm,
             )
+        elif event.action == 3 and filename == "CONNECT_ATTEMPT":
+            # Connect not in blocked_ips: check cmdline for blocked domains (argv-based)
+            ip_int = getattr(event, "blocked_ip", 0)
+            ip_str = _ip_int_to_str(ip_int) if ip_int else "0.0.0.0"
+            _check_connect_domain_rules(ns_pid, comm, ip_str)
         elif event.action == 1:
             # Blocked by BPF — find the matching rule ID
             rule_id = _find_rule_id_for_block(filename)
@@ -235,12 +241,10 @@ def _print_event(cpu, data, size):
             )
         elif event.action == 2:
             # Post-exec (sys_exit_execve): /proc/cmdline now has the new argv.
-            
-            # Layer 4: Userspace file hash check
-            _check_malicious_hashes(ns_pid, comm, filename)
-            
-            # Layer 2: Userspace deny_rules check
+            # Check deny_rules (incl. domains) FIRST — fast, no file I/O
             matched_rule = _check_deny_rules(ns_pid, comm, filename)
+            if not matched_rule:
+                _check_malicious_hashes(ns_pid, comm, filename)
             if not matched_rule:
                 logger.debug(
                     "[observed] pid=%d uid=%d comm=%s file=%s",
@@ -315,6 +319,51 @@ def _check_malicious_hashes(pid: int, comm: str, filename: str) -> str | None:
 
     return None
 
+# ---------------------------------------------------------------------------
+# Connect-time domain check (argv-based, no DNS)
+# ---------------------------------------------------------------------------
+
+def _check_connect_domain_rules(pid: int, comm: str, ip_str: str) -> str | None:
+    """At connect(): read cmdline, check for blocked domains. Kill if match.
+    Provides a second chance to block when exec-time check raced."""
+    if not _blocked_domains:
+        return None
+    try:
+        with open(f"/proc/{pid}/cmdline", "rb") as f:
+            raw = f.read(4096)
+    except OSError:
+        return None
+
+    cmdline = raw.replace(b"\x00", b" ").decode(errors="replace").strip()
+    if not cmdline:
+        return None
+
+    # Exempt our own alert dispatch
+    if "openclaw agent" in cmdline and "ClawEDR Alert" in cmdline:
+        return None
+
+    for rule_id, domain in _blocked_domains.items():
+        if domain in cmdline:
+            block_logger.warning(
+                "BLOCKED [%s] (domain=connect) pid=%d comm=%s target=%s cmdline=%s",
+                rule_id, pid, comm, ip_str, cmdline[:150],
+            )
+            dispatch_alert_async(
+                rule_id=rule_id,
+                action="deny_rule",
+                target=cmdline[:200],
+                pid=pid,
+                comm=comm,
+            )
+            try:
+                os.kill(pid, 9)
+            except OSError:
+                pass
+            return rule_id
+
+    return None
+
+
 # ---------------------------------------------------------------------------
 # Layer 2: deny_rules — userspace argv matching
 # ---------------------------------------------------------------------------
@@ -348,6 +397,12 @@ def _check_deny_rules(pid: int, comm: str, filename: str) -> str | None:
     except (OSError, IndexError, ValueError):
         pass
 
+    # Exempt our own alert dispatch: openclaw agent --message "ClawEDR Alert..."
+    # The message embeds the blocked cmdline and rule descriptions (e.g. "port 4444")
+    # which can falsely match LIN-001 and cause an alert cascade.
+    if "openclaw agent" in cmdline and "ClawEDR Alert" in cmdline:
+        return None
+
     for rule_id, rule in _deny_rules.items():
         pattern = rule.get("match", "")
         if not pattern:
@@ -637,30 +692,36 @@ def _apply_blocked_paths(policy: dict, exempted: set[str]) -> None:
 # ---------------------------------------------------------------------------
 
 def _apply_deny_rules(policy: dict, exempted: set[str]) -> None:
-    """Load deny_rules for argv matching. blocked_domains are injected as domain-in-cmdline rules."""
+    """Load deny_rules for argv matching. blocked_domains are injected as domain-in-cmdline rules.
+    Domain rules are loaded FIRST so they take precedence over generic LIN-* rules (e.g. LIN-001
+    *4444* would otherwise match alert messages that mention port 4444)."""
     global _deny_rules
     raw = policy.get("deny_rules", {})
     if isinstance(raw, dict) and "linux" in raw:
         raw = raw.get("linux", {})
 
     _deny_rules = {}
     skipped = 0
-    for rule_id, rule in raw.items():
+
+    # Load domain rules FIRST so they match before generic rules (curl to blocked domain -> DOM-xxx)
+    global _blocked_domains
+    _blocked_domains = {}
+    for rule_id, domain in policy.get("blocked_domains", {}).items():
         if rule_id in exempted:
-            logger.info("Skipping exempted deny_rule %s", rule_id)
             skipped += 1
             continue
-        if isinstance(rule, dict) and rule.get("match"):
-            _deny_rules[rule_id] = rule
+        _blocked_domains[rule_id] = domain
+        _deny_rules[rule_id] = {"match": f"*{domain}*", "rule": "blocked_domain", "scope": "argv"}
 
-    # Inject blocked_domains as argv deny_rules (match cmdline containing domain)
-    for rule_id, domain in policy.get("blocked_domains", {}).items():
+    for rule_id, rule in raw.items():
         if rule_id in exempted:
+            logger.info("Skipping exempted deny_rule %s", rule_id)
             skipped += 1
             continue
         if rule_id in _deny_rules:
             continue
-        _deny_rules[rule_id] = {"match": f"*{domain}*", "rule": "blocked_domain", "scope": "argv"}
+        if isinstance(rule, dict) and rule.get("match"):
+            _deny_rules[rule_id] = rule
 
     logger.info("Loaded %d deny_rules for userspace matching (%d skipped)",
                 len(_deny_rules), skipped)
@@ -690,7 +751,7 @@ def _apply_blocked_ips(policy: dict, exempted: set[str]) -> None:
     global _ip_to_rule_id
     if _bpf_instance is None:
         return
-        
+
     raw_ips = policy.get("blocked_ips", {})
 
     _ip_to_rule_id = {}
@@ -807,7 +868,7 @@ def _stop(signum, frame):
 
         if _bpf_instance is not None:
             try:
-                _bpf_instance.perf_buffer_poll(timeout=50)
+                _bpf_instance.perf_buffer_poll(timeout=5)
             except Exception:
                 logger.exception("Crash in perf_buffer_poll")