Say where the telemetry stops, instead of showing an empty table

Sentora/build_agent.ps1

@@ -182,6 +182,7 @@ $hiddenImports = @(
     # builds cleanly and dies on the first channel connect, which is the worst
     # place to find it.
     "modules.version",
+    "modules.telemetry_health",
     "websocket",
     "winpty",
     "winpty.ptyprocess",

Sentora/build_agent.sh

@@ -205,6 +205,7 @@ step "Running PyInstaller..."
     --hidden-import modules.console \
     --hidden-import modules.link \
     --hidden-import modules.version \
+    --hidden-import modules.telemetry_health \
     --hidden-import websocket \
     --hidden-import modules.db \
     --exclude-module torch \

Sentora/main.py

@@ -22,6 +22,7 @@
 import modules.agent_paths as agent_paths
 import modules.console as console
 import modules.link as link
+import modules.telemetry_health as telemetry_health
 
 import logging
 import builtins
@@ -539,7 +540,16 @@ def send_table(table: str):
             s.sendall(struct.pack('!Q', fsize))
             s.sendall(data_bytes)
 
+        # Marked sent because `sendall` returned, which means the bytes
+        # reached the socket buffer. The ingest protocol has no reply, so
+        # this is the strongest thing the agent can know - and it is weaker
+        # than it reads. A server that accepts a batch and stores none of it
+        # looks identical from here, which is exactly how four tables came to
+        # be permanently empty while this line said "sent (50 rows)" every
+        # cycle. `modules/telemetry_health` exists so the server can put its
+        # own row count beside this number.
         mark_sent(table, [r['id'] for r in rows])
+        telemetry_health.record_send(table, len(rows))
         if debug:
             # The IP/OS/HOST/MAC banner used to be repeated here. It is
             # constant for the life of the process and was the single largest
@@ -548,6 +558,7 @@ def send_table(table: str):
             print(f"[+] {table} sent ({len(rows)} rows)")
 
     except Exception as e:
+        telemetry_health.record_send_failure(table, e)
         if debug:
             print(f"[!] Sending error ({table}): {e}")
 
@@ -563,13 +574,40 @@ def db_sender_loop():
         time.sleep(10)
 
 
+#: How many consecutive failures between repeat reports, once a collector has
+#: announced itself broken. Loud on the first, then rare - a collector that
+#: has been failing for a day should not be writing a line every two minutes.
+_FAILURE_REPEAT_EVERY = 30
+
+
 def periodic_wrapped(func, interval: int, name: str):
+    """Run a collector forever, and say when it stops working.
+
+    The failure report used to be behind `if debug`, which nothing sets. A
+    collector that raised on its first line failed silently every interval
+    for the life of the agent, and the only symptom was an empty table in the
+    console - indistinguishable from a host that genuinely has nothing to
+    report. Hardware and network inventory were both empty for exactly this
+    reason, and nothing anywhere said so.
+
+    Rate-limited rather than unconditional, because the same argument that
+    makes silence wrong makes a line every two minutes useless: it would bury
+    the schema errors and the auth rejections that also live in this log.
+    """
+    failures = 0
     while True:
         try:
             func()
+            if failures:
+                print(f"[+] {name} recovered after {failures} failure(s)", flush=True)
+            failures = 0
         except Exception as e:
-            if debug:
-                print(f"[!] Error: ({name}): {e}")
+            failures += 1
+            if failures == 1 or failures % _FAILURE_REPEAT_EVERY == 0:
+                print(f"[!] {name} failed ({failures} consecutive): "
+                      f"{type(e).__name__}: {e}", flush=True)
+                if debug:
+                    traceback.print_exc()
         time.sleep(interval)
 
 
@@ -1924,6 +1962,13 @@ def dispatch_channel_request(method: str, path: str, body) -> tuple[dict, int]:
     if path == "/health":
         return {"ok": True, "agent": AGENT_NAME}, 200
 
+    if path == "/telemetry/health":
+        # What this agent holds and believes it has shipped, per table. The
+        # server puts its own row counts beside these; neither half alone can
+        # tell a quiet host from a broken pipeline.
+        return {"ok": True, "agent": AGENT_NAME,
+                "tables": telemetry_health.report(TABLES)}, 200
+
     # Named rather than a bare 404: a path the agent does not implement is
     # usually a server that is newer than this build, and saying so is the
     # difference between "upgrade the agent" and an afternoon of guessing.

Sentora/modules/edr_enforcer.py

@@ -69,12 +69,23 @@ def check_fim():
             })
 
 def track_network():
+    """Established outbound connections, with the process behind each.
+
+    `psutil.net_connections` raises AccessDenied where the agent is not
+    privileged enough to see other processes' sockets. That is a real
+    condition worth reporting rather than an empty table: the caller isolates
+    and logs it, and an operator can act on "the agent cannot read the
+    connection table" in a way they cannot act on a blank page.
+    """
     for conn in psutil.net_connections(kind='inet'):
         if conn.status == 'ESTABLISHED' and conn.raddr:
             try:
                 proc = psutil.Process(conn.pid)
                 name = proc.name()
             except Exception:
+                # The process ended between listing and lookup, which is
+                # ordinary on a busy host. The connection is still worth
+                # recording.
                 name = "Unknown"
 
             insert_record_enc("network_connections", {
@@ -150,53 +161,90 @@ def monitor_processes():
             continue
 
 def get_hardware_inventory():
+    # Nothing is swallowed here any more. An empty hardware tab is
+    # indistinguishable from a host with no hardware, so a collector that
+    # cannot run has to say why - the console cannot infer it.
     if IS_WINDOWS:
         ps_cmd = "[Console]::OutputEncoding = [System.Text.Encoding]::UTF8; Get-PnpDevice -PresentOnly | Select-Object FriendlyName, InstanceId | ConvertTo-Json"
-        try:
-            res = subprocess.run(["powershell", "-NoProfile", "-Command", ps_cmd], capture_output=True, text=True, encoding='utf-8', errors='replace')
-            if res.returncode == 0 and res.stdout:
-                data = json.loads(res.stdout)
-                for item in (data if isinstance(data, list) else [data]):
-                    if not item or not item.get("FriendlyName"): continue
-                    insert_record_enc("hardware_inventory", {
-                        "type": "pnp",
-                        "name": item.get("FriendlyName"),
-                        "product_id": item.get("InstanceId", "N/A"),
-                        "serial_number": item.get("InstanceId", "N/A"),
-                        "status": "active"
-                    })
-        except Exception: pass
+        res = subprocess.run(["powershell", "-NoProfile", "-Command", ps_cmd], capture_output=True, text=True, encoding='utf-8', errors='replace')
+        if res.returncode != 0:
+            raise RuntimeError(
+                f"Get-PnpDevice exited {res.returncode}: "
+                f"{(res.stderr or '').strip()[:200]}")
+        if not (res.stdout or "").strip():
+            raise RuntimeError("Get-PnpDevice returned nothing")
+        data = json.loads(res.stdout)
+        for item in (data if isinstance(data, list) else [data]):
+            if not item or not item.get("FriendlyName"):
+                continue
+            insert_record_enc("hardware_inventory", {
+                "type": "pnp",
+                "name": item.get("FriendlyName"),
+                "product_id": item.get("InstanceId", "N/A"),
+                "serial_number": item.get("InstanceId", "N/A"),
+                "status": "active"
+            })
     else:
-        try:
-            res = subprocess.run(["lsusb"], capture_output=True, text=True, encoding='utf-8', errors='replace')
-            for line in res.stdout.splitlines():
+        res = subprocess.run(["lsusb"], capture_output=True, text=True, encoding='utf-8', errors='replace')
+        if res.returncode != 0:
+            raise RuntimeError(
+                f"lsusb exited {res.returncode}: "
+                f"{(res.stderr or '').strip()[:200]}")
+        for line in res.stdout.splitlines():
+            if line.strip():
                 insert_record_enc("hardware_inventory", {
                     "type": "usb",
                     "name": line.strip()
                 })
-        except Exception: pass
 
 def monitor_registry():
-    if not IS_WINDOWS: return
+    if not IS_WINDOWS:
+        return
     ps_cmd = "Get-ItemProperty HKLM:\\Software\\Microsoft\\Windows\\CurrentVersion\\Run | ConvertTo-Json"
-    try:
-        res = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True, encoding='utf-8', errors='replace')
-        if res.returncode == 0 and res.stdout:
-            data = json.loads(res.stdout)
-            for k, v in data.items():
-                if k not in ["PSPath", "PSParentPath", "PSChildName", "PSDrive", "PSProvider"]:
-                    insert_record_enc("registry_logs", {
-                        "hive": "HKLM",
-                        "key_path": "Software\\Microsoft\\Windows\\CurrentVersion\\Run",
-                        "value_name": k,
-                        "value_data": str(v),
-                        "status": "monitored"
-                    })
-    except Exception: pass
+    res = subprocess.run(["powershell", "-NoProfile", "-Command", ps_cmd], capture_output=True, text=True, encoding='utf-8', errors='replace')
+    if res.returncode != 0:
+        raise RuntimeError(
+            f"Get-ItemProperty exited {res.returncode}: "
+            f"{(res.stderr or '').strip()[:200]}")
+    if not (res.stdout or "").strip():
+        return                      # an empty Run key is a real answer
+    data = json.loads(res.stdout)
+    for k, v in data.items():
+        if k not in ["PSPath", "PSParentPath", "PSChildName", "PSDrive", "PSProvider"]:
+            insert_record_enc("registry_logs", {
+                "hive": "HKLM",
+                "key_path": "Software\\Microsoft\\Windows\\CurrentVersion\\Run",
+                "value_name": k,
+                "value_data": str(v),
+                "status": "monitored"
+            })
 
 def main():
-    check_fim()
-    track_network()
-    monitor_processes()
-    get_hardware_inventory()
-    monitor_registry()
+    """Every collector, each isolated from the others.
+
+    They used to be five bare calls in a row, so the first one to raise took
+    the rest of the cycle with it - and since the caller retries on a timer,
+    it took them again every two minutes, permanently.
+
+    That is why the console showed no hardware and no network inventory at
+    the same time: `track_network` and `get_hardware_inventory` sit second and
+    fourth, and something ahead of them was throwing on every pass. An empty
+    table reads as "this host has nothing to report", which is the one thing
+    it did not mean.
+
+    One failing collector now costs its own data and nothing else, and says
+    so. `periodic_wrapped` rate-limits the repeat, so a permanently broken
+    collector is loud once rather than every two minutes forever.
+    """
+    for name, collector in (
+        ("fim", check_fim),
+        ("network", track_network),
+        ("processes", monitor_processes),
+        ("hardware", get_hardware_inventory),
+        ("registry", monitor_registry),
+    ):
+        try:
+            collector()
+        except Exception as e:
+            print(f"[edr] {name} collector failed: {type(e).__name__}: {e}",
+                  flush=True)

Sentora/modules/telemetry_health.py

@@ -0,0 +1,105 @@
+"""What this agent has collected, and what it believes it has shipped.
+
+Three bugs in one day had the same shape: every layer reported success and
+the chain was broken anyway, and the only symptom was an empty table -
+indistinguishable from a host that genuinely has nothing to report.
+
+    a collector raised on its first line, so the four after it never ran
+    the server accepted a batch and discarded every row of it
+    the agent marked rows sent as soon as `sendall` returned
+
+The last one is why none of it was visible from here. `sendall` returning
+means the bytes reached the operating system's socket buffer. It says nothing
+about whether the server stored them, and the ingest protocol has no reply -
+so the agent cannot know, and has always reported success.
+
+This module does not fix that. It makes the two halves *comparable*: the agent
+says what it holds and what it has shipped, the server says what it holds, and
+the difference names the broken link. `[+] network_connections sent (50 rows)`
+against a server table with zero rows is not ambiguous once somebody puts the
+two numbers next to each other - the whole difficulty was that nobody ever did.
+
+Counts are read from the database rather than kept as counters. A counter
+resets when the agent restarts and drifts whenever anything writes without
+going through it; the tables are the truth, and reading them costs two
+queries per cycle.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+#: Per table: when the last send succeeded, how many rows it carried, and the
+#: last error if there was one. Small, and the only thing here that is not
+#: read straight from the database.
+_send_state: dict[str, dict] = {}
+_lock = threading.Lock()
+
+
+def record_send(table: str, rows: int) -> None:
+    with _lock:
+        state = _send_state.setdefault(table, {})
+        state["last_sent_at"] = time.time()
+        state["last_sent_rows"] = int(rows)
+        state["last_error"] = None
+
+
+def record_send_failure(table: str, error: Exception | str) -> None:
+    """A send that raised. Kept rather than logged only, because the console
+    asking "why is this table empty" needs the answer, and the answer is on
+    the endpoint."""
+    with _lock:
+        state = _send_state.setdefault(table, {})
+        state["last_error"] = f"{type(error).__name__}: {error}" \
+            if isinstance(error, Exception) else str(error)
+        state["last_error_at"] = time.time()
+
+
+def _counts(table: str) -> tuple[int, int]:
+    """(rows held, rows not yet shipped) for one table.
+
+    A table that does not exist is `(0, 0)` and not an error: the agent's
+    schema gains tables over releases, and a missing one is a real state the
+    report should show rather than an exception that hides every other table
+    behind it.
+    """
+    from modules.db import get_conn
+
+    try:
+        with get_conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(f"SELECT COUNT(*), COUNT(*) FILTER (WHERE sent = FALSE) "
+                            f"FROM {table}")
+                row = cur.fetchone()
+                return int(row[0] or 0), int(row[1] or 0)
+    except Exception:
+        return 0, 0
+
+
+def report(tables) -> dict:
+    """Everything this agent knows about its own telemetry, per table.
+
+    Deliberately flat and dumb. The judgement about what the numbers *mean*
+    lives on the server, where the other half of the comparison is - putting
+    it here would mean an agent deciding whether its own data arrived.
+    """
+    with _lock:
+        state = {t: dict(v) for t, v in _send_state.items()}
+
+    out = {}
+    for table in tables:
+        held, unsent = _counts(table)
+        sent = state.get(table, {})
+        out[table] = {
+            "held": held,
+            "unsent": unsent,
+            # Rows this agent believes it has shipped. Belief, not fact:
+            # nothing acknowledges an ingest batch.
+            "shipped": max(held - unsent, 0),
+            "last_sent_at": sent.get("last_sent_at"),
+            "last_sent_rows": sent.get("last_sent_rows"),
+            "last_error": sent.get("last_error"),
+            "last_error_at": sent.get("last_error_at"),
+        }
+    return out