From 5a62c262c13a7e6e48bbbb0fdd64abb6d8327e94 Mon Sep 17 00:00:00 2001 From: ClaudioWayne <35531629+ClaudioWayne@users.noreply.github.com> Date: Wed, 23 Sep 2026 10:19:37 +0000 Subject: [PATCH 1/5] Restore Top Detections on the dashboard (#3254) --- web/dashboard/test_dashboard_scope.py | 79 ++++++++++++++++++++ web/dashboard/views.py | 11 +++ web/templates/dashboard/_top_detections.html | 18 +++++ web/templates/dashboard/index.html | 23 +----- 4 files changed, 112 insertions(+), 19 deletions(-) mode change 100644 => 100755 web/dashboard/test_dashboard_scope.py mode change 100644 => 100755 web/dashboard/views.py create mode 100755 web/templates/dashboard/_top_detections.html mode change 100644 => 100755 web/templates/dashboard/index.html diff --git a/web/dashboard/test_dashboard_scope.py b/web/dashboard/test_dashboard_scope.py old mode 100644 new mode 100755 index da43c868033..0017a2ec61d --- a/web/dashboard/test_dashboard_scope.py +++ b/web/dashboard/test_dashboard_scope.py @@ -6,6 +6,17 @@ +@pytest.fixture(autouse=True) +def dashboard_detections(monkeypatch): + """Keep dashboard view tests independent of the live report database.""" + from unittest.mock import Mock + import dashboard.views as dv + + detections = Mock(return_value=[]) + monkeypatch.setattr(dv, "top_detections", detections) + return detections + + @pytest.mark.django_db def test_dashboard_entitled_scopes(cape_db, mt_enabled, monkeypatch): from dashboard.views import entitled_scopes @@ -269,3 +280,71 @@ def fake_render(request, template, context=None): assert '
' in html # 'mine' has no done tasks -> its estimate alert is suppressed. assert "Mine —" not in html + + +@pytest.mark.django_db +@pytest.mark.parametrize("detections", [[], False, [{"family": "ExampleFamily", "total": 7}]]) +def test_index_global_top_detections(monkeypatch, mt_disabled, dashboard_detections, detections): + dashboard_detections.return_value = detections + user = User.objects.create_user("top-global") + context, html = _call_index(monkeypatch, user, _FakeDB()) + + dashboard_detections.assert_called_once_with() + assert context["panels"][0]["top_detections"] == detections + assert ("Top Detections" in html) == bool(detections) + if detections: + assert '/analysis/search/detections:ExampleFamily' in html + assert '>7' in html + + +@pytest.mark.django_db +def test_index_top_detections_disabled(monkeypatch, mt_disabled): + import dashboard.views as dv + import lib.cuckoo.common.web_utils as wu + from unittest.mock import Mock + + aggregate = Mock(side_effect=AssertionError("Disabled detections must not query MongoDB")) + monkeypatch.setattr(wu.web_cfg.general, "top_detections", False) + monkeypatch.setattr(wu, "mongo_aggregate", aggregate, raising=False) + monkeypatch.setattr(dv, "top_detections", wu.top_detections) + user = User.objects.create_user("top-disabled") + context, html = _call_index(monkeypatch, user, _FakeDB()) + + assert context["panels"][0]["top_detections"] is False + assert "Top Detections" not in html + aggregate.assert_not_called() + + +@pytest.mark.django_db +def test_index_scoped_top_detections(monkeypatch, mt_enabled, cape_db, dashboard_detections): + from users.models import Tenant, UserProfile + + tenant = Tenant.objects.create(slug="top-tenant", name="Top Tenant") + user = User.objects.create_user("top-scoped") + profile = UserProfile.objects.get(user=user) + profile.tenant = tenant + profile.save() + user = User.objects.get(pk=user.pk) + results = [ + [{"family": "PublicFamily", "total": 3}], + [{"family": "TenantFamily", "total": 2}], + [{"family": "PersonalFamily", "total": 1}], + ] + dashboard_detections.side_effect = results + context, html = _call_index(monkeypatch, user, _FakeDB()) + + assert [p["top_detections"] for p in context["panels"]] == results + calls = dashboard_detections.call_args_list + assert [call.kwargs["scope_match"] for call in calls] == [ + {"info.visibility": "public"}, + {"info.tenant_id": tenant.id, "info.visibility": "tenant"}, + {"info.user_id": user.id}, + ] + assert all(call.kwargs["viewer"].user_id == user.id for call in calls) + assert all(call.kwargs["viewer"].tenant_id == tenant.id for call in calls) + assert html.count("Top Detections") == 3 + for panel_html, detections in zip(html.split('
')[1:], results): + assert f'/analysis/search/detections:{detections[0]["family"]}' in panel_html + for other in results: + if other != detections: + assert other[0]["family"] not in panel_html diff --git a/web/dashboard/views.py b/web/dashboard/views.py old mode 100644 new mode 100755 index 9f54ccf5190..d971e307e29 --- a/web/dashboard/views.py +++ b/web/dashboard/views.py @@ -13,6 +13,8 @@ sys.path.append(settings.CUCKOO_PATH) +from lib.cuckoo.common.tenancy_optional import scope_match +from lib.cuckoo.common.web_utils import top_detections from lib.cuckoo.core.database import Database from lib.cuckoo.core.data.task import TASK_COMPLETED, TASK_REPORTED try: @@ -168,5 +170,14 @@ def index(request): "estimate_day": estimate_day, }) + for panel in panels: + if panel["scope"] == "global": + panel["top_detections"] = top_detections() + else: + panel["top_detections"] = top_detections( + scope_match=scope_match(panel["scope"], v), + viewer=v, + ) + data = {"title": "Dashboard", "panels": panels} return render(request, "dashboard/index.html", data) diff --git a/web/templates/dashboard/_top_detections.html b/web/templates/dashboard/_top_detections.html new file mode 100755 index 00000000000..463c48a968c --- /dev/null +++ b/web/templates/dashboard/_top_detections.html @@ -0,0 +1,18 @@ +{% if p.top_detections %} +
+
+
Top Detections
+
+
+
+ {% for block in p.top_detections %} + + {{block.total}} + {{block.family}} + + {% endfor %} +
+
+
+{% endif %} diff --git a/web/templates/dashboard/index.html b/web/templates/dashboard/index.html old mode 100644 new mode 100755 index 228f5a605b7..62ad083084d --- a/web/templates/dashboard/index.html +++ b/web/templates/dashboard/index.html @@ -25,6 +25,8 @@
+{% include "dashboard/_top_detections.html" with p=p only %} +
@@ -85,24 +87,7 @@
Analysis States {% else %}{% with p=panels.0 %} -{% if report.top_detections %} -
-
-
Top Detections
-
-
-
- {% for block in report.top_detections %} - - {{block.total}} - {{block.family}} - - {% endfor %} -
-
-
-{% endif %} +{% include "dashboard/_top_detections.html" with p=p only %}
@@ -170,4 +155,4 @@
Analysis States
-{% endwith %}{% endif %}{% endblock %} +{% endwith %}{% endif %}{% endblock %} \ No newline at end of file From dde6f4d596d82c1603f99c90655d40beaf26d5af Mon Sep 17 00:00:00 2001 From: Karl Hiramoto Date: Wed, 23 Sep 2026 12:20:31 +0200 Subject: [PATCH 2/5] Harden PyInstArchive against path traversal in _extractPyz and parseTOC (#3253) --- .../common/integrations/pyinstxtractor.py | 45 +++-- tests/test_pyinstxtractor_path_traversal.py | 169 ++++++++++++++++++ 2 files changed, 203 insertions(+), 11 deletions(-) create mode 100644 tests/test_pyinstxtractor_path_traversal.py diff --git a/lib/cuckoo/common/integrations/pyinstxtractor.py b/lib/cuckoo/common/integrations/pyinstxtractor.py index 53e69fbf103..13a0aa0ad99 100644 --- a/lib/cuckoo/common/integrations/pyinstxtractor.py +++ b/lib/cuckoo/common/integrations/pyinstxtractor.py @@ -227,17 +227,19 @@ def parseTOC(self): log.warning("[!] File name %s contains invalid bytes. Using random name %s", name, newName) name = newName - while "%" in name: + for _ in range(10): + if "%" not in name: + break new_name = urllib.parse.unquote(name) if new_name == name: break name = new_name # Prevent writing outside the extraction directory - if name.startswith("/"): - name = name.lstrip("/") - name = name.replace("\\", "/") - name = name.replace("..", "__") + name = name.replace("\0", "").replace("\\", "/") + name = name.replace("..", "__").lstrip("/") + parts = [p for p in name.split("/") if p and p not in (".", "..")] + name = "/".join(parts) if len(name) == 0: name = str(uniquename()) @@ -283,10 +285,11 @@ def extractFiles(self): os.mkdir(extractionDir) # os.chdir(extractionDir) + dest_root = os.path.realpath(self.destination_folder) for entry in self.tocList: - destination_entry = os.path.abspath(os.path.join(self.destination_folder, entry.name)) - if not destination_entry.startswith(os.path.abspath(self.destination_folder) + os.sep): + destination_entry = os.path.realpath(os.path.join(dest_root, entry.name)) + if not destination_entry.startswith(dest_root + os.sep): log.warning("[!] Path traversal attempt detected. Skipping %s", entry.name) continue self.fPtr.seek(entry.position, os.SEEK_SET) @@ -323,7 +326,7 @@ def extractFiles(self): # if we don't have the pyc header yet, fix them in a later pass self.barePycList.append(final_filename) - elif entry.typeCmprsData == (b"M", b"m") and not self.only_entrypoints: + elif entry.typeCmprsData in (b"M", b"m") and not self.only_entrypoints: # M -> ARCHIVE_ITEM_PYPACKAGE # m -> ARCHIVE_ITEM_PYMODULE # packages and modules are pyc files with their header intact @@ -379,6 +382,7 @@ def _extractPyz(self, name): # Create a directory for the contents of the pyz if not os.path.exists(dirName): os.mkdir(dirName) + dir_root = os.path.realpath(dirName) with open(name, "rb") as f: pyzMagic = f.read(4) @@ -427,12 +431,31 @@ def _extractPyz(self, name): # for Python > 3.3 some keys are bytes object some are str object fileName = fileName.decode("utf-8") - # Prevent writing outside dirName + fileName = str(fileName).replace("\0", "") + for _ in range(10): + if "%" not in fileName: + break + new_name = urllib.parse.unquote(fileName) + if new_name == fileName: + break + fileName = new_name + + # Prevent writing outside dirName (including leading '.', '/', or '\') + fileName = fileName.replace("\\", os.path.sep).replace("/", os.path.sep) fileName = fileName.replace("..", "__").replace(".", os.path.sep) + parts = [p for p in fileName.split(os.path.sep) if p and p not in (".", "..")] + if not parts: + parts = [str(uniquename())] + safe_rel = os.path.join(*parts) + if ispkg == 1: - filePath = os.path.join(dirName, fileName, "__init__.pyc") + filePath = os.path.realpath(os.path.join(dir_root, safe_rel, "__init__.pyc")) else: - filePath = os.path.join(dirName, fileName + ".pyc") + filePath = os.path.realpath(os.path.join(dir_root, safe_rel + ".pyc")) + + if not filePath.startswith(dir_root + os.sep): + log.warning("[!] Path traversal attempt detected in PYZ archive. Skipping %s", key) + continue fileDir = os.path.dirname(filePath) if not os.path.exists(fileDir): diff --git a/tests/test_pyinstxtractor_path_traversal.py b/tests/test_pyinstxtractor_path_traversal.py new file mode 100644 index 00000000000..e6364ce19e0 --- /dev/null +++ b/tests/test_pyinstxtractor_path_traversal.py @@ -0,0 +1,169 @@ +import io +import marshal +import os +import struct +import sys +import zlib + +from lib.cuckoo.common.integrations.pyinstxtractor import CTOCEntry, PyInstArchive + + +def _build_toc_entry(entry_pos: int, data: bytes, type_flag: bytes, name: bytes, compressed: int = 1) -> bytes: + cmprsd = zlib.compress(data) if compressed else data + header_size = struct.calcsize("!iIIIBc") + name_padded = name + b"\x00" + entry_size = header_size + len(name_padded) + return struct.pack( + f"!iIIIBc{len(name_padded)}s", + entry_size, + entry_pos, + len(cmprsd), + len(data), + compressed, + type_flag, + name_padded, + ) + + +def test_parse_toc_and_extract_files_path_traversal(tmp_path): + dest_dir = tmp_path / "extracted" + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + + payload = b"MARSHALED_CODE_PAYLOAD" + compressed_payload = zlib.compress(payload) + + entries_raw = b"".join( + [ + _build_toc_entry(0, payload, b"s", b"../../../../outside/pwn_rel"), + _build_toc_entry(0, payload, b"s", b"%2e%2e/%2e%2e/outside/pwn_url"), + _build_toc_entry(0, payload, b"s", b"\\..\\..\\outside\\pwn_win"), + _build_toc_entry(0, payload, b"s", str(outside_dir / "pwn_abs").encode()), + _build_toc_entry(0, payload, b"M", b"pkg/module_ok"), + ] + ) + + stream = io.BytesIO(compressed_payload + entries_raw) + arch = PyInstArchive( + { + "file": str(tmp_path / "dummy.exe"), + "destination_folder": str(dest_dir), + "entry_points": False, + } + ) + arch.fPtr = stream + arch.overlayPos = 0 + arch.tableOfContentsPos = len(compressed_payload) + arch.tableOfContentsSize = len(entries_raw) + arch.pymaj = sys.version_info.major + arch.pymin = sys.version_info.minor + arch.pycMagic = b"\x42\x0d\x0d\x0a" + + arch.parseTOC() + arch.extractFiles() + + # Verify nothing was written to outside_dir + assert list(outside_dir.iterdir()) == [] + + # Verify all created files reside strictly under dest_dir + dest_real = os.path.realpath(str(dest_dir)) + extracted_files = [] + for root, _, files in os.walk(dest_dir): + for fname in files: + full_path = os.path.realpath(os.path.join(root, fname)) + assert full_path.startswith(dest_real + os.sep) + extracted_files.append(os.path.relpath(full_path, dest_real)) + + assert "pkg/module_ok.pyc" in extracted_files + + +def test_extract_pyz_leading_dot_and_slash_traversal(tmp_path): + """Verify _extractPyz cannot escape dirName when TOC keys start with '.', '/', or '\\'.""" + dest_dir = tmp_path / "extracted" + dest_dir.mkdir() + outside_target = tmp_path / "escaped_from_pyz" + outside_target.mkdir() + + payload = zlib.compress(b"PYZ_BYTECODE") + # Build a synthetic PYZ archive in-memory on disk + pyz_path = dest_dir / "archive.pyz" + + # Header: b"PYZ\0" (4B) + magic (4B) + toc_pos (4B) + payload + marshaled TOC + data_offset = 12 + # Before this fix, a leading single dot like ".tmp.pwn" turned into "/tmp/pwn" via .replace(".", os.path.sep), + # causing os.path.join(dirName, fileName + ".pyc") to discard dirName and write to /tmp/pwn.pyc! + leading_dot_key = "." + str(outside_target).lstrip("/").replace("/", ".") + ".dot_escape" + leading_slash_key = str(outside_target / "slash_escape") + leading_backslash_key = "\\" + str(outside_target / "backslash_escape").lstrip("/").replace("/", "\\") + + toc_dict = { + leading_dot_key.encode(): (0, data_offset, len(payload)), + leading_slash_key.encode(): (0, data_offset, len(payload)), + leading_backslash_key.encode(): (1, data_offset, len(payload)), + b"../../outside/rel_escape": (0, data_offset, len(payload)), + b"legit.pkg.submodule": (0, data_offset, len(payload)), + } + marshaled_toc = marshal.dumps(toc_dict) + toc_pos = data_offset + len(payload) + + with open(pyz_path, "wb") as f: + f.write(b"PYZ\0") + f.write(b"\x42\x0d\x0d\x0a") + f.write(struct.pack("!i", toc_pos)) + f.write(payload) + f.write(marshaled_toc) + + arch = PyInstArchive( + { + "file": str(tmp_path / "dummy.exe"), + "destination_folder": str(dest_dir), + "entry_points": False, + } + ) + arch.pymaj = sys.version_info.major + arch.pymin = sys.version_info.minor + arch.pycMagic = b"\x42\x0d\x0d\x0a" + + arch._extractPyz(str(pyz_path)) + + # Ensure nothing escaped into outside_target + assert list(outside_target.iterdir()) == [] + + # Ensure everything written is strictly inside archive.pyz_extracted + pyz_extracted_dir = pyz_path.parent / "archive.pyz_extracted" + assert pyz_extracted_dir.is_dir() + assert (pyz_extracted_dir / "legit" / "pkg" / "submodule.pyc").is_file() + + +def test_extract_files_blocks_symlink_escape(tmp_path): + dest_dir = tmp_path / "extracted" + dest_dir.mkdir() + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + + # Create a symlink inside dest_dir pointing to outside_dir + symlink_dir = dest_dir / "sublink" + symlink_dir.symlink_to(outside_dir, target_is_directory=True) + + payload = b"SYMLINK_PAYLOAD" + compressed = zlib.compress(payload) + + arch = PyInstArchive( + { + "file": str(tmp_path / "dummy.exe"), + "destination_folder": str(dest_dir), + "entry_points": True, + } + ) + arch.fPtr = io.BytesIO(compressed) + arch.pymaj = sys.version_info.major + arch.pymin = sys.version_info.minor + arch.pycMagic = b"\x42\x0d\x0d\x0a" + arch.tocList = [ + CTOCEntry(0, len(compressed), len(payload), 1, b"s", "sublink/escaped_via_symlink"), + ] + + arch.extractFiles() + + # Verify symlink escape was blocked and nothing was written to outside_dir + assert list(outside_dir.iterdir()) == [] From 55ff0d10d602da392399758f366a93c863f8600f Mon Sep 17 00:00:00 2001 From: Karl Hiramoto Date: Wed, 23 Sep 2026 12:24:00 +0200 Subject: [PATCH 3/5] Fix NFS mount path traversal and restrict NFS hosts to private networks (#3252) * Fix arbitrary NFS mount path traversal in fstab.py and dist.py * Restrict NFS mounts to private networks (ip_address.is_private) and allowed_networks CIDRs * Address review feedback on remove_nfs_entry, nfs_host, and IP type annotations --- conf/default/distributed.conf.default | 7 ++ tests/test_fstab_path_traversal.py | 168 ++++++++++++++++++++++++++ utils/dist.py | 88 +++++++++++++- utils/fstab.py | 76 +++++++++++- 4 files changed, 329 insertions(+), 10 deletions(-) create mode 100644 tests/test_fstab_path_traversal.py diff --git a/conf/default/distributed.conf.default b/conf/default/distributed.conf.default index f35d5feacf4..1b6d3badd2f 100644 --- a/conf/default/distributed.conf.default +++ b/conf/default/distributed.conf.default @@ -29,6 +29,13 @@ submit_only = no # Path will be $CAPE_ROOT/$nfs_mount_folder. Ex: /opt/CAPEv2/workers mount_folder = workers fstab_socket = /tmp/cape-fstab +# Optional comma-separated list of private CIDR subnets allowed for NFS mounts +# (e.g., 10.0.0.0/8, 192.168.1.0/24). When empty, any private network IP +# (ip_address.is_private, excluding loopback/link-local) is allowed. +# Note: When nfs = yes, POST /node rejects registration (HTTP 400) if the NFS +# host does not resolve to a private IP (or pass a separate private `nfs_host` +# in the POST /node payload if the node `url` uses a public hostname). +allowed_networks = # Google Cloud Platform [GCP] diff --git a/tests/test_fstab_path_traversal.py b/tests/test_fstab_path_traversal.py new file mode 100644 index 00000000000..633658aa8b1 --- /dev/null +++ b/tests/test_fstab_path_traversal.py @@ -0,0 +1,168 @@ +from unittest.mock import patch +import os +import pytest + +from utils import fstab + + +@pytest.mark.parametrize( + "malicious_folder", + [ + "/etc", + "/etc/cron.d", + "../../etc", + "../workers_evil", + "worker/../../etc", + ".", + "..", + "", + "-o", + "worker\nattacker:/opt/CAPEv2 /etc nfs defaults 0 0", + "worker space", + ], +) +def test_fstab_rejects_path_traversal_worker_folder(malicious_folder, tmp_path): + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_mount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + with pytest.raises(ValueError): + fstab.add_nfs_entry("10.0.0.1", malicious_folder) + mock_write.assert_not_called() + mock_mount.assert_not_called() + + +@pytest.mark.parametrize( + "malicious_hostname", + [ + "/etc", + "../../etc", + "attacker.com /etc nfs defaults 0 0 #", + "attacker.com\n10.0.0.2:/opt/CAPEv2 /etc", + "-flag", + "", + ".", + "..", + "8.8.8.8", + "1.1.1.1", + "127.0.0.1", + "::1", + "169.254.169.254", + "0.0.0.0", + "2001:4860:4860::8888", + ], +) +def test_fstab_rejects_invalid_or_non_private_hostname(malicious_hostname, tmp_path): + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_mount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + fstab.dist_conf.NFS.allowed_networks = "" + with pytest.raises(ValueError): + fstab.add_nfs_entry(malicious_hostname, "worker1") + with pytest.raises(ValueError): + fstab.remove_nfs_entry(malicious_hostname, "worker1") + mock_write.assert_not_called() + mock_mount.assert_not_called() + + +def test_fstab_rejects_domain_resolving_to_public_ip(tmp_path): + fake_addrinfo = [(2, 1, 6, "", ("93.184.216.34", 0))] + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab.socket, "getaddrinfo", return_value=fake_addrinfo), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_mount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + fstab.dist_conf.NFS.allowed_networks = "" + with pytest.raises(ValueError, match="non-private IP"): + fstab.add_nfs_entry("attacker.com", "worker1") + mock_write.assert_not_called() + mock_mount.assert_not_called() + + +def test_fstab_enforces_allowed_networks_cidr(tmp_path): + workers_dir = tmp_path / "workers" + workers_dir.mkdir() + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab, "path_mount_point", return_value=False), + patch.object(fstab, "path_read_file", return_value="# /etc/fstab\n"), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_mount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + fstab.dist_conf.NFS.allowed_networks = "10.128.0.0/16, 172.16.10.0/24" + + # Private IP outside allowed_networks is rejected + with pytest.raises(ValueError, match="outside configured allowed_networks"): + fstab.add_nfs_entry("192.168.1.10", "cape-worker-1") + mock_write.assert_not_called() + + # Private IP inside allowed_networks is accepted + fstab.add_nfs_entry("10.128.1.25", "cape-worker-1") + mock_write.assert_called_once() + mock_mount.assert_called_once() + + +def test_fstab_allows_valid_worker_within_mount_folder(tmp_path): + workers_dir = tmp_path / "workers" + workers_dir.mkdir() + expected_worker_path = os.path.realpath(str(workers_dir / "cape-worker-1")) + + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab, "path_mount_point", return_value=False), + patch.object(fstab, "path_read_file", return_value="# /etc/fstab\n"), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_mount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + fstab.dist_conf.NFS.allowed_networks = "" + fstab.add_nfs_entry("192.168.1.10", "cape-worker-1") + + mock_write.assert_called_once() + written_fstab = mock_write.call_args[0][1] + assert f"192.168.1.10:/opt/CAPEv2 {expected_worker_path} nfs" in written_fstab + mock_mount.assert_called_once_with(["mount", expected_worker_path]) + + +def test_remove_nfs_entry_requires_worker_folder_and_uses_exact_prefix_match(tmp_path): + workers_dir = tmp_path / "workers" + workers_dir.mkdir() + worker1_path = os.path.realpath(str(workers_dir / "cape-worker-1")) + worker10_path = os.path.realpath(str(workers_dir / "cape-worker-10")) + + # Calling without worker_folder raises TypeError + with pytest.raises(TypeError): + fstab.remove_nfs_entry("10.0.0.1") + + initial_fstab = "\n".join( + [ + "# /etc/fstab", + f"10.0.0.10:/opt/CAPEv2 {worker10_path} nfs _netdev 0 0", + f"10.0.0.1:/opt/CAPEv2 {worker1_path} nfs _netdev 0 0", + ] + ) + + with ( + patch.object(fstab, "CUCKOO_ROOT", str(tmp_path)), + patch.object(fstab, "path_read_file", return_value=initial_fstab), + patch.object(fstab, "path_write_file") as mock_write, + patch.object(fstab.subprocess, "check_output") as mock_umount, + ): + fstab.dist_conf.NFS.mount_folder = "workers" + fstab.dist_conf.NFS.allowed_networks = "" + fstab.remove_nfs_entry("10.0.0.1", "cape-worker-1") + + mock_write.assert_called_once() + updated_fstab = mock_write.call_args[0][1] + # 10.0.0.10 entry must remain untouched (no prefix collision) + assert f"10.0.0.10:/opt/CAPEv2 {worker10_path} nfs" in updated_fstab + assert f"10.0.0.1:/opt/CAPEv2 {worker1_path} nfs" not in updated_fstab + mock_umount.assert_called_once_with(["umount", worker1_path]) diff --git a/utils/dist.py b/utils/dist.py index 5d43f9266e8..efc6f45e9c7 100644 --- a/utils/dist.py +++ b/utils/dist.py @@ -6,11 +6,13 @@ import argparse import hashlib +import ipaddress import logging import os import re import queue import shutil +import socket import sys import threading import time @@ -328,6 +330,60 @@ def node_get_report(task_id, fmt, url, apikey, stream=False): log.critical("Error fetching report (task #%d, node %s): %s", task_id, url, e) +VALID_NODE_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$") + + +def validate_node_name(name: str) -> str: + if ( + not isinstance(name, str) + or not VALID_NODE_NAME_RE.fullmatch(name) + or name in (".", "..") + or name.startswith("-") + ): + raise ValueError(f"Invalid node name: {name!r}") + return name + + +def _is_private_nfs_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + return bool( + ip.is_private + and not (ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_multicast) + ) + + +def validate_nfs_hostname(hostname: str) -> str: + if not isinstance(hostname, str) or not hostname or hostname in (".", "..") or hostname.startswith("-"): + raise ValueError(f"Invalid NFS hostname: {hostname!r}") + + clean_host = hostname.strip("[]") + try: + ip = ipaddress.ip_address(clean_host) + except ValueError: + if not VALID_NODE_NAME_RE.fullmatch(clean_host): + raise ValueError(f"Invalid NFS hostname: {hostname!r}") + try: + addr_info = socket.getaddrinfo(clean_host, None, proto=socket.IPPROTO_TCP) + except OSError as exc: + raise ValueError(f"Unable to resolve NFS hostname {hostname!r}: {exc}") from exc + if not addr_info: + raise ValueError(f"Unable to resolve NFS hostname: {hostname!r}") + resolved_ips = [ipaddress.ip_address(info[4][0]) for info in addr_info] + if not all(_is_private_nfs_ip(resolved_ip) for resolved_ip in resolved_ips): + raise ValueError(f"NFS hostname {hostname!r} resolves to a non-private IP address") + ip = resolved_ips[0] + + if not _is_private_nfs_ip(ip): + raise ValueError(f"NFS host {hostname!r} ({ip}) is not a private network IP address") + + allowed_cidrs = getattr(dist_conf.NFS, "allowed_networks", "") or "" + if allowed_cidrs.strip(): + networks = [ipaddress.ip_network(cidr.strip(), strict=False) for cidr in allowed_cidrs.split(",") if cidr.strip()] + if networks and not any(ip in net for net in networks): + raise ValueError(f"NFS host {hostname!r} ({ip}) is outside configured allowed_networks") + + return f"[{ip}]" if ip.version == 6 else str(ip) + + def node_get_report_nfs(task_id, worker_name, main_task_id) -> bool: """ Retrieves a report from a worker node via NFS and copies it to the main task's analysis directory. @@ -346,7 +402,11 @@ def node_get_report_nfs(task_id, worker_name, main_task_id) -> bool: Logs: Error messages if the worker node is not mounted, the file does not exist, or if there is an exception during copying. """ - worker_path = os.path.join(CUCKOO_ROOT, dist_conf.NFS.mount_folder, str(worker_name)) + worker_name = validate_node_name(str(worker_name)) + base_dir = os.path.realpath(os.path.join(CUCKOO_ROOT, dist_conf.NFS.mount_folder)) + worker_path = os.path.realpath(os.path.join(base_dir, worker_name)) + if os.path.commonpath([base_dir, worker_path]) != base_dir or worker_path == base_dir: + raise ValueError(f"Worker path escapes mount_folder: {worker_name!r}") if not path_mount_point(worker_path): log.error("[-] Worker: %s is not mounted to: %s!", worker_name, worker_path) @@ -1849,6 +1909,7 @@ class NodeRegister(BaseModel): url: str apikey: str = "" enabled: Optional[bool] = None + nfs_host: Optional[str] = None class NodeUpdate(BaseModel): url: Optional[str] = None @@ -1882,6 +1943,20 @@ def get_nodes(): @app.post("/node") def post_node(payload: NodeRegister): + try: + validate_node_name(payload.name) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + nfs_host = None + if NFS_FETCH: + hostname = payload.nfs_host or urlparse(payload.url).hostname or urlparse(payload.url).netloc.split(":")[0] + if hostname != main_server_name: + try: + nfs_host = validate_nfs_hostname(hostname) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + with session() as db: node_exist = False # On autoscaling we might get the same name but different IP for server. Kinda PUT friendly POST @@ -1919,11 +1994,9 @@ def post_node(payload: NodeRegister): db.add(node) db.commit() - if NFS_FETCH: + if NFS_FETCH and nfs_host: # Add entry to /etc/fstab, create folder and mount server - hostname = urlparse(payload.url).netloc.split(":")[0] - if hostname != main_server_name: - send_socket_command(dist_conf.NFS.fstab_socket, "add_entry", *[hostname, payload.name]) + send_socket_command(dist_conf.NFS.fstab_socket, "add_entry", *[nfs_host, payload.name]) return dict(name=payload.name, machines=machines, exitnodes=exitnodes) @@ -2159,6 +2232,11 @@ def register_node_cli(name, url, apikey, enabled): if not name or not url: print("Error: Registering a node requires both --node and --url ") sys.exit(1) + try: + validate_node_name(name) + except ValueError as exc: + print(f"Error: {exc}") + sys.exit(1) with session() as db: node_exist = False node = db.scalar(select(Node).where(Node.name == name)) diff --git a/utils/fstab.py b/utils/fstab.py index 2fc8fd29805..31d54db30df 100644 --- a/utils/fstab.py +++ b/utils/fstab.py @@ -8,9 +8,11 @@ import argparse import errno import grp +import ipaddress import json import logging.handlers import os +import re import signal import socket import stat @@ -38,9 +40,68 @@ log.addHandler(ch) log.setLevel(logging.INFO) +VALID_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$") + + +def _is_private_nfs_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + return bool( + ip.is_private + and not (ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_multicast) + ) + + +def _validate_hostname(hostname: str) -> str: + if not isinstance(hostname, str) or not hostname or hostname in (".", "..") or hostname.startswith("-"): + raise ValueError(f"Invalid NFS hostname: {hostname!r}") + + clean_host = hostname.strip("[]") + try: + ip = ipaddress.ip_address(clean_host) + except ValueError: + if not VALID_NAME_RE.fullmatch(clean_host): + raise ValueError(f"Invalid NFS hostname: {hostname!r}") + try: + addr_info = socket.getaddrinfo(clean_host, None, proto=socket.IPPROTO_TCP) + except OSError as exc: + raise ValueError(f"Unable to resolve NFS hostname {hostname!r}: {exc}") from exc + if not addr_info: + raise ValueError(f"Unable to resolve NFS hostname: {hostname!r}") + resolved_ips = [ipaddress.ip_address(info[4][0]) for info in addr_info] + if not all(_is_private_nfs_ip(resolved_ip) for resolved_ip in resolved_ips): + raise ValueError(f"NFS hostname {hostname!r} resolves to a non-private IP address") + ip = resolved_ips[0] + + if not _is_private_nfs_ip(ip): + raise ValueError(f"NFS host {hostname!r} ({ip}) is not a private network IP address") + + allowed_cidrs = getattr(dist_conf.NFS, "allowed_networks", "") or "" + if allowed_cidrs.strip(): + networks = [ipaddress.ip_network(cidr.strip(), strict=False) for cidr in allowed_cidrs.split(",") if cidr.strip()] + if networks and not any(ip in net for net in networks): + raise ValueError(f"NFS host {hostname!r} ({ip}) is outside configured allowed_networks") + + return f"[{ip}]" if ip.version == 6 else str(ip) + + +def _resolve_worker_path(worker_folder: str) -> str: + if ( + not isinstance(worker_folder, str) + or not VALID_NAME_RE.fullmatch(worker_folder) + or worker_folder in (".", "..") + or worker_folder.startswith("-") + ): + raise ValueError(f"Invalid worker folder name: {worker_folder!r}") + + base_dir = os.path.realpath(os.path.join(CUCKOO_ROOT, dist_conf.NFS.mount_folder)) + worker_path = os.path.realpath(os.path.join(base_dir, worker_folder)) + if os.path.commonpath([base_dir, worker_path]) != base_dir or worker_path == base_dir: + raise ValueError(f"Worker mount path escapes mount_folder: {worker_folder!r}") + return worker_path + def add_nfs_entry(hostname: str, worker_folder: str): - worker_path = os.path.abspath(os.path.join(CUCKOO_ROOT, dist_conf.NFS.mount_folder, worker_folder)) + hostname = _validate_hostname(hostname) + worker_path = _resolve_worker_path(worker_folder) if not path_exists(worker_path): path_mkdir(worker_path, parent=True, mode=0o755) @@ -52,7 +113,11 @@ def add_nfs_entry(hostname: str, worker_folder: str): # new line strip if fstab[-1] == "": fstab = fstab[:-1] - if any(hostname in entry for entry in fstab if not entry.startswith("#")): + if any( + entry.startswith(f"{hostname}:") or f" {worker_path} nfs " in entry + for entry in fstab + if not entry.startswith("#") + ): return # hostname:/opt/CAPEv2 /opt/CAPEv2/2 nfs _netdev,nofail,noatime,nolock,intr,tcp,actimeo=1800,x-systemd.automount,x-systemd.mount-timeout=30s 0 0 @@ -65,13 +130,14 @@ def add_nfs_entry(hostname: str, worker_folder: str): print("add_nfs_entry error on mount: %s", str(e)) -def remove_nfs_entry(hostname: str): - worker_path = os.path.join(CUCKOO_ROOT, dist_conf.NFS.mount_folder, hostname) +def remove_nfs_entry(hostname: str, worker_folder: str): + hostname = _validate_hostname(hostname) + worker_path = _resolve_worker_path(worker_folder) with lock: fstab = path_read_file("/etc/fstab", mode="text").split("\n") for entry in fstab: - if entry.startswith(hostname) and " nfs " in entry: + if entry.startswith(f"{hostname}:") and f" {worker_path} nfs " in entry: fstab.remove(entry) _ = path_write_file("/etc/fstab", "\n".join(fstab), mode="text") break From d199f22c2a7abcd49e12a4b86278522dc729a92f Mon Sep 17 00:00:00 2001 From: ClaudioWayne <35531629+ClaudioWayne@users.noreply.github.com> Date: Wed, 23 Sep 2026 11:05:49 +0000 Subject: [PATCH 4/5] The report views MongoDB projection omitted url_analysis, leaving the submitted URL unavailable to the template even though it existed in the database. (#3255) Add url_analysis to the projection so URL analysis reports display the submitted URL again. --- web/analysis/views.py | 1 + 1 file changed, 1 insertion(+) diff --git a/web/analysis/views.py b/web/analysis/views.py index 9641c11fe0e..2c8c6d41ac5 100644 --- a/web/analysis/views.py +++ b/web/analysis/views.py @@ -2883,6 +2883,7 @@ def report(request, task_id): "info": 1, "target": 1, "signatures": 1, + "url_analysis": 1, "malscore": 1, "malstatus": 1, "detections": 1, From 6b15e76f28cd83fd1187da6c9aaaad2174255356 Mon Sep 17 00:00:00 2001 From: doomedraven Date: Wed, 23 Sep 2026 13:41:52 +0200 Subject: [PATCH 5/5] remove mcp as it has it own dedicated place now as CAPE-mcp --- mcp/__init__.py | 0 mcp/filters.py | 31 --- mcp/server.py | 642 ------------------------------------------------ 3 files changed, 673 deletions(-) delete mode 100644 mcp/__init__.py delete mode 100644 mcp/filters.py delete mode 100644 mcp/server.py diff --git a/mcp/__init__.py b/mcp/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 diff --git a/mcp/filters.py b/mcp/filters.py deleted file mode 100644 index faa36ef074a..00000000000 --- a/mcp/filters.py +++ /dev/null @@ -1,31 +0,0 @@ -# Configuration for MCP server search filters -# You can modify this dictionary to include or exclude specific fields in the lean report -# Injested by Agents to give a quick overview - -lean_search_filters = { - "info": 1, - "virustotal_summary": 1, - "detections.family": 1, - "malfamily": 1, - "malfamily_tag": 1, - "malscore": 1, - "network.pcap_sha256": 1, - "network.domains.domain": 1, - "network.http.uri": 1, - "signatures.name": 1, - "signatures.description": 1, - "signatures.severity": 1, - "CAPE": 1, - "behavior.summary.mutexes": 1, - "behavior.summary.executed_commands": 1, - "mlist_cnt": 1, - "f_mlist_cnt": 1, - "target.file.clamav": 1, - "target.file.sha256": 1, - "suri_tls_cnt": 1, - "suri_alert_cnt": 1, - "suri_http_cnt": 1, - "suri_file_cnt": 1, - "trid": 1, - "_id": 0, -} diff --git a/mcp/server.py b/mcp/server.py deleted file mode 100644 index fd5f7d41699..00000000000 --- a/mcp/server.py +++ /dev/null @@ -1,642 +0,0 @@ -import json -import os -import sys -import mimetypes -import re -from typing import Any, Dict - -# Ensure CAPE root is in path for lib imports -CAPE_ROOT = os.path.join(os.path.abspath(os.path.dirname(__file__)), "..") -sys.path.append(CAPE_ROOT) - -try: - import httpx - from fastmcp import FastMCP -except ImportError: - sys.exit("poetry run pip install .[mcp]") - -try: - from lib.cuckoo.common.config import Config - from lib.cuckoo.common.web_utils import ( - search_term_map, - perform_search_filters, - hash_searches, - normalized_lower_terms, - ) -except ImportError: - sys.exit("Could not import lib.cuckoo.common.config. Ensure you are running from CAPE root.") - -# Initialize CAPE Config -api_config = Config("api") - -# Configuration from Environment or Config File -# Run with: CAPE_API_URL=http://127.0.0.1:8000/apiv2 CAPE_API_TOKEN=your_token poetry run python mcp/server.py -API_URL = os.environ.get("CAPE_API_URL") -if not API_URL: - # Try to get from api.conf [api] url - try: - base_url = api_config.api.url.rstrip("/") - API_URL = f"{base_url}/apiv2" - except AttributeError: - API_URL = "http://127.0.0.1:8000/apiv2" - -API_TOKEN = os.environ.get("CAPE_API_TOKEN", "") - -# Proactively map enabled MCP tools. Default is NO. -ENABLED_MCP_TOOLS = set() -for section_name in api_config.get_config(): - if section_name == "api": - continue - try: - section = api_config.get(section_name) - if getattr(section, "mcp", False): - ENABLED_MCP_TOOLS.add(section_name) - except Exception: - continue - -def check_mcp_enabled(section: str) -> bool: - """Check if a specific section is enabled for MCP.""" - return section in ENABLED_MCP_TOOLS - -def mcp_tool(section: str): - """ - Conditional decorator that only registers the tool with FastMCP - if the corresponding section is enabled in api.conf. - """ - def decorator(func): - if check_mcp_enabled(section): - return mcp.tool()(func) - return func - return decorator - -def is_auth_required() -> bool: - """Check if token authorization is enabled globally.""" - try: - return api_config.api.token_auth_enabled - except AttributeError: - return False - -# Startup Check: Warn if Auth is enabled but no default token is provided -if is_auth_required() and not API_TOKEN: - print("WARNING: Token authentication is enabled in api.conf, but CAPE_API_TOKEN is not set.", file=sys.stderr) - print(" All MCP tool calls must include a valid 'token' argument.", file=sys.stderr) - -# Initialize FastMCP -mcp = FastMCP("cape-sandbox") - -# Security: Restrict file submission to a specific directory -# Defaults to current working directory if not set -ALLOWED_SUBMISSION_DIR = os.environ.get("CAPE_ALLOWED_SUBMISSION_DIR", os.getcwd()) - -def get_headers(token: str = "") -> Dict[str, str]: - headers = {} - auth_token = token if token else API_TOKEN - - if auth_token: - headers["Authorization"] = f"Token {auth_token}" - return headers - -async def _request(method: str, endpoint: str, token: str = "", **kwargs) -> Any: - # Auth Check - if is_auth_required(): - auth_token = token if token else API_TOKEN - if not auth_token: - return {"error": True, "message": "Authentication required but no token provided."} - - url = f"{API_URL.rstrip('/')}/{endpoint.lstrip('/')}" - async with httpx.AsyncClient() as client: - try: - response = await client.request(method, url, headers=get_headers(token), **kwargs) - # We don't raise_for_status immediately to handle API errors gracefully in JSON - if response.status_code >= 400: - try: - return response.json() - except json.JSONDecodeError: - return {"error": True, "message": f"HTTP {response.status_code}", "body": response.text} - - try: - return response.json() - except json.JSONDecodeError: - return {"error": False, "data": response.text} - except httpx.HTTPStatusError as e: - return {"error": True, "message": str(e), "body": e.response.text} - except Exception as e: - return {"error": True, "message": str(e)} - -async def _download_file(endpoint: str, destination: str, default_filename: str = "downloaded_file.bin", token: str = "") -> str: - """Helper to download a file from an API endpoint.""" - # Auth Check - if is_auth_required(): - auth_token = token if token else API_TOKEN - if not auth_token: - return json.dumps({"error": True, "message": "Authentication required but no token provided."}, indent=2) - - if not os.path.isdir(destination): - return json.dumps({"error": True, "message": "Destination directory does not exist"}) - - url = f"{API_URL.rstrip('/')}/{endpoint.lstrip('/')}" - headers = get_headers(token) - - async with httpx.AsyncClient() as client: - try: - async with client.stream("GET", url, headers=headers) as response: - if response.status_code != 200: - content = await response.read() - return json.dumps({"error": True, "message": f"HTTP {response.status_code}", "body": content.decode('utf-8', errors='ignore')}, indent=2) - - filename = default_filename - content_disposition = response.headers.get("content-disposition") - if content_disposition: - match = re.search(r'filename="?([^"]+)"?', content_disposition) - if match: - filename = os.path.basename(match.group(1)) - - filepath = os.path.join(destination, filename) - - with open(filepath, "wb") as f: - async for chunk in response.aiter_bytes(): - f.write(chunk) - - return json.dumps({"error": False, "message": f"Saved to {filepath}", "path": filepath}, indent=2) - except Exception as e: - return json.dumps({"error": True, "message": str(e)}, indent=2) - -def _build_submission_data(**kwargs) -> Dict[str, str]: - """Helper to build submission data dictionary, handling type conversions.""" - data = {} - for key, value in kwargs.items(): - # Skip empty values (None, "", 0, False) to match original behavior - if not value: - continue - - if isinstance(value, bool): - data[key] = "1" - elif isinstance(value, int): - data[key] = str(value) - else: - data[key] = value - return data - -# --- Tasks Creation --- - -@mcp_tool("filecreate") -async def submit_file( - file_path: str, - machine: str = "", - package: str = "", - options: str = "", - tags: str = "", - priority: int = 1, - timeout: int = 0, - platform: str = "", - memory: bool = False, - enforce_timeout: bool = False, - clock: str = "", - custom: str = "", - token: str = "" -) -> str: - """ - Submit a local file for analysis. - """ - # Auth Check (Manual check needed here because we stream file) - if is_auth_required(): - auth_token = token if token else API_TOKEN - if not auth_token: - return json.dumps({"error": True, "message": "Authentication required but no token provided."}) - - if not os.path.exists(file_path): - return json.dumps({"error": True, "message": "File not found"}) - - # Security check: Ensure file is within allowed directory - abs_file_path = os.path.abspath(file_path) - abs_allowed_dir = os.path.abspath(ALLOWED_SUBMISSION_DIR) - - if not abs_file_path.startswith(abs_allowed_dir): - return json.dumps({ - "error": True, - "message": f"Security Violation: File submission is restricted to {abs_allowed_dir}" - }) - - filename = os.path.basename(file_path) - mime_type, _ = mimetypes.guess_type(file_path) - if not mime_type: - mime_type = "application/octet-stream" - - data = _build_submission_data( - machine=machine, package=package, options=options, tags=tags, - priority=priority, timeout=timeout, platform=platform, - memory=memory, enforce_timeout=enforce_timeout, clock=clock, - custom=custom - ) - - url = f"{API_URL.rstrip('/')}/tasks/create/file/" - - async with httpx.AsyncClient() as client: - try: - with open(file_path, "rb") as f: - files = {"file": (filename, f, mime_type)} - response = await client.post(url, data=data, files=files, headers=get_headers(token)) - try: - result = response.json() - except json.JSONDecodeError: - result = {"error": response.status_code >= 400, "data": response.text} - except Exception as e: - result = {"error": True, "message": str(e)} - - return json.dumps(result, indent=2) - -@mcp_tool("urlcreate") -async def submit_url( - url: str, - machine: str = "", - package: str = "", - options: str = "", - tags: str = "", - priority: int = 1, - timeout: int = 0, - platform: str = "", - memory: bool = False, - enforce_timeout: bool = False, - clock: str = "", - custom: str = "", - token: str = "" -) -> str: - """Submit a URL for analysis.""" - data = {"url": url} - data.update(_build_submission_data( - machine=machine, package=package, options=options, tags=tags, - priority=priority, timeout=timeout, platform=platform, - memory=memory, enforce_timeout=enforce_timeout, clock=clock, - custom=custom - )) - - result = await _request("POST", "tasks/create/url/", token=token, data=data) - return json.dumps(result, indent=2) - -@mcp_tool("dlnexeccreate") -async def submit_dlnexec( - url: str, - machine: str = "", - package: str = "", - options: str = "", - tags: str = "", - priority: int = 1, - token: str = "" -) -> str: - """Submit a URL for Download & Execute analysis.""" - data = {"dlnexec": url} - data.update(_build_submission_data( - machine=machine, package=package, options=options, tags=tags, priority=priority - )) - - result = await _request("POST", "tasks/create/dlnexec/", token=token, data=data) - return json.dumps(result, indent=2) - -@mcp_tool("staticextraction") -async def submit_static( - file_path: str, - priority: int = 1, - options: str = "", - token: str = "" -) -> str: - """Submit a file for static extraction only.""" - # Auth Check (Manual check needed here because we stream file) - if is_auth_required(): - auth_token = token if token else API_TOKEN - if not auth_token: - return json.dumps({"error": True, "message": "Authentication required but no token provided."}) - - if not os.path.exists(file_path): - return json.dumps({"error": True, "message": "File not found"}) - - # Security check: Ensure file is within allowed directory - abs_file_path = os.path.abspath(file_path) - abs_allowed_dir = os.path.abspath(ALLOWED_SUBMISSION_DIR) - - if not abs_file_path.startswith(abs_allowed_dir): - return json.dumps({ - "error": True, - "message": f"Security Violation: File submission is restricted to {abs_allowed_dir}" - }) - - filename = os.path.basename(file_path) - mime_type, _ = mimetypes.guess_type(file_path) - if not mime_type: - mime_type = "application/octet-stream" - - data = _build_submission_data(priority=priority, options=options) - - url = f"{API_URL.rstrip('/')}/tasks/create/static/" - - async with httpx.AsyncClient() as client: - try: - with open(file_path, "rb") as f: - files = {"file": (filename, f, mime_type)} - response = await client.post(url, data=data, files=files, headers=get_headers(token)) - try: - result = response.json() - except json.JSONDecodeError: - result = {"error": response.status_code >= 400, "data": response.text} - except Exception as e: - result = {"error": True, "message": str(e)} - - return json.dumps(result, indent=2) - -# --- Task Management & Search --- - -def get_lean_cape_report(raw_cape_json): - """Filters a 50MB CAPE report down to a 500-token LLM payload.""" - return { - "score": raw_cape_json.get("info", {}).get("score", 0), - "family": raw_cape_json.get("malfamily") or raw_cape_json.get("detections", {}).get("family") or "Unknown", - "extracted_configs": raw_cape_json.get("CAPE", []), - "high_severity_signatures": [ - {"name": sig["name"], "desc": sig["description"]} - for sig in raw_cape_json.get("signatures", []) - if isinstance(sig, dict) and sig.get("severity", 0) >= 3 - ], - "network": { - "domains": [d["domain"] for d in raw_cape_json.get("network", {}).get("domains", [])] if isinstance(raw_cape_json.get("network", {}).get("domains"), list) else [], - "http_uris": [h["uri"] for h in raw_cape_json.get("network", {}).get("http", [])] if isinstance(raw_cape_json.get("network", {}).get("http"), list) else [], - }, - "indicators": { - "mutexes": raw_cape_json.get("behavior", {}).get("summary", {}).get("mutexes", []) if isinstance(raw_cape_json.get("behavior", {}).get("summary"), dict) else [], - "commands": raw_cape_json.get("behavior", {}).get("summary", {}).get("executed_commands", []) if isinstance(raw_cape_json.get("behavior", {}).get("summary"), dict) else [] - } - } - -def _apply_lean_report(result): - if isinstance(result, dict): - if result.get("error") is False and "data" in result: - if isinstance(result["data"], list): - result["data"] = [get_lean_cape_report(item) for item in result["data"]] - elif isinstance(result["data"], dict): - result["data"] = get_lean_cape_report(result["data"]) - elif "info" in result: - return get_lean_cape_report(result) - elif isinstance(result, list): - return [get_lean_cape_report(item) for item in result] - return result - -@mcp_tool("tasksearch") -async def search_task(hash_value: str, lean: bool = True, token: str = "") -> str: - """Search for tasks by MD5, SHA1, or SHA256.""" - if not re.match(r"^[a-fA-F0-9]+$", hash_value): - return json.dumps({"error": True, "message": "Invalid hash value provided. Only hexadecimal characters are allowed."}, indent=2) - - algo = "md5" - if len(hash_value) == 40: - algo = "sha1" - elif len(hash_value) == 64: - algo = "sha256" - - result = await _request("GET", f"tasks/search/{algo}/{hash_value}/", token=token) - if lean: - result = _apply_lean_report(result) - return json.dumps(result, indent=2) - -@mcp_tool("extendedtasksearch") -async def extended_search(option: str, argument: str, lean: bool = True, token: str = "") -> str: - """ - Search tasks using extended options. - Options include: id, name, type, string, ssdeep, crc32, file, command, resolvedapi, key, mutex, domain, ip, signature, signame, etc. - """ - data = {"option": option, "argument": argument} - if lean: - data["lean"] = True - result = await _request("POST", "tasks/extendedsearch/", token=token, data=data) - if lean: - result = _apply_lean_report(result) - return json.dumps(result, indent=2) - -@mcp_tool("extendedtasksearch") -async def get_search_info() -> str: - """ - Retrieve the available advanced search terms, filters, and hash types. - Use this information to construct valid queries for `extended_search`. - """ - return json.dumps({ - "search_term_map": search_term_map, - "perform_search_filters": perform_search_filters, - "hash_searches": hash_searches, - "normalized_lower_terms": normalized_lower_terms - }, indent=2, default=str) - -@mcp_tool("tasklist") -async def list_tasks(limit: int = 10, offset: int = 0, status: str = "", token: str = "") -> str: - """List tasks with optional limit, offset and status filter.""" - params = {} - if status: - params["status"] = status - - endpoint = f"tasks/list/{limit}/{offset}/" - result = await _request("GET", endpoint, token=token, params=params) - return json.dumps(result, indent=2) - -@mcp_tool("taskview") -async def view_task(task_id: int, token: str = "") -> str: - """Get details of a specific task.""" - result = await _request("GET", f"tasks/view/{task_id}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("taskresched") -async def reschedule_task(task_id: int, token: str = "") -> str: - """Reschedule a task.""" - result = await _request("GET", f"tasks/reschedule/{task_id}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("taskreprocess") -async def reprocess_task(task_id: int, token: str = "") -> str: - """Reprocess a task.""" - result = await _request("GET", f"tasks/reprocess/{task_id}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("taskstatus") -async def get_task_status(task_id: int, token: str = "") -> str: - """Get the status of a task.""" - result = await _request("GET", f"tasks/status/{task_id}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("tasks_latest") -async def get_latest_tasks(hours: int = 24, token: str = "") -> str: - """Get IDs of tasks finished in the last X hours.""" - result = await _request("GET", f"tasks/get/latests/{hours}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("statistics") -async def get_statistics(days: int = 7, token: str = "") -> str: - """Get task statistics for the last X days.""" - result = await _request("GET", f"tasks/statistics/{days}/", token=token) - return json.dumps(result, indent=2) - -# --- Reports & IOCs --- - -@mcp_tool("taskreport") -async def get_task_report(task_id: int, format: str = "json", token: str = "") -> str: - """Get the analysis report for a task (json, lite, maec, metadata, lean).""" - allowed_formats = {"json", "lite", "maec", "metadata", "lean"} - if format not in allowed_formats: - return json.dumps({"error": True, "message": f"Invalid format provided. Allowed formats: {', '.join(allowed_formats)}"}, indent=2) - - if format == "lean": - data = {"option": "id", "argument": str(task_id), "lean": True} - result = await _request("POST", "tasks/extendedsearch/", token=token, data=data) - - # Extract the single task report from the search results - if isinstance(result, dict) and not result.get("error") and isinstance(result.get("data"), list): - if len(result["data"]) > 0: - result["data"] = result["data"][0] - else: - result = {"error": True, "message": "Task report not found via lean search."} - - result = _apply_lean_report(result) - return json.dumps(result, indent=2) - - result = await _request("GET", f"tasks/get/report/{task_id}/{format}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("taskiocs") -async def get_task_iocs(task_id: int, detailed: bool = False, token: str = "") -> str: - """Get IOCs for a task.""" - endpoint = f"tasks/get/iocs/{task_id}/" - if detailed: - endpoint += "detailed/" - result = await _request("GET", endpoint, token=token) - return json.dumps(result, indent=2) - -@mcp_tool("capeconfig") -async def get_task_config(task_id: int, token: str = "") -> str: - """Get the extracted malware configuration for a task.""" - result = await _request("GET", f"tasks/get/config/{task_id}/", token=token) - return json.dumps(result, indent=2) - -# --- File Downloads --- - -@mcp_tool("taskscreenshot") -async def download_task_screenshot(task_id: int, destination: str, screenshot_id: str = "all", token: str = "") -> str: - """Download task screenshots (zip or single image).""" - return await _download_file(f"tasks/get/screenshot/{task_id}/{screenshot_id}/", destination, f"{task_id}_screenshots.zip", token=token) - -@mcp_tool("taskpcap") -async def download_task_pcap(task_id: int, destination: str, token: str = "") -> str: - """Download the PCAP file for a task.""" - return await _download_file(f"tasks/get/pcap/{task_id}/", destination, f"{task_id}_dump.pcap", token=token) - -@mcp_tool("tasktlspcap") -async def download_task_tlspcap(task_id: int, destination: str, token: str = "") -> str: - """Download the TLS PCAP file for a task.""" - return await _download_file(f"tasks/get/tlspcap/{task_id}/", destination, f"{task_id}_tls.pcap", token=token) - -@mcp_tool("taskevtx") -async def download_task_evtx(task_id: int, destination: str, token: str = "") -> str: - """Download the EVTX logs for a task.""" - return await _download_file(f"tasks/get/evtx/{task_id}/", destination, f"{task_id}_evtx.zip", token=token) - -@mcp_tool("taskdropped") -async def download_task_dropped(task_id: int, destination: str, token: str = "") -> str: - """Download dropped files for a task.""" - return await _download_file(f"tasks/get/dropped/{task_id}/", destination, f"{task_id}_dropped.zip", token=token) - -@mcp_tool("taskselfextracted") -async def download_self_extracted_files(task_id: int, destination: str, tool: str = "all", token: str = "") -> str: - """Download self-extracted files for a task.""" - return await _download_file(f"tasks/get/selfextracted/{task_id}/{tool}/", destination, f"{task_id}_selfextracted_{tool}.zip", token=token) - -@mcp_tool("tasksurifile") -async def download_task_surifile(task_id: int, destination: str, token: str = "") -> str: - """Download Suricata files for a task.""" - return await _download_file(f"tasks/get/surifile/{task_id}/", destination, f"{task_id}_surifiles.zip", token=token) - -@mcp_tool("taskmitmdump") -async def download_task_mitmdump(task_id: int, destination: str, token: str = "") -> str: - """Download mitmdump HAR file for a task.""" - return await _download_file(f"tasks/get/mitmdump/{task_id}/", destination, f"{task_id}_dump.har", token=token) - -@mcp_tool("payloadfiles") -async def download_task_payloadfiles(task_id: int, destination: str, token: str = "") -> str: - """Download CAPE payload files.""" - return await _download_file(f"tasks/get/payloadfiles/{task_id}/", destination, f"{task_id}_payloads.zip", token=token) - -@mcp_tool("procdumpfiles") -async def download_task_procdumpfiles(task_id: int, destination: str, token: str = "") -> str: - """Download CAPE procdump files.""" - return await _download_file(f"tasks/get/procdumpfiles/{task_id}/", destination, f"{task_id}_procdumps.zip", token=token) - -@mcp_tool("taskprocmemory") -async def download_task_procmemory(task_id: int, destination: str, pid: str = "all", token: str = "") -> str: - """Download process memory dumps.""" - return await _download_file(f"tasks/get/procmemory/{task_id}/{pid}/", destination, f"{task_id}_procmemory.zip", token=token) - -@mcp_tool("taskfullmemory") -async def download_task_fullmemory(task_id: int, destination: str, token: str = "") -> str: - """Download full VM memory dump.""" - return await _download_file(f"tasks/get/fullmemory/{task_id}/", destination, f"{task_id}_fullmemory.dmp", token=token) - -# --- Files & Machines --- - -@mcp_tool("fileview") -async def view_file(hash_value: str, hash_type: str = "sha256", token: str = "") -> str: - """View information about a file in the database.""" - if not re.match(r"^[a-fA-F0-9]+$", hash_value): - return json.dumps({"error": True, "message": "Invalid hash value provided. Only hexadecimal characters are allowed."}, indent=2) - return await _request("GET", f"files/view/{hash_type}/{hash_value}/", token=token) - -@mcp_tool("sampledl") -async def download_sample(hash_value: str, destination: str, hash_type: str = "sha256", token: str = "") -> str: - """Download a sample from the database.""" - if not re.match(r"^[a-fA-F0-9]+$", hash_value): - return json.dumps({"error": True, "message": "Invalid hash value provided. Only hexadecimal characters are allowed."}, indent=2) - return await _download_file(f"files/get/{hash_type}/{hash_value}/", destination, f"{hash_value}.bin", token=token) - -@mcp_tool("machinelist") -async def list_machines(token: str = "") -> str: - """List available analysis machines.""" - result = await _request("GET", "machines/list/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("machineview") -async def view_machine(name: str, token: str = "") -> str: - """View details of a specific machine.""" - result = await _request("GET", f"machines/view/{name}/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("list_exitnodes") -async def list_exitnodes(token: str = "") -> str: - """List available exit nodes.""" - result = await _request("GET", "exitnodes/", token=token) - return json.dumps(result, indent=2) - -@mcp_tool("cuckoostatus") -async def get_cuckoo_status(token: str = "") -> str: - """Get the status of the CAPE host.""" - result = await _request("GET", "cuckoo/status/", token=token) - return json.dumps(result, indent=2) - -@mcp.tool() -async def verify_auth(token: str = "") -> str: - """ - Verify if the provided API token is valid. - Useful for checking authentication status before performing other operations. - """ - # We use a lightweight endpoint like cuckoo status to check auth - result = await _request("GET", "cuckoo/status/", token=token) - - if isinstance(result, dict) and result.get("error"): - return json.dumps({"authenticated": False, "message": "Invalid token or authentication failed.", "details": result}, indent=2) - - return json.dumps({"authenticated": True, "message": "Token is valid.", "user": "Authenticated User"}, indent=2) - -if __name__ == "__main__": - import argparse - - parser = argparse.ArgumentParser(description="CAPE MCP Server") - parser.add_argument("--transport", choices=["stdio", "sse", "streamable-http", "http"], default=os.environ.get("CAPE_MCP_TRANSPORT", "stdio"), help="Transport protocol (default: stdio)") - parser.add_argument("--host", default=os.environ.get("CAPE_MCP_HOST", "127.0.0.1"), help="Host to bind for HTTP/SSE (default: 127.0.0.1)") - parser.add_argument("--port", type=int, default=int(os.environ.get("CAPE_MCP_PORT", "9004")), help="Port to bind for HTTP/SSE (default: 9004)") - args = parser.parse_args() - - if args.transport in ["sse", "streamable-http", "http"]: - print(f"Starting {args.transport} server on {args.host}:{args.port}", file=sys.stderr) - mcp.run(transport=args.transport, host=args.host, port=args.port) - else: - mcp.run(transport="stdio")