From 646409951db535576eea5e801e3b5a47f0f2b8b0 Mon Sep 17 00:00:00 2001 From: d3vhex Date: Mon, 7 Sep 2026 03:00:36 +0300 Subject: [PATCH] Watch the platform itself, and encrypt what we said we encrypted Seven writers stored fields declared encrypted in the clear, and plaintext reads back identically to ciphertext so nothing could see it; the portscanner had been dead every cycle on a SQL string it built by hand. --- .env.example | 8 + Sentora/main.py | 11 +- Sentora/modules/console.py | 46 +++- Sentora/modules/db.py | 36 ++- Sentora/modules/enc_db.py | 30 +++ Sentora/modules/fim.py | 17 +- Sentora/modules/inventory.py | 11 +- Sentora/modules/portscanner/portscanner.py | 36 +-- Sentora/modules/resource_checker/disks.py | 6 +- app.py | 173 ++++++++++++++- core/agent_link.py | 24 +- core/self_defence.py | 173 +++++++++++++++ docker-compose.yaml | 20 +- docs/production-deployment.md | 40 ++-- frontend/src/pages/AuditLogs.tsx | 106 ++++++++- frontend/src/services/api.ts | 5 + tests/test_agent_sql_parameters.py | 107 +++++++++ tests/test_auth_wiring.py | 56 +++++ tests/test_console_open_deadline.py | 129 +++++++++++ tests/test_encryption_actually_applied.py | 126 +++++++++++ tests/test_key_mismatch_visibility.py | 136 ++++++++++++ tests/test_self_defence.py | 241 +++++++++++++++++++++ tests/test_tls_documentation.py | 36 ++- 23 files changed, 1515 insertions(+), 58 deletions(-) create mode 100644 core/self_defence.py create mode 100644 tests/test_agent_sql_parameters.py create mode 100644 tests/test_console_open_deadline.py create mode 100644 tests/test_encryption_actually_applied.py create mode 100644 tests/test_key_mismatch_visibility.py create mode 100644 tests/test_self_defence.py diff --git a/.env.example b/.env.example index cbeb694..6de6a51 100644 --- a/.env.example +++ b/.env.example @@ -110,6 +110,14 @@ AGENT_SHARED_SECRET= # un-migrated agent is refused rather than silently accepted. # INGEST_TLS_PORT=5011 # INGEST_TLS_REQUIRED=0 +# +# Closing the plaintext path takes both of these. INGEST_TLS_REQUIRED +# stops the listener inside the container; Docker goes on publishing +# the host port regardless, so a connection is still accepted by the +# proxy and the batch written into a socket nobody reads - which looks +# exactly like a server that stored it. Bind the published port to +# loopback to make the refusal real. +# INGEST_PLAINTEXT_BIND=127.0.0.1 # SESSION_COOKIE_SECURE follows TLS_ENABLED when it is not set. It used to # default to 0 regardless, so turning on TLS and leaving this alone gave an diff --git a/Sentora/main.py b/Sentora/main.py index 0116315..8c568b4 100644 --- a/Sentora/main.py +++ b/Sentora/main.py @@ -627,10 +627,13 @@ def send_alert(source, severity, message, metadata=None): return rec["dup_fp"] = fingerprint - if hasattr(enc_db, "insert_record_enc"): - enc_db.insert_record_enc("events_alert", rec) - else: - insert_record("events_alert", rec) + # No fallback to the plain insert. `events_alert` has `source` and + # `message` declared encrypted, so writing one unencrypted stores the + # alert text in the clear - and plaintext and ciphertext render + # identically downstream, so nothing would ever say it happened. + # `insert_record_enc` has always existed; the guard only described a + # way to fail quietly. + enc_db.insert_record_enc("events_alert", rec) except Exception as e: print(f"[Alert] Failed to send alert: {e}") diff --git a/Sentora/modules/console.py b/Sentora/modules/console.py index 152ae10..d633bb5 100644 --- a/Sentora/modules/console.py +++ b/Sentora/modules/console.py @@ -1165,6 +1165,14 @@ def new_direct_session(argv: list[str] | None = None): raise ConsoleUnavailable(" | ".join(failures) or describe_unavailable()) +#: The console strategy that worked last time in this process, if any. +#: +#: Process-local and forgotten on restart, which is the right lifetime: it +#: records how *this* process was started, and a restarted agent may have been +#: started differently. +_WORKING_STRATEGY = None + + def new_session(argv: list[str] | None = None) -> "PtySession | WinPtySession": """The right session for this host, or a reason there is none. @@ -1180,6 +1188,32 @@ def new_session(argv: list[str] | None = None) -> "PtySession | WinPtySession": if argv: return WinPtySession(argv) + # What worked last time, first. + # + # This chain is not cheap when it fails. On a host where the agent runs as + # a service, the in-process pseudoconsole dies with STATUS_CONTROL_C_EXIT + # and the session helper then times out after twelve seconds - so roughly + # fifteen seconds elapse before the strategy that actually works is even + # tried. The server gives up on a console after fifteen. Measured on a + # live host: 17:25:03 the first attempt, 17:25:18 the fall through to + # pipes, and by then nobody was listening. + # + # Whether this process can host a pseudoconsole is a property of how it + # was started. It cannot change while it is running, so paying fifteen + # seconds to rediscover it on every console open is fifteen seconds spent + # proving something already known. + global _WORKING_STRATEGY + if _WORKING_STRATEGY is not None: + try: + return _WORKING_STRATEGY() + except ConsoleUnavailable as e: + # The remembered answer stopped being true - a helper that has + # since died, a session that logged out. Fall through and work it + # out again rather than reporting a failure that may be stale. + print(f"[console] the strategy that worked last time no longer " + f"does ({e}); trying the others", flush=True) + _WORKING_STRATEGY = None + # Try here first. It costs a few hundred milliseconds and it is the only # path that works when the agent is not a service - a developer running # it from a terminal, for instance. @@ -1189,14 +1223,18 @@ def new_session(argv: list[str] | None = None) -> "PtySession | WinPtySession": # fail, which is the path whose whole purpose is to explain itself. direct_error = "" try: - return new_direct_session() + session = new_direct_session() + _WORKING_STRATEGY = new_direct_session + return session except ConsoleUnavailable as e: direct_error = str(e) print(f"[console] no console in this process ({direct_error}); " f"trying the user's session", flush=True) try: - return HelperSession() + session = HelperSession() + _WORKING_STRATEGY = HelperSession + return session except ConsoleUnavailable as e: helper_error = str(e) print(f"[console] no console in the user's session either " @@ -1205,7 +1243,9 @@ def new_session(argv: list[str] | None = None) -> "PtySession | WinPtySession": # Last resort, and honest about it - see PipeSession. try: - return PipeSession() + session = PipeSession() + _WORKING_STRATEGY = PipeSession + return session except ConsoleUnavailable as pipe_error: raise ConsoleUnavailable( f"no console of any kind could be started. " diff --git a/Sentora/modules/db.py b/Sentora/modules/db.py index 0d55039..3baaa7e 100644 --- a/Sentora/modules/db.py +++ b/Sentora/modules/db.py @@ -221,11 +221,41 @@ def reoffer_recent(table: str, limit: int) -> int: The resend is safe because the server deduplicates: rows it already holds are recognised and skipped rather than stored twice. """ - query = (f"UPDATE {table} SET sent = FALSE WHERE id IN " - f"(SELECT id FROM {table} ORDER BY id DESC LIMIT %s)") + from modules import enc_db + with get_conn() as conn: + with conn.cursor(cursor_factory=DictCursor) as cur: + cur.execute(f"SELECT * FROM {table} ORDER BY id DESC LIMIT %s", + (limit,)) + candidates = cur.fetchall() + + # Skip what this agent can no longer read itself. + # + # The Fernet key comes from the server's bootstrap and there is no + # rotation path, so an agent that has outlived one key holds rows it + # cannot decrypt - and neither can the server. Offering them again + # imports rows that render as `` for + # ever, spending a bounded recovery budget on data nobody can read. + # + # Found by doing it: the first re-offer on a live host moved about two + # thousand such rows onto the server, and the readable/unreadable + # boundary in the result was exact. + keep = [] + for row in candidates: + try: + if enc_db.is_readable(table, dict(row)): + keep.append(row["id"]) + except Exception: + # No key configured yet, or a shape this does not understand. + # Offering it is the older behaviour and the safer default: + # the cost is an unreadable row, not a lost one. + keep.append(row["id"]) + + if not keep: + return 0 with conn.cursor() as cur: - cur.execute(query, (limit,)) + cur.execute(f"UPDATE {table} SET sent = FALSE WHERE id = ANY(%s)", + (keep,)) changed = cur.rowcount conn.commit() return changed if changed and changed > 0 else 0 diff --git a/Sentora/modules/enc_db.py b/Sentora/modules/enc_db.py index 51375dd..c155157 100644 --- a/Sentora/modules/enc_db.py +++ b/Sentora/modules/enc_db.py @@ -137,6 +137,36 @@ def _dec_value(v: Any) -> Any: return v +def is_readable(table: str, row: Dict[str, Any]) -> bool: + """Whether this agent can still decrypt its own stored row. + + It cannot, for anything encrypted before the Fernet key last changed. The + key comes from the server's bootstrap, and there is no rotation path - so + an agent that has outlived one key holds a local history it can no longer + read, and neither can the server. + + That matters when rows are offered again after a server-side reset. Those + rows arrive, store, and render as `` + for ever: a recovery that fills the console with markers and spends its + bounded budget on rows nobody can read. Measured on a live host, the + boundary was exact - the newest 28 rows readable, every older one not. + + Rows with nothing encrypted are readable by definition, which is the + common case and costs one dictionary lookup. + """ + fields = ENCRYPT_FIELDS_MAP.get(table) or [] + for field in fields: + value = row.get(field) + if not isinstance(value, str) or not value.startswith(ENC_PREFIX): + continue # plaintext, or absent + if _dec_value(value) is value: + # `_dec_value` hands the input straight back when the token will + # not open. Identity rather than equality: a value that decrypts + # to itself is vanishingly unlikely and would be harmless anyway. + return False + return True + + def _encrypt_row(table: str, data: Dict[str, Any]) -> Dict[str, Any]: return {k: (_enc_value(v) if _should_encrypt(table, k) else v) for k, v in data.items()} diff --git a/Sentora/modules/fim.py b/Sentora/modules/fim.py index e2e087f..e4e8497 100644 --- a/Sentora/modules/fim.py +++ b/Sentora/modules/fim.py @@ -29,7 +29,7 @@ from watchdog.events import FileSystemEventHandler from . import agent_paths -from .db import insert_record +from .enc_db import insert_record_enc DEFAULT_MONITOR_PATHS = [ "/etc", @@ -157,7 +157,20 @@ def process(self, path, status): return current_hash = calculate_sha256(path) if status != "deleted" else "DELETED" - insert_record("fim_data", { + # `insert_record_enc`, not `insert_record`. + # + # `enc_db.ENCRYPT_FIELDS_MAP` lists fim_data's `path` and + # `hash_sha256` as encrypted and the server lists them as fields to + # decrypt, and this writer - which produces almost every fim_data row + # there is - used the plain insert. So the file paths of every + # monitored host were stored in the clear, locally and on the server, + # while both ends of the design said otherwise. + # + # It was invisible because plaintext and ciphertext render identically: + # the server's `decrypt_value` hands anything without the `enc::` + # prefix straight back, so the console showed a perfect table. The only + # way to see it was to ask what the stored bytes actually start with. + insert_record_enc("fim_data", { "path": path, "hash_sha256": current_hash or "ERROR", "status": status, diff --git a/Sentora/modules/inventory.py b/Sentora/modules/inventory.py index a31b3c4..7852e6f 100644 --- a/Sentora/modules/inventory.py +++ b/Sentora/modules/inventory.py @@ -5,6 +5,7 @@ import subprocess from datetime import datetime from .db import insert_record +from .enc_db import insert_record_enc def get_cpu_info(): cpu_name = platform.processor() @@ -193,26 +194,26 @@ def scan_inventory(): gpu = get_gpu_info() mobo = get_mobo_info() - insert_record("hardware_inventory", { + insert_record_enc("hardware_inventory", { "type": "cpu", "name": cpu, "vendor_id": "CPU", "product_id": "Model", "serial_number": "N/A", "status": "active", "timestamp": timestamp, "sent": False }) - insert_record("hardware_inventory", { + insert_record_enc("hardware_inventory", { "type": "ram", "name": ram, "vendor_id": "System", "product_id": "Memory", "serial_number": "N/A", "status": "active", "timestamp": timestamp, "sent": False }) - insert_record("hardware_inventory", { + insert_record_enc("hardware_inventory", { "type": "gpu", "name": gpu, "vendor_id": "GPU", "product_id": "Display", "serial_number": "N/A", "status": "active", "timestamp": timestamp, "sent": False }) - insert_record("hardware_inventory", { + insert_record_enc("hardware_inventory", { "type": "motherboard", "name": mobo, "vendor_id": "System", "product_id": "BaseBoard", "serial_number": "N/A", "status": "active", "timestamp": timestamp, "sent": False }) for part in psutil.disk_partitions(): try: - insert_record("hardware_inventory", { + insert_record_enc("hardware_inventory", { "type": "disk", "name": part.device, "vendor_id": part.fstype, "product_id": part.mountpoint, "serial_number": "N/A", "status": "active", "timestamp": timestamp, "sent": False diff --git a/Sentora/modules/portscanner/portscanner.py b/Sentora/modules/portscanner/portscanner.py index 335719f..c6dba7d 100644 --- a/Sentora/modules/portscanner/portscanner.py +++ b/Sentora/modules/portscanner/portscanner.py @@ -382,23 +382,29 @@ ], } -def normalize(value): - return value.replace("'", "''") if isinstance(value, str) else value - def is_duplicate(port, protocol, service, product, version): - protocol = normalize(protocol) - service = normalize(service) - product = normalize(product) - version = normalize(version) + """Have we already recorded this exact service on this port? + + Parameterised, and it was not. The condition used to be assembled by + interpolation with a hand-rolled `normalize()` that doubled single quotes, + and on this host it produced + + [!] portscanner failed: SyntaxError: unterminated quoted string + at or near "'" + + every cycle - so no port scan result was written at all. Which input broke + it hardly matters: nmap hands back product and version strings containing + whatever a service chose to advertise, an attacker-controlled banner among + them, and the answer to that is never a better escaping function. - condition = ( - f"port = {port} AND " - f"protocol = '{protocol}' AND " - f"service = '{service}' AND " - f"product = '{product}' AND " - f"version = '{version}'" - ) - return bool(fetch_where(TABLE, condition)) + A security product that builds SQL by string concatenation is running the + pattern it exists to detect elsewhere. `fetch_where` has taken a `params` + tuple all along. + """ + condition = ("port = %s AND protocol = %s AND service = %s " + "AND product = %s AND version = %s") + params = (port, protocol, service, product, version) + return bool(fetch_where(TABLE, condition, params)) async def smart_recv(reader, size=8192, max_attempts=5): diff --git a/Sentora/modules/resource_checker/disks.py b/Sentora/modules/resource_checker/disks.py index ed094b9..62bffa6 100644 --- a/Sentora/modules/resource_checker/disks.py +++ b/Sentora/modules/resource_checker/disks.py @@ -3,6 +3,7 @@ import platform from datetime import datetime from modules.db import insert_record, fetch_where, update_record, delete_all +from modules.enc_db import insert_record_enc def bytes_to_gb(byte_value): return round(byte_value / (1024**3), 2) @@ -19,7 +20,10 @@ def send_local_alert(severity, message, metadata=None): } if metadata: rec["message"] += f" | {json.dumps(metadata)}" - insert_record("events_alert", rec) + # `source` and `message` are declared encrypted, so the plain + # insert stored this alert in the clear - and plaintext reads + # back identically to ciphertext, so nothing said so. + insert_record_enc("events_alert", rec) except Exception as e: print(f"[DiskMonitor] Failed to send alert: {e}") diff --git a/app.py b/app.py index 0f6f794..54bf38f 100644 --- a/app.py +++ b/app.py @@ -1,6 +1,7 @@ from sanic import Request, Sanic import bcrypt import aiomysql +import inspect import secrets import os, pathlib import stat @@ -115,6 +116,7 @@ def password_complexity(cls, v: str): from core import config_validation from core import login_guard from core import netloc +from core import self_defence from core import threat_feeds from core import tls as product_tls from core import version as product_version @@ -2007,9 +2009,39 @@ def _close_sync_quiet(conn): pass async def _close_async_quiet(conn): + """Return a connection, whatever kind it is, without ever raising. + + This used to read + + await conn.ensure_closed() if hasattr(conn, "ensure_closed") else conn.close() + + and `await` binds tighter than a conditional expression, so Python parses + that as `(await conn.ensure_closed()) if ... else conn.close()`. The else + branch was never awaited - and `_PooledConn.close` is a coroutine, so the + connection went back to the pool only when the garbage collector got round + to `__del__`, with + + RuntimeWarning: coroutine '_PooledConn.close' was never awaited + + the only sign. Not a leak, because `__del__` catches it, but a pool that + hands back its slots on GC timing is a pool that runs out under exactly + the load that makes GC late. + + `close()` first, deliberately. On `_PooledConn` that is the method that + releases the slot; `ensure_closed` would reach through `__getattr__` to the + underlying connection and shut the socket without ever telling the pool - + turning a warning into the slot leak it was only pretending to be. + """ + if conn is None: + return try: - if conn is not None: - await conn.ensure_closed() if hasattr(conn, "ensure_closed") else conn.close() + result = conn.close() + if inspect.isawaitable(result): + await result + elif hasattr(conn, "ensure_closed"): + # aiomysql: `close()` is synchronous and only marks it; this is + # the one that actually waits for the socket. + await conn.ensure_closed() except Exception: pass @@ -2435,6 +2467,7 @@ def decrypt_row_fields(row: dict, encrypted_fields: list, fernet: Fernet) -> dic except Exception: decoded = pt.decode("utf-8") out[field] = decoded + _decrypt_counters["decrypted"] += 1 except InvalidToken: # Returning the raw ciphertext made a key mismatch look exactly # like a field nobody decrypted — the operator sees `enc::gAAA...` @@ -2451,9 +2484,46 @@ def decrypt_row_fields(row: dict, encrypted_fields: list, fernet: Fernet) -> dic f"if it was registered against an older key.", flush=True, ) + _decrypt_counters["undecryptable"] += 1 out[field] = "" return out + +#: How many stored values this process has read, and how many it could not. +#: +#: Counted per field, and counted at all, because the scale of a key mismatch +#: was invisible. One log line per field name was the whole signal - a `set` +#: of names, printed once each - so a table where 86% of the messages could +#: not be read produced exactly the same output as one bad row. Measured on a +#: live host it was 173 of 200 in `siem_events`, 197 of 200 in `events_alert`. +#: +#: Process-local and reset on restart, deliberately. This is a health number, +#: not a ledger: "how much of what I am serving right now is unreadable". +_decrypt_counters = {"decrypted": 0, "undecryptable": 0} + + +def decryption_health() -> dict: + """What proportion of encrypted values this process could actually read. + + Zero of zero is not a problem, and has to say so rather than dividing. + """ + ok = _decrypt_counters["decrypted"] + bad = _decrypt_counters["undecryptable"] + total = ok + bad + return { + "decrypted": ok, + "undecryptable": bad, + "unreadable_percent": round(bad * 100.0 / total, 1) if total else 0.0, + "detail": ( + "Values encrypted with a Fernet key this server no longer holds. " + "There is no rotation path, so these cannot be recovered - the " + "number matters for knowing how much of the history is readable, " + "not as something to fix." + if bad else + "Every encrypted value read so far opened with the current key." + ), + } + async def stream_from_db_dec(table: str, agent: str, connect_db_for_agent, encrypted_fields: list = None, @@ -5102,6 +5172,13 @@ async def login(request): await log_login_attempt(username, "local", "failure", f"locked out: {locked}", ip) print(f"[auth] lockout {username!r} from {ip}: {locked}", flush=True) + # The conclusion, not every attempt. One failed password is not an + # incident and storing it as one is how a security view becomes + # noise; the threshold has already been crossed by the time this + # branch runs, and that is the thing worth putting in front of + # somebody. + await _platform_event("LOGIN_LOCKOUT", subject=username, + source_ip=ip, detail=locked) return response.json({ "status": "error", "message": "Too many failed attempts. Try again later.", @@ -7330,6 +7407,64 @@ async def get_telemetry_health(request, agent): }) +@app.route("/api/platform/events") +@require_permission("manage_system") +async def get_platform_events(request): + """Attacks on this platform, as opposed to attacks on the fleet. + + Every one of these was already detected. A lockout went to a container + log, a rejected agent key went to a container log, a permission denial + went to a table nobody opens unless they are already investigating - so + the machine that watches every host in the fleet was the one nobody + watched. + + Folded rows, not raw attempts: a brute-force is one row whose + `occurrences` climbs, because a view that files four hundred rows for one + attack is the thing that hides it. + """ + hours = _positive_int(request.args.get("hours"), 24, 24 * 30) + + def _read(): + with sync_mysql_conn("userdb") as conn: + cursor = conn.cursor(dictionary=True) + try: + cursor.execute(self_defence.DDL) + cursor.execute( + "SELECT kind, severity, subject, source_ip, detail, " + " occurrences, first_seen, last_seen " + " FROM platform_events " + " WHERE last_seen > (NOW() - INTERVAL %s HOUR) " + " ORDER BY last_seen DESC LIMIT 500", (hours,)) + return cursor.fetchall() or [] + finally: + cursor.close() + + try: + rows = await asyncio.to_thread(_read) + except Exception as e: + return sanic_json({"status": "error", "message": str(e)}, status=500) + + for row in rows: + # What the kind means, sent with it. An operator seeing + # FLEET_KEY_ON_CHANNEL for the first time should not have to go + # looking for what it implies. + row["explanation"] = self_defence.describe(row["kind"])[1] + for field in ("first_seen", "last_seen"): + if row.get(field) is not None: + row[field] = row[field].isoformat() + + return sanic_json({ + "status": "success", + "window_hours": hours, + "events": rows, + "summary": self_defence.summarise(rows), + # Not an attack, and it belongs here anyway: this is the view that + # answers "is the platform itself healthy", and history nobody can + # decrypt is a fact about the platform rather than about any host. + "decryption": decryption_health(), + }) + + def _threat_trend(days: int = 14) -> list: """Detections per day across the fleet, in two counts that differ. @@ -9776,6 +9911,31 @@ def _purge(): }) +async def _platform_event(kind: str, *, subject: str = "", + source_ip: str = "", detail: str = "") -> None: + """Raise one event about an attack on this platform, and never fail. + + Swallowed on purpose, and this is the one place in the file where that is + right: every caller is in the middle of refusing something. A server that + will not reject a bad login because it could not journal the rejection has + turned a monitoring gap into an outage. + """ + def _write(): + try: + with sync_mysql_conn("userdb") as conn: + cursor = conn.cursor() + try: + self_defence.record(cursor, kind, subject=subject, + source_ip=source_ip, detail=detail) + conn.commit() + finally: + cursor.close() + except Exception as e: + print(f"[self-defence] could not record {kind}: {e}", flush=True) + + await asyncio.to_thread(_write) + + AGENT_LINKS = agent_link.LinkRegistry() @@ -9794,8 +9954,17 @@ async def agent_link_socket(request, ws): """ agent = await asyncio.to_thread(_validate_agent_auth_sync, request) if not agent or agent == "*": + # Recorded, not only printed. A container log is where this went, and + # nobody reads a container log until they already suspect something - + # which is the wrong order for the one signal that says an agent key + # is being used after it was revoked. + kind = ("FLEET_KEY_ON_CHANNEL" if agent == "*" + else "AGENT_KEY_REJECTED") print(f"[agent-link] refused a connection from {_client_ip(request)}", flush=True) + await _platform_event(kind, source_ip=_client_ip(request), + detail="a channel was opened with a key this " + "server will not accept") await ws.close(code=1008, reason="unauthorized") return diff --git a/core/agent_link.py b/core/agent_link.py index 1d34558..59e2d7d 100644 --- a/core/agent_link.py +++ b/core/agent_link.py @@ -73,6 +73,25 @@ # operator is not left watching a spinner over a host that has gone quiet. DEFAULT_TIMEOUT_S = 20.0 +# How long to wait for an agent to actually have a console or a screen. +# +# Longer than the ordinary request timeout, because opening one is not an +# ordinary request: on Windows the agent tries a pseudoconsole in its own +# process, then a helper in the interactive session, then a shell behind +# pipes, and only the last of those works when the agent runs as a service. +# +# This was 15s and the chain takes about that long to fail its way to the +# strategy that works. Measured on a live host: the attempt began at +# 17:25:03, the fall through to pipes happened at 17:25:18, and the console +# came up a second or two later - to nobody, because the server had already +# given up and told the operator the agent "would not open a console". It +# would; it was still trying. +# +# The agent now remembers which strategy worked, so the common case is a +# second or two. This deadline covers the first open after a restart, which +# is the one that has to pay for the discovery. +STREAM_OPEN_TIMEOUT_S = 45.0 + # Refused rather than queued past this. A caller that cannot be served now is # better told so: the alternative is an unbounded map fed by whatever the # agent chooses not to answer. @@ -232,12 +251,15 @@ def deliver(self, frame: dict) -> bool: # -- streams ----------------------------------------------------------- async def open_stream(self, kind: str, args: dict | None = None, - timeout: float = 15.0) -> StreamChannel: + timeout: float = STREAM_OPEN_TIMEOUT_S) -> StreamChannel: """Ask the agent to start a console or a screen, and wait for it. Waiting matters: "the frame was sent" and "the agent has a shell" are different facts, and the console spent a while reporting the first as though it were the second. + + The deadline has to be longer than the work it is waiting for, and it + was not - see `STREAM_OPEN_TIMEOUT_S`. """ if self._closed: raise LinkError(f"{self.agent} is not connected") diff --git a/core/self_defence.py b/core/self_defence.py new file mode 100644 index 0000000..0dc9da2 --- /dev/null +++ b/core/self_defence.py @@ -0,0 +1,173 @@ +"""Attacks on the platform itself, recorded where somebody will see them. + +The server already detects most of this. It just does not tell anyone. + +A brute-force against the console produces `login_logs` rows and eventually a +lockout, and the console's alert view shows nothing. An agent key that has been +revoked keeps trying and the server prints + + [agent-link] refused a connection from 203.0.113.9 + +to a container log. An operator hitting agent routes they have no permission +for lands in `audit_logs` as `PROXY_DENIED`, in a table nobody opens unless +they are already investigating something. + +So the platform watches every host in the fleet and is the one machine nobody +watches. That is worth fixing on its own, and it is worth fixing *here* rather +than by adding another table nobody reads: these become events of the same +shape as everything else, so they sort, filter and alert beside the telemetry +the console already surfaces. + +Three rules this module exists to keep: + +**Recording must never break the thing it is recording.** Every writer here +swallows its own failure. A platform that refuses a login because it could not +journal the refusal has turned a monitoring gap into an outage. + +**A severity is a claim, so it has to be earned.** One failed password is not +an incident and must not be stored as one; the fifth from the same address in +a minute is. The counting already happens in `core.login_guard`, and this +module records the *conclusion* rather than every attempt. + +**Never store a credential.** These records describe attempts on secrets, and +the obvious fields to include - the password tried, the agent key presented - +are exactly the ones that turn an audit trail into a second breach. What is +kept is enough to act on: who, from where, how many, and what they were after. +""" +from __future__ import annotations + +import datetime + + +#: The event kinds this platform raises about itself. +#: +#: Named rather than free text. A severity that varies by call site is a +#: severity nobody can filter on, and the first version of anything like this +#: always grows six spellings of "login failure". +KINDS: dict[str, tuple[str, str]] = { + # kind: (severity, what it means) + "LOGIN_LOCKOUT": ( + "HIGH", + "Repeated failed logins crossed the lockout threshold. Somebody is " + "guessing, or an integration is using a credential that has changed."), + "AGENT_KEY_REJECTED": ( + "HIGH", + "A channel was opened with a key this server does not recognise. Any " + "agent whose key was revoked will do this; so will anyone who found a " + "key that no longer works."), + "FLEET_KEY_ON_CHANNEL": ( + "CRITICAL", + "The fleet-wide secret was presented on the agent channel, which is " + "refused. That channel carries /self_destruct, so a leaked master key " + "must not open one against every endpoint at once."), + "PERMISSION_DENIED": ( + "MEDIUM", + "An authenticated operator reached for something their role does not " + "allow. One is a mis-click; a run of them is somebody mapping what " + "they can touch."), + "ENROLMENT_TOKEN_REUSED": ( + "HIGH", + "An enrolment token was presented after it had already been used. The " + "token is a one-time credential, so a second use means it leaked."), + "TLS_DOWNGRADE_REFUSED": ( + "MEDIUM", + "An agent tried the plaintext ingest port after this deployment " + "required TLS."), +} + +#: Severity order, so a caller can compare without hardcoding a list. +SEVERITY_RANK = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3, "INFO": 4} + + +def describe(kind: str) -> tuple[str, str]: + """(severity, explanation) for a kind, or a safe default. + + Unknown kinds are MEDIUM rather than dropped. A caller that invents a name + is a bug, but losing the event is worse than filing it under the wrong + heading - and MEDIUM is visible without crying wolf. + """ + return KINDS.get(kind, ("MEDIUM", "Unrecognised platform event.")) + + +DDL = """ +CREATE TABLE IF NOT EXISTS platform_events ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + kind VARCHAR(64) NOT NULL, + severity VARCHAR(16) NOT NULL, + subject VARCHAR(255) NULL, + source_ip VARCHAR(64) NULL, + detail TEXT NULL, + occurrences INT UNSIGNED NOT NULL DEFAULT 1, + first_seen TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP, + last_seen TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP, + KEY idx_pe_kind (kind), + KEY idx_pe_seen (last_seen), + KEY idx_pe_group (kind, subject, source_ip) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 +""" + +#: How long a repeat folds into the previous row instead of making a new one. +#: +#: Without this a brute-force writes a row per attempt and the view that is +#: supposed to make the attack visible becomes the thing that hides it. With +#: it, one attack is one row whose `occurrences` climbs - which is also the +#: number an operator actually wants. +COALESCE_WINDOW_MINUTES = 10 + + +def record(cursor, kind: str, *, subject: str = "", source_ip: str = "", + detail: str = "") -> None: + """File one platform security event, folding repeats together. + + Takes a cursor rather than opening its own connection: these are raised + from inside request handlers that already hold one, and a second + connection per failed login is how recording an attack becomes a way to + amplify it. + """ + severity, _ = describe(kind) + subject = (subject or "")[:255] + source_ip = (source_ip or "")[:64] + + cursor.execute(DDL) + cursor.execute( + "UPDATE platform_events SET occurrences = occurrences + 1, " + " last_seen = CURRENT_TIMESTAMP, detail = %s " + " WHERE kind = %s AND subject = %s AND source_ip = %s " + " AND last_seen > (NOW() - INTERVAL %s MINUTE) " + " ORDER BY id DESC LIMIT 1", + (detail[:1000], kind, subject, source_ip, COALESCE_WINDOW_MINUTES)) + if cursor.rowcount: + return + + cursor.execute( + "INSERT INTO platform_events (kind, severity, subject, source_ip, detail) " + "VALUES (%s, %s, %s, %s, %s)", + (kind, severity, subject, source_ip, detail[:1000])) + + +def summarise(rows) -> dict: + """Counts by severity for a set of rows, for the console's header. + + Sums `occurrences` rather than counting rows, because the rows are folded: + "3 events" where one of them is 400 failed logins is the wrong number to + put in front of somebody. + """ + by_severity: dict[str, int] = {} + attempts = 0 + for row in rows: + severity = (row.get("severity") or "MEDIUM").upper() + count = int(row.get("occurrences") or 1) + by_severity[severity] = by_severity.get(severity, 0) + count + attempts += count + return { + "by_severity": by_severity, + "attempts": attempts, + "worst": min(by_severity, key=lambda s: SEVERITY_RANK.get(s, 9), + default=None), + } + + +def since(hours: int = 24) -> datetime.datetime: + """The window boundary, as a naive UTC datetime to match the columns.""" + return (datetime.datetime.now(datetime.timezone.utc) + .replace(tzinfo=None) - datetime.timedelta(hours=hours)) diff --git a/docker-compose.yaml b/docker-compose.yaml index 12ffc36..60a0dac 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -187,9 +187,23 @@ services: # that went quiet looks exactly like a host with nothing to say. # # server.py names every agent still arriving in the clear, once each. - # When that list is empty, set INGEST_TLS_REQUIRED=1 and drop the 5001 - # line below; a connection refused is a symptom somebody can see. - - "5001:5001" + # When that list is empty, close the plaintext path with two settings in + # .env, both of which are needed: + # + # INGEST_TLS_REQUIRED=1 stops the listener inside the container + # INGEST_PLAINTEXT_BIND=127.0.0.1 stops Docker publishing the port + # + # The second is not tidiness. INGEST_TLS_REQUIRED alone closes the + # listener while Docker goes on answering on the host port: the proxy + # accepts the connection, the batch is written into a socket nobody + # reads, and the close looks exactly like a server that stored + # everything. Verified on a live stack - the bytes were accepted and + # nothing was ingested. + # + # Bound rather than removed, because this file is tracked and a fresh + # install that is not on TLS yet still needs 5001. Same mechanism as + # BIND_ADDR above. + - "${INGEST_PLAINTEXT_BIND:-0.0.0.0}:5001:5001" - "5011:5011" volumes: # Read-only, and required. The agent encrypts `message` before sending; diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 7d60ef2..0d5647d 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -407,21 +407,33 @@ Do it in this order: [!] web-01 is sending telemetry in the clear on port 5001. Its logs, hostnames and process names cross the network unencrypted... ``` -5. **Close the plaintext path** when nothing is named any more: set - `INGEST_TLS_REQUIRED=1` **and remove the `"5001:5001"` line from - `docker-compose.yaml`.** - -> **Both halves of step 5, or you get the fourth thing that was wrong.** -> `INGEST_TLS_REQUIRED=1` closes the listener *inside the container*; Docker -> goes on publishing the host port. A connection is then accepted by the -> proxy, the batch is written into a socket nobody reads, and the close looks -> exactly like a server that stored everything. Verified on a live stack: the -> bytes were accepted, nothing was ingested, and the batch reported as sent. +5. **Close the plaintext path** when nothing is named any more. Two + settings in `.env`, and both are needed: + + ```ini + INGEST_TLS_REQUIRED=1 # stops the listener in the container + INGEST_PLAINTEXT_BIND=127.0.0.1 # stops Docker publishing the port + ``` + +> **Both, or you get the fourth thing that was wrong.** +> `INGEST_TLS_REQUIRED=1` closes the listener *inside the container*; +> Docker goes on publishing the host port. A connection is then accepted +> by the proxy, the batch is written into a socket nobody reads, and the +> close looks exactly like a server that stored everything. Verified on a +> live stack: the bytes were accepted, nothing was ingested, and the batch +> reported as sent. > -> The agent now refuses to be fooled by this — once a server has acknowledged -> a batch, silence from it is treated as a failure and the rows are kept — but -> a refused connection is still the honest answer, and that means removing the -> port mapping. +> The agent no longer falls for this - once a server has acknowledged a +> batch, silence from it keeps the rows - but a refused connection is +> still the honest answer, and that means unpublishing the port. + + Verify from another machine, not from the server itself: the loopback + binding still answers there, which is the point of it. + + ```bash + nc -vz soc.example.com 5001 # expect: connection refused + nc -vz soc.example.com 5011 # expect: succeeded + ``` Setting `INGEST_TLS_REQUIRED=1` without `TLS_ENABLED=1` is a startup failure: it would close the plaintext port and open nothing, leaving an ingest service diff --git a/frontend/src/pages/AuditLogs.tsx b/frontend/src/pages/AuditLogs.tsx index 2bbf6a6..d5e02c9 100644 --- a/frontend/src/pages/AuditLogs.tsx +++ b/frontend/src/pages/AuditLogs.tsx @@ -1,7 +1,13 @@ import React, { useState, useEffect, useMemo } from 'react'; import { User, Activity, Globe, Search, Eye } from 'lucide-react'; import { adminService, agentService } from '../services/api'; -import { Card } from '../components/ui'; +import { Card, Badge } from '../components/ui'; + +/** Platform events carry a severity, and a severity is a claim - so it takes + * the semantic colours rather than a series palette. */ +const SEVERITY_TONE: Record = { + CRITICAL: 'critical', HIGH: 'high', MEDIUM: 'medium', LOW: 'low', INFO: 'neutral', +}; import { CategoryBars } from '../components/ui/charts'; const chartNote: React.CSSProperties = { @@ -27,9 +33,20 @@ const AuditLogs: React.FC = () => { const [loading, setLoading] = useState(true); const [searchTerm, setSearchTerm] = useState(''); const [selected, setSelected] = useState(null); + /* Attacks on this platform. Separate state from the audit log because it + answers a different question: that one records what operators did, this + records what was attempted against them. */ + const [platform, setPlatform] = useState(null); + const [decryption, setDecryption] = useState(null); useEffect(() => { fetchLogs(); + adminService.getPlatformEvents(24) + .then((r) => { setPlatform(r.events || []); setDecryption(r.decryption || null); }) + // A role with manage_users but not manage_system gets 403 here. That is + // not an error worth showing - the section simply does not apply to + // them - so it stays hidden rather than rendering a failure. + .catch(() => setPlatform(null)); }, []); const fetchLogs = (q: string = '*') => { @@ -75,6 +92,93 @@ const AuditLogs: React.FC = () => {

Comprehensive history of all administrative actions, resource modifications, and system changes.

+ {/* The platform watches every host in the fleet, and until recently + nothing watched the platform. A lockout went to a container log, a + rejected agent key went to a container log - so the one machine an + attacker has to get through was the one with no view of its own. + + Shown even when empty, which is a correction of the first version. + Hiding it read well as an argument - a permanently empty panel is + furniture people learn to skip - and it recreated the exact failure + the rest of this work spent its time removing: an operator who never + sees the panel cannot tell "nothing has attacked us" from "this + platform does not watch itself". On a security view, a quiet + twenty-four hours is a finding. + + `null` is the third state and stays hidden: the role cannot read + this, so it is not an empty result, it is not their question. */} + {platform !== null && ( + +

+ Failed logins past the lockout threshold, agent keys this server + does not recognise, and operators reaching past their role. + Repeats are folded: one attack is one row whose count climbs, not + four hundred rows that bury it. +

+ {/* Not an attack, and it belongs on this panel anyway: history + nobody can decrypt is a fact about the platform. It was counted + and never shown - one log line per field name, so a table where + 86% of the messages were unreadable produced the same output as + one bad row. */} + {!!decryption && decryption.undecryptable > 0 && ( +

+ + {decryption.unreadable_percent}% of encrypted values read since + this server started could not be decrypted + {' '} + ({decryption.undecryptable} of{' '} + {decryption.undecryptable + decryption.decrypted}). {decryption.detail} +

+ )} + + {platform.length === 0 ? ( +

+ Nothing in the last 24 hours. This panel is live — an empty one + means no attempts were detected, not that nothing is watching. +

+ ) : ( +
+ + + + + + + + + + + + {platform.map((e, i) => ( + + + + + + + + ))} + +
WhatSubjectFromCountLast seen
+ {e.kind} +
+ {e.explanation} +
+ 		{e.subject || '—'}{e.source_ip || '—'} + {e.occurrences} + + {formatTs(e.last_seen)} +
+
+ )} + + )} + {/* An audit log is read after something went wrong, and by then the question is never "what happened" but "who was doing what". A hundred rows in reverse-chronological order answers the first and diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index c6b26ef..25bb430 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -256,6 +256,11 @@ export const adminService = { getLoginLogs: () => api.get('/login-logs').then(res => res.data), getAuditLogs: () => api.get('/audit-logs').then(res => res.data), + // Attacks on the platform itself, as opposed to on the fleet. Separate from + // the audit log: that records what operators did, this records what was + // attempted against them. + getPlatformEvents: (hours = 24) => + api.get('/api/platform/events', { params: { hours } }).then(res => res.data), }; export default api; diff --git a/tests/test_agent_sql_parameters.py b/tests/test_agent_sql_parameters.py new file mode 100644 index 0000000..8b877d2 --- /dev/null +++ b/tests/test_agent_sql_parameters.py @@ -0,0 +1,107 @@ +"""The agent never builds a SQL predicate out of a value. + +`portscanner.is_duplicate` did, with a hand-rolled `normalize()` that doubled +single quotes, and on a live host it failed every cycle: + + [!] portscanner failed (1 consecutive): SyntaxError: unterminated quoted + string at or near "'" + +so no port scan result was written at all. The collector had been silently +dead, which - being a table that produces rows rarely - looked exactly like a +host with no open ports. + +Which input broke it hardly matters. nmap returns product and version strings +containing whatever a service chose to advertise, and a banner is +attacker-controlled by definition. The answer to that is not a better +escaping function. + +A security product assembling SQL by concatenation is running the pattern it +exists to detect elsewhere. Every driver here takes parameters; the helpers +have taken a `params` tuple all along. +""" +from __future__ import annotations + +import ast +import pathlib +import re + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +AGENT = ROOT / "Sentora" + +#: An interpolated *identifier* is fine and unavoidable: the table name comes +#: from this project's own fixed list, and no driver lets you parameterise it. +#: What must never be interpolated is a value. +_PREDICATE = re.compile(r"\b(?:WHERE|AND|OR|SET)\b[^\n]*?=\s*'?\{", re.I) + + +def _interpolated_predicates(): + found = [] + for path in sorted(AGENT.rglob("*.py")): + try: + tree = ast.parse(path.read_text(encoding="utf-8", errors="ignore")) + except SyntaxError: + continue + for node in ast.walk(tree): + if not isinstance(node, ast.JoinedStr): + continue + # Reconstruct with `{...}` where the substitutions are, so a + # placeholder can be told from a literal. + text = "".join( + part.value if isinstance(part, ast.Constant) else "{}" + for part in node.values) + if _PREDICATE.search(text): + found.append((path, node.lineno, text.strip()[:100])) + return found + + +CASES = [pytest.param(p, ln, t, id=f"{p.name}:{ln}") + for p, ln, t in _interpolated_predicates()] + + +@pytest.mark.parametrize("path,line,text", CASES or + [pytest.param(None, 0, "", id="none-found")]) +def test_no_value_is_interpolated_into_a_predicate(path, line, text): + """Quoting is the driver's job. Doing it by hand is how a service banner + becomes a syntax error, and on a less lucky day something worse.""" + assert path is None, ( + f"{path.name}:{line} builds a SQL predicate by interpolation: {text!r}\n" + f"Pass values as parameters - every helper in modules/db.py takes a " + f"`params` tuple." + ) + + +def test_the_scan_can_still_see_sql(): + """A detector that has stopped matching anything passes silently. This + holds it to a pattern it must keep recognising.""" + # Double-quoted outer string so the SQL quotes inside need no escaping; + # the first version escaped them into a syntax error and tested nothing. + sample = 'f"SELECT * FROM {table} WHERE name = \'{value}\'"' + tree = ast.parse(sample) + node = next(n for n in ast.walk(tree) if isinstance(n, ast.JoinedStr)) + text = "".join(part.value if isinstance(part, ast.Constant) else "{}" + for part in node.values) + assert _PREDICATE.search(text), "the scan no longer recognises the bug" + + +def test_the_helpers_offer_parameters(): + """The fix has to be available, or the rule above is a rule nobody can + follow.""" + 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 == "fetch_where") + assert any(a.arg == "params" for a in fn.args.args) + + +def test_the_portscanner_uses_them(): + """The specific caller that was failing, pinned - it is the one with + attacker-controlled input, since nmap reports whatever a service + advertises.""" + 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.FunctionDef) and n.name == "is_duplicate") + body = ast.unparse(fn) + assert "%s" in body, "the duplicate check builds its own literals again" + assert "params" in body diff --git a/tests/test_auth_wiring.py b/tests/test_auth_wiring.py index 2ef0281..14dbdc1 100644 --- a/tests/test_auth_wiring.py +++ b/tests/test_auth_wiring.py @@ -330,3 +330,59 @@ def test_the_agent_channel_is_declared_rather_than_counted_as_session_only(tree) a false reassurance in the opposite direction from the one this file exists to prevent.""" assert "agent_link_socket" in _literal_set(tree, "_PUBLIC_HANDLERS") + + +# -------------------------------------------------------------------------- +# A permission has to exist to protect anything +# -------------------------------------------------------------------------- + +def _seeded_permissions() -> set: + """The permission names `db/init_userdb.sql` actually creates.""" + import re + + sql = (APP_PY.parent / "db" / "init_userdb.sql").read_text(encoding="utf-8") + block = sql[sql.index("INSERT INTO permissions"):] + block = block[:block.index(";")] + return set(re.findall(r"\('([a-z_]+)'", block)) + + +def _guarded_permissions() -> set: + """Permission names from actual decorators, read off the AST. + + Not from the source text. `_handler_qualname`'s docstring contains a + worked example of decorator ordering that uses `require_permission("x")`, + and a regex over the file reads that prose as two unreachable routes - + the seventh time in this suite that matching text has found a comment + warning about the very thing it was looking for. + """ + import ast + + names = set() + tree = ast.parse(APP_PY.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for decorator in node.decorator_list: + if (isinstance(decorator, ast.Call) + and getattr(decorator.func, "id", "") == "require_permission" + and decorator.args + and isinstance(decorator.args[0], ast.Constant)): + names.add(decorator.args[0].value) + return names + + +def test_every_guard_names_a_permission_that_exists(): + """A route guarded by a permission nothing grants is a route nobody can + reach - including the admin, and including whoever wrote it. + + It fails closed, which is the right direction, but it fails *silently*: + the route 404s or 403s for everyone and looks like a bug in the feature + rather than a typo in its guard. `read_audit` was invented for the + platform-events view and never existed anywhere; the view would have been + unreachable for every role in the product. + """ + invented = sorted(_guarded_permissions() - _seeded_permissions()) + assert not invented, ( + f"these guards name permissions that db/init_userdb.sql never " + f"creates, so the routes behind them are unreachable: {invented}" + ) diff --git a/tests/test_console_open_deadline.py b/tests/test_console_open_deadline.py new file mode 100644 index 0000000..33440cf --- /dev/null +++ b/tests/test_console_open_deadline.py @@ -0,0 +1,129 @@ +"""The deadline has to be longer than the work it is waiting for. + +The console reported + + the agent would not open a console: DESKTOP-EVS8H9J-3 did not open a + console within 15s + +against an agent that was opening one. On Windows the agent tries three +strategies in order - a pseudoconsole in its own process, a helper in the +interactive session, then a shell behind pipes - and only the last works when +the agent runs as a service. The first fails fast; the second times out after +twelve seconds. Measured on a live host: + + 17:25:03,382 no console in this process; trying the user's session + 17:25:18,548 falling back to a shell behind pipes + 17:24:48,854 shell alive: powershell -NoLogo -NoProfile + +Fifteen seconds to reach the strategy that works, against a fifteen second +deadline. The console was never broken; nobody was still listening when it +came up. + +Two things follow, and both are here: a deadline that covers the worst case, +and an agent that does not pay for the discovery twice. +""" +from __future__ import annotations + +import ast +import pathlib + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +LINK = ROOT / "core" / "agent_link.py" +CONSOLE = ROOT / "Sentora" / "modules" / "console.py" + + +def _const(path: pathlib.Path, name: str): + node = next(n for n in ast.parse(path.read_text(encoding="utf-8")).body + if isinstance(n, ast.Assign) + and any(getattr(t, "id", "") == name for t in n.targets)) + return ast.literal_eval(node.value) + + +def test_the_deadline_covers_the_agents_worst_case(): + """The chain costs about fifteen seconds before the working strategy is + even tried, so anything at or near fifteen is a deadline that expires + exactly when the answer is about to arrive.""" + timeout = _const(LINK, "STREAM_OPEN_TIMEOUT_S") + assert timeout >= 30, ( + f"opening a stream gives up after {timeout}s; the Windows console " + f"chain takes ~15s to reach the strategy that works" + ) + + +def test_it_is_longer_than_an_ordinary_request(): + """Opening a console is not an ordinary request and must not inherit an + ordinary request's patience.""" + assert (_const(LINK, "STREAM_OPEN_TIMEOUT_S") + > _const(LINK, "DEFAULT_TIMEOUT_S")) + + +def test_the_deadline_is_named_not_inlined(): + """It was a literal default in the signature, where the reasoning had + nowhere to live and nothing could assert on it.""" + fn = next(n for n in ast.walk(ast.parse(LINK.read_text(encoding="utf-8"))) + if isinstance(n, ast.AsyncFunctionDef) and n.name == "open_stream") + assert "STREAM_OPEN_TIMEOUT_S" in ast.unparse(fn.args) + + +def test_no_caller_reintroduces_a_shorter_one(): + """A default that every call site overrides is decoration.""" + import re + + app = (ROOT / "app.py").read_text(encoding="utf-8") + for call in re.findall(r"open_stream\([^)]*\)", app): + assert "timeout" not in call, f"a caller sets its own deadline: {call}" + + +# -------------------------------------------------------------------------- +# And the agent stops paying for the discovery +# -------------------------------------------------------------------------- + +def test_the_working_strategy_is_remembered(): + """Whether this process can host a pseudoconsole is a property of how it + was started; it cannot change while it runs. Rediscovering it on every + console open spends fifteen seconds proving something already known.""" + source = CONSOLE.read_text(encoding="utf-8") + assert "_WORKING_STRATEGY" in source + + fn = next(n for n in ast.walk(ast.parse(source)) + if isinstance(n, ast.FunctionDef) and n.name == "new_session") + body = ast.unparse(fn) + remembered = body.index("_WORKING_STRATEGY") + first_attempt = body.index("new_direct_session") + assert remembered < first_attempt, \ + "the remembered strategy is tried after the chain, which saves nothing" + + +def test_a_stale_memory_falls_back_rather_than_failing(): + """A helper that has since died, a session that logged out. Reporting that + as "no console" would turn a cache into an outage.""" + fn = next(n for n in ast.walk(ast.parse(CONSOLE.read_text(encoding="utf-8"))) + if isinstance(n, ast.FunctionDef) and n.name == "new_session") + body = ast.unparse(fn) + guard = body[:body.index("no console in this process")] + assert "_WORKING_STRATEGY = None" in guard, \ + "a strategy that stopped working is never forgotten" + + +def test_every_strategy_records_itself(): + """Remembering only one of them leaves the others paying full price.""" + body = ast.unparse(next( + n for n in ast.walk(ast.parse(CONSOLE.read_text(encoding="utf-8"))) + if isinstance(n, ast.FunctionDef) and n.name == "new_session")) + for strategy in ("new_direct_session", "HelperSession", "PipeSession"): + assert f"_WORKING_STRATEGY = {strategy}" in body, \ + f"{strategy} succeeding is not remembered" + + +def test_the_memory_does_not_survive_a_restart(): + """It records how *this* process was started. A restarted agent may have + been started differently - from a service, from a terminal - and a cache + that outlived that would be confidently wrong.""" + source = CONSOLE.read_text(encoding="utf-8") + assert "_WORKING_STRATEGY = None" in source + for persisted in ("json.dump", "open(", "pickle"): + block = source[source.index("_WORKING_STRATEGY = None"):] + block = block[:block.index("def new_session")] + assert persisted not in block, "the strategy is written to disk" diff --git a/tests/test_encryption_actually_applied.py b/tests/test_encryption_actually_applied.py new file mode 100644 index 0000000..35550e3 --- /dev/null +++ b/tests/test_encryption_actually_applied.py @@ -0,0 +1,126 @@ +"""A field declared encrypted has to be written through the encrypting insert. + +`enc_db.ENCRYPT_FIELDS_MAP` said fim_data's `path` and `hash_sha256` were +encrypted. `core.telemetry_crypto.ENCRYPTED_FIELDS` said the same, so the +server was ready to decrypt them. And `fim.py` - which produces almost every +fim_data row there is - called the plain `insert_record`. + +So the file paths of every monitored host were stored in the clear, in the +agent's database and on the server, while both ends of the design said +otherwise. 1,081 rows on one host, none of them encrypted. + +Nothing could see it. The server's `decrypt_value` hands anything without the +`enc::` prefix straight back, by design - that is what makes a mixed table +readable - so plaintext and ciphertext render *identically* in the console. +The table looked perfect. The only way to notice was to ask what the stored +bytes actually begin with. + +This file removes the need to notice. +""" +from __future__ import annotations + +import ast +import pathlib + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +AGENT = ROOT / "Sentora" +ENC_DB = AGENT / "modules" / "enc_db.py" +SERVER_CRYPTO = ROOT / "core" / "telemetry_crypto.py" + + +def _map(path: pathlib.Path, name: str) -> dict: + node = next(n for n in ast.parse(path.read_text(encoding="utf-8")).body + if isinstance(n, (ast.Assign, ast.AnnAssign)) + and name in ast.unparse(n.target if isinstance(n, ast.AnnAssign) + else n.targets[0])) + return ast.literal_eval(node.value) + + +AGENT_MAP = _map(ENC_DB, "ENCRYPT_FIELDS_MAP") + + +def _inserts() -> list[tuple[pathlib.Path, int, str, str]]: + """(file, line, table, which insert) for every insert in the agent.""" + found = [] + for path in AGENT.rglob("*.py"): + try: + tree = ast.parse(path.read_text(encoding="utf-8", errors="ignore")) + except SyntaxError: + continue + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + fn = getattr(node.func, "id", "") or getattr(node.func, "attr", "") + if fn not in ("insert_record", "insert_record_enc"): + continue + if node.args and isinstance(node.args[0], ast.Constant): + found.append((path, node.lineno, node.args[0].value, fn)) + return found + + +ENCRYPTED_WRITES = [ + pytest.param(path, line, table, fn, + id=f"{path.name}:{line}:{table}") + for path, line, table, fn in _inserts() + if table in AGENT_MAP +] + + +@pytest.mark.parametrize("path,line,table,fn", ENCRYPTED_WRITES) +def test_an_encrypted_table_is_written_through_the_encrypting_insert( + path, line, table, fn): + """The plain insert stores exactly what it is given. For a table with + declared encrypted fields that means storing a secret in the clear, and + nothing downstream can tell.""" + assert fn == "insert_record_enc", ( + f"{path.name}:{line} writes '{table}' with {fn}(), and " + f"{sorted(AGENT_MAP[table])} are declared encrypted - so those values " + f"are stored in the clear and render identically to encrypted ones" + ) + + +def test_there_is_at_least_one_such_write(): + """A parametrised test over an empty list passes silently. If the scan + stops finding inserts - a rename, a helper, a different import style - + this is the only thing that says so.""" + assert ENCRYPTED_WRITES, "the insert scan found nothing to check" + + +# -------------------------------------------------------------------------- +# The two maps have to agree +# -------------------------------------------------------------------------- + +SERVER_MAP = _map(SERVER_CRYPTO, "ENCRYPTED_FIELDS") + + +@pytest.mark.parametrize("table", sorted(AGENT_MAP)) +def test_everything_the_agent_encrypts_the_server_decrypts(table): + """The agent encrypting a field the server does not decrypt is the worst + of the four combinations: the console shows `enc::gAAAA...` and the data + is there and unreadable, with nothing saying why.""" + assert table in SERVER_MAP, f"the server never decrypts anything in {table}" + missing = sorted(set(AGENT_MAP[table]) - set(SERVER_MAP[table])) + assert not missing, ( + f"the agent encrypts {table}.{missing} and the server does not " + f"decrypt it, so the console will show ciphertext" + ) + + +def test_the_server_side_extras_are_deliberate(): + """The reverse - the server decrypting a field the agent leaves alone - is + harmless, because `decrypt_value` returns anything without the `enc::` + prefix untouched. It is recorded rather than asserted away, so the drift + stays visible instead of accumulating. + """ + drift = {} + for table, fields in SERVER_MAP.items(): + extra = sorted(set(fields) - set(AGENT_MAP.get(table, ()))) + if extra: + drift[table] = extra + # Harmless today. If this ever grows past the known pair, somebody has + # changed one map and not the other. + assert set(drift) <= {"siem_events", "events_alert"}, ( + f"the two encryption maps have drifted further apart: {drift}" + ) diff --git a/tests/test_key_mismatch_visibility.py b/tests/test_key_mismatch_visibility.py new file mode 100644 index 0000000..864813a --- /dev/null +++ b/tests/test_key_mismatch_visibility.py @@ -0,0 +1,136 @@ +"""How much of the stored history can this server actually read? + +The answer was invisible. A key mismatch printed one line per field name - +guarded by a `set`, so once each for the life of the process - and then +returned `` per value. Which meant a table +where 86% of the messages could not be read produced exactly the same output +as a table with one bad row. + +Measured on a live host: 173 of 200 `siem_events` messages, 197 of 200 +`events_alert` messages, 172 of 200 `critical_files` paths. The Fernet key +comes from the server's bootstrap and there is no rotation path, so everything +encrypted under a previous key is gone - that is not fixable, and it is +exactly why the *quantity* has to be visible rather than the fact. + +The second half of this file is about not making it worse. Offering rows again +after a server reset was importing rows the agent itself can no longer read. +""" +from __future__ import annotations + +import ast +import pathlib +import sys + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +APP = ROOT / "app.py" +AGENT_DB = ROOT / "Sentora" / "modules" / "db.py" +ENC_DB = ROOT / "Sentora" / "modules" / "enc_db.py" + + +# -------------------------------------------------------------------------- +# The scale, not just the fact +# -------------------------------------------------------------------------- + +def _health(): + """`decryption_health` and its counters, compiled on their own.""" + tree = ast.parse(APP.read_text(encoding="utf-8")) + fn = next(n for n in tree.body + if isinstance(n, ast.FunctionDef) and n.name == "decryption_health") + counters = next(n for n in tree.body + if isinstance(n, ast.Assign) + and any(getattr(t, "id", "") == "_decrypt_counters" + for t in n.targets)) + namespace: dict = {} + exec(compile(ast.Module(body=[counters, fn], type_ignores=[]), "app.py", "exec"), + namespace) + return namespace + + +def test_nothing_read_is_not_a_problem(): + """Zero of zero has to say so rather than divide.""" + ns = _health() + assert ns["decryption_health"]()["unreadable_percent"] == 0.0 + + +def test_the_proportion_is_reported(): + """"3 failures" is meaningless without the denominator; the number that + matters is what fraction of the history is readable.""" + ns = _health() + ns["_decrypt_counters"]["decrypted"] = 27 + ns["_decrypt_counters"]["undecryptable"] = 173 + health = ns["decryption_health"]() + assert health["unreadable_percent"] == 86.5 + assert health["undecryptable"] == 173 + + +def test_a_healthy_server_says_so_differently(): + """An operator should be able to tell "nothing unreadable" from "not + measured", which is the same distinction this whole codebase keeps + getting wrong.""" + ns = _health() + ns["_decrypt_counters"]["decrypted"] = 100 + clean = ns["decryption_health"]()["detail"] + ns["_decrypt_counters"]["undecryptable"] = 1 + broken = ns["decryption_health"]()["detail"] + assert clean != broken + assert "no rotation path" in broken or "cannot be recovered" in broken + + +def test_successes_are_counted_too(): + """Counting only failures gives a denominator of failures, and every + deployment then reads as 100% unreadable.""" + source = APP.read_text(encoding="utf-8") + assert '_decrypt_counters["decrypted"] += 1' in source + + +def test_the_console_shows_it(): + """Counted and never surfaced is where this started.""" + page = (ROOT / "frontend" / "src" / "pages" / "AuditLogs.tsx").read_text( + encoding="utf-8") + assert "unreadable_percent" in page + assert "undecryptable" in page + + +# -------------------------------------------------------------------------- +# Recovery must not import what nobody can read +# -------------------------------------------------------------------------- + +def test_the_agent_can_tell_whether_it_still_reads_a_row(): + sys.path.insert(0, str(ROOT / "Sentora")) + from cryptography.fernet import Fernet + from modules import enc_db + + enc_db.set_encrypt_fields_map({"siem_events": ["message"]}, merge=True) + + enc_db.set_fernet_key(Fernet.generate_key()) + old_row = {"message": enc_db._enc_value("under the previous key")} + enc_db.set_fernet_key(Fernet.generate_key()) + new_row = {"message": enc_db._enc_value("under the current key")} + + assert enc_db.is_readable("siem_events", old_row) is False + assert enc_db.is_readable("siem_events", new_row) is True + assert enc_db.is_readable("siem_events", {"message": "plaintext"}) is True + assert enc_db.is_readable("disk_usage", {"percent": 12}) is True + + +def test_the_re_offer_skips_what_it_cannot_read(): + """Offering them again imports rows that render as + `` for ever, and spends a bounded + recovery budget on data nobody can read. The first re-offer on a live host + moved about two thousand of them onto the server.""" + body = ast.unparse(next( + n for n in ast.walk(ast.parse(AGENT_DB.read_text(encoding="utf-8"))) + if isinstance(n, ast.FunctionDef) and n.name == "reoffer_recent")) + assert "is_readable" in body + + +def test_an_unknown_shape_is_offered_rather_than_dropped(): + """No key configured yet, or a row this does not understand. The cost of + offering it is an unreadable row; the cost of dropping it is a lost one, + and this whole module exists to stop the second.""" + fn = next(n for n in ast.walk(ast.parse(AGENT_DB.read_text(encoding="utf-8"))) + if isinstance(n, ast.FunctionDef) and n.name == "reoffer_recent") + handler = next(h for h in ast.walk(fn) if isinstance(h, ast.ExceptHandler)) + assert "keep.append" in ast.unparse(handler) diff --git a/tests/test_self_defence.py b/tests/test_self_defence.py new file mode 100644 index 0000000..a58ec4c --- /dev/null +++ b/tests/test_self_defence.py @@ -0,0 +1,241 @@ +"""The platform watches every host in the fleet. Nobody watched the platform. + +Every one of these was already detected and none of it was visible. A +brute-force against the console produced `login_logs` rows and a lockout, and +the alert view showed nothing. A revoked agent key retrying produced + + [agent-link] refused a connection from 203.0.113.9 + +in a container log, which nobody reads until they already suspect something - +the wrong order for the one signal that says a key is being used after it was +taken away. + +So these become events of the same shape as everything else the console +surfaces, and this file pins the three properties that stop the cure being +worse than the disease. +""" +from __future__ import annotations + +import ast +import pathlib + +import pytest + +ROOT = pathlib.Path(__file__).resolve().parent.parent +APP = ROOT / "app.py" + +from core import self_defence # noqa: E402 + + +class _FakeCursor: + """Enough of a DB-API cursor to exercise the fold. + + `rowcount` is set by each statement, the way a real driver does it - the + first version seeded it in the constructor, and the DDL that runs first + then cleared it before the UPDATE could be read. The fake was wrong and + the code was right, which is the more expensive way round. + """ + + def __init__(self, updates_match: bool = False): + self.statements: list[tuple[str, tuple]] = [] + self.updates_match = updates_match + self.rowcount = 0 + + def execute(self, sql, params=()): + self.statements.append((" ".join(sql.split()), params)) + verb = sql.strip().split()[0].upper() + if verb == "UPDATE": + self.rowcount = 1 if self.updates_match else 0 + else: + self.rowcount = 0 + + +# -------------------------------------------------------------------------- +# A severity is a claim +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("kind", sorted(self_defence.KINDS)) +def test_every_kind_has_a_severity_and_an_explanation(kind): + """An operator seeing FLEET_KEY_ON_CHANNEL for the first time should not + have to go looking for what it implies.""" + severity, explanation = self_defence.describe(kind) + assert severity in self_defence.SEVERITY_RANK + assert len(explanation) > 40, f"{kind} explains nothing" + + +def test_an_unknown_kind_is_filed_rather_than_dropped(): + """A caller inventing a name is a bug, but losing the event is worse than + filing it under the wrong heading.""" + severity, _ = self_defence.describe("SOMETHING_NEW") + assert severity == "MEDIUM" + + +def test_the_fleet_key_is_the_worst_thing_here(): + """It opens a channel that carries /self_destruct against every endpoint + at once. Nothing else on this list is in that category.""" + ranks = {k: self_defence.SEVERITY_RANK[self_defence.describe(k)[0]] + for k in self_defence.KINDS} + assert ranks["FLEET_KEY_ON_CHANNEL"] == min(ranks.values()) + + +# -------------------------------------------------------------------------- +# One attack is one row +# -------------------------------------------------------------------------- + +def test_a_repeat_folds_into_the_previous_row(): + """A brute-force writing a row per attempt makes the view that is supposed + to reveal the attack into the thing that buries it.""" + cursor = _FakeCursor(updates_match=True) + self_defence.record(cursor, "LOGIN_LOCKOUT", subject="admin", + source_ip="203.0.113.9", detail="5 in 60s") + kinds = [sql.split()[0] for sql, _ in cursor.statements] + assert "INSERT" not in kinds, "a repeat created a second row" + assert "UPDATE" in kinds + + +def test_a_first_sighting_creates_a_row(): + cursor = _FakeCursor(updates_match=False) + self_defence.record(cursor, "LOGIN_LOCKOUT", subject="admin", + source_ip="203.0.113.9") + assert any(sql.startswith("INSERT INTO platform_events") + for sql, _ in cursor.statements) + + +def test_the_fold_is_bounded_in_time(): + """Without a window, an attack from a year ago and one happening now are + the same row and the timestamps say nothing.""" + cursor = _FakeCursor(updates_match=True) + self_defence.record(cursor, "LOGIN_LOCKOUT", subject="a", source_ip="b") + update = next(sql for sql, _ in cursor.statements if sql.startswith("UPDATE")) + assert "INTERVAL" in update and "last_seen >" in update + + +def test_the_summary_counts_attempts_not_rows(): + """"3 events" where one of them is 400 failed logins is the wrong number + to put in front of somebody.""" + summary = self_defence.summarise([ + {"severity": "HIGH", "occurrences": 400}, + {"severity": "MEDIUM", "occurrences": 2}, + ]) + assert summary["attempts"] == 402 + assert summary["worst"] == "HIGH" + + +# -------------------------------------------------------------------------- +# Never store a credential +# -------------------------------------------------------------------------- + +def test_nothing_here_records_a_secret(): + """These records describe attempts on secrets, and the obvious fields to + include - the password tried, the key presented - are exactly the ones + that turn an audit trail into a second breach.""" + source = (ROOT / "core" / "self_defence.py").read_text(encoding="utf-8") + columns = source[source.index("CREATE TABLE"):source.index("ENGINE=InnoDB")] + for forbidden in ("password", "agent_key", "token", "secret", "hash"): + assert forbidden not in columns.lower(), \ + f"the platform event table has a `{forbidden}` column" + + +# -------------------------------------------------------------------------- +# Recording must never break what it records +# -------------------------------------------------------------------------- + +def test_the_writer_swallows_its_own_failure(): + """Every caller is in the middle of refusing something. A server that + will not reject a bad login because it could not journal the rejection has + turned a monitoring gap into an outage.""" + fn = next(n for n in ast.walk(ast.parse(APP.read_text(encoding="utf-8"))) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + and n.name == "_platform_event") + body = ast.unparse(fn) + assert "except Exception" in body + assert "raise" not in body + + +def test_it_does_not_block_the_event_loop(): + """A synchronous MySQL write inside a request handler stalls every other + request on the worker - and this one runs on the login path, which is + exactly where an attacker controls the rate.""" + fn = next(n for n in ast.walk(ast.parse(APP.read_text(encoding="utf-8"))) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + and n.name == "_platform_event") + assert "to_thread" in ast.unparse(fn) + + +# -------------------------------------------------------------------------- +# Detected and recorded is only half of it +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("kind", ["LOGIN_LOCKOUT", "AGENT_KEY_REJECTED", + "FLEET_KEY_ON_CHANNEL"]) +def test_the_detections_are_actually_wired(kind): + """A vocabulary nothing raises is a vocabulary. These three are already + detected in app.py and used to stop at a print.""" + assert f'"{kind}"' in APP.read_text(encoding="utf-8"), \ + f"{kind} is defined and never raised" + + +def test_there_is_somewhere_to_read_them(): + """Recording without surfacing repeats the original mistake in a new + table.""" + source = APP.read_text(encoding="utf-8") + assert "/api/platform/events" in source + + +# -------------------------------------------------------------------------- +# Surfaced, not just stored +# -------------------------------------------------------------------------- + +FRONTEND = ROOT / "frontend" / "src" + + +def test_the_console_reads_them(): + """The failure this whole module exists to fix was "detected, recorded + somewhere nobody looks". Adding a table and stopping there would be the + same mistake in a new place.""" + api = (FRONTEND / "services" / "api.ts").read_text(encoding="utf-8") + assert "/api/platform/events" in api, "nothing in the console fetches them" + + page = (FRONTEND / "pages" / "AuditLogs.tsx").read_text(encoding="utf-8") + assert "getPlatformEvents" in page + assert "occurrences" in page, \ + "the fold is invisible, so one attack reads as one attempt" + + +def test_a_role_without_the_permission_is_not_shown_an_error(): + """`manage_users` reaches the audit log and not this. That is not a + failure for them - the section does not apply - and rendering it as one + trains people to ignore red.""" + page = (FRONTEND / "pages" / "AuditLogs.tsx").read_text(encoding="utf-8") + fetch = page[page.index("getPlatformEvents"):] + fetch = fetch[:fetch.index("}, []);")] + assert ".catch(" in fetch, "a 403 surfaces as a broken page" + + +def test_severity_uses_the_semantic_colours(): + """A severity is a claim. Drawing CRITICAL in whatever colour the third + row happened to get is the chart lying about its own data.""" + page = (FRONTEND / "pages" / "AuditLogs.tsx").read_text(encoding="utf-8") + assert "SEVERITY_TONE" in page + assert "CRITICAL: 'critical'" in page + + +def test_an_empty_result_is_shown_rather_than_hidden(): + """The first version hid the panel when there was nothing in it, and the + argument for that read well: a permanently empty panel is furniture people + learn to skip. + + It was wrong, and wrong in exactly the way the rest of this work exists to + prevent. An operator who never sees the panel cannot tell "nothing has + attacked us" from "this platform does not watch itself" - the same + ambiguity as an empty telemetry table, on the one view whose whole job is + to remove it. On a security view a quiet twenty-four hours is a finding, + so it is stated. + + `null` stays hidden, and that is a different case: the role cannot read + this, so it is not an empty result, it is not their question. + """ + page = (FRONTEND / "pages" / "AuditLogs.tsx").read_text(encoding="utf-8") + assert "platform !== null" in page, "the panel is hidden on an empty result again" + assert "platform?.length &&" not in page, "the panel still renders only when it has rows" + assert "not that nothing is watching" in page, "an empty panel does not say what empty means" diff --git a/tests/test_tls_documentation.py b/tests/test_tls_documentation.py index b601781..9ed00ea 100644 --- a/tests/test_tls_documentation.py +++ b/tests/test_tls_documentation.py @@ -125,16 +125,44 @@ def test_the_ports_table_covers_the_tls_listener(): def test_closing_the_plaintext_path_documents_both_halves(): """`INGEST_TLS_REQUIRED=1` closes the listener inside the container while Docker goes on publishing the port. A connection is then accepted by the - proxy and the batch written into a socket nobody reads - which is why the - port mapping has to go too, and why the guide has to say so.""" + proxy and the batch written into a socket nobody reads - so the port has + to be unpublished too, and the guide has to say both. + + Asserted as the property rather than the mechanism. The first version + checked for the literal `"5001:5001"`, because the instruction then was to + delete that line from a tracked compose file - which is a bad thing to ask + an operator to do, and the moment it became an environment variable this + test failed against a guide that had just got better. + """ section = _tls_section() - assert "INGEST_TLS_REQUIRED=1" in section - assert '"5001:5001"' in section, ( + assert "INGEST_TLS_REQUIRED=1" in section, \ + "the guide never says how to stop the listener" + assert "INGEST_PLAINTEXT_BIND" in section, ( "the guide tells the operator to require TLS without telling them to " "stop publishing the plaintext port" ) +def test_unpublishing_the_port_is_something_compose_can_actually_do(): + """A documented setting that no file reads is what this whole suite exists + for.""" + import yaml + + compose = yaml.safe_load(COMPOSE.read_text(encoding="utf-8")) + published = compose["services"]["ingest"]["ports"] + plaintext = [p for p in published if str(p).endswith("5001:5001")] + assert plaintext, "the plaintext port is no longer published at all" + assert "INGEST_PLAINTEXT_BIND" in str(plaintext[0]), ( + "the guide names INGEST_PLAINTEXT_BIND and compose ignores it, so " + "following the migration leaves the port open" + ) + + # And the default still publishes, because a fresh install that is not on + # TLS yet needs 5001 to work without editing a tracked file. + assert ":-0.0.0.0}" in str(plaintext[0]), \ + "the default now hides the plaintext port from every new deployment" + + def test_the_agent_side_of_that_trap_is_implemented(): """The guide claims the agent will not be fooled by an accepted-and- discarded batch. That claim depends on it remembering that this server has