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." + )