From 4e32876368466a20870663ee826ede31f3f7208d Mon Sep 17 00:00:00 2001 From: d3vhex Date: Sun, 20 Sep 2026 22:55:35 +0300 Subject: [PATCH] Show what the host is listening on, and stop the search index holding ciphertext Every ingested table was indexed with the row exactly as received, so the field operators search on held Fernet blobs and matched nothing, while snapshot copies accumulated for ever. --- app.py | 20 ++++ core/opensearch.py | 80 +++++++++++-- docs/production-deployment.md | 24 ++++ frontend/src/pages/Assets.tsx | 72 +++++++++++- scripts/prune_search_indices.py | 98 ++++++++++++++++ server.py | 45 ++++++- tests/test_assets_columns.py | 2 +- tests/test_ingest_dedup.py | 80 +++++++++++-- tests/test_opensearch_indexing.py | 189 ++++++++++++++++++++++++++++++ 9 files changed, 587 insertions(+), 23 deletions(-) create mode 100644 scripts/prune_search_indices.py create mode 100644 tests/test_opensearch_indexing.py diff --git a/app.py b/app.py index 77d077f..3bd1db1 100644 --- a/app.py +++ b/app.py @@ -2792,6 +2792,26 @@ async def get_network_inventory(request, agent): return await stream_from_db_dec("network_connections", agent, connect_db_for_agent, encrypted_fields=fields) +@require_permission("read_telemetry") +@app.route("/api/agent//inventory/ports") +async def get_listening_ports(request, agent): + # `network_inventory` was collected every ten minutes, shipped, + # deduplicated, stored - and read by nothing at all. It was the last table + # in `ALLOWED_TABLES` in that state, after `security_audit` (which had no + # collector) and `software_inventory` (which had no reader). + # + # It is not the same data as the Network tab above. `network_connections` + # is conversations already underway; this is what the host is listening + # on, which is the question "what can be reached from outside" - and it is + # a better answer than `portscan_result` gives, because a scan infers the + # surface from outside while this is the kernel's own list, with the owning + # process named. + # + # Plaintext, so no fields: nothing in it is sensitive on its own and the + # port has to be indexable. + return await stream_from_db_dec("network_inventory", agent, + connect_db_for_agent) + @require_permission("read_telemetry") @app.route("/api/agent//fim") async def get_fim_data(request, agent): diff --git a/core/opensearch.py b/core/opensearch.py index b99e5db..537a41c 100644 --- a/core/opensearch.py +++ b/core/opensearch.py @@ -3,6 +3,8 @@ from datetime import datetime from opensearchpy import OpenSearch, RequestsHttpConnection +from core import telemetry_crypto + OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://opensearch:9200") OPENSEARCH_USER = os.getenv("OPENSEARCH_USER", "admin") OPENSEARCH_PASS = os.getenv("OPENSEARCH_PASSWORD", "admin") @@ -15,20 +17,84 @@ connection_class=RequestsHttpConnection ) +#: Tables whose rows are indexed for search, and the reason the rest are not. +#: +#: Every table used to be indexed, which was never a decision - `index_log` was +#: called unconditionally for whatever ingest had just stored. Two costs came +#: out of that: +#: +#: - **the index held ciphertext.** `siem_events.message` went in as +#: `enc::gAAAA...`, so `message:powershell` matched 0 of 69 documents while +#: `source:PowerShell` matched 52 of the same 69. Nine indices and roughly +#: 863,000 documents were unsearchable on the only field that mattered. +#: - **it grew without bound.** A snapshot table is emptied and rewritten in +#: MySQL on every batch; the index kept every copy. 426,768 +#: `hardware_inventory` documents for a CPU, some RAM and four disks. +#: +#: The rule is what the search is *for*. An index answers "find this, anywhere +#: in the fleet, in this window" - which suits an append-only event log and +#: does not suit a table whose answer is "the current list for this host". +#: State tables have a route and a view each, and a view gives an exact answer +#: where a time-windowed full-text search gives a vague one. +INDEXED_TABLES = frozenset({ + "siem_events", "events_alert", # the log, and what the rules said + "process_events", "registry_logs", # EDR observations + "fim_data", "critical_files", # file and permission changes + "security_audit", # posture findings, hunted fleet-wide + "soar_actions", # what was done in response + "portscan_result", # small, and "who has 3389 open" is + # exactly an index question + "audit_logs", # platform, not agent: who did what +}) + +#: Deliberately not indexed, with the view that answers the question instead. +#: Kept beside the allowlist so the reason arrives with the name, and asserted +#: against `server.ALLOWED_TABLES` by `tests/test_opensearch_indexing.py` - a +#: table added to ingest and to neither of these is a silent decision. +NOT_INDEXED = { + "hardware_inventory": "Assets > Hardware. 426,768 documents for one host's components.", + "software_inventory": "Assets > Software.", + "network_inventory": "Assets > Ports.", + "network_connections": "Assets > Network. A snapshot, rewritten every batch.", + "packages": "Input to the vulnerability scanner; surfaced as vulnerabilities_report.", + "docker_containers": "Its own view, and a snapshot.", + "disk_usage": "Dashboard metrics, not events.", + "resource_usage": "Dashboard metrics, not events.", + "vulnerabilities_report": "Written server-side by scanners/vuln.py, which never called this.", +} + + async def index_log(agent: str, table: str, item: dict): + """Index one row for search, decrypted, if this table is searched at all. + + **Decrypted here and nowhere upstream.** `core/telemetry_crypto` lists + where agent telemetry is turned back into plaintext - the prompt and the + correlation fields - and where it is not: the stored row and the broker + message. The index was a fifth destination that list never mentioned, and + it got ciphertext by default rather than by choice. + + Indexing plaintext is a real cost, not a free fix: `sentora-logs-*` becomes + a readable copy of the telemetry it covers, so it has to be treated as + sensitive - see `docs/production-deployment.md`. The alternative was a hunt + surface that cannot match a process name, which is not a security control, + only a broken feature that reports success. + + A value that will not decrypt is indexed as it stands. `decrypt_value` + counts those, and a missing key is reported rather than guessed at: an + index quietly holding ciphertext is what this change exists to end. """ - Index a single log entry into OpenSearch. - Index name pattern: sentora-logs- - """ + if table not in INDEXED_TABLES: + return None + try: index_name = f"sentora-logs-{table.replace('_', '-')}" - - doc = dict(item) + + doc = telemetry_crypto.decrypt_item(table, dict(item)) doc["agent_name"] = agent doc["@timestamp"] = datetime.now().isoformat() - + doc.pop("id", None) - + response = client.index( index=index_name, body=doc, diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 5ba9dd2..277dfe6 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -139,6 +139,30 @@ authenticate in the default configuration: OpenSearch runs with log the platform has collected. Publish through the same reverse proxy and auth as `app` instead. +**OpenSearch holds plaintext telemetry, and MySQL does not.** This is the one +place in the stack where that is true, so it changes what a compromise of the +volume yields: + +| Store | Volume | What a copy of it gives an attacker | +| :--- | :--- | :--- | +| MySQL | `mysql_data` | Ciphertext for the encrypted columns. Needs `data/fernet.key` as well to be readable. | +| RabbitMQ | — | Ciphertext. A queued message is data at rest for as long as it is queued, so it is left encrypted. | +| **OpenSearch** | `opensearch_data` | **Readable log lines, process command lines, registry values and file paths** — for the event tables it indexes. | + +That is a deliberate trade, not an oversight. The index used to hold +`enc::gAAAA...` in the field operators search on, which meant +`message:powershell` matched none of the events that contained it while +`source:PowerShell` matched 52 — a hunting surface that reports success and +finds nothing is not a security control. `core/opensearch.INDEXED_TABLES` +limits the plaintext to the tables a search is actually for; the inventory and +snapshot tables are excluded and answered by their own views. + +So: treat `opensearch_data` and port 9200 as you would the database, keep +`data/fernet.key` on a different trust boundary from the index if you can, and +run `scripts/prune_search_indices.py` once after upgrading — the indices +written before the allowlist existed are still on disk, roughly 1.1 million +documents of stale inventory copies. + The "Open Dashboards" button in Log Explorer links to port 5601 on the server's hostname, so it works from the host itself; remote operators need that proxy. diff --git a/frontend/src/pages/Assets.tsx b/frontend/src/pages/Assets.tsx index f01f4f7..ca0333b 100644 --- a/frontend/src/pages/Assets.tsx +++ b/frontend/src/pages/Assets.tsx @@ -3,8 +3,9 @@ import { Database, Search, Cpu, - Package, - Globe, + Package, + Globe, + EthernetPort, Server, RefreshCw, ChevronRight, @@ -31,7 +32,7 @@ const STATE_TONE: Record = { const Assets: React.FC = () => { const [agents, setAgents] = useState([]); const [selectedAgent, setSelectedAgent] = useState(null); - const [activeTab, setActiveTab] = useState<'hardware' | 'software' | 'network'>('hardware'); + const [activeTab, setActiveTab] = useState<'hardware' | 'software' | 'network' | 'ports'>('hardware'); const [data, setData] = useState([]); const [loading, setLoading] = useState(false); const [searchTerm, setSearchTerm] = useState(''); @@ -86,6 +87,10 @@ const Assets: React.FC = () => { const field = activeTab === 'hardware' ? (i: any) => i.type : activeTab === 'software' ? (i: any) => i.vendor + // Every listening socket is in state LISTEN, so grouping ports by + // state draws one bar. The useful question is which process owns + // the surface. + : activeTab === 'ports' ? (i: any) => i.process_name : (i: any) => i.state; const counts = new Map(); @@ -106,8 +111,11 @@ const Assets: React.FC = () => { ? { title: 'Hardware by type', note: 'What the inventory is made of. A category that is missing entirely usually means the collector for it failed rather than the host has none.' } : activeTab === 'software' - ? { title: 'Packages by vendor', + ? { title: 'Software by vendor', note: 'Who wrote the software on this host. A long tail of one-off vendors is where unmanaged software lives.' } + : activeTab === 'ports' + ? { title: 'Listening ports by process', + note: 'The host’s own account of what can be reached from outside, with the process that owns each socket. A listener you cannot name is the one to look at first.' } : { title: 'Connections by state', note: 'Listening sockets are attack surface; established ones are traffic already flowing. The mix is the shape of what this host is doing.' }; @@ -202,10 +210,15 @@ const Assets: React.FC = () => { {/* Main Content */}
{/* Tabs */} -
+ {/* flexWrap, because a fourth tab is what takes this row + past a narrow viewport. `width: fit-content` does not + shrink below its content, so without wrapping the last + tab is simply unreachable on a phone. */} +
setActiveTab('hardware')} icon={} label="Hardware" /> setActiveTab('software')} icon={} label="Software" /> setActiveTab('network')} icon={} label="Network" /> + setActiveTab('ports')} icon={} label="Ports" />
@@ -262,6 +275,18 @@ const Assets: React.FC = () => { Remote State } + {/* No State column and no Remote: + every row here is a listener, so + both are constant. A column with + the same value on every row spends + width to say nothing. */} + {activeTab === 'ports' && <> + Port + Protocol + Bound To + Process + PID + } @@ -308,6 +333,13 @@ const Assets: React.FC = () => { } + {activeTab === 'ports' && <> + {item.local_port} + {item.protocol} + + {item.process_name} + {item.pid} + } ))} @@ -342,6 +374,36 @@ const TabButton = ({ active, onClick, icon, label }: any) => ( ); +/** What a bind address means for reach, which is the only thing that makes one + * listening port more interesting than another. + * + * `0.0.0.0` and `::` accept from any interface; `127.0.0.1` and `::1` cannot + * be reached off the host at all. Rendering both as a bare address leaves the + * operator to make that distinction 80 rows at a time, and it is the whole + * question they came to the tab with. */ +const LOCAL_ONLY = new Set(['127.0.0.1', '::1', 'localhost']); +const ANY_INTERFACE = new Set(['0.0.0.0', '::', '*', '']); + +const Reach = ({ address }: { address: string }) => { + const addr = (address ?? '').trim(); + const exposed = ANY_INTERFACE.has(addr); + const local = LOCAL_ONLY.has(addr); + return ( + + {addr || '0.0.0.0'} + + {exposed ? 'all interfaces' : local ? 'loopback' : 'one interface'} + + + ); +}; + const Th = ({ children }: any) => ( {children} ); diff --git a/scripts/prune_search_indices.py b/scripts/prune_search_indices.py new file mode 100644 index 0000000..cdfd9d0 --- /dev/null +++ b/scripts/prune_search_indices.py @@ -0,0 +1,98 @@ +"""Remove the search indices Sentora no longer writes. + +`core/opensearch.INDEXED_TABLES` is an allowlist now. Before it existed, every +ingested table was indexed, which produced two piles nobody asked for: + + sentora-logs-hardware-inventory 426,768 documents + sentora-logs-packages 255,534 + sentora-logs-software-inventory 165,297 + sentora-logs-network-connections 165,068 + sentora-logs-docker-containers 63,364 + sentora-logs-network-inventory 38,142 + sentora-logs-resource-usage 26,181 + sentora-logs-disk-usage 2,630 + +Those are snapshot and inventory tables. MySQL empties and rewrites them on +every batch; the index kept every copy and none of the deletes, so what looks +like history is a pile of stale rows with no way to tell which one is current. +Each has a route and a view that answers the question exactly. + +**Not run automatically, and not from the server.** Deleting an index is not +reversible and these may be the backing store for saved visualisations in +OpenSearch Dashboards - which this script cannot see and must not assume +about. So it prints, asks, and only then deletes. + + docker compose exec sentora-server python scripts/prune_search_indices.py + docker compose exec sentora-server python scripts/prune_search_indices.py --yes + +Nothing here touches MySQL. The rows are still there, and the views still read +them; this removes only the duplicate search copy. +""" +from __future__ import annotations + +import argparse +import sys + +sys.path.insert(0, ".") + +from core import opensearch as os_utils # noqa: E402 + + +def index_for(table: str) -> str: + return f"sentora-logs-{table.replace('_', '-')}" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--yes", action="store_true", + help="delete without asking") + args = parser.parse_args() + + client = os_utils.client + candidates = [] + + for table, reason in sorted(os_utils.NOT_INDEXED.items()): + index = index_for(table) + try: + if not client.indices.exists(index=index): + continue + count = client.count(index=index).get("count", 0) + except Exception as exc: + print(f"[!] {index}: could not be read ({exc}); skipping") + continue + candidates.append((index, count, reason)) + + if not candidates: + print("Nothing to prune: no index exists for a table that is no " + "longer written.") + return 0 + + total = sum(count for _, count, _ in candidates) + print(f"{len(candidates)} index/indices, {total:,} documents:\n") + for index, count, reason in candidates: + print(f" {index:<40} {count:>10,} {reason}") + + print("\nThe rows stay in MySQL. This removes the search copy only.") + + if not args.yes: + # input(), deliberately. A destructive default is how a maintenance + # script becomes the thing that lost the data. + answer = input("\nDelete these indices? [y/N] ").strip().lower() + if answer not in ("y", "yes"): + print("Nothing deleted.") + return 1 + + failed = 0 + for index, count, _ in candidates: + try: + client.indices.delete(index=index) + print(f"[+] deleted {index} ({count:,} documents)") + except Exception as exc: + failed += 1 + print(f"[!] {index}: {exc}") + + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server.py b/server.py index 3e2ea3f..b149bab 100644 --- a/server.py +++ b/server.py @@ -626,13 +626,32 @@ def _repair_rows_stored_without_a_fingerprint(cursor, db_name: str) -> None: The guard is therefore a primary-key lookup, and the marker is written after the work rather than before: an interrupted repair resumes instead of recording itself as done. + + **Bounded to rows that existed when it started**, which is not a detail. + An agent that has not been rebuilt still sends these tables with no + `dup_fp`, so an unbounded "delete what has no fingerprint" deletes each + fresh batch as it lands: the table stays empty, the console reads "this + host has no software", and the repair never converges so it never stops. + The high-water mark is taken once, per table, and stored - so old rows go, + new ones stay whatever they look like, and upgrading the agent is what + fixes the duplicates rather than a condition for seeing any data at all. """ cursor.execute(""" CREATE TABLE IF NOT EXISTS ingest_migration ( name VARCHAR(64) NOT NULL PRIMARY KEY, + value BIGINT NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB """) + # An earlier build of this created the table without `value`. + cursor.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = %s AND table_name = 'ingest_migration'", + (db_name,), + ) + if "value" not in {str(r[0]).lower() for r in cursor.fetchall()}: + cursor.execute("ALTER TABLE ingest_migration ADD COLUMN value BIGINT NULL") + cursor.execute("SELECT 1 FROM ingest_migration WHERE name = %s", (_DEDUP_REPAIR,)) if cursor.fetchone(): return @@ -649,22 +668,42 @@ def _repair_rows_stored_without_a_fingerprint(cursor, db_name: str) -> None: if table not in present: continue + # The newest row when this table was first looked at. Recorded rather + # than recomputed, because recomputing it each pass would sweep in + # everything that arrived in between - which is the whole failure this + # bound exists to prevent. + mark_name = f"{_DEDUP_REPAIR}:{table}"[:64] + cursor.execute("SELECT value FROM ingest_migration WHERE name = %s", + (mark_name,)) + row = cursor.fetchone() + if row is None: + cursor.execute(f"SELECT COALESCE(MAX(id), 0) FROM `{table}`") + high_water = int((cursor.fetchone() or [0])[0] or 0) + cursor.execute( + "INSERT IGNORE INTO ingest_migration (name, value) VALUES (%s, %s)", + (mark_name, high_water)) + else: + high_water = int(row[0] or 0) + # Keep the lowest id per fingerprint. The derived table is not # decoration: MySQL refuses a subquery that reads the table being # deleted from unless it is materialised through one. cursor.execute( f"DELETE FROM `{table}` " - f"WHERE dup_fp IS NOT NULL AND id NOT IN (" + f"WHERE id <= %s AND dup_fp IS NOT NULL AND id NOT IN (" f" SELECT keep_id FROM (" f" SELECT MIN(id) AS keep_id FROM `{table}` " f" WHERE dup_fp IS NOT NULL GROUP BY dup_fp" f" ) AS keep" - f") LIMIT {_REPAIR_CHUNK}" + f") LIMIT {_REPAIR_CHUNK}", + (high_water,) ) collapsed = cursor.rowcount cursor.execute( - f"DELETE FROM `{table}` WHERE dup_fp IS NULL LIMIT {_REPAIR_CHUNK}") + f"DELETE FROM `{table}` WHERE id <= %s AND dup_fp IS NULL " + f"LIMIT {_REPAIR_CHUNK}", + (high_water,)) dropped = cursor.rowcount if collapsed or dropped: diff --git a/tests/test_assets_columns.py b/tests/test_assets_columns.py index 8a235f5..0191498 100644 --- a/tests/test_assets_columns.py +++ b/tests/test_assets_columns.py @@ -31,7 +31,7 @@ APP = ROOT / "app.py" SCHEMA = ROOT / "db" / "init.sql" -TABS = ("hardware", "software", "network") +TABS = ("hardware", "software", "network", "ports") #: Keys the API adds or the page derives, which no table is expected to hold. NOT_FROM_THE_TABLE: set = set() diff --git a/tests/test_ingest_dedup.py b/tests/test_ingest_dedup.py index 4a4a146..6146504 100644 --- a/tests/test_ingest_dedup.py +++ b/tests/test_ingest_dedup.py @@ -386,30 +386,46 @@ def test_a_table_is_never_both_snapshotted_and_deduplicated(): REPAIR_CHUNK = _literal(SERVER, "_REPAIR_CHUNK") +HIGH_WATER = 163984 # what software_inventory actually held + + class _FakeCursor: """Enough of a MySQL cursor to watch what the repair does, in order. `deleted` is what a DELETE reports back: 0 for a database with nothing left to clean, a positive number for one still working through the pile. + `recorded_mark` stands for a second pass - the high-water mark is already + stored and must be reused rather than taken again. """ - def __init__(self, *, marker_present=False, tables=(), deleted=0): + def __init__(self, *, marker_present=False, tables=(), deleted=0, + recorded_mark=None): self.sql: list = [] + self.params: list = [] self._marker = marker_present self._tables = tables self._deleted = deleted + self._recorded = recorded_mark self._rows: list = [] self.rowcount = 0 def execute(self, sql, params=()): flat = " ".join(sql.split()) self.sql.append(flat) + self.params.append(params) low = flat.lower() self.rowcount = self._deleted if low.startswith("delete from `") else 0 - if "from ingest_migration" in low: + + if low.startswith("select 1 from ingest_migration"): self._rows = [(1,)] if self._marker else [] + elif low.startswith("select value from ingest_migration"): + self._rows = [(self._recorded,)] if self._recorded is not None else [] + elif "information_schema.columns" in low: + self._rows = [("value",)] # already migrated elif "information_schema.tables" in low: self._rows = [(t,) for t in self._tables] + elif "coalesce(max(id)" in low: + self._rows = [(HIGH_WATER,)] else: self._rows = [] @@ -436,13 +452,17 @@ def test_the_repair_does_nothing_once_it_has_run(): assert not any("DELETE" in s for s in cursor.sql) +#: The final marker, as distinct from the per-table high-water rows, which are +#: also written into `ingest_migration`. +DONE_MARKER = "INSERT IGNORE INTO ingest_migration (name) VALUES" + + def test_the_repair_marks_itself_done_only_at_the_end(): """Claiming the marker first would turn a half-finished repair into a permanent one - the rows stay, and nothing ever comes back to them.""" cursor = _FakeCursor(tables=RE_REPORTED_TABLES) _repair()(cursor, "agent_db") - written = [i for i, s in enumerate(cursor.sql) - if s.startswith("INSERT IGNORE INTO ingest_migration")] + written = [i for i, s in enumerate(cursor.sql) if s.startswith(DONE_MARKER)] assert written, "the repair never records that it ran" assert written[0] == len(cursor.sql) - 1 @@ -458,13 +478,60 @@ def test_the_repair_handles_both_kinds_of_stored_row(table): assert any("GROUP BY dup_fp" in s and "NOT IN" in s for s in mine), \ "duplicates sharing a fingerprint are not collapsed" - assert any(f"DELETE FROM `{table}` WHERE dup_fp IS NULL" in s for s in mine), \ + assert any(s.startswith(f"DELETE FROM `{table}`") and "dup_fp IS NULL" in s + for s in mine), \ "rows stored before the agent set a fingerprint are left behind" assert any(s.startswith("INSERT IGNORE INTO ingest_fingerprint") for s in mine), \ "the survivors' fingerprints are never recorded, so the next send " \ "stores every one of them again" +# -------------------------------------------------------------------------- +# The repair must not delete what arrives while it is running +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("table", RE_REPORTED_TABLES) +def test_no_delete_reaches_past_the_high_water_mark(table): + """An agent that has not been rebuilt still sends these tables with no + `dup_fp`. Unbounded, "delete what has no fingerprint" deletes each fresh + batch as it lands: the table stays permanently empty, the console reads + "this host has no software", and the repair never converges so it never + stops. Measured here first - `network_inventory` went to 0 rows and stayed + there while the old agent kept filling it. + """ + cursor = _FakeCursor(tables=RE_REPORTED_TABLES) + _repair()(cursor, "agent_db") + deletes = [s for s in cursor.sql if s.startswith(f"DELETE FROM `{table}`")] + assert deletes, f"nothing deletes from {table}" + for statement in deletes: + assert "id <= %s" in statement, ( + f"unbounded delete on {table}: {statement}. It will take rows that " + f"arrived after the repair started." + ) + + +def test_the_high_water_mark_is_taken_before_anything_is_deleted(): + cursor = _FakeCursor(tables=("packages",)) + _repair()(cursor, "agent_db") + took = next(i for i, s in enumerate(cursor.sql) if "COALESCE(MAX(id)" in s) + first_delete = next(i for i, s in enumerate(cursor.sql) + if s.startswith("DELETE FROM `packages`")) + assert took < first_delete + + +def test_a_later_pass_reuses_the_stored_mark_rather_than_taking_a_new_one(): + """Recomputing it each pass would sweep in everything that arrived in + between, which is the whole failure the bound exists to prevent - so a + second pass must read the stored value and not look at MAX(id) again.""" + cursor = _FakeCursor(tables=("packages",), recorded_mark=500) + _repair()(cursor, "agent_db") + assert not any("COALESCE(MAX(id)" in s for s in cursor.sql), \ + "the high-water mark is taken again on every pass" + delete = next(i for i, s in enumerate(cursor.sql) + if s.startswith("DELETE FROM `packages`")) + assert cursor.params[delete] == (500,) + + def test_the_repair_rebuilds_fingerprints_only_once_a_table_is_clean(): """The other order records fingerprints for rows that are about to be deleted in the next chunk, which leaves the table both emptied and unable @@ -481,8 +548,7 @@ def test_a_pass_with_rows_left_does_not_mark_itself_done(): started with and never looks at them again.""" cursor = _FakeCursor(tables=RE_REPORTED_TABLES, deleted=REPAIR_CHUNK) _repair()(cursor, "agent_db") - assert not any(s.startswith("INSERT IGNORE INTO ingest_migration") - for s in cursor.sql) + assert not any(s.startswith(DONE_MARKER) for s in cursor.sql) @pytest.mark.parametrize("table", RE_REPORTED_TABLES) diff --git a/tests/test_opensearch_indexing.py b/tests/test_opensearch_indexing.py new file mode 100644 index 0000000..0cc4a09 --- /dev/null +++ b/tests/test_opensearch_indexing.py @@ -0,0 +1,189 @@ +"""What reaches the search index, and in what form. + +Two bugs lived in one unconditional line. `server.py` called +`os_utils.index_log(agent, table, item)` for whatever ingest had just stored, +with `item` exactly as received - so: + + - **the index held ciphertext.** Measured on the running deployment: + + siem-events / message 71/71 enc::gAAAA... + registry-logs / value_data 300/300 + fim-data / path 300/300 + network-connections / process 300/300 + hardware-inventory / name 300/300 + packages / package 300/300 + + and the consequence, over the same 69 documents: + + message:*powershell* -> 0 hits + source:*PowerShell* -> 52 hits + + `core/telemetry_crypto` documents four destinations and which of them get + plaintext. The index was a fifth that the list never mentioned. + + - **it grew without bound.** A snapshot table is emptied and rewritten in + MySQL on every batch; the index kept every copy ever sent. 426,768 + `hardware_inventory` documents on one host, for a CPU, some RAM and four + disks. + +So the index is now an allowlist, and what goes into it is decrypted. +""" +from __future__ import annotations + +import ast +import pathlib + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +SERVER = ROOT / "server.py" +OS_MODULE = ROOT / "core" / "opensearch.py" +CRYPTO = ROOT / "core" / "telemetry_crypto.py" + + +def _literal(source: pathlib.Path, name: str): + """Read a module-level constant without importing the module. + + `frozenset({...})` is a call, not a literal, so `literal_eval` refuses it - + unwrap the one argument and evaluate that instead. + """ + for node in ast.parse(source.read_text(encoding="utf-8")).body: + if not (isinstance(node, ast.Assign) and any( + getattr(t, "id", "") == name for t in node.targets)): + continue + value = node.value + if isinstance(value, ast.Call) and getattr(value.func, "id", "") in ( + "frozenset", "set", "tuple", "list", "dict") and value.args: + value = value.args[0] + return ast.literal_eval(value) + raise AssertionError(f"{name} is not a module-level literal in {source.name}") + + +ALLOWED_TABLES = set(_literal(SERVER, "ALLOWED_TABLES")) +SNAPSHOT_TABLES = set(_literal(SERVER, "SNAPSHOT_TABLES")) +INDEXED_TABLES = set(_literal(OS_MODULE, "INDEXED_TABLES")) +NOT_INDEXED = _literal(OS_MODULE, "NOT_INDEXED") +ENCRYPTED_FIELDS = _literal(CRYPTO, "ENCRYPTED_FIELDS") + +#: Indexed but not ingested from an agent. `audit_logs` is the platform's own +#: trail, written by app.py's audit helper straight to `index_log`. +NOT_FROM_AN_AGENT = {"audit_logs"} + + +# -------------------------------------------------------------------------- +# Every ingested table is a decision, one way or the other +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("table", sorted(ALLOWED_TABLES)) +def test_an_ingested_table_is_either_indexed_or_explained(table): + """A table added to ingest and to neither list is a silent decision. + + Silent in the direction that matters: `index_log` returns early for + anything it does not recognise, so the table is simply absent from search + and nothing anywhere says so. That is the shape of every bug in this + repository's history - a gap that renders identically to a quiet system. + """ + if table in INDEXED_TABLES: + return + assert table in NOT_INDEXED, ( + f"{table} is ingested, is not indexed, and no reason is recorded. Add " + f"it to INDEXED_TABLES, or to NOT_INDEXED with the view that answers " + f"the question instead." + ) + assert NOT_INDEXED[table].strip(), f"{table} has an empty reason" + + +def test_the_two_lists_do_not_overlap(): + assert not (INDEXED_TABLES & set(NOT_INDEXED)) + + +def test_nothing_is_excused_that_is_not_ingested(): + """A stale entry in NOT_INDEXED reads as a considered decision about a + table that no longer exists.""" + stale = sorted(set(NOT_INDEXED) - ALLOWED_TABLES - {"vulnerabilities_report"}) + assert not stale, f"NOT_INDEXED mentions tables ingest does not accept: {stale}" + + +def test_every_indexed_table_is_ingested_or_named_as_platform_data(): + unexplained = sorted(INDEXED_TABLES - ALLOWED_TABLES - NOT_FROM_AN_AGENT) + assert not unexplained, ( + f"{unexplained} are indexed but never ingested; nothing writes them." + ) + + +# -------------------------------------------------------------------------- +# The two properties that were actually broken +# -------------------------------------------------------------------------- + +def test_what_is_indexed_is_decrypted_first(): + """The bug, as a structural check. + + Asserted against the AST rather than the source text: a comment explaining + that the document is decrypted is exactly what this file would have matched + while the code did nothing of the kind. + """ + tree = ast.parse(OS_MODULE.read_text(encoding="utf-8")) + fn = next(n for n in ast.walk(tree) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + and n.name == "index_log") + + calls = {n.func.attr for n in ast.walk(fn) + if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)} + assert "decrypt_item" in calls, ( + "index_log indexes the row as received. Encrypted columns arrive as " + "enc::gAAAA..., so the field an operator searches on holds ciphertext " + "and matches nothing." + ) + + +def test_the_allowlist_is_checked_before_anything_is_sent(): + """The guard has to be the first thing, not a filter after the document is + built - a half-built document that then returns is how a later edit turns + the guard back off without failing anything.""" + tree = ast.parse(OS_MODULE.read_text(encoding="utf-8")) + fn = next(n for n in ast.walk(tree) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + and n.name == "index_log") + body = [n for n in fn.body if not isinstance(n, ast.Expr)] # drop docstring + first = body[0] + assert isinstance(first, ast.If), "index_log does not start with the guard" + assert "INDEXED_TABLES" in ast.dump(first.test) + assert any(isinstance(n, ast.Return) for n in ast.walk(first)) + + +@pytest.mark.parametrize("table", sorted(INDEXED_TABLES - {"audit_logs"})) +def test_an_indexed_table_with_encrypted_columns_is_covered_by_the_crypto_map(table): + """`decrypt_item` decrypts the fields `ENCRYPTED_FIELDS` names for a table + and nothing else. A table indexed but missing from that map goes in as + ciphertext again, through the new code path rather than the old one. + """ + from_agent_encrypted = { + "siem_events", "events_alert", "fim_data", "registry_logs", + "process_events", "critical_files", "security_audit", + } + if table not in from_agent_encrypted: + return # written in the clear; nothing to decrypt + assert table in ENCRYPTED_FIELDS, ( + f"{table} is indexed and the agent encrypts it, but " + f"telemetry_crypto.ENCRYPTED_FIELDS does not list it, so decrypt_item " + f"leaves it as ciphertext." + ) + + +# -------------------------------------------------------------------------- +# Why the state tables are out +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("table", sorted(SNAPSHOT_TABLES)) +def test_a_snapshot_table_is_not_indexed(table): + """A snapshot is emptied and rewritten in MySQL on every batch, so the + table holds one current picture. The index has no such mechanism and keeps + every picture ever sent - which is where 426,768 hardware_inventory + documents came from, and why they are not a history anybody can use: the + deletes are not in them. + """ + assert table not in INDEXED_TABLES, ( + f"{table} is a snapshot table. Indexing it accumulates one document " + f"per entity per collection cycle, for ever, with no record of the " + f"deletes that made it a snapshot." + )