diff --git a/FILESYSTEM_COVERAGE.txt b/FILESYSTEM_COVERAGE.txt new file mode 100644 index 0000000..defa969 --- /dev/null +++ b/FILESYSTEM_COVERAGE.txt @@ -0,0 +1,36 @@ +Dateisystem-Erkennungsabdeckung – Quellenabgleich 05.10.2026 + +Keine vollständige Erfassung aller weltweit bekannten oder privaten Herstellerregeln. +Alle Treffer sind Fundkandidaten, kein automatischer Kompromittierungsbeweis. + +Neu: elf eigenständige Inhaltsregeln mit Quellenzuordnung in +src/netscaler_triage/data/filesystem_rules.json. Ergänzte Konfigurationspfade: +/nsconfig/httpd.conf, /flash/nsconfig/httpd.conf, /nsconfig/crontab, +/nsconfig/rc.netscaler, /etc/master.passwd, /etc/inetd.conf sowie zusätzliche +Cron-Verzeichnisse und Konfigurationsdateien. Zusätzlich sh.log und nsumond +bei der optionalen Logauswahl. Reguläre Dateien werden unabhängig vom Namen gelesen. + +Quellen und abgeleitete Abdeckung: +Google/Mandiant: PHP-Handler und Engine-Aktivierung, Web-Routing, Headerkanäle, +Cron-Bereinigung, SUID-Shell, weitere getarnte Endungen und Konfigurationspfade. +https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances +LevelBlue/SpiderLabs: CSS-Aliase, sec_monitor, Reverse-Shell-/Staging-Referenzen. +https://www.levelblue.com/blogs/spiderlabs-blog/citrix-netscaler-cve-2026-88771-observed-exploitation-artifacts-and-hunt-indicators +Arctic Wolf: nsmon-Cron-Muster; vorhandene Pfad-/Hash-/Netzwerkindikatoren abgeglichen. +https://raw.githubusercontent.com/rtkwlf/wolf-tools/main/pack_alerts/202609-citrix-netscaler-active-exploitation-cve-2026-88771/README.md +Rapid7: /vpn/c-Staging und .ctxs.receiver-Hash bereits enthalten. +https://www.rapid7.com/blog/post/etr-zero-day-exploitation-of-citrix-netscaler-adc-and-gateway-cve-2026-88771-and-cve-2026-88772/ + +Öffentliche Hash-, Pfad- und Netzwerkindikatoren werden aus dem versionierten +Indikatorpaket geladen. Vertrauliche Ergänzungen bleiben außerhalb des Repositorys. + +Weiterhin nicht vollständig automatisiert: Auflösung aller Apache-Includes und +Aliasziele; semantischer Herstellervergleich; Binär-/RAM-Analyse; sämtliche +verschachtelten Container; Nachweis von Loglöschung; Ende-zu-Ende-Korrelation mit +externen Firewall- und Identitätsdaten. Generische SSH-/Start-Konfigurationen sind +Prüfkontext, nicht automatisch eine veröffentlichte kampagnenspezifische IOC. +Die neuen Regeln ersetzen keine vollständige Datenflussanalyse einer Webshell. +404 und fehlende temporäre Dateien sind keine sicheren Misserfolgsbeweise. + +149 automatisierte Tests bestanden. Dies bestätigt die dokumentierte Funktion, +keine universelle Erkennungsrate oder vollständige Unversehrtheit. diff --git a/README.md b/README.md index dfe9849..1caca59 100644 --- a/README.md +++ b/README.md @@ -1,40 +1,265 @@ # NetScaler Triage -Offline forensic triage for Citrix NetScaler ADC and Gateway logs. Point it at a -folder or archive to find suspicious requests, public indicators, encoded -payloads, update records and related events across logs. +Offline-Analyse von Citrix-NetScaler-Logs und Dateisystem-Sicherungen. Das Werkzeug +sucht nach Angriffsversuchen, bekannten IOCs, Webshell-Mustern und Persistenzhinweisen. +Es verknüpft Logereignisse und speichert Fundstellen mit Originalkontext. -Original inputs are not modified. Payloads are decoded as data, never executed -or downloaded. A finding is an investigation lead, **not proof of compromise**. +Originaldaten bleiben unverändert. Verdächtige Inhalte werden niemals ausgeführt +oder heruntergeladen. Ein Treffer ist ein Prüfhinweis, kein automatischer Beweis +einer Kompromittierung. Auch ein Scan ohne Treffer beweist kein sauberes System. -## Quick start +## 1. Installieren -Requires **Python 3.14 or newer**. +Voraussetzung: **Python 3.14 oder neuer**. Das Repository herunterladen und ein +Terminal in dessen Hauptordner öffnen. Alle folgenden lokalen Befehle werden dort +ausgeführt. Installation kann Internetzugriff benötigen; die Analyse läuft offline. + +macOS/Linux: ```sh -git clone https://tea.karlcom.de/karl/NetScaler_forensik.git -cd NetScaler_forensik python3 -m venv .venv source .venv/bin/activate python -m pip install . -netscaler-triage ./logs.7z +netscaler-triage --help ``` -On Windows PowerShell, create and activate the environment with: +Windows PowerShell: ```powershell py -3.14 -m venv .venv .venv\Scripts\Activate.ps1 python -m pip install . -netscaler-triage .\logs.7z +netscaler-triage --help ``` -Installation may require internet access. Analysis runs offline. +Die folgenden Eingabenamen sind Beispiele und müssen durch die eigenen Dateien +ersetzt werden. Ausgabeordner müssen neu sein und außerhalb der untersuchten +Verzeichnisse liegen. Originale und Ergebnisse vertraulich aufbewahren. -### Update an existing installation +## 2. Logs prüfen -Activate the same virtual environment used to run the tool, then run from the -repository directory: +Einen Logordner oder ein Logarchiv untersuchen: + +```sh +netscaler-triage ./logs --output ./log-ergebnis +netscaler-triage ./logs.7z --output ./archiv-ergebnis +``` + +Mehrere Exporte gemeinsam analysieren: + +```sh +netscaler-triage ./appliance-a ./appliance-b export.zip --output ./gemeinsames-ergebnis +``` + +Unterstützt werden ZIP einschließlich ZipCrypto/AES, TAR, TAR.GZ/TGZ, GZIP und 7z; +verschachtelte Archive innerhalb der Verarbeitungsgrenzen. Bei verschlüsselten +Archiven fragt `--password` das Passwort verdeckt ab: + +```sh +netscaler-triage ./logs.7z --password --output ./verschluesselt-ergebnis +``` + +Die Zuordnung zu Appliances wird aus der Ordnerstruktur abgeleitet. Bei ungewöhnlich +aufgebauten Exporten das Vorbereitungsmanifest kontrollieren. + +### Ergebnisse der Loganalyse lesen + +Zuerst `SUMMARY.txt` öffnen: Abdeckung, Trefferzahlen und Untersuchungslücken. +Danach `FINDINGS.csv`: Treffer mit Erklärung und Verweis auf die Originalquelle. +`IOCS.csv` enthält Suchkandidaten für weitere Prüfungen, keine automatische Sperrliste. + +Unter `details/` liegen vollständige Ereignisse, Dekodierungen, Korrelationen, +Prüfsummen und der SQLite-Index. `details/findings.jsonl` enthält auch Kontext und +Verarbeitungsfehler; zum Durchsehen der Treffer zunächst `FINDINGS.csv` verwenden. +Hash-Treffer stehen in `details/file_hash_hits.json`. Vorbereitete Eingabekopien +liegen gegebenenfalls unter `inputs/`. + +CSV-Zellen werden gegen Tabellenformeln abgesichert; exakte Originalwerte bleiben +in JSON erhalten. Eigene Tests und Referenzindikatoren bleiben gekennzeichnet. +Bei abgebrochenen Läufen kann nur `summary.json` im Ausgabeordner vorliegen. + +## 3. Dateisystem-Sicherung prüfen + +Die Erweiterung `filesystem` liest eine TAR.GZ-Sicherung direkt, ohne sie vollständig +zu entpacken. Das spart zusätzlichen Speicherplatz: + +```sh +netscaler-triage filesystem ./snapshot.tar.gz --output ./dateisystem-ergebnis +``` + +Bereits entpackte Dateien sind ebenfalls möglich. Als Eingabe den Wurzelordner mit +`etc/`, `var/` und `netscaler/` angeben, nicht den übergeordneten Sammelordner: + +```sh +netscaler-triage filesystem ./export-root --output ./ordner-ergebnis +``` + +Der Scanner prüft reguläre Dateien einschließlich versteckter Dateien auf: + +- Bekannte Pfade, Dateinamen und SHA-256-Werte aus den IOC-Listen. +- Webshell-Inhalte, PHP in ungewöhnlichen Dateitypen und Ausführungsfunktionen. +- Verdächtige HTTP-Zuordnungen, Cron-/Startkonfigurationen und Reverse-Shell-Muster. +- Payloads, Hinweise auf Konfigurationsbereitstellung sowie bekannte IPs, Domains und URLs. +- Dateirechte, Besitzer, Zeitstempel und Linkziele; Symlinks werden nicht verfolgt. + +Dateiinhalte und gzip-Inhalte werden nach Mustern durchsucht, auch enthaltene Logs. +Für zusätzliche strukturierte Auswertung und Korrelation ausgewählter System-, +Authentifizierungs-, HTTP- und Call-Home-Logs `--analyze-logs` ergänzen: + +```sh +netscaler-triage filesystem ./snapshot.tar.gz --output ./dateisystem-mit-logs --analyze-logs --timezone Europe/Berlin +``` + +Die Zeitzone muss zum Gerät passen. Ergebnisse der zusätzlichen Loganalyse stehen +unter `loganalyse/`. Dafür werden höchstens 16 MiB temporäre Original-Logkopien +angelegt und anschließend entfernt. Fehlende Jahresangaben werden nicht geraten. +Diese Logauswertung kann unvollständig sein, obwohl der Dateisystemscan fertig ist. + +### Eigene IOCs und früheren Stand ergänzen + +Zusätzliche IOCs als CSV laden; Format siehe Abschnitt 5: + +```sh +netscaler-triage filesystem ./snapshot.tar.gz --output ./ioc-ergebnis --ioc-file ./private/IOC.csv +``` + +Mit einem früheren Dateiverzeichnis vergleichen: + +```sh +netscaler-triage filesystem ./snapshot.tar.gz --output ./vergleich-ergebnis --baseline ./previous-files +``` + +Ein früherer Stand ist keine unabhängig bestätigte, saubere Herstellerreferenz. + +### Mehrere Sicherungen parallel prüfen + +Im Repository-Hauptordner mit aktivierter Python-Umgebung ausführen. Dieses Beispiel +verwendet zwei Prozesse und funktioniert auch ohne Shell-Platzhalter: + +```sh +python scripts/analyze_filesystem_exports.py ./export-a.tar.gz ./export-b.tar.gz --output ./batch-ergebnis --workers 2 +``` + +Es sind ein bis vier Prozesse möglich. Optional `--ioc-file ./private/IOC.csv` +ergänzen. Ergebnisse liegen pro Archiv in einem Unterordner; `batch_summary.json` +fasst die Exitcodes zusammen. Die Batch-Prüfung sucht auch in Log-Inhalten, startet +aber keine zusätzliche strukturierte Logkorrelation über `--analyze-logs`. + +### Ergebnisse der Dateisystemprüfung lesen + +Zuerst `summary.json` prüfen: Vollständigkeit, Fehler, Grenzen und Integritätsprüfung. +Dann die folgenden Dateien auswerten: + +- `artifact_checks.json`: bekannte Pfade und Dateinamen, vorhanden oder nicht gefunden. +- `findings.jsonl`: Inhalts- und IOC-Treffer mit Kontext, Position und gegebenenfalls Hash. +- `inventory.jsonl`: Dateiinventar mit SHA-256 und Metadaten. +- `selected_content.jsonl`: begrenzte Textauszüge ausgewählter kritischer Dateien. +- `baseline_changes.jsonl`: Änderungen gegenüber einer übergebenen Vergleichsbasis. + +Ein IOC in einem Log oder Installationspaket beweist kein aktives Implantat. +PHP-Bibliotheken und reguläre Herstellerdateien können ebenfalls Treffer erzeugen. +Fundstellen vor einer Bewertung im Zusammenhang prüfen. + +## 4. Dateisystem auf dem NetScaler sichern + +**Nur diesen Exportbefehl auf der Appliance ausführen**, in der BSD-Shell als root. +Er erzeugt ein neues Archiv unter `/var/tmp`. Genügend freien Speicher vorsehen. +Quelldateien werden nicht gelöscht; ein bereits vorhandener Zielname wird abgelehnt. + +```sh +sh -c 'f="/var/tmp/netscaler-filesystem-$(date -u +%Y%m%dT%H%M%SZ).tar.gz"; test ! -e "$f" || exit 1; umask 077; tar -czf "$f" --exclude=./dev --exclude=./proc --exclude=./kern --exclude=".$f" --exclude="./var/tmp/netscaler-filesystem*.tar.gz" -C / . && sha256 "$f"' +``` + +Das Archiv anschließend über den vorhandenen sicheren Übertragungsweg auf den +Analyse-Rechner kopieren und dort wie in Abschnitt 3 prüfen. Fehlerausgabe und +Exitstatus des Exports aufbewahren. Eine fehlgeschlagene Sicherung ist nicht vollständig. + +Das Archiv enthält möglicherweise Zugangsdaten und private Schlüssel. Die Sicherung +eines laufenden Systems ist kein konsistentes Datenträgerabbild. Der Exportbefehl +wurde im dokumentierten Testumfang nicht auf einer echten Appliance ausgeführt. + +## 5. Regeln und IOCs ergänzen + +Vertrauliche Regelpakete unter `private/` speichern; dieser Ordner ist von Git und +Release-Archiven ausgeschlossen. Keine Kundendaten oder Analyseergebnisse ins +öffentliche Repository übernehmen. Die Beispiele hier sind synthetisch. + +CSV-Beispiel für zusätzliche IOCs: + +```csv +type,value,status,source,description +ipv4,192.0.2.42,reference_only,local-review,Matching events require review +path,/example/suspicious-file,reference_only,local-review,Check file context +``` + +Originalwerte verwenden, keine entschärften Schreibweisen wie `[.]`. +Die Dateisystemprüfung lädt eine zusätzliche CSV über `--ioc-file`. +Die Loganalyse unterstützt darüber hinaus Sigma YAML, JSON, CSV und TXT über +`--config`, auch rekursiv aus Verzeichnissen: + +```sh +netscaler-triage rules --config ./private/my-rules +netscaler-triage ./logs --config ./private/my-rules --output ./eigene-regeln-ergebnis +``` + +`--config` kann wiederholt werden. TXT erlaubt einen Indikator pro Zeile. +Die unterstützte Sigma-Teilmenge steht in [RULES.txt](RULES.txt); nicht unterstützte +Funktionen werden gemeldet. Regelpakete werden nicht als Programmcode ausgeführt. + +Quellen und Stand der öffentlichen IOC-Sammlung: +[THREAT_INTELLIGENCE.txt](THREAT_INTELLIGENCE.txt) und +[config/threat_intel.json](config/threat_intel.json). +Konkrete Dateisystemregeln und Abdeckung: +[FILESYSTEM_COVERAGE.txt](FILESYSTEM_COVERAGE.txt). + +## Grenzen und Exitcodes + +Exitcode 0 bedeutet erfolgreiche Verarbeitung innerhalb der dokumentierten Grenzen, +nicht „nicht kompromittiert“. Bei jedem unvollständigen Lauf die Fehler prüfen: + +- Loganalyse: Exitcode 2 bei Fehlern oder unvollständiger Verarbeitung. +- Dateisystemprüfung: Exitcode 1 bei gemeldeten Scanproblemen; Exitcode 2 bei allgemeinen CLI-Fehlern. +- Batch-Helfer: Exitcode 1, sobald mindestens ein Einzelscan einen Fehlerstatus liefert. + +Die Loganalyse hat standardmäßig 2 GiB Vorbereitungsbudget, 100.000 Archiveinträge, +Verschachtelungstiefe 3, 1 MiB je Zeile und 512 MiB entpackte Daten je Logdatei. +Binäre Appliance-Logs benötigen gegebenenfalls einen externen Decoder. + +Der Dateisystemscanner begrenzt rohe Dateidaten und entpackte gzip-Daten auf jeweils +128 GiB, die Ausgabe auf 128 MiB und Archive auf eine Million Einträge. Pro Regel, +Ansicht und Datei speichert er höchstens 24 Fundauszüge; weitere Treffer werden +gezählt. Ausgewählte Textkopien sind auf jeweils 256 KiB begrenzt. Der Batch-Helfer +setzt das Ausgabelimit pro Archiv auf 256 MiB. Optionen: `filesystem --help-all`. + +Eingebettete TAR-/ZIP-/7z-/Paketdateien werden nicht vollständig rekursiv inventarisiert. +Gzip-Daten werden durchsucht, ohne daraus auf eine aktiv installierte Datei zu schließen. +Archivfehler und erreichte Byte-/Ausgabegrenzen werden als Untersuchungslücken gemeldet. + +Bei TAR.GZ wird das Originalarchiv während der Verarbeitung und nochmals danach +gehasht; außerdem werden Größe, Änderungszeit und Inode verglichen. Bei einem +Verzeichnis gibt es keinen Archivhash oder atomaren Schnappschuss. Entpacken kann +Metadaten verändern. Originalarchive deshalb erhalten. + +Live-Sicherungen enthalten weder den vollständigen RAM noch gelöschte Dateien. +Keine IOC-Liste deckt unbekannte Angriffe vollständig ab. Es gibt keine automatische +Wiederaufnahme unterbrochener Scans und kein hartes Gesamtlimit für RAM oder CPU. + +## Zeitangaben in Logs + +Jahr und Zeitzone nur mit belegten Geräteeinstellungen ergänzen: + +```sh +netscaler-triage scan --input ADC-A=./logs --year 2026 --timezone ADC-A=UTC +``` + +Name, Jahr und Zeitzone sind Beispiele. Explizite Zeitstempel bleiben erhalten; +mehrdeutige Sommerzeitangaben werden nicht stillschweigend aufgelöst. +Für mehrere Geräte oder dateispezifische Jahre siehe [examples/time.json](examples/time.json) +und `--time-config`. Annahmen stehen in `details/configuration.json`. + +## Aktualisieren, Hilfe und Entwicklung + +Bei einem Git-Checkout die virtuelle Umgebung aktivieren und im Repository ausführen: ```sh git pull @@ -42,145 +267,23 @@ python -m pip install --upgrade . netscaler-triage rules ``` -For an environment managed by uv, use `uv sync --locked` after `git pull`. -Version 1.0.1 fixes installation failures reporting `invalid perl operator: (?=` -or `SigmaRegularExpressionError` when loading the bundled rules. pySigma 2 changed -its default validation engine to RE2, which rejects lookaheads. The package now -requires compatible pySigma 1.x; detection patterns and timed matching are -unchanged. Reinstalling with the command above also downgrades an incompatible -pySigma installation. Rerun any scan that stopped with this error. +Bei uv-Verwaltung stattdessen nach `git pull` mit `uv sync --locked` aktualisieren. +Bei älteren Installationen mit `invalid perl operator: (?=` oder +`SigmaRegularExpressionError` neu installieren und betroffene Scans wiederholen. +Die kompatible pySigma-Version wird durch die Paketabhängigkeiten festgelegt. -## Everyday use - -```sh -# Unpacked logs, an archive, or multiple appliance exports -netscaler-triage ./logs -netscaler-triage export.tar.gz -netscaler-triage ./appliance-a ./appliance-b export.zip - -# Discover inputs in the current directory -netscaler-triage - -# Encrypted archive: prompt for a password without displaying it -netscaler-triage export.7z --password -``` - -ZIP (including ZipCrypto/AES), TAR, TAR.GZ/TGZ, GZIP and 7z inputs are supported, -including nested archives within preparation limits. Unsupported formats are -reported. `--password VALUE` is also supported, but can expose the password in -shell history or process listings. Passwords are not stored in results. - -The tool prints the result directory. Use `--output PATH` to choose a new one; -existing results are not overwritten. Appliance grouping is inferred from -directories; review the preparation manifest for unusual export layouts. - -## Read the results - -Start with these three files: - -```text -triage-results-.../ - SUMMARY.txt Start here - FINDINGS.csv Review matched events - IOCS.csv Search candidates for your SIEM - details/ Full evidence and machine-readable exports - inputs/ Prepared input copies, when preparation was needed -``` - -| File | Purpose | -| --- | --- | -| `SUMMARY.txt` | Readable overview: coverage, counts, limitations and what to open next | -| `FINDINGS.csv` | Only rule/indicator matches, with explanations and original source references | -| `IOCS.csv` | SIEM search candidates; not an automatic blocklist | - -All technical exports are under `details/`: full events in `findings.jsonl`, -URL/Base64 decoding, updates, correlations, checksums, coverage and the SQLite -index. Prepared input copies, if needed, remain in `inputs/`. - -`details/findings.jsonl` includes context and processing errors as well as matches; -use `FINDINGS.csv` to browse matched events. A match is not proof of execution. -Original-file SHA-256 matches are counted separately in `SUMMARY.txt`; their -evidence is in `details/file_hash_hits.json`, even when no log event matches. -The CSV keeps authorized-test and reference statuses visible. Spreadsheet-sensitive -cells are prefixed with an apostrophe; exact original values remain in JSON. - -This layout applies to CLI scans. Direct Python calls to `engine.scan` retain -flat machine-readable exports. Failed or interrupted scans can also retain the -flat layout; inspect their `summary.json` before using partial results. - -## Add indicators and Sigma rules - -Put your packs in a directory and validate them before scanning: - -```sh -netscaler-triage rules --config ./my-rules -netscaler-triage ./logs --config ./my-rules -``` - -Directories load Sigma YAML, JSON, CSV and TXT recursively. Repeat `--config` -to combine packs. No Python changes are needed to add an IP, path or rule. -For example, an IOC CSV can contain: - -```csv -type,value,status,source,description -ipv4,192.0.2.42,reference_only,local-review,Investigate matching events -path,/example/suspicious-file,reference_only,local-review,Check file context -``` - -These are synthetic examples. Use original, non-defanged values for matching. -TXT accepts one value per line; unambiguous IP addresses, SHA-256 hashes, URLs -and Unix paths are inferred. Other values remain literal strings. - -Sigma is parsed with pySigma and evaluated locally using the subset documented -in [RULES.txt](RULES.txt). Unsupported features fail explicitly. Sigma metadata -does not bypass detection conditions. Rule packs are data, not executable plugins. - -Defaults include 47 Sigma rules and 246 public reference indicators. Attribution -and limitations are recorded in [THREAT_INTELLIGENCE.txt](THREAT_INTELLIGENCE.txt) -and [config/threat_intel.json](config/threat_intel.json). This is a dated reference -collection, not a promise of exhaustive intelligence or zero false positives. - -Keep confidential packs in `private/`, excluded from Git and release archives. -Load them explicitly. Never put customer logs or generated findings in a public -repository. Examples under [examples/](examples/) use synthetic data. - -## Limits and interpretation - -- Exit `0`: configured processing completed; **not** a clean-system verdict. -- Exit `2`: an error or incomplete processing; inspect `SUMMARY.txt` and - `details/summary.json`. If the scan failed before presentation, inspect - `summary.json` at the result root instead. -- Missing years/timezones are not guessed. See time configuration below. -- Binary appliance logs may need an external decoder. Decode failures can also come from ordinary text. -- Damaged inputs and regex timeouts are reported. Archive path traversal, unsafe links and exhausted quotas stop preparation. -- Processing uses one worker, incremental reads and disk-backed SQLite. There is no hard process-wide RAM/CPU quota or automatic resume. -- Default limits: 2 GiB preparation budget, 100,000 archive entries, depth 3, 1 MiB per line and 512 MiB expanded per log file. Advanced options can raise limits. - -Tests were executed on macOS/ARM64. Windows and Linux instructions are provided, -but those platforms have not yet been validated in the recorded test environment. - -## Time configuration - -Use known device settings when logs omit their year or timezone: - -```sh -netscaler-triage scan --input ADC-A=./logs --year 2026 --timezone ADC-A=UTC -``` - -The appliance name, year and timezone above are examples. Explicit timestamps -are retained; ambiguous daylight-saving times remain unresolved. Use -`--time-config` for per-device settings and per-file year overrides; see -[examples/time.json](examples/time.json). The effective assumptions are recorded -in `details/configuration.json`. - -## Help and development +Weitere Hilfe: ```sh netscaler-triage --help netscaler-triage scan --help-all -netscaler-triage decode payload.txt -netscaler-triage inventory ./logs +netscaler-triage filesystem --help-all +python scripts/analyze_filesystem_exports.py --help +``` +Entwicklungsprüfungen: + +```sh uv sync --locked uv run python -W error -m unittest discover -s tests uv run mypy @@ -189,12 +292,13 @@ uv run ruff format --check src tests typing_stubs uv build ``` -See [RULES.txt](RULES.txt) for rule-pack semantics and -[VALIDATION.txt](VALIDATION.txt) for test scope and known coverage limits. +Tests wurden auf macOS/ARM64 ausgeführt. Windows und Linux wurden im dokumentierten +Testumfang nicht validiert. Details stehen in [VALIDATION.txt](VALIDATION.txt). +Direkte Python-Aufrufe von `engine.scan` liefern flache technische Exporte statt +der oben beschriebenen CLI-Berichtsstruktur. -## License +## Lizenz -Project code is licensed under the [MIT License](LICENSE.txt). -Third-party dependencies and referenced external materials retain their own -licenses. Source attribution does not imply affiliation with Citrix or the -referenced researchers, and does not relicense their original publications. +Der Programmcode steht unter der [MIT-Lizenz](LICENSE.txt). Für Abhängigkeiten und +zitierte Materialien gelten deren eigene Lizenzen. Quellenangaben bedeuten keine +Zugehörigkeit zu Citrix oder den genannten Forschenden. diff --git a/VALIDATION.txt b/VALIDATION.txt index ee93058..70f49e3 100644 --- a/VALIDATION.txt +++ b/VALIDATION.txt @@ -177,3 +177,14 @@ package with Python warnings treated as errors. A new regression checks the reported lookahead pattern against raw and decoded text, including negative cases. Ruff lint, formatting and strict mypy across source and tests passed. This verification did not rerun the complete private forensic datasets. + +2026-10-05: Abgeschlossene Erweiterung der Dateisystemanalyse +TAR.GZ und entpackte Exportverzeichnisse; Inhaltsmuster, bekannte SHA-256, +Netzwerk-IOCs, zusätzliche Fall-IOC.csv, Metadaten und Artefakt-Prüfliste. +Elf zusätzliche quellendokumentierte Regeln. Exportbefehl und paralleler +Batch-Aufruf im README, CLI-Hilfe einschließlich Grenzen aktualisiert. +149 Tests erfolgreich, darunter elf Dateisystemtests. Ruff/Mypy geprüft. +Die Erweiterung verarbeitet Archive ohne Ausführung enthaltener Programme. +Unvollständige gzip-Streams werden als Verarbeitungsfehler ausgewiesen. +Zusätzliche strukturierte Logkorrelation ist optional; der Batch-Helfer aktiviert sie nicht. +Erkennungsgrenzen: FILESYSTEM_COVERAGE.txt. Validierung ist keine Clean-Garantie. diff --git a/pyproject.toml b/pyproject.toml index a9ffef0..0b0a56a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,7 @@ netscaler-triage = "netscaler_triage.cli:main" packages = ["src/netscaler_triage"] [tool.hatch.build.targets.sdist] -include = ["src/", "tests/", "typing_stubs/", "examples/", "config/threat_intel.json", "README.md", "RULES.txt", "VALIDATION.txt", "THREAT_INTELLIGENCE.txt", "LICENSE.txt", "uv.lock", ".python-version", ".gitignore", "pyproject.toml"] +include = ["src/", "tests/", "typing_stubs/", "examples/", "config/threat_intel.json", "README.md", "RULES.txt", "VALIDATION.txt", "THREAT_INTELLIGENCE.txt", "FILESYSTEM_COVERAGE.txt", "scripts/analyze_filesystem_exports.py", "LICENSE.txt", "uv.lock", ".python-version", ".gitignore", "pyproject.toml"] [dependency-groups] dev = [ diff --git a/scripts/analyze_filesystem_exports.py b/scripts/analyze_filesystem_exports.py new file mode 100644 index 0000000..cbf906b --- /dev/null +++ b/scripts/analyze_filesystem_exports.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Run the installed filesystem scanner concurrently on explicitly named exports.""" + +import argparse +import concurrent.futures +import hashlib +import json +import subprocess +import sys +from pathlib import Path + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("archives", nargs="+", type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--ioc-file", type=Path) + parser.add_argument("--workers", type=int, default=4) + args = parser.parse_args() + if args.workers < 1 or args.workers > 4: + parser.error("workers must be 1-4") + names = [p.name.removesuffix(".tar.gz").removesuffix(".tgz") for p in args.archives] + if len(set(names)) != len(names): + parser.error("Archive names must be unique") + if args.output.exists(): + parser.error("Output must be new") + for archive in args.archives: + if not archive.is_file(): + parser.error(f"Not a readable regular archive: {archive.name}") + args.output.mkdir(parents=True, mode=0o700) + if args.ioc_file: + # Preserve the exact small case rule input used for this run. + data = args.ioc_file.read_bytes() + (args.output / "case-iocs.csv").write_bytes(data) + args.ioc_file = args.output / "case-iocs.csv" + code = Path(__file__).resolve().parents[1] / "src/netscaler_triage" + manifest = { + "archives": [p.name for p in args.archives], + "workers": args.workers, + "tool_sha256": { + str(p.relative_to(code)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in [ + code / "filesystem.py", + code / "cli.py", + code / "data/filesystem_rules.json", + code / "data/threat_intel.json", + ] + }, + } + (args.output / "run_manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") + + def run(item: tuple[Path, str]) -> dict: + archive, name = item + command = [ + sys.executable, + "-m", + "netscaler_triage", + "filesystem", + str(archive), + "--output", + str(args.output / name), + "--max-output-bytes", + str(256 * 1024**2), + "--json", + ] + if args.ioc_file: + command.extend(["--ioc-file", str(args.ioc_file)]) + print(json.dumps({"started": archive.name}), flush=True) + with ( + (args.output / (name + ".stdout.json")).open("w") as stdout, + (args.output / (name + ".progress.log")).open("w") as stderr, + ): + result = subprocess.run(command, stdout=stdout, stderr=stderr, check=False) + row = {"archive": archive.name, "output": name, "exit_code": result.returncode} + print(json.dumps(row), flush=True) + return row + + with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool: + results = list(pool.map(run, zip(args.archives, names, strict=True))) + (args.output / "batch_summary.json").write_text(json.dumps(results, indent=2) + "\n") + return int(any(r["exit_code"] != 0 for r in results)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/netscaler_triage/cli.py b/src/netscaler_triage/cli.py index f93f8cb..66a6217 100644 --- a/src/netscaler_triage/cli.py +++ b/src/netscaler_triage/cli.py @@ -86,6 +86,7 @@ def main(argv: list[str] | None = None) -> int: "decode", "indicators", "inventory", + "filesystem", "rules", "--help", "-h", @@ -187,6 +188,26 @@ def main(argv: list[str] | None = None) -> int: ) rules_parser.add_argument("--no-default-patterns", action="store_true") rules_parser.add_argument("--regex-timeout", type=float, default=0.05) + fs = commands.add_parser( + "filesystem", help="Inspect a TAR.GZ or exported filesystem directory without extraction" + ) + fs.add_argument("source", type=Path, help="TAR.GZ archive or exported filesystem directory") + fs.add_argument( + "--output", + type=Path, + required=True, + help="Required new output directory outside the evidence tree", + ) + fs.add_argument("--baseline", type=Path) + fs.add_argument("--ioc-file", type=Path, help="Additional case IOC CSV") + fs.add_argument( + "--analyze-logs", + action="store_true", + help="Analyze selected logs using temporary copies capped at 16 MiB", + ) + fs.add_argument("--timezone", help="IANA timezone for selected log analysis") + fs.add_argument("--max-expanded-bytes", type=int, default=128 * 1024**3) + fs.add_argument("--max-output-bytes", type=int, default=128 * 1024**2) inv = commands.add_parser( "inventory", help="Create a SHA-256 inventory with paths relative to the input root" ) @@ -215,6 +236,10 @@ def main(argv: list[str] | None = None) -> int: "root": "Log directory to inventory recursively", } examples = { + "filesystem": ( + "Example: netscaler-triage filesystem snapshot.tar.gz " + "--output ./filesystem-results --baseline ./previous-files" + ), "scan": ( "Example: netscaler-triage scan ./logs\nBuilt-in patterns, safe " "decoding, update detection and correlation run automatically. Missing " @@ -258,6 +283,32 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) result: object try: + if args.command == "filesystem": + from .filesystem import scan_filesystem + + result = scan_filesystem( + args.source, + args.output, + baseline=args.baseline, + ioc_file=args.ioc_file, + max_expanded_bytes=args.max_expanded_bytes, + max_output_bytes=args.max_output_bytes, + ) + if args.analyze_logs and result["complete_scan"]: + from .filesystem import scan_snapshot_logs + + result["log_analysis"] = scan_snapshot_logs( + args.source, args.output, timezone=args.timezone + ) + display_result(result, True) + return ( + 0 + if result["complete_scan"] + and result["source_unchanged"] is not False + and not result["issues"] + and result.get("log_analysis", {}).get("status") != "partial" + else 1 + ) if args.command == "rules": from .matching import Matcher diff --git a/src/netscaler_triage/data/filesystem_rules.json b/src/netscaler_triage/data/filesystem_rules.json new file mode 100644 index 0000000..4da6357 --- /dev/null +++ b/src/netscaler_triage/data/filesystem_rules.json @@ -0,0 +1,102 @@ +{ + "updated": "2026-10-05", + "classification": "review_candidate", + "rules": [ + { + "id": "php_nonstandard_handler", + "regex": "\\b(?:addhandler|addtype|sethandler)\\b[^\\r\\n]{0,250}application/x-httpd-php[^\\r\\n]{0,100}\\.(?:deb|sig|ico|rpm|tgz|html|dat|png|so|sh)\\b", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "application/x-httpd-php" + ] + }, + { + "id": "php_engine_enabled", + "regex": "\\bphp_(?:flag|admin_flag)\\s+engine\\s+on\\b", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "php_flag", + "php_admin_flag" + ] + }, + { + "id": "web_alias_routing", + "regex": "\\b(?:alias|aliasmatch|rewriterule)\\b[^\\r\\n]{0,800}(?:/vpn/(?:media|theme|images)|\\.sig|\\.ctxs\\.receiver|\\.local_journal|receiver\\.min|logonuisimple)", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "alias", + "rewriterule" + ] + }, + { + "id": "css_webshell_alias", + "regex": "(?:receiver\\.min(?:\\.[a-f0-9]*)?\\.css|logonuisimple\\.html\\.style\\.min(?:\\.[a-f0-9]+)?\\.css)", + "source": "https://www.levelblue.com/blogs/spiderlabs-blog/citrix-netscaler-cve-2026-88771-observed-exploitation-artifacts-and-hunt-indicators", + "required_any_literal": [ + "receiver.min", + "logonuisimple" + ] + }, + { + "id": "nsmon_cron", + "regex": "^[ \\t]*\\*/5[ \\t]+\\*[ \\t]+\\*[ \\t]+\\*[ \\t]+\\*[^\\r\\n]{0,200}\\bperl\\b[^\\r\\n]{0,200}nsmon\\.pl", + "source": "https://raw.githubusercontent.com/rtkwlf/wolf-tools/main/pack_alerts/202609-citrix-netscaler-active-exploitation-cve-2026-88771/README.md", + "required_any_literal": [ + "nsmon.pl" + ] + }, + { + "id": "privileged_monitor_account", + "regex": "\\b(?:add|set|bind)\\s+system\\s+user\\s+sec_monitor\\b[^\\r\\n]*", + "source": "https://www.levelblue.com/blogs/spiderlabs-blog/citrix-netscaler-cve-2026-88771-observed-exploitation-artifacts-and-hunt-indicators", + "required_any_literal": [ + "sec_monitor" + ] + }, + { + "id": "command_channel_header", + "regex": "\\b(?:http[_-])?(?:nsc[_-](?:ldap|clienttype)|x[_-]ux(?:[_-][0-9]+)?)\\b", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "ldap", + "clienttype", + "ux" + ] + }, + { + "id": "cron_scrubbing_reference", + "regex": "(?:preg_replace|sed)[^\\r\\n]{0,800}(?:crontab|/vpn/scripts/linux)", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "preg_replace", + "sed" + ] + }, + { + "id": "shell_privilege_change", + "regex": "\\bchmod\\s+(?:u\\+s|[2467][0-7]{3})\\s+/bin/sh\\b", + "source": "https://cloud.google.com/blog/topics/threat-intelligence/defending-against-active-exploitation-of-citrix-netscaler-adc-and-gateway-appliances", + "required_any_literal": [ + "chmod" + ] + }, + { + "id": "reverse_shell_reference", + "regex": "/dev/tcp/|\\b(?:nc|ncat)\\s+[^\\r\\n]{0,100}-e\\s+/bin/(?:ba)?sh|\\bdup2\\s*\\(", + "source": "https://www.levelblue.com/blogs/spiderlabs-blog/citrix-netscaler-cve-2026-88771-observed-exploitation-artifacts-and-hunt-indicators", + "required_any_literal": [ + "/dev/tcp/", + "nc", + "dup2" + ] + }, + { + "id": "configuration_staging_reference", + "regex": "\\b(?:tar|cat)\\b[^\\r\\n]{0,500}/(?:flash/)?nsconfig[^\\r\\n]{0,500}(?:/var/netscaler|update_result)", + "source": "https://www.levelblue.com/blogs/spiderlabs-blog/citrix-netscaler-cve-2026-88771-observed-exploitation-artifacts-and-hunt-indicators", + "required_any_literal": [ + "nsconfig" + ] + } + ] +} diff --git a/src/netscaler_triage/filesystem.py b/src/netscaler_triage/filesystem.py new file mode 100644 index 0000000..888ce81 --- /dev/null +++ b/src/netscaler_triage/filesystem.py @@ -0,0 +1,790 @@ +"""Bounded, non-extracting TAR filesystem triage. Findings are review candidates.""" + +from __future__ import annotations + +import csv +import gzip +import hashlib +import io +import json +import os +import re +import stat +import sys +import tarfile +import time +import zlib +from collections import Counter +from collections.abc import Iterator +from contextlib import contextmanager +from datetime import UTC, datetime +from pathlib import Path, PurePosixPath +from typing import Any, BinaryIO + +BLOCK = 1024 * 1024 +OVERLAP = 4096 +RARE = re.compile( + rb"\.ctxs\.receiver|\.local_journal|\.uxd(?:port|lock)|nsmon\.pl|sec_monitor|http_nsc_ldap|http_nsc_clienttype|http_x_ux|uxd_idle_exit|nsc_tass|rhfajaf1h992|7489a0f93c67fa5cdaeb4b921d90594d|update_c08937\.pl|update_result_3567cs\.tgz|id009\.txt|ww009\.php|wbnsisqmfj\.png" +) +EXEC = re.compile( + rb"shell_exec\s*\(|passthru\s*\(|proc_open\s*\(|popen\s*\(|(?:eval|system|assert)\s*\(" +) +INPUT = re.compile(rb"\$_(?:get|post|request|cookie|server)|http_nsc_|http_x_ux") +PERSIST = re.compile( + rb"nsmon|sec_monitor|auto_prepend|auto_append|aliasmatch|addhandler|sethandler|authorizedkeys|forcecommand|permitrootlogin|(?:curl|wget|fetch)\s|/var/tmp/|/tmp/|dup2|pty\.spawn" +) +LOG = re.compile( + rb"ns_monuploadd_err\.pl|pitboss.*?ppe.*?(?:unexpectedly died|missed too many heartbeats)" + rb"|ssl_handshake_failure|orphan rings|not restarting nsppe" +) +NAMES = { + ".ctxs.receiver", + ".local_journal", + ".nsmon", + ".uxdport", + ".uxdlock", + "nsmon.pl", + "nsg64.deb", + "nsgclient.sig", + "e6ee7c85.sig", + "main.py", + "1.py", + ".s", + "update_c08937.pl", + "update_result_3567cs.tgz", + "xua.html", + "insight-new.js", + "id009.txt", + "ww009.php", + "wbNsIsqmFj.png", + "lula", + "boom", +} +EXACT = { + "var/netscaler/gui/vpn/c", + "var/tmp/sh", + "var/vpn/theme/x.php", + "var/vpn/themes/logon.php", + "var/netscaler/logon/logon.php", +} +CAPTURE = { + "etc/httpd.conf", + "nsconfig/httpd.conf", + "flash/nsconfig/httpd.conf", + "nsconfig/crontab", + "nsconfig/rc.netscaler", + "etc/master.passwd", + "etc/inetd.conf", + "etc/syslog.conf", + "etc/crontab", + "etc/passwd", + "etc/sshd_config", + "nsconfig/ns.conf", + "flash/nsconfig/ns.conf", + "flash/nsconfig/rc.netscaler", + "flash/nsconfig/crontab", + "netscaler/ns_monuploadd_err.pl", + "var/python/bin/customsnmpd", + "netscaler/websocketd/websocket-ssh-deamon.php", + "etc/rc.conf", + "etc/rc.local", + "root/.ssh/authorized_keys", +} + + +def normalized(name: str) -> str: + """Normalize only harmless leading './'; never resolve archive paths on disk.""" + while name.startswith("./"): + name = name[2:] + return name + + +def critical(path: str) -> bool: + return ( + path in CAPTURE + or path.startswith( + ("var/cron/tabs/", "var/spool/cron/", "etc/cron.d/", "etc/rc.d/", "usr/local/etc/rc.d/") + ) + or ( + path.startswith(("etc/", "nsconfig/", "flash/nsconfig/")) + and (path.endswith(".conf") or "/httpd" in path) + ) + or path.endswith(("/authorized_keys", "/.htaccess", "/.user.ini")) + ) + + +def context(path: str) -> str: + if path.startswith(("var/log/", "var/nslog/")): + return "log" + if path.startswith(("var/nsinstall/", "var/ns_sys_backup/", "var/tmp/support/")): + return "installer_or_backup" + if critical(path): + return "configuration_or_program" + if path.startswith(("netscaler/", "var/netscaler/", "var/vpn/")): + return "web_or_vendor_program" + return "other" + + +class HashReader(io.RawIOBase): + def __init__(self, raw: BinaryIO) -> None: + self.raw = raw + self.digest = hashlib.sha256() + self.count = 0 + + def read(self, size: int | None = -1) -> bytes: + data = self.raw.read(-1 if size is None else size) + self.digest.update(data) + self.count += len(data) + return data + + +class DirectoryArchive: + """Read an exported tree without following links or opening special devices.""" + + def __init__(self, root: Path) -> None: + self.root = root + self.current: tuple[int, str, os.stat_result] | None = None + self.handle: BinaryIO | None = None + + def __iter__(self) -> Iterator[tarfile.TarInfo]: + def fail(error: OSError) -> None: + raise error + + for parent, dirs, files, fd in os.fwalk(self.root, follow_symlinks=False, onerror=fail): + for name in sorted(dirs + files): + st = os.stat(name, dir_fd=fd, follow_symlinks=False) + relative = str((Path(parent) / name).relative_to(self.root)) + member = tarfile.TarInfo(relative) + member.mode, member.uid, member.gid = stat.S_IMODE(st.st_mode), st.st_uid, st.st_gid + member.mtime = st.st_mtime + if stat.S_ISLNK(st.st_mode): + member.type = tarfile.SYMTYPE + member.linkname = os.readlink(name, dir_fd=fd) + elif stat.S_ISDIR(st.st_mode): + member.type = tarfile.DIRTYPE + elif stat.S_ISREG(st.st_mode): + member.size = st.st_size + else: + member.type = tarfile.FIFOTYPE if stat.S_ISFIFO(st.st_mode) else b"?" + self.current = (fd, name, st) + try: + yield member + if self.handle is not None: + after = os.fstat(self.handle.fileno()) + if (st.st_size, st.st_mtime_ns, st.st_ctime_ns) != ( + after.st_size, + after.st_mtime_ns, + after.st_ctime_ns, + ): + raise ValueError("Source file changed during reading: " + relative) + finally: + if self.handle is not None: + self.handle.close() + self.handle = None + + def extractfile(self, member: tarfile.TarInfo) -> BinaryIO: + assert self.current is not None + fd, name, before = self.current + opened = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd) + current = os.fstat(opened) + if not stat.S_ISREG(current.st_mode) or (current.st_dev, current.st_ino) != ( + before.st_dev, + before.st_ino, + ): + os.close(opened) + raise ValueError("Source entry changed before opening: " + member.name) + self.handle = os.fdopen(opened, "rb") + return self.handle + + +@contextmanager +def filesystem_input(source: Path) -> Iterator[tuple[Any, Any, Any]]: + if source.is_dir(): + directory = DirectoryArchive(source) + try: + yield directory, None, None + finally: + if directory.handle is not None: + directory.handle.close() + else: + with source.open("rb", buffering=8 * BLOCK) as raw: + reader = HashReader(raw) + with ( + gzip.GzipFile(fileobj=reader) as expanded, + tarfile.open(fileobj=expanded, mode="r|", stream=True) as archive, # type: ignore[call-overload] + ): + yield archive, reader, expanded + + +def scan_filesystem( + source: Path, + output: Path, + *, + baseline: Path | None = None, + ioc_file: Path | None = None, + max_expanded_bytes: int = 128 * 1024**3, + max_output_bytes: int = 128 * 1024**2, +) -> dict[str, Any]: + """Hash every regular member, inspect raw/gzip bytes, retain no full archive copies.""" + source = source.resolve(strict=True) + before = source.stat() + is_directory = stat.S_ISDIR(before.st_mode) + if not is_directory and not stat.S_ISREG(before.st_mode): + raise ValueError("Input must be a regular archive file or exported filesystem directory") + if is_directory and (source == output.resolve() or source in output.resolve().parents): + raise ValueError("Output must not be inside the evidence directory") + if output.exists() or output.is_symlink(): + raise ValueError("Output must be a new directory") + if baseline is not None: + baseline = baseline.resolve(strict=True) + if baseline == output.resolve() or baseline in output.resolve().parents: + raise ValueError("Output must not be inside the baseline") + if min(max_expanded_bytes, max_output_bytes) < 1: + raise ValueError("Limits must be positive") + output.mkdir(parents=True, mode=0o700) + os.chmod(output, 0o700) + intel = json.loads((Path(__file__).parent / "data/threat_intel.json").read_text()) + custom_ioc_hash = None + if ioc_file is not None: + custom_bytes = ioc_file.read_bytes() + custom_ioc_hash = hashlib.sha256(custom_bytes).hexdigest() + for indicator in csv.DictReader(io.StringIO(custom_bytes.decode("utf-8-sig"))): + if indicator.get("type") not in {"ip", "ipv4", "domain", "url", "path", "sha256"}: + continue + if indicator.get("status", "").lower() in {"excluded", "own_test", "authorized_test"}: + continue + if not indicator.get("value", "").strip(): + continue + intel["indicators"].append(indicator) + rules_doc = json.loads((Path(__file__).parent / "data/filesystem_rules.json").read_text()) + extra_patterns = [ + (r["id"], re.compile(r["regex"].encode(), re.I | re.M)) for r in rules_doc["rules"] + ] + rule_gates = { + r["id"]: tuple(t.encode() for t in r.get("required_any_literal", [])) + for r in rules_doc["rules"] + } + known = {x["value"].lower(): x for x in intel["indicators"] if x["type"] == "sha256"} + known_paths = {x["value"].lstrip("/") for x in intel["indicators"] if x["type"] == "path"} + network_values = sorted( + { + x["value"].lower() + for x in intel["indicators"] + if x["type"] in {"ip", "ipv4", "domain", "url"} + } + ) + network_pattern = ( + re.compile( + b"|".join(re.escape(v.encode()) for v in sorted(network_values, key=len, reverse=True)) + ) + if network_values + else None + ) + counts: Counter[str] = Counter() + issues: list[dict[str, Any]] = [] + seen: set[str] = set() + baseline_hashes: dict[str, str] = {} + started_at = datetime.now(UTC).isoformat() + started = time.monotonic() + last_progress = started + handles = { + name: (output / f"{name}.jsonl").open("x", encoding="utf-8") + for name in ("inventory", "findings", "selected_content", "baseline_changes") + } + + def emit(channel: str, data: dict[str, Any]) -> None: + line = json.dumps(data, ensure_ascii=True) + "\n" + size = len(line.encode()) + if counts["output_bytes"] + size > max_output_bytes: + raise ValueError("Output limit reached; incomplete analysis") + handles[channel].write(line) + counts["output_bytes"] += size + + def finding(path: str, rule: str, **details: Any) -> None: + counts[f"finding_{rule}"] += 1 + emit( + "findings", + { + "path": path, + "context": context(path), + "rule": rule, + "classification": "review_candidate", + **details, + }, + ) + + class View: + def __init__(self, path: str, label: str) -> None: + self.path, self.label = path, label + self.total = self.lines = 0 + self.tail = b"" + self.matches: Counter[str] = Counter() + self.flags: set[str] = set() + self.digest = hashlib.sha256() + + def feed(self, chunk: bytes) -> None: + if not chunk: + return + self.digest.update(chunk) + data = self.tail + chunk + lower = data.lower() + base = self.total - len(self.tail) + if b" None: + counts[f"{self.label}_bytes_scanned"] += self.total + if self.digest.hexdigest() in known: + finding( + self.path, + "known_sha256", + view=self.label, + sha256=self.digest.hexdigest(), + indicator=known[self.digest.hexdigest()], + ) + if "php" in self.flags and self.path.lower().endswith( + ( + ".deb", + ".sig", + ".ico", + ".png", + ".css", + ".rpm", + ".tgz", + ".html", + ".dat", + ".so", + ".sh", + ) + ): + finding(self.path, "php_in_disguised_file", view=self.label) + if {"php", "exec", "http_input"} <= self.flags: + finding( + self.path, + "php_input_and_execution", + view=self.label, + note="Co-occurrence only; manually trace data flow before confirming webshell", + ) + if "socket_dup2" in self.flags and "customsnmpd" in self.path: + finding(self.path, "customsnmpd_socket_dup2", view=self.label) + for rule, n in self.matches.items(): + if n > 24: + finding( + self.path, + "excerpt_limit", + view=self.label, + affected_rule=rule, + total_matches=n, + saved_excerpts=24, + ) + + archive_sha = None + complete = False + try: + with filesystem_input(source) as (archive, reader, expanded): + for member in archive: + path = normalized(member.name) + counts["members"] += 1 + if counts["members"] > 1000000: + raise ValueError("Member limit reached") + if time.monotonic() - last_progress > 30: + print( + json.dumps( + { + "members": counts["members"], + "raw_member_GiB": round(counts["raw_bytes_scanned"] / 1024**3, 2), + "compressed_GiB_read": round(reader.count / 1024**3, 2) + if reader + else None, + } + ), + file=sys.stderr, + flush=True, + ) + last_progress = time.monotonic() + row: dict[str, Any] = { + "path": path, + "member_name": member.name, + "size": member.size, + "mtime_epoch": member.mtime, + "mode": oct(member.mode), + "uid": member.uid, + "gid": member.gid, + "uname": member.uname, + "gname": member.gname, + "type": member.type.decode("ascii", "replace"), + "linkname": member.linkname, + "pax_headers": member.pax_headers, + "archive_header_offset": member.offset, + } + if path in seen: + finding(path, "duplicate_member") + seen.add(path) + if path.startswith("/") or ".." in PurePosixPath(path).parts: + finding(path, "unsafe_archive_path", note="Recorded only, never extracted") + if PurePosixPath(path).name in NAMES or path in EXACT or path in known_paths: + finding(path, "known_artifact_name", metadata=row) + if member.isfile() and member.mode & 0o6000: + finding( + path, + "suid_sgid", + mode=oct(member.mode), + uid=member.uid, + note="Vendor binaries can legitimately have these bits", + ) + if member.isfile(): + counts["regular_files"] += 1 + view = View(path, "raw") + unpacked: View | None = None + decompressor: Any = None + compressed_bad = False + captured = bytearray() + capture = critical(path) + stream = archive.extractfile(member) + if stream is None: + raise ValueError("Regular member has no stream") + while chunk := stream.read(BLOCK): + if view.total == 0: + if chunk.startswith(b"\x1f\x8b"): + unpacked = View(path, "gzip") + decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS) + counts["gzip_members"] += 1 + if chunk.startswith( + (b"PK\x03\x04", b"7z\xbc\xaf\x27\x1c") + ) or path.endswith((".tar", ".tgz", ".tar.gz", ".zip", ".7z", ".deb")): + finding( + path, + "embedded_container", + note=( + "Raw/gzip bytes inspected; " + "nested members not individually parsed" + ), + ) + view.feed(chunk) + if capture and len(captured) < 262144: + captured.extend(chunk[: 262144 - len(captured)]) + if decompressor is not None and not compressed_bad: + pending = chunk + try: + while pending: + if decompressor.eof: + pending = pending.lstrip(b"\0") + if not pending: + break + decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS) + data = decompressor.decompress(pending, BLOCK) + counts["expanded_bytes"] += len(data) + if counts["expanded_bytes"] > max_expanded_bytes: + raise ValueError("Nested gzip expanded-byte limit reached") + assert unpacked is not None + unpacked.feed(data) + pending = ( + decompressor.unused_data + if decompressor.eof + else decompressor.unconsumed_tail + ) + except zlib.error as exc: + compressed_bad = True + issues.append( + { + "path": path, + "error": "nested_gzip_error", + "detail": str(exc), + } + ) + if counts["raw_bytes_scanned"] + view.total > max_expanded_bytes: + raise ValueError("Raw member byte limit reached") + view.finish() + row["sha256"] = view.digest.hexdigest() + row["content_flags"] = sorted(view.flags) + if view.total != member.size: + raise ValueError("Member size mismatch") + if unpacked is not None: + if not compressed_bad and not decompressor.eof: + issues.append({"path": path, "error": "truncated_nested_gzip"}) + unpacked.finish() + row["gzip_complete"] = not compressed_bad and decompressor.eof + row["gzip_sha256"] = unpacked.digest.hexdigest() + row["gzip_bytes"] = unpacked.total + if capture: + emit( + "selected_content", + { + "path": path, + "sha256": row["sha256"], + "size": member.size, + "truncated": member.size > len(captured), + "text": captured.decode("utf-8", "replace"), + }, + ) + if ( + baseline is not None + and not path.startswith("/") + and ".." not in PurePosixPath(path).parts + ): + reference = baseline / path + resolved = reference.resolve() + if ( + baseline in resolved.parents + and reference.is_file() + and not reference.is_symlink() + ): + counts["baseline_compared"] += 1 + key = str(resolved) + if key not in baseline_hashes: + with reference.open("rb") as f: + baseline_hashes[key] = hashlib.file_digest( + f, "sha256" + ).hexdigest() + row["baseline_same"] = row["sha256"] == baseline_hashes[key] + if not row["baseline_same"]: + counts["baseline_changed"] += 1 + emit( + "baseline_changes", + { + "path": path, + "sha256": row["sha256"], + "baseline_sha256": baseline_hashes[key], + "size": member.size, + "note": ( + "Historical comparison, " + "not a trusted clean-build baseline" + ), + }, + ) + else: + counts["no_baseline_file"] += 1 + emit("inventory", row) + if expanded is not None: + # Drain gzip to validate its CRC/trailer, then hash any trailing raw bytes. + while expanded.read(BLOCK): + counts["outer_trailing_expanded_blocks"] += 1 + while reader.read(BLOCK): + pass + archive_sha = reader.digest.hexdigest() + complete = True + except (OSError, ValueError, EOFError, tarfile.TarError) as exc: + issues.append({"error": type(exc).__name__, "detail": str(exc)}) + finally: + for handle in handles.values(): + handle.close() + after = source.stat() + final_sha = None + if not is_directory: + with source.open("rb", buffering=8 * BLOCK) as verification_stream: + final_sha = hashlib.file_digest(verification_stream, "sha256").hexdigest() # type: ignore[arg-type] + unchanged = ( + None + if is_directory + else ( + (before.st_size, before.st_mtime_ns, before.st_ino) + == (after.st_size, after.st_mtime_ns, after.st_ino) + and archive_sha == final_sha + ) + ) + result = { + "source_name": source.name, + "source_kind": "directory" if is_directory else "tar.gz", + "rule_revision": rules_doc["updated"], + "custom_ioc_sha256": custom_ioc_hash, + "known_hash_count": len(known), + "network_indicator_count": len(network_values), + "rule_ids": [r["id"] for r in rules_doc["rules"]], + "complete_scan": complete, + "source_size": before.st_size, + "archive_sha256": archive_sha, + "verification_sha256": final_sha, + "source_unchanged": unchanged, + "complete_outer_archive": complete if not is_directory else None, + "counts": dict(counts), + "issues": issues, + "duration_seconds": round(time.monotonic() - started, 2), + "started_at_utc": started_at, + "source_mtime_utc": datetime.fromtimestamp(before.st_mtime, UTC).isoformat(), + "limitations": [ + "Live filesystem copy, no RAM/deleted blocks; source tools may be compromised", + ( + "Directory mode: hashes describe observed files; " + "no atomic tree integrity guarantee; " + "host metadata may differ from appliance metadata" + ), + ( + "Nested gzip byte streams inspected; " + "nested TAR/ZIP/7z/package member metadata not recursively parsed" + ), + ( + "At most 24 excerpts per rule/view/file; " + "complete matching continues, excess counts recorded" + ), + ( + "Critical text captures limited to 256 KiB each; " + "generic encrypted/obfuscated payloads can evade rules" + ), + "Baselines are historical evidence, not manufacturer-signed clean images", + ( + "Symlinks/hardlinks recorded, never followed or executed; " + "payload bytes hashed only at regular members" + ), + ], + } + artifact_checks = { + "scope": "Outer archive members or exported directory; no historical absence claim", + "scan_complete": complete, + "issues": issues, + "paths": [{"path": p, "present": p in seen} for p in sorted(known_paths | EXACT)], + "names": [ + {"name": n, "paths": sorted(p for p in seen if PurePosixPath(p).name == n)} + for n in sorted(NAMES) + ], + } + (output / "artifact_checks.json").write_text(json.dumps(artifact_checks, indent=2) + "\n") + (output / "summary.json").write_text(json.dumps(result, indent=2, ensure_ascii=True) + "\n") + return result + + +def read_selected_members( + source: Path, paths: set[str], *, max_total_bytes: int = 32 * 1024**2 +) -> list[tuple[dict[str, Any], bytes]]: + """Read explicit regular members into bounded memory, never extract or follow links. + + All matching occurrences are retained so duplicate member names are not silently resolved. + Intended for focused manual content review after the inventory, not bulk extraction. + """ + result: list[tuple[dict[str, Any], bytes]] = [] + total = 0 + with filesystem_input(source) as (archive, _reader, _expanded): + for member in archive: + path = normalized(member.name) + if path not in paths: + continue + metadata = { + "path": path, + "type": member.type.decode("ascii", "replace"), + "size": member.size, + "mtime_epoch": member.mtime, + "mode": oct(member.mode), + "uid": member.uid, + "linkname": member.linkname, + "archive_header_offset": member.offset, + } + if not member.isfile(): + result.append((metadata, b"")) + continue + if total + member.size > max_total_bytes: + raise ValueError("Selected-member memory limit exceeded") + handle = archive.extractfile(member) + if handle is None: + raise ValueError("Regular member has no stream") + data = handle.read(member.size) + if len(data) != member.size: + raise ValueError("Incomplete selected member") + total += len(data) + metadata["sha256"] = hashlib.sha256(data).hexdigest() + result.append((metadata, data)) + return result + + +def scan_snapshot_logs( + source: Path, + output: Path, + *, + timezone: str | None = None, + max_log_bytes: int = 16 * 1024**2, +) -> dict[str, Any]: + """Run the existing log engine on selected small logs; always remove temporary copies.""" + import tempfile + + from .engine import scan + + names = re.compile( + r"^(?:ns\.log|messages|notice\.log|auth|secure|cron|callhome|httpaccess|httperror|nsvpn|bash|shell|sh\.log|nsumond)" + ) + paths: set[str] = set() + with (output / "inventory.jsonl").open() as inventory: + for line in inventory: + row = json.loads(line) + path = row["path"] + if ( + row["type"] == "0" + and path.startswith(("var/log/", "var/nslog/")) + and "/db/" not in path + and names.match(PurePosixPath(path).name) + ): + paths.add(path) + rows = read_selected_members(source, paths, max_total_bytes=max_log_bytes) + manifest = [] + with tempfile.TemporaryDirectory(prefix="netscaler-selected-logs-") as directory: + root = Path(directory) + for entry, data in rows: + path = entry["path"] + if entry["type"] != "0": + continue + if path.startswith("/") or ".." in PurePosixPath(path).parts: + raise ValueError("Unsafe selected log path") + target = root / path + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("xb") as handle: + handle.write(data) + os.chmod(target, 0o600) + manifest.append(entry) + (output / "selected_log_manifest.json").write_text( + json.dumps(manifest, indent=2, ensure_ascii=True) + "\n" + ) + result = scan( + [("FILESYSTEM", root)], + output / "loganalyse", + [], + year=None, + timezones={"FILESYSTEM": timezone} if timezone else {}, + max_file_bytes=64 * 1024**2, + ) + return result diff --git a/tests/test_filesystem.py b/tests/test_filesystem.py new file mode 100644 index 0000000..1870b58 --- /dev/null +++ b/tests/test_filesystem.py @@ -0,0 +1,284 @@ +import gzip +import hashlib +import io +import json +import tarfile +import tempfile +import unittest +from pathlib import Path + +from netscaler_triage.filesystem import BLOCK, read_selected_members, scan_filesystem + + +class FilesystemTests(unittest.TestCase): + def make_archive(self, path: Path, entries: dict[str, bytes]) -> None: + with tarfile.open(path, "w:gz") as archive: + for name, data in entries.items(): + info = tarfile.TarInfo(name) + info.size = len(data) + info.mode = 0o6555 if name == "bin/sh" else 0o644 + archive.addfile(info, io.BytesIO(data)) + link = tarfile.TarInfo("escape-link") + link.type, link.linkname = tarfile.SYMTYPE, "/etc/passwd" + archive.addfile(link) + + def test_streaming_evidence_hashes_links_gzip_and_baseline(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "snapshot.tgz" + inert = b" None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"a": b"inert"}) + data = bytearray(source.read_bytes()) + data[-8] ^= 255 + source.write_bytes(data) + result = scan_filesystem(source, root / "result") + self.assertFalse(result["complete_outer_archive"]) + self.assertTrue(result["issues"]) + + def test_nested_gzip_error_recorded(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"bad.gz": gzip.compress(b"inert")[:-5]}) + result = scan_filesystem(source, root / "result") + self.assertTrue(result["complete_outer_archive"]) + self.assertEqual(result["issues"][0]["error"], "truncated_nested_gzip") + + def test_byte_limit_and_existing_output(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"a": b"x" * 100}) + result = scan_filesystem(source, root / "result", max_expanded_bytes=10) + self.assertFalse(result["complete_outer_archive"]) + with self.assertRaises(ValueError): + scan_filesystem(source, root / "result") + + def test_targeted_read_is_bounded_and_does_not_follow_links(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"./etc/test": b"inert evidence"}) + rows = read_selected_members(source, {"etc/test", "escape-link"}) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0][1], b"inert evidence") + self.assertEqual(rows[1][1], b"") + self.assertEqual(rows[1][0]["linkname"], "/etc/passwd") + with self.assertRaises(ValueError): + read_selected_members(source, {"etc/test"}, max_total_bytes=2) + + def test_cli_streaming_route(self) -> None: + from contextlib import redirect_stdout + + from netscaler_triage.cli import main + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"a": b"inert"}) + with redirect_stdout(io.StringIO()): + status = main(["filesystem", str(source), "--output", str(root / "result")]) + self.assertEqual(status, 0) + self.assertTrue((root / "result" / "summary.json").is_file()) + + def test_log_review_cleans_temporary_copies_on_error(self) -> None: + from unittest.mock import patch + + from netscaler_triage.filesystem import scan_snapshot_logs + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "input.tgz" + self.make_archive(source, {"var/log/ns.log": b"inert log\n"}) + scan_filesystem(source, root / "result") + seen: list[Path] = [] + + def fail(*args: object, **kwargs: object) -> dict[str, object]: + inputs = args[0] + assert isinstance(inputs, list) + temporary = inputs[0][1] + assert isinstance(temporary, Path) + seen.append(temporary) + self.assertEqual((temporary / "var/log/ns.log").read_bytes(), b"inert log\n") + raise ValueError("simulated engine failure") + + with ( + patch("netscaler_triage.engine.scan", side_effect=fail), + self.assertRaises(ValueError), + ): + scan_snapshot_logs(source, root / "result") + self.assertEqual(len(seen), 1) + self.assertFalse(seen[0].exists()) + + def test_directory_matches_archive_and_never_follows_links(self) -> None: + import os + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "export" + (source / "var/tmp").mkdir(parents=True) + sample = b"HTTP_X_UX nsmon.pl 149.104.78.141\n" + (source / "var/tmp/sample").write_bytes(sample) + (source / "outside").symlink_to(root / "private") + (root / "private").write_text("secret never read") + os.mkfifo(source / "pipe") + result = scan_filesystem(source, root / "results") + self.assertTrue(result["complete_scan"]) + self.assertIsNone(result["source_unchanged"]) + self.assertIsNone(result["archive_sha256"]) + self.assertEqual(result["counts"]["regular_files"], 1) + self.assertEqual(result["issues"], []) + rows = [ + json.loads(line) + for line in (root / "results/inventory.jsonl").read_text().splitlines() + ] + self.assertEqual(next(r for r in rows if r["path"] == "outside")["type"], "2") + findings = [ + json.loads(line) + for line in (root / "results/findings.jsonl").read_text().splitlines() + ] + self.assertTrue(any(r["rule"] == "known_network_indicator" for r in findings)) + selected = read_selected_members(source, {"var/tmp/sample", "outside"}) + self.assertEqual( + next(data for meta, data in selected if meta["path"] == "var/tmp/sample"), sample + ) + self.assertEqual( + next(data for meta, data in selected if meta["path"] == "outside"), b"" + ) + with self.assertRaises(ValueError): + scan_filesystem(source, source / "results") + + def test_network_indicator_token_boundaries(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "export" + source.mkdir() + (source / "a").write_text("149.104.78.1419 1149.104.78.141\n149.104.78.141:443\n") + scan_filesystem(source, root / "results") + hits = [ + json.loads(line) + for line in (root / "results/findings.jsonl").read_text().splitlines() + if json.loads(line)["rule"] == "known_network_indicator" + ] + self.assertEqual(len(hits), 1) + self.assertEqual(hits[0]["line"], 2) + + def test_published_persistence_rules_and_config_paths(self) -> None: + from netscaler_triage.filesystem import critical + + samples = { + "php_nonstandard_handler": "AddType application/x-httpd-php .rpm", + "php_engine_enabled": "php_flag engine on", + "web_alias_routing": "AliasMatch /vpn/media/example.ico /vendor/example.sig", + "css_webshell_alias": "LogonUISimple.html.style.min.ab12.css", + "nsmon_cron": "*/5 * * * * root perl /var/tmp/.nsmon/nsmon.pl", + "privileged_monitor_account": "bind system user sec_monitor superuser", + "command_channel_header": "X-UX-12", + "cron_scrubbing_reference": "# sed /vpn/scripts/linux /etc/crontab", + "shell_privilege_change": "# chmod 6555 /bin/sh", + "reverse_shell_reference": "# /dev/tcp/", + "configuration_staging_reference": ( + "# cat /flash/nsconfig/ns.conf > /var/netscaler/example" + ), + } + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "export" + source.mkdir() + for rule, sample in samples.items(): + (source / rule).write_text(sample + "\n") + scan_filesystem(source, root / "results") + hits = [ + json.loads(line) + for line in (root / "results/findings.jsonl").read_text().splitlines() + ] + for rule in samples: + self.assertTrue(any(h["path"] == rule and h["rule"] == rule for h in hits), rule) + self.assertTrue(all(h["classification"] == "review_candidate" for h in hits)) + for path in [ + "nsconfig/httpd.conf", + "flash/nsconfig/httpd.conf", + "nsconfig/crontab", + "var/spool/cron/root", + "etc/cron.d/custom", + "etc/master.passwd", + ]: + self.assertTrue(critical(path), path) + + def test_case_ioc_csv_and_artifact_matrix(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "export" + source.mkdir() + (source / "evidence").write_text("192.0.2.55 192.0.2.56\n") + iocs = root / "case.csv" + iocs.write_text( + "type,value,status\nip,192.0.2.55,observed_attempt\n" + "ip,192.0.2.56,own_test\npath,/missing-artifact,reference_only\n" + ) + result = scan_filesystem(source, root / "results", ioc_file=iocs) + self.assertEqual( + result["custom_ioc_sha256"], hashlib.sha256(iocs.read_bytes()).hexdigest() + ) + hits = [ + json.loads(line) + for line in (root / "results/findings.jsonl").read_text().splitlines() + ] + markers = {h.get("marker") for h in hits} + self.assertIn("192.0.2.55", markers) + self.assertNotIn("192.0.2.56", markers) + checks = json.loads((root / "results/artifact_checks.json").read_text()) + self.assertTrue(checks["scan_complete"]) + self.assertIn({"path": "missing-artifact", "present": False}, checks["paths"]) diff --git a/tests/test_release_hygiene.py b/tests/test_release_hygiene.py index c3ec797..2d87f3b 100644 --- a/tests/test_release_hygiene.py +++ b/tests/test_release_hygiene.py @@ -44,6 +44,8 @@ class ReleaseHygieneTests(unittest.TestCase): "RULES.txt", "VALIDATION.txt", "THREAT_INTELLIGENCE.txt", + "FILESYSTEM_COVERAGE.txt", + "scripts/analyze_filesystem_exports.py", "LICENSE.txt", "config/threat_intel.json", "uv.lock",