QR data extraction (kevoreilly#2480)

* Update screenshots.py

* Add QR code URL extraction to screenshots module

Introduces optional QR code URL extraction in Windows screenshots. Adds configuration option 'screenshots_qr' and corresponding UI checkbox, enabling automatic detection and opening of URLs from QR codes found in screenshots if OpenCV is available.

* Add QR code URL extraction from screenshots

Introduces QR code detection using OpenCV in the deduplication process. Extracted URLs from QR codes in screenshots are collected and stored in results. Adds a new signature module to report these URLs.

* Update screenshots.py

* Update deduplication.py

* Update test_analysis_manager.py

* Update test_analysis_manager.py

* Update test_analysis_manager.py

* Update test_analysis_manager.py

Author: doomedraven

analyzer/windows/modules/auxiliary/screenshots.py

@@ -3,14 +3,29 @@
 # See the file 'docs/LICENSE' for copying permission.
 
 import logging
+import os
 import time
+from contextlib import suppress
 from io import BytesIO
 from threading import Thread
 
+try:
+    from PIL import Image
+except ImportError:
+    pass
+
 from lib.api.screenshot import Screenshot
 from lib.common.abstracts import Auxiliary
 from lib.common.results import NetlogFile
 
+HAVE_CV2 = False
+with suppress(ImportError):
+    import cv2
+    import numpy as np
+
+    HAVE_CV2 = True
+
+
 log = logging.getLogger(__name__)
 
 SHOT_DELAY = 1
@@ -20,13 +35,34 @@
 SKIP_AREA = None
 
 
+def handle_qr_codes(image_data):
+    """Extract URL from QR code if present."""
+    if not HAVE_CV2:
+        return None
+
+    try:
+        image = Image.open(image_data)
+        # Convert PIL image to BGR numpy array for OpenCV
+        img = cv2.cvtColor(np.array(image.convert("RGB")), cv2.COLOR_RGB2BGR)
+        detector = cv2.QRCodeDetector()
+        extracted, _, _ = detector.detectAndDecode(img)
+        # Simple URL detection
+        if extracted and "://" in extracted[:10]:
+            return extracted
+    except Exception as e:
+        log.debug("Error in handle_qr_codes: %s", e)
+
+    return None
+
+
 class Screenshots(Auxiliary, Thread):
     """Take screenshots."""
 
     def __init__(self, options, config):
         Auxiliary.__init__(self, options, config)
         Thread.__init__(self)
         self.enabled = config.screenshots_windows
+        self.screenshots_qr = getattr(config, "screenshots_qr", False)
         self.do_run = self.enabled
 
     def stop(self):
@@ -62,7 +98,17 @@ def run(self):
                 img_current.save(tmpio, format="JPEG")
                 tmpio.seek(0)
 
-                # now upload to host from the StringIO
+                if self.screenshots_qr and HAVE_CV2:
+                    url = handle_qr_codes(tmpio)
+                    if url:
+                        log.info("QR code detected with URL: %s", url)
+                        try:
+                            # os.startfile is Windows only and usually works for URLs
+                            os.startfile(url)
+                        except Exception as e:
+                            log.error("Failed to open QR URL: %s", e)
+                    tmpio.seek(0)
+
                 nf = NetlogFile()
                 nf.init(f"shots/{str(img_counter).rjust(4, '0')}.jpg")
                 for chunk in tmpio:

conf/default/auxiliary.conf.default

@@ -32,6 +32,7 @@ procmon = no
 recentfiles = no
 screenshots_windows = yes
 screenshots_linux = yes
+screenshots_qr = no
 sysmon_windows = no
 sysmon_linux = no
 tlsdump = yes

modules/processing/deduplication.py

@@ -17,6 +17,14 @@
 except ImportError:
     log.error("Missed dependency: poetry run pip install ImageHash")
 
+HAVE_CV2 = False
+try:
+    import cv2
+
+    HAVE_CV2 = True
+except ImportError:
+    print("Missed dependency: poetry run pip install opencv-python")
+
 try:
     from PIL import Image
 
@@ -38,6 +46,23 @@ def reindex_screenshots(shots_path):
         os.rename(old_path, new_path)
 
 
+def handle_qr_codes(image_path):
+    if not HAVE_CV2:
+        return None
+    try:
+        # cv2.imread handles file path directly
+        img = cv2.imread(image_path)
+        if img is None:
+            return None
+        detector = cv2.QRCodeDetector()
+        extracted, points, straight_qrcode = detector.detectAndDecode(img)
+        if extracted and "://" in extracted[:10]:
+            return extracted
+    except Exception as e:
+        log.error("Error detecting QR in %s: %s", image_path, e)
+    return None
+
+
 class Deduplicate(Processing):
     """Deduplicate screenshots."""
 
@@ -116,6 +141,18 @@ def hashfunc(img):
             screenshots = sorted(self.deduplicate_images(userpath=shots_path, hashfunc=hashfunc))
             shots = [re.sub(r"\.(png|jpg)$", "", screenshot) for screenshot in screenshots]
 
+            if HAVE_CV2:
+                qr_urls = set()
+                for img_name in os.listdir(shots_path):
+                    if not img_name.lower().endswith((".jpg", ".png")):
+                        continue
+                    url = handle_qr_codes(os.path.join(shots_path, img_name))
+                    if url:
+                        qr_urls.add(url)
+
+                if qr_urls:
+                    self.results["qr_urls"] = list(qr_urls)
+
         except Exception as e:
             log.error(e)
 

modules/signatures/qr_urls.py

@@ -0,0 +1,19 @@
+
+from lib.cuckoo.common.abstracts import Signature
+
+class QRUrls(Signature):
+    name = "qr_urls"
+    description = "URLs extracted from QR codes in screenshots"
+    severity = 1
+    categories = ["info"]
+    authors = ["DoomedRaven"]
+    minimum = "1.3"
+    evented = False
+
+    def run(self):
+        qr_urls = self.results.get("qr_urls")
+        if qr_urls:
+            for url in qr_urls:
+                self.data.append({"url": url})
+            return True
+        return False

tests/test_analysis_manager.py

@@ -10,7 +10,7 @@
 from sqlalchemy import select
 
 from lib.cuckoo.common.abstracts import Machinery
-from lib.cuckoo.common.config import ConfigMeta
+from lib.cuckoo.common.config import Config, ConfigMeta
 from lib.cuckoo.core.analysis_manager import AnalysisManager
 from lib.cuckoo.core.database import TASK_RUNNING, Guest, Machine, Task, _Database
 from lib.cuckoo.core.machinery_manager import MachineryManager
@@ -303,7 +303,12 @@ def screenshot(self2, label, path):
         assert "no machine is used" in caplog.text
 
     def test_build_options(
-        self, db: _Database, tmp_path: pathlib.Path, task: Task, machine: Machine, machinery_manager: MachineryManager
+        self,
+        db: _Database,
+        tmp_path: pathlib.Path,
+        task: Task,
+        machine: Machine,
+        machinery_manager: MachineryManager,
     ):
         with db.session.begin():
             task = db.session.merge(task)
@@ -315,57 +320,38 @@ def test_build_options(
 
         analysis_man = AnalysisManager(task=task, machine=machine, machinery_manager=machinery_manager)
         opts = analysis_man.build_options()
-        assert opts == {
-            "amsi": False,
-            "browser": True,
-            "browsermonitor": False,
+
+        expected_opts = {
             "category": "file",
             "clock": datetime.datetime(2099, 1, 1, 9, 1, 1),
-            "curtain": False,
-            "digisig": True,
-            "disguise": True,
             "do_upload_max_size": 0,
-            "during_script": False,
             "enable_trim": 0,
             "enforce_timeout": 1,
-            "evtx": False,
             "exports": "",
-            "filecollector": True,
             "file_name": "sample.py",
-            "file_pickup": False,
             "file_type": "Python script, ASCII text executable",
-            "human_linux": False,
-            "human_windows": True,
             "id": task.id,
             "ip": "5.6.7.8",
             "options": "foo=bar",
             "package": "foo",
-            "permissions": False,
             "port": "2043",
-            "pre_script": False,
-            "procmon": False,
-            "recentfiles": False,
-            "screenshots_linux": True,
-            "screenshots_windows": True,
-            "sslkeylogfile": False,
-            "sysmon_linux": False,
-            "sysmon_windows": False,
             "target": str(tmp_path / "sample.py"),
             "terminate_processes": False,
             "timeout": 10,
-            "tlsdump": True,
-            "tracee_linux": False,
             "upload_max_size": 100000000,
-            "usage": False,
-            "windows_static_route": False,
-            "windows_static_route_gateway": "192.168.1.1",
-            "dns_etw": False,
-            "wmi_etw": False,
-            "watchdownloads": False,
         }
+        # Dynamically load auxiliary modules from Config to ensure test stays in sync with configuration changes
+        expected_opts.update(Config("auxiliary").auxiliary_modules)
+
+        assert opts == expected_opts
 
     def test_build_options_pe(
-        self, db: _Database, tmp_path: pathlib.Path, task: Task, machine: Machine, machinery_manager: MachineryManager
+        self,
+        db: _Database,
+        tmp_path: pathlib.Path,
+        task: Task,
+        machine: Machine,
+        machinery_manager: MachineryManager,
     ):
         sample_location = get_test_object_path(
             pathlib.Path("data/core/5dd87d3d6b9d8b4016e3c36b189234772661e690c21371f1eb8e018f0f0dec2b")
@@ -380,54 +366,31 @@ def test_build_options_pe(
 
         analysis_man = AnalysisManager(task=task, machine=machine, machinery_manager=machinery_manager)
         opts = analysis_man.build_options()
-        assert opts == {
-            "amsi": False,
-            "browser": True,
-            "browsermonitor": False,
+
+        expected_opts = {
             "category": "file",
             "clock": datetime.datetime(2099, 1, 1, 9, 1, 1),
-            "curtain": False,
-            "digisig": True,
-            "disguise": True,
             "do_upload_max_size": 0,
-            "during_script": False,
             "enable_trim": 0,
             "enforce_timeout": 1,
-            "evtx": False,
             "exports": "",
-            "filecollector": True,
             "file_name": sample_location.name,
-            "file_pickup": False,
             "file_type": "PE32 executable (console) Intel 80386, for MS Windows",
-            "human_linux": False,
-            "human_windows": True,
             "id": task.id,
             "ip": "5.6.7.8",
             "options": "",
             "package": "file",
-            "permissions": False,
             "port": "2043",
-            "pre_script": False,
-            "procmon": False,
-            "recentfiles": False,
-            "screenshots_linux": True,
-            "screenshots_windows": True,
-            "sslkeylogfile": False,
-            "sysmon_linux": False,
-            "sysmon_windows": False,
             "target": str(sample_location),
             "terminate_processes": False,
             "timeout": 10,
-            "tlsdump": True,
-            "tracee_linux": False,
             "upload_max_size": 100000000,
-            "usage": False,
-            "windows_static_route": False,
-            "windows_static_route_gateway": "192.168.1.1",
-            "dns_etw": False,
-            "wmi_etw": False,
-            "watchdownloads": False,
         }
+        # Dynamically load auxiliary modules from Config to ensure test stays in sync with configuration changes
+        expected_opts.update(Config("auxiliary").auxiliary_modules)
+
+        assert opts == expected_opts
+
 
     def test_category_checks(
         self, db: _Database, task: Task, machine: Machine, machinery_manager: MachineryManager, mocker: MockerFixture

web/submission/views.py

@@ -352,6 +352,9 @@ def index(request, task_id=None, resubmit_hash=None):
         if request.POST.get("unpack"):
             options += "unpack=yes,"
 
+        if request.POST.get("screenshots_qr"):
+            options += "screenshots_qr=yes,"
+
         job_category = False
         if request.POST.get("job_category"):
             job_category = request.POST.get("job_category")

web/templates/submission/index.html

@@ -598,7 +598,15 @@ <h5 class="mb-0 text-white"><i class="fas fa-cogs me-2 text-primary"></i>Advance
                                         id="duringScript" name="during_script">
                                 </div>
                                 {% endif %}
-
+                                <div class="form-check">
+                                    <label>
+                                        <input type="checkbox" name="screenshots_qr" /> QR URL extraction <span class="text-muted"><small>(Extract and open URLs from QR codes in screenshots)</small></span>
+                                    </label>
+                                </div>
+                                <div class="form-check">
+                                    <label>
+                                        <input type="checkbox" name="static"/> Try to extract config without VM <span class="text-muted"><small>(Submit to VM if not extracted)</small></span>
+                                    </label>
                                 <div class="mb-3">
                                     <label for="form_custom" class="text-white-50">Custom Field</label>
                                     <input type="text" class="form-control bg-dark text-white border-secondary"