Merge pull request #98 from Ostorlab/feature/Expose_the_OS_in_the_fin…

…gerprint

Expose the OS in the fingerprint from nmap

README.md

@@ -41,6 +41,7 @@ Supported agent flags:
 * `timing_template` (`-Tx`): Template of timing settings (T0, T1, ... T5)..
 * `script_default` (`-sC`): Script scan, equivalent to --script=default.
 * `scripts` (`--script`): List of scripts to run using Nmap.
+* `os` (`--os`): Enable OS detection
 
 
  ### Install directly from OXO agent store

agent/nmap_agent.py

@@ -189,6 +189,7 @@ def _scan_domain(self, domain_name: str) -> Tuple[Dict[str, Any], str]:
             scripts=self.args.get("scripts"),
             script_default=self.args.get("script_default", False),
             version_detection=self.args.get("version_info", False),
+            os_detection=self.args.get("os", False),
         )
         client = nmap_wrapper.NmapWrapper(options)
         logger.info("scanning domain %s with options %s", domain_name, options)
@@ -372,6 +373,21 @@ def _emit_fingerprints(
                 else:
                     raise ValueError(f"Incorrect ip version {version}")
 
+                if host.get("os", {}).get("osmatch") is not None:
+                    os_match_highest = host.get("os").get("osmatch", {})[0]
+                    fingerprint_data = {
+                        "host": host.get("address", {}).get("@addr"),
+                        "library_type": "OS",
+                        "library_name": os_match_highest.get("osclass", {}).get(
+                            "@osfamily"
+                        ),
+                        "library_version": os_match_highest.get("osclass").get(
+                            "@osgen"
+                        ),
+                        "detail": os_match_highest.get("@name"),
+                    }
+                    self.emit(selector, fingerprint_data)
+
                 for data in generators.get_services(scan_results):
                     if data.get("product") is not None:
                         logger.debug("sending results to selector %s", selector)

agent/nmap_options.py

@@ -53,6 +53,7 @@ class NmapOptions:
         default_factory=lambda: ["default", "banner"]
     )
     version_detection: bool = True
+    os_detection: bool = True
     port_scanning_techniques: List[PortScanningTechnique] = dataclasses.field(
         default_factory=lambda: [
             PortScanningTechnique.TCP_SYN,
@@ -61,8 +62,15 @@ class NmapOptions:
     no_ping: bool = True
     privileged: Optional[bool] = None
 
+    def _set_os_detection_option(self) -> List[str]:
+        """Appends the os detection option to the list of nmap options."""
+        command_options = []
+        if self.os_detection is True:
+            command_options.append("-O")
+        return command_options
+
     def _set_version_detection_option(self) -> List[str]:
-        """Appends the  option to the list of nmap options."""
+        """Appends the version detection option to the list of nmap options."""
         command_options = []
         if self.version_detection is True:
             command_options.append("-sV")
@@ -144,6 +152,7 @@ def _run_scripts_command(self, scripts: List[str]) -> List[str]:
     def command_options(self) -> List[str]:
         """Computes the list of nmap options."""
         command_options = []
+        command_options.extend(self._set_os_detection_option())
         command_options.extend(self._set_version_detection_option())
         command_options.extend(self._set_dns_resolution_option())
         command_options.extend(self._set_ports_option())

tests/nmap_agent_test.py

@@ -742,3 +742,71 @@ def testAgent_whenServiceWithProductAndVersion_fingerprintMessageShouldHaveLibra
     assert fingerprint_msg.data["port"] == 22
     assert fingerprint_msg.data["library_name"] == "OpenSSH"
     assert fingerprint_msg.data["library_version"] == "7.4"
+
+
+def testAgent_whenHostHaveOs_fingerprintMessageShouldHaveOs(
+    nmap_test_agent: nmap_agent.NmapAgent,
+    agent_mock: List[message.Message],
+    ipv4_msg: message.Message,
+    agent_persist_mock: Dict[Union[str, bytes], Union[str, bytes]],
+    mocker: plugin.MockerFixture,
+) -> None:
+    """Ensure the agents emits the detected library name with its version."""
+    del agent_persist_mock
+    product_fake_output = {
+        "nmaprun": {
+            "host": {
+                "address": {"@addr": "127.0.0.1", "@addrtype": "ipv4"},
+                "ports": {
+                    "port": {
+                        "@portid": "22",
+                        "@protocol": "tcp",
+                        "state": {
+                            "@state": "open",
+                            "@reason": "syn-ack",
+                            "@reason_ttl": "0",
+                        },
+                        "service": {
+                            "@name": "ssh",
+                            "@product": "OpenSSH",
+                            "@version": "7.4",
+                            "cpe": "cpe:/a:openbsd:openssh:7.4",
+                        },
+                    }
+                },
+                "os": {
+                    "osmatch": [
+                        {
+                            "@name": "Microsoft Windows 10 1511",
+                            "@accuracy": "88",
+                            "@line": "69505",
+                            "osclass": {
+                                "@type": "specialized",
+                                "@vendor": "Microsoft",
+                                "@osfamily": "Windows",
+                                "@osgen": "10",
+                                "@accuracy": "88",
+                                "cpe": "cpe:/o:microsoft:windows_10:1511",
+                            },
+                        }
+                    ]
+                },
+            }
+        }
+    }
+
+    mocker.patch(
+        "agent.nmap_wrapper.NmapWrapper.scan_hosts",
+        return_value=(product_fake_output, ""),
+    )
+
+    nmap_test_agent.process(ipv4_msg)
+
+    assert len(agent_mock) == 4
+    assert agent_mock[0].selector == "v3.asset.ip.v4.port.service"
+    assert agent_mock[1].selector == "v3.report.vulnerability"
+    fingerprint_msg = agent_mock[2]
+    assert fingerprint_msg.selector == "v3.fingerprint.ip.v4.service.library"
+    assert fingerprint_msg.data["host"] == "127.0.0.1"
+    assert fingerprint_msg.data["library_name"] == "Windows"
+    assert fingerprint_msg.data["library_version"] == "10"

tests/nmap_wrapper_test.py

@@ -31,6 +31,7 @@ def testNmapWrapper_whenFastMode_returnCommand(
 
     assert command == [
         "nmap",
+        "-O",
         "-sV",
         "-n",
         "-F",
@@ -68,6 +69,7 @@ def testNmapWrapper_whenTopPortsUsed_returnCommand(
 
     assert command == [
         "nmap",
+        "-O",
         "-sV",
         "-n",
         "--top-ports",
@@ -106,6 +108,7 @@ def testNmapWrapper_whenAllTopPortsUsed_returnCommand(
 
     assert command == [
         "nmap",
+        "-O",
         "-sV",
         "-n",
         "-p",