Send the two tables the agent collected and never shipped, and stop a NUL byte killing every port scan

Sentora/db/init.sql

@@ -26,23 +26,34 @@ CREATE UNIQUE INDEX IF NOT EXISTS uq_cf_dup_not_null
   ON critical_files (dup_fp) WHERE dup_fp IS NOT NULL;
 
 -- ========== portscan_result ==========
+-- target_ip, state and banner are new here and are not new on the server: its
+-- table has held all three since the beginning, so three columns were NULL on
+-- every row that ever arrived, because the agent had nowhere to put them. A
+-- port list that does not say which host, or whether the port answered, is a
+-- list of numbers.
 CREATE TABLE IF NOT EXISTS portscan_result (
     id         SERIAL PRIMARY KEY,
+    target_ip  TEXT,
     port       INTEGER,
     protocol   TEXT,
+    state      TEXT,
     service    TEXT,
     product    TEXT,
     version    TEXT,
+    banner     TEXT,
     scanned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     sent       BOOLEAN DEFAULT FALSE,
     dup_fp     CHAR(64)
 );
 ALTER TABLE portscan_result
+    ADD COLUMN IF NOT EXISTS target_ip  TEXT,
     ADD COLUMN IF NOT EXISTS port       INTEGER,
     ADD COLUMN IF NOT EXISTS protocol   TEXT,
+    ADD COLUMN IF NOT EXISTS state      TEXT,
     ADD COLUMN IF NOT EXISTS service    TEXT,
     ADD COLUMN IF NOT EXISTS product    TEXT,
     ADD COLUMN IF NOT EXISTS version    TEXT,
+    ADD COLUMN IF NOT EXISTS banner     TEXT,
     ADD COLUMN IF NOT EXISTS scanned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     ADD COLUMN IF NOT EXISTS sent       BOOLEAN DEFAULT FALSE,
     ADD COLUMN IF NOT EXISTS dup_fp     CHAR(64);

Sentora/main.py

@@ -19,7 +19,7 @@
 import ipaddress
 
 from modules.db import (insert_record, fetch_unsent, mark_sent, fetch_one,
-                        reoffer_recent, prune_sent)
+                        reoffer_recent, prune_sent, prune_unsent)
 import modules.enc_db as enc_db
 import modules.screen_capture as screen_capture
 import modules.agent_paths as agent_paths
@@ -479,6 +479,22 @@ def _handle_server_epoch(epoch: str) -> None:
     'process_events',
     'hardware_inventory',
     'docker_containers',
+    # `software_inventory` and `network_inventory` are the opposite of the
+    # `security_audit` case below, and they were missing for longer.
+    #
+    # `modules/inventory.py` has been writing both every cycle since it was
+    # written - "Inventory: Scanned 326 apps and 68 open ports" is in the log
+    # every twenty minutes - and neither table was on this list, so not one row
+    # was ever offered to the server. On this host that is 368,902 rows of
+    # software inventory and 78,224 of network inventory sitting unsent in the
+    # agent's own database, while the console showed both tables empty: a
+    # machine with no software installed and nothing listening.
+    #
+    # The server was ready the whole time. Both names are in its
+    # `ALLOWED_TABLES` and its `DEDUP_TABLES`, so it has been prepared to
+    # receive and deduplicate two tables nothing ever sent it.
+    'software_inventory',
+    'network_inventory',
     # `security_audit` was here and is not any more. Nothing in this agent
     # ever wrote a row to it - no collector, not a broken one - so it was
     # shipped empty every cycle and the console showed it permanently as
@@ -828,6 +844,14 @@ def db_sender_loop():
 #: quietly capped by whatever the cleaner happened to leave behind.
 _RETENTION_ROWS = 5000
 
+#: The second bound, on rows the server has *not* got.
+#:
+#: Four times the sent limit, because this one must never bite a table that
+#: is simply behind. The send loop moves fifty rows every few seconds, so a
+#: healthy table stays far below this even through a long outage; only a table
+#: nothing is shipping climbs here, and two of them did - see `prune_unsent`.
+_UNSENT_BACKLOG_ROWS = 20000
+
 
 def retention_loop():
     """Keep the agent's local database from growing for ever.
@@ -849,16 +873,25 @@ def retention_loop():
         # doing is shipping the backlog, not deleting it.
         time.sleep(3600)
         total = 0
+        abandoned = 0
         for table in TABLES:
             try:
                 total += prune_sent(table, _RETENTION_ROWS)
+                abandoned += prune_unsent(table, _UNSENT_BACKLOG_ROWS)
             except Exception as e:
                 # One table failing must not stop the rest - the same rule the
                 # collectors follow, and for the same reason.
                 print(f"[!] retention failed on {table}: {e}", flush=True)
         if total:
             print(f"[*] Retention: removed {total} already-shipped row(s); "
                   f"keeping the newest {_RETENTION_ROWS} per table", flush=True)
+        if abandoned:
+            # Louder than the line above, and phrased as a loss, because it
+            # is one: these rows were collected and never delivered.
+            print(f"[!] Retention: abandoned {abandoned} row(s) that were "
+                  f"never sent, keeping the newest {_UNSENT_BACKLOG_ROWS} per "
+                  f"table. A table reaching this is not being shipped.",
+                  flush=True)
 
 
 #: How many consecutive failures between repeat reports, once a collector has

Sentora/modules/db.py

@@ -19,6 +19,46 @@ def get_conn():
     )
 
 
+def scrub_nuls(value):
+    r"""Remove NUL bytes from anything about to be bound as a parameter.
+
+    PostgreSQL text cannot hold `\x00`. Not "should not" - the wire protocol
+    terminates strings with it, so psycopg2 refuses the statement outright:
+
+        A string literal cannot contain NUL (0x00) characters.
+
+    Which is fine until you remember what this agent collects. Every hour the
+    port scanner opens a socket to each listening port and reads whatever comes
+    back, and a binary protocol answers with binary. `decode(errors='ignore')`
+    does not help: it drops bytes that are not *valid*, and `\x00` is a
+    perfectly valid code point. So the banner keeps it, the service and product
+    strings are cut from the banner, and the insert dies.
+
+    It had been dying every hour on this host - ten times in the current log -
+    and it took the whole scan with it, because the exception left the loop
+    that was still writing the other ports. 108 rows, none newer than the day
+    before, and the console showed a Ports tab that looked merely quiet.
+
+    Stripping loses a byte that could not have been stored anyway; the row it
+    saves is one that is currently lost entirely. Applied to every bound
+    parameter rather than to inserts alone, because the scanner's duplicate
+    check passes the same product string to a SELECT and that raises first.
+
+    Text only. `bytea` stores NUL without complaint, and a binary column is
+    where a raw capture belongs, so stripping bytes here would corrupt the one
+    case that is already correct.
+    """
+    if isinstance(value, str):
+        return value.replace("\x00", "")
+    if isinstance(value, list):
+        return [scrub_nuls(v) for v in value]
+    if isinstance(value, tuple):
+        return tuple(scrub_nuls(v) for v in value)
+    if isinstance(value, dict):
+        return {k: scrub_nuls(v) for k, v in value.items()}
+    return value
+
+
 def _schema_path():
     """db/init.sql, whether running from source or from the PyInstaller bundle."""
     import sys
@@ -175,7 +215,7 @@ def delete_all(table: str):
 def insert_record(table: str, data: dict):
     columns = ','.join(data.keys())
     placeholders = ','.join(['%s'] * len(data))
-    values = list(data.values())
+    values = scrub_nuls(list(data.values()))
     query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
     with get_conn() as conn:
         with conn.cursor() as cur:
@@ -301,14 +341,52 @@ def prune_sent(table: str, keep: int) -> int:
     return removed if removed and removed > 0 else 0
 
 
+def prune_unsent(table: str, keep: int) -> int:
+    """Abandon unsent rows beyond the newest `keep`, and say how many.
+
+    `prune_sent` deliberately refuses to touch an unsent row, on the grounds
+    that it has not reached the server and deleting it is data loss. That is
+    right for a table that is being delivered and wrong for one that is not,
+    and the agent had two of the latter: `software_inventory` and
+    `network_inventory` were collected every cycle and were not on the
+    shipping list, so they accumulated 368,902 and 78,224 unsent rows here
+    while both tables read as empty on the server.
+
+    Nothing reported it. `prune_sent` skipped them because they were unsent,
+    the send loop skipped them because they were not listed, and the only
+    visible symptom was a host that appeared to have no software installed.
+
+    So there is a second bound, far above the first. A table being delivered
+    never reaches it - the send loop drains fifty rows every few seconds, and
+    an outage long enough to build this backlog is one where the newest rows
+    are the ones worth keeping anyway. A table that is not being delivered
+    reaches it within a day and then stops growing.
+
+    `keep` is the newest rows, not the oldest, and that is a real choice:
+    `fetch_unsent` has no ORDER BY, so a backlog drains roughly oldest-first
+    and a month-old software list would ship ahead of today's.
+    """
+    query = (
+        f"DELETE FROM {table} WHERE sent = FALSE AND id < "
+        f"(SELECT id FROM {table} WHERE sent = FALSE "
+        f" ORDER BY id DESC LIMIT 1 OFFSET %s)"
+    )
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(query, (keep,))
+            removed = cur.rowcount
+        conn.commit()
+    return removed if removed and removed > 0 else 0
+
+
 def fetch_one(table: str, where: str = "1=1", params: tuple = (), order_by: str = None):
     query = f"SELECT * FROM {table} WHERE {where}"
     if order_by:
         query += f" ORDER BY {order_by}"
     query += " LIMIT 1"
     with get_conn() as conn:
         with conn.cursor(cursor_factory=DictCursor) as cur:
-            cur.execute(query, params)
+            cur.execute(query, scrub_nuls(params))
             row = cur.fetchone()
     return row
 
@@ -332,14 +410,14 @@ def fetch_where(table: str, where: str = "1=1", params: tuple = (), order_by: st
         params = params + (limit,)
     with get_conn() as conn:
         with conn.cursor(cursor_factory=DictCursor) as cur:
-            cur.execute(query, params)
+            cur.execute(query, scrub_nuls(params))
             rows = cur.fetchall()
     return rows
 
 
 def update_record(table: str, data: dict, where: str, params: tuple = ()):
     sets = ','.join([f"{k}=%s" for k in data])
-    values = list(data.values()) + list(params)
+    values = scrub_nuls(list(data.values()) + list(params))
     query = f"UPDATE {table} SET {sets} WHERE {where}"
     with get_conn() as conn:
         with conn.cursor() as cur:

Sentora/modules/portscanner/portscanner.py

@@ -598,8 +598,8 @@ async def scan_ports(target, timeout):
     while not result_queue.empty():
         res = await result_queue.get()
         port, protocol, service, product, version, banner = res
-        results.append((port, protocol, service, product, version))
-        
+        results.append((port, protocol, service, product, version, banner))
+
         banner_preview = banner[:80] + '...' if banner and len(banner) > 80 else banner
         print(f"[+] {port:5d}/tcp - {service:20s} | {product:20s} | v{version}")
         if banner_preview:
@@ -622,24 +622,52 @@ async def main_async():
     try:
         total_found = 0
         total_saved = 0
+        total_failed = 0
         results = await scan_ports(target, timeout)
-        
-        for port, protocol, service, product, version in results:
+
+        for port, protocol, service, product, version, banner in results:
             total_found += 1
-            if not is_duplicate(port, protocol, service, product, version):
-                insert_record(
-                    TABLE,
-                    {
-                        'port': port,
-                        'protocol': protocol,
-                        'service': service,
-                        'product': product,
-                        'version': version,
-                        'scanned_at': datetime.now(timezone.utc).replace(tzinfo=None),
-                    },
-                )
-                total_saved += 1
-
+            # Per row, not per scan.
+            #
+            # This loop used to sit inside the try below, so the first row
+            # that would not insert ended the scan and took every port after
+            # it. One binary banner with a NUL in it cost the whole hour's
+            # work, every hour. `db.scrub_nuls` is the fix for that particular
+            # input; this is the fix for the shape of the failure, which will
+            # come back with some other byte.
+            try:
+                if not is_duplicate(port, protocol, service, product, version):
+                    insert_record(
+                        TABLE,
+                        {
+                            'target_ip': target,
+                            'port': port,
+                            'protocol': protocol,
+                            # Only a port that accepted a connection is
+                            # enqueued, so every row here is open. The column
+                            # existed and was NULL on all 108 rows, which
+                            # makes a port list say nothing about whether the
+                            # port answered.
+                            'state': 'open',
+                            'service': service,
+                            'product': product,
+                            'version': version,
+                            # Kept, and bounded. The banner is the evidence
+                            # for the identification - without it `Unknown-*`
+                            # is a dead end - but it is remote input, so a
+                            # service that answers with a megabyte does not
+                            # get to write a megabyte per port per hour.
+                            'banner': (banner or '')[:2048] or None,
+                            'scanned_at': datetime.now(timezone.utc).replace(tzinfo=None),
+                        },
+                    )
+                    total_saved += 1
+            except Exception as e:
+                total_failed += 1
+                if total_failed == 1:
+                    print(f"[!] {port}/{protocol} not stored: "
+                          f"{type(e).__name__}: {e}")
+
         end_time = datetime.now(timezone.utc).replace(tzinfo=None)
         duration = (end_time - start_time).total_seconds()
 
@@ -649,7 +677,9 @@ async def main_async():
         print(f"  Duration: {duration:.2f} seconds ({duration/60:.2f} minutes)")
         print(f"  Ports Found: {total_found}")
         print(f"  New Records Saved: {total_saved}")
-        print(f"  Duplicates Skipped: {total_found - total_saved}")
+        print(f"  Duplicates Skipped: {total_found - total_saved - total_failed}")
+        if total_failed:
+            print(f"  Not Stored: {total_failed}")
         print(f"{'='*70}\n")
     except KeyboardInterrupt:
         print("\n[!] Scan interrupted by user.")