Create watchdownloads.py

analyzer/windows/modules/auxiliary/watchdownloads.py

@@ -0,0 +1,79 @@
+# Copyright (C) 2025 Xiang Chen
+# This file is part of CAPE Sandbox
+# See the file 'docs/LICENSE' for copying permission.
+
+import os
+import logging
+import time
+from threading import Thread
+
+from lib.common.abstracts import Auxiliary
+from lib.common.results import upload_to_host
+
+log = logging.getLogger(__name__)
+
+folders_to_monitor = [
+    os.path.join(os.environ["HOMEPATH"], "downloads"),
+]
+
+
+HAVE_WATCHDOG = False
+try:
+    from watchdog.events import FileSystemEvent, FileSystemEventHandler, EVENT_TYPE_DELETED
+    from watchdog.observers import Observer
+
+    class MyEventHandler(FileSystemEventHandler):
+        def on_any_event(self, event: FileSystemEvent) -> None:
+            if event.event_type == EVENT_TYPE_DELETED:
+                return
+            try:
+                filename = os.path.basename(event.src_path)
+                if not filename.endswith(('.part', 'desktop.ini')):
+                    log.info("Monitor uploading %s", filename)
+                    upload_to_host(event.src_path, f"files/{filename}")
+            except Exception as e:
+                log.exception("Can't upload new file %s to host. %s", event.src_path, str(e))
+
+
+    HAVE_WATCHDOG = True
+except ImportError as e:
+    log.debug(
+        f"Could not load auxiliary module WatchDownloads due to '{e}'"
+    )
+
+
+class WatchDownloads(Auxiliary, Thread):
+    """Collect CPU/memory usage info from monitored processes"""
+
+    def __init__(self, options, config):
+        Auxiliary.__init__(self, options, config)
+        Thread.__init__(self)
+        self.enabled = self.config.watchdownloads
+        self.do_run = True
+
+    def stop(self):
+        """Stop collecting info"""
+        self.do_run = False
+
+    def run(self):
+        """Run capturing of info.
+        @return: operation status.
+        """
+        if not self.enabled:
+            return False
+
+        event_handler = MyEventHandler()
+        observer = Observer()
+        for folder in folders_to_monitor:
+            log.info("Monitoring %s", folder)
+            observer.schedule(event_handler, folder, recursive=True)
+        observer.start()
+
+        try:
+            while self.do_run:
+                time.sleep(1)
+        finally:
+            observer.stop()
+            observer.join()
+
+        return True

conf/default/auxiliary.conf.default

@@ -47,6 +47,7 @@ sslkeylogfile = no
 browsermonitor = no
 wmi_etw = no
 dns_etw = no
+watchdownloads = no
 
 [AzSniffer]
 # Enable or disable the use of Azure Network Watcher packet capture feature, disable standard sniffer if this is in use to not create concurrent .pcap files

tests/test_analysis_manager.py

@@ -356,6 +356,7 @@ def test_build_options(
             "windows_static_route_gateway": "192.168.1.1",
             "dns_etw": False,
             "wmi_etw": False,
+            "watchdownloads": False,
         }
 
     def test_build_options_pe(
@@ -420,6 +421,7 @@ def test_build_options_pe(
             "windows_static_route_gateway": "192.168.1.1",
             "dns_etw": False,
             "wmi_etw": False,
+            "watchdownloads": False,
         }
 
     def test_category_checks(