From 656bbefd9433f8a329e05385e6984ab9db7a171a Mon Sep 17 00:00:00 2001 From: d3vhex Date: Tue, 8 Sep 2026 15:38:45 +0300 Subject: [PATCH] Send the two tables the agent collected and never shipped, and stop a NUL byte killing every port scan Strip NUL from every bound parameter, store target_ip/state/banner, keep one bad row from ending a scan, add software_inventory and network_inventory to TABLES, bound the unsent backlog, and add the invariants that would have caught all of it. --- Sentora/db/init.sql | 11 +++ Sentora/main.py | 35 +++++++- Sentora/modules/db.py | 86 +++++++++++++++++++- Sentora/modules/portscanner/portscanner.py | 68 +++++++++++----- tests/test_agent_sql_parameters.py | 87 ++++++++++++++++++++ tests/test_ingest_schema_coverage.py | 92 +++++++++++++++++++++- 6 files changed, 353 insertions(+), 26 deletions(-) diff --git a/Sentora/db/init.sql b/Sentora/db/init.sql index 4d1cf25..ea2a746 100644 --- a/Sentora/db/init.sql +++ b/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); diff --git a/Sentora/main.py b/Sentora/main.py index 8c568b4..330b3d9 100644 --- a/Sentora/main.py +++ b/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,9 +873,11 @@ 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. @@ -859,6 +885,13 @@ def retention_loop(): 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 diff --git a/Sentora/modules/db.py b/Sentora/modules/db.py index 3baaa7e..4bafd46 100644 --- a/Sentora/modules/db.py +++ b/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,6 +341,44 @@ 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: @@ -308,7 +386,7 @@ def fetch_one(table: str, where: str = "1=1", params: tuple = (), order_by: str 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: diff --git a/Sentora/modules/portscanner/portscanner.py b/Sentora/modules/portscanner/portscanner.py index c6dba7d..c94f233 100644 --- a/Sentora/modules/portscanner/portscanner.py +++ b/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.") diff --git a/tests/test_agent_sql_parameters.py b/tests/test_agent_sql_parameters.py index 8b877d2..e2b0058 100644 --- a/tests/test_agent_sql_parameters.py +++ b/tests/test_agent_sql_parameters.py @@ -105,3 +105,90 @@ def test_the_portscanner_uses_them(): body = ast.unparse(fn) assert "%s" in body, "the duplicate check builds its own literals again" assert "params" in body + + +# -------------------------------------------------------------------------- +# Parameterising it was necessary and not sufficient +# -------------------------------------------------------------------------- + +def _db_func(name: str): + """Lift one function out of the agent's db module without importing it.""" + source = (AGENT / "modules" / "db.py").read_text(encoding="utf-8") + tree = ast.parse(source) + fn = next(n for n in tree.body + if isinstance(n, ast.FunctionDef) and n.name == name) + ns: dict = {} + exec(compile(ast.Module(body=[fn], type_ignores=[]), "db.py", "exec"), ns) + return ns[name] + + +def test_a_nul_byte_does_not_reach_postgres(): + r"""The next thing that broke after the quoting was fixed. + + PostgreSQL text cannot contain `\x00` - the protocol terminates strings + with it - so psycopg2 refuses the whole statement: + + A string literal cannot contain NUL (0x00) characters. + + The port scanner reads a raw banner off every listening port, and a binary + protocol answers in binary. `decode(errors='ignore')` keeps the NUL, + because it drops bytes that are not *valid* and `\x00` is valid. So the + scan ran every hour, found its ports, and died on the write. Ten times in + the current log, against a table whose newest row was the previous day. + """ + scrub = _db_func("scrub_nuls") + assert scrub("SSH-2.0-OpenSSH\x008.9") == "SSH-2.0-OpenSSH8.9" + assert scrub(("a\x00b", 5, None)) == ("ab", 5, None) + assert scrub(["x\x00"]) == ["x"] + assert scrub({"banner": "\x00\x00"}) == {"banner": ""} + + +def test_binary_is_left_alone(): + """`bytea` stores NUL without complaining, and a raw capture is what a + binary column is for. Stripping there would corrupt the one case that was + already right.""" + scrub = _db_func("scrub_nuls") + assert scrub(b"\x00\x01\x00") == b"\x00\x01\x00" + + +@pytest.mark.parametrize("func", ["insert_record", "update_record", + "fetch_where", "fetch_one"]) +def test_every_bound_parameter_is_scrubbed(func): + """Inserts are not enough. The scanner's duplicate check passes the same + product string to a SELECT, and that raises first - so the row never + reached the insert that was supposedly the problem.""" + source = (AGENT / "modules" / "db.py").read_text(encoding="utf-8") + fn = next(n for n in ast.parse(source).body + if isinstance(n, ast.FunctionDef) and n.name == func) + assert "scrub_nuls" in ast.unparse(fn), f"{func} binds unscrubbed values" + + +def test_the_scan_survives_a_row_it_cannot_store(): + """The insert loop used to sit inside the try that reports the scan as + failed, so the first unstorable row ended the scan and took every port + after it with it. One banner cost the whole hour.""" + source = (AGENT / "modules" / "portscanner" / "portscanner.py").read_text( + encoding="utf-8") + fn = next(n for n in ast.parse(source).body + if isinstance(n, ast.AsyncFunctionDef) and n.name == "main_async") + loop = next(n for n in ast.walk(fn) if isinstance(n, ast.For)) + assert any(isinstance(n, ast.Try) for n in loop.body), ( + "a failed row ends the scan again" + ) + + +@pytest.mark.parametrize("column", ["target_ip", "state", "banner"]) +def test_the_scanner_fills_the_columns_the_server_has(column): + """All three existed on the server and were NULL on all 108 rows, because + the agent's own table did not have them and the insert did not mention + them. A port list that does not say which host, or whether the port + answered, is a list of numbers.""" + source = (AGENT / "modules" / "portscanner" / "portscanner.py").read_text( + encoding="utf-8") + fn = next(n for n in ast.parse(source).body + if isinstance(n, ast.AsyncFunctionDef) and n.name == "main_async") + assert f"'{column}'" in ast.unparse(fn), f"{column} is not written" + + schema = (AGENT / "db" / "init.sql").read_text(encoding="utf-8") + block = schema.split("portscan_result", 1)[1].split(";", 1)[0] + assert column in block, f"{column} is not in the agent's own table" diff --git a/tests/test_ingest_schema_coverage.py b/tests/test_ingest_schema_coverage.py index 1dabc21..de1ec51 100644 --- a/tests/test_ingest_schema_coverage.py +++ b/tests/test_ingest_schema_coverage.py @@ -54,8 +54,23 @@ def _set_literal(name: str) -> set: def _schema_tables(path: pathlib.Path) -> set: - return set(re.findall( - r"CREATE TABLE IF NOT EXISTS [`\"]?(\w+)", path.read_text(encoding="utf-8"))) + """Table names, from SQL only. + + Comments go first. The agent schema documents one of its own migrations + with the line + + -- Existing agents: CREATE TABLE IF NOT EXISTS leaves the old CHECK … + + and this used to read a table called `leaves` out of it. Harmless where + the result is filtered against `ALLOWED_TABLES`, and not harmless in + `test_the_agent_ships_every_table_it_collects` below, which asks a + question about every name it finds. + """ + from tests.test_agent_schema_migration import _load_func + + strip = _load_func(ROOT / "Sentora" / "modules" / "db.py", "_strip_comments") + return set(re.findall(r"CREATE TABLE IF NOT EXISTS [`\"]?(\w+)", + strip(path.read_text(encoding="utf-8")))) ALLOWED = _set_literal("ALLOWED_TABLES") @@ -102,6 +117,79 @@ def test_the_agent_collects_nothing_the_server_will_not_take(): ) +# Tables in the agent's own schema that it deliberately does not ship, and +# why. Anything not named here has to be on `TABLES` in main.py. +# +# The list is short on purpose. A table the agent creates is a table something +# was expected to write to, so "we do not send this" is a decision, and a +# decision that is not written down is indistinguishable from the bug this +# test exists to catch. +AGENT_LOCAL_ONLY = { + "automations": + "A work queue the server pushes down and the agent updates in place. " + "It travels the other way, through /automations/report.", + "security_audit": + "No collector was ever written for it. The table is kept so one can " + "be added without a schema change; the claim that this agent reports " + "it is not - see the note on TABLES in main.py.", +} + + +def test_the_agent_ships_every_table_it_collects(): + """The direction nothing checked, and it had been wrong for months. + + `modules/inventory.py` wrote `software_inventory` and `network_inventory` + every cycle. Neither was on `TABLES`, so the send loop never offered them. + `prune_sent` never touched them either, because it will not delete an + unsent row - so they grew: 368,902 rows and 78,224 rows in the agent's own + database, none of them ever leaving the host. + + Every layer was consistent with itself. The collector logged "Scanned 326 + apps and 68 open ports". The server had both names in `ALLOWED_TABLES` and + `DEDUP_TABLES`, ready to receive them. The console showed two empty + tables, which is what a machine with no software installed looks like. + + So the question this asks is not "does the server accept it" - that test + is above and it passed throughout. It is "does the agent send it at all". + """ + main = (ROOT / "Sentora" / "main.py").read_text(encoding="utf-8") + block = re.search(r"^TABLES = \[(.*?)^\]", main, re.S | re.M) + assert block, "TABLES is no longer a list literal in the agent's main.py" + shipped = set(re.findall(r"['\"]([a-z_]+)['\"]", block.group(1))) + + unshipped = sorted(_schema_tables(AGENT_SCHEMA) - shipped - AGENT_LOCAL_ONLY.keys()) + assert not unshipped, ( + f"the agent's schema has {unshipped} and TABLES does not, so anything " + f"written to them stays on the host for ever. Add them to TABLES, or " + f"to AGENT_LOCAL_ONLY with the reason." + ) + + +def test_nothing_is_excused_from_shipping_that_no_longer_exists(): + """The excuse list rots the same way any list does.""" + stale = sorted(AGENT_LOCAL_ONLY.keys() - _schema_tables(AGENT_SCHEMA)) + assert not stale, f"AGENT_LOCAL_ONLY names tables the schema dropped: {stale}" + + +def test_an_unshipped_table_cannot_grow_for_ever(): + """The second half of the same failure. + + `prune_sent` refuses to delete an unsent row - correctly, for a table that + is being delivered. For one that is not, it means the local database grows + until the disk does, on a machine this product is supposed to be watching + rather than filling. `prune_unsent` is the bound. + """ + db = (ROOT / "Sentora" / "modules" / "db.py").read_text(encoding="utf-8") + assert "sent = FALSE AND id <" in db, ( + "prune_unsent no longer bounds the unsent backlog" + ) + main = (ROOT / "Sentora" / "main.py").read_text(encoding="utf-8") + tree = ast.parse(main) + called = {n.func.id for n in ast.walk(tree) + if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)} + assert "prune_unsent" in called, "the retention loop never calls it" + + def test_a_rejected_row_does_not_keep_its_fingerprint(): """The fingerprint is written before the row it stands for.