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