[nmap] fix: correct parser when target selector key is manual (#4412)

Signed-off-by: Antoine MAZEAS <antoine.mazeas@filigran.io>
Co-authored-by: Antoine MAZEAS <antoine.mazeas@filigran.io>

Author: savacano28

.circleci/config.yml

@@ -38,7 +38,7 @@ jobs:
       - slack/notify:
           event: fail
           template: basic_fail_1
-  test:
+  test-nuclei:
     working_directory: ~/openaev
     docker:
       - image: cimg/python:3.13
@@ -62,6 +62,30 @@ jobs:
           working_directory: ~/openaev/nuclei
           name: Tests for nuclei injector
           command: python -m unittest
+  test-nmap:
+    working_directory: ~/openaev
+    docker:
+      - image: cimg/python:3.13
+    steps:
+      - checkout
+      - setup_remote_docker
+      - run:
+          working_directory: ~/openaev/nmap
+          name: Install dependencies for nmap injector
+          command: pip install -r requirements.txt
+      - run:
+          working_directory: ~/openaev/nmap
+          name: Overwrite pyoaev with correct version from CI
+          command: |
+            if [ "${CIRCLE_BRANCH}" = "main" ]; then
+              pip install --force-reinstall git+https://github.com/OpenAEV-Platform/client-python.git@main
+            else
+              pip install --force-reinstall git+https://github.com/OpenAEV-Platform/client-python.git@release/current
+            fi
+      - run:
+          working_directory: ~/openaev/nmap
+          name: Tests for nmap injector
+          command: python -m unittest
   build_1:
     working_directory: ~/openaev
     docker:
@@ -245,7 +269,8 @@ workflows:
     jobs:
       - ensure_formatting
       - linter
-      - test
+      - test-nmap
+      - test-nuclei
       - build_1:
           filters:
             tags:
@@ -256,7 +281,8 @@ workflows:
           requires:
             - ensure_formatting
             - linter
-            - test
+            - test-nmap
+            - test-nuclei
           filters:
             branches:
               only:
@@ -265,7 +291,8 @@ workflows:
           requires:
             - ensure_formatting
             - linter
-            - test
+            - test-nmap
+            - test-nuclei
           filters:
             branches:
               only:

injector_common/injector_common/targets.py

@@ -64,7 +64,11 @@ def extract_targets(
             )
 
         elif selector_key == "manual":
-            targets = [t.strip() for t in content[TARGETS_KEY].split(",") if t.strip()]
+            targets = list(
+                dict.fromkeys(
+                    [t.strip() for t in content[TARGETS_KEY].split(",") if t.strip()]
+                )
+            )
 
         else:
             raise ValueError("No targets provided for this injection")

nmap/src/helpers/nmap_output_parser.py

@@ -1,9 +1,15 @@
 from typing import Dict
 
+from injector_common.targets import TargetExtractionResult
+
 
 class NmapOutputParser:
-    def parse(data: Dict, result: str, asset_list: []) -> Dict:
+    @staticmethod
+    def parse(data: Dict, result: str, target_results: TargetExtractionResult) -> Dict:
         """Parse nmap results and extract open ports."""
+        asset_list = list(target_results.ip_to_asset_id_map.values())
+        targets = target_results.targets or []
+
         run = result["nmaprun"]
         if not isinstance(run["host"], list):
             run["host"] = [run["host"]]
@@ -27,7 +33,10 @@ def parse(data: Dict, result: str, asset_list: []) -> Dict:
                             port_result["host"] = host["address"]["@addr"]
                         else:
                             port_result["asset_id"] = None
-                            port_result["host"] = asset_list[idx]
+                            if idx < len(targets):
+                                port_result["host"] = targets[idx]
+                            else:
+                                port_result["host"] = None
                         ports_scans_results.append(port_result)
 
         return {

nmap/src/openaev_nmap.py

@@ -53,16 +53,15 @@ def nmap_execution(self, start: float, data: Dict) -> Dict:
         target_results = Targets.extract_targets(
             selector_key, selector_property, data, self.helper
         )
-        asset_list = list(target_results.ip_to_asset_id_map.values())
         # Deduplicate targets
-        unique_targets = list(dict.fromkeys(target_results.targets))
+        targets = target_results.targets
         # Handle empty targets as an error
-        if not unique_targets:
+        if not targets:
             message = f"No target identified for the property {TargetProperty[selector_property.upper()].value}"
             raise ValueError(message)
 
         # Build Arguments to execute
-        nmap_args = NmapCommandBuilder.build_args(contract_id, unique_targets)
+        nmap_args = NmapCommandBuilder.build_args(contract_id, targets)
 
         self.helper.injector_logger.info(
             "Executing nmap with command: " + " ".join(nmap_args)
@@ -90,7 +89,7 @@ def nmap_execution(self, start: float, data: Dict) -> Dict:
         jc = NmapProcess.js_execute(["jc", "--xml", "-p"], nmap_result)
         result = json.loads(jc.stdout.decode("utf-8").strip())
 
-        return NmapOutputParser.parse(data, result, asset_list)
+        return NmapOutputParser.parse(data, result, target_results)
 
     def process_message(self, data: Dict) -> None:
         start = time.time()

nmap/test/__init__.py

@@ -0,0 +1,98 @@
+from unittest import TestCase
+
+from src.helpers.nmap_output_parser import NmapOutputParser
+
+from injector_common.targets import TargetExtractionResult
+
+
+class NmapOutputParserTest(TestCase):
+    def setUp(self):
+        self.result_single_host = {
+            "nmaprun": {
+                "host": {
+                    "address": {"@addr": "172.16.5.10"},
+                    "ports": {
+                        "port": [
+                            {
+                                "@portid": "21",
+                                "state": {"@state": "open"},
+                                "service": {"@name": "ftp"},
+                            }
+                        ]
+                    },
+                }
+            }
+        }
+
+    # -------------------------------
+    # Tests
+    # -------------------------------
+
+    def test_parse_target_assets(self):
+        """Ensure target_selector='assets' uses asset_list and sets asset_id."""
+        data = {"injection": {"inject_content": {"target_selector": "assets"}}}
+
+        result = NmapOutputParser.parse(
+            data,
+            self.result_single_host,
+            TargetExtractionResult(
+                ip_to_asset_id_map={"172.16.5.10": "asset-123"}, targets=[]
+            ),
+        )
+
+        scan = result["outputs"]["scan_results"][0]
+
+        self.assertEqual(scan["asset_id"], "asset-123")
+        self.assertEqual(scan["host"], "172.16.5.10")
+        self.assertEqual(scan["port"], 21)
+        self.assertEqual(scan["service"], "ftp")
+
+    def test_parse_target_asset_groups(self):
+        """Ensure target_selector='asset-groups' also uses asset_list."""
+        data = {"injection": {"inject_content": {"target_selector": "asset-groups"}}}
+
+        result = NmapOutputParser.parse(
+            data,
+            self.result_single_host,
+            TargetExtractionResult(
+                ip_to_asset_id_map={"172.16.5.10": "group-asset-555"},
+                targets=["172.16.5.10"],
+            ),
+        )
+
+        scan = result["outputs"]["scan_results"][0]
+
+        self.assertEqual(scan["asset_id"], "group-asset-555")
+        self.assertEqual(scan["host"], "172.16.5.10")
+        self.assertEqual(scan["port"], 21)
+        self.assertEqual(scan["service"], "ftp")
+
+    def test_parse_target_manual(self):
+        """Ensure target_selector='manual' sets asset_id=None."""
+        data = {"injection": {"inject_content": {"target_selector": "manual"}}}
+
+        result = NmapOutputParser.parse(
+            data,
+            self.result_single_host,
+            TargetExtractionResult(ip_to_asset_id_map={}, targets=["172.16.5.10"]),
+        )
+
+        scan = result["outputs"]["scan_results"][0]
+
+        self.assertIsNone(scan["asset_id"])
+        self.assertEqual(scan["host"], "172.16.5.10")
+
+    def test_parse_target_manual_None_values(self):
+        """Ensure target_selector='manual' sets asset_id=None."""
+        data = {"injection": {"inject_content": {"target_selector": "manual"}}}
+
+        result = NmapOutputParser.parse(
+            data,
+            self.result_single_host,
+            TargetExtractionResult(ip_to_asset_id_map={}, targets=[]),
+        )
+
+        scan = result["outputs"]["scan_results"][0]
+
+        self.assertIsNone(scan["asset_id"])
+        self.assertIsNone(scan["host"])

nmap/test/helpers/__init__.py

@@ -79,16 +79,14 @@ def nuclei_execution(self, start: float, data: Dict) -> Dict:
             selector_key, selector_property, data, self.helper
         )
         # Deduplicate targets
-        unique_targets = list(dict.fromkeys(target_results.targets))
+        targets = target_results.targets
         # Handle empty targets as an error
-        if not unique_targets:
+        if not targets:
             message = f"No target identified for the property {TargetProperty[selector_property.upper()].value}"
             raise ValueError(message)
         # Build Arguments to execute
-        nuclei_args = self.command_builder.build_args(
-            contract_id, content, unique_targets
-        )
-        input_data = "\n".join(unique_targets).encode("utf-8")
+        nuclei_args = self.command_builder.build_args(contract_id, content, targets)
+        input_data = "\n".join(targets).encode("utf-8")
 
         self.helper.injector_logger.info(
             "Executing nuclei with: " + " ".join(nuclei_args)