documentation and better variable names

Author: Aarush289

nettacker/core/lib/socket.py

@@ -95,6 +95,21 @@ def tcp_connect_send_and_receive(self, host, port, timeout):
         }
 
     def tcp_and_udp_scan(self, host, port: int, timeout=5):
+        """
+        Probe a port with the nmap-service-probes engine to fingerprint the running service.
+
+        Tries plain TCP first, falls back to TLS if the port rejects plaintext, then falls
+        back to UDP. Returns a dict with "service", "ssl_flag", and "log" keys, or None if
+        the port could not be identified as open.
+
+        Args:
+            host: target host/IP
+            port: target port
+            timeout: per-probe timeout in seconds
+
+        Returns:
+            A result dict, or None if the port appears closed/unresponsive.
+        """
         # `timeout` follows the same convention as every other method here
         # (tcp_connect_only, tcp_connect_send_and_receive, socket_icmp): it's
         # in seconds, as set by the module yaml (e.g. "timeout: 3"). The

nettacker/probing/engine.py

@@ -26,7 +26,7 @@
 )
 
 
-def Interpret(value: bytes | str, endian: str) -> int:
+def interpret(value: bytes | str, endian: str) -> int:
     """
     Interpret up to 8 bytes as unsigned integer
     endian: '>' = big-endian, '<' = little-endian
@@ -40,7 +40,7 @@ def Interpret(value: bytes | str, endian: str) -> int:
     return int.from_bytes(value, byteorder=byteorder, signed=False)
 
 
-def Printable(value: bytes | str) -> str:
+def printable(value: bytes | str) -> str:
     """
     Make a string printable:
     - Remove NULLs
@@ -71,7 +71,7 @@ def apply_subst(match_obj, regex_match):
     return value.replace(old, new)
 
 
-def expand_SUBST(template: str, regex_match):
+def expand_subst(template: str, regex_match):
     while True:
         m = SUBST_RE.search(template)
         if not m:
@@ -82,24 +82,24 @@ def expand_SUBST(template: str, regex_match):
     return template
 
 
-def expand_I(template: str, match):
+def expand_i(template: str, match):
     def repl(m):
         idx = int(m.group(1))
         endian = m.group(2)
         try:
             captured = match.group(idx)
         except IndexError:
             return ""
-        return str(Interpret(captured, endian))
+        return str(interpret(captured, endian))
 
     return I_RE.sub(repl, template)
 
 
-def expand_P(template: str, regex_match):
+def expand_p(template: str, regex_match):
     def repl(m):
         i = int(m.group(1))
         try:
-            return Printable(regex_match.group(i))
+            return printable(regex_match.group(i))
         except IndexError:
             return ""
 
@@ -123,14 +123,16 @@ def repl(m):
 
 
 def expand_template(template: str, regex_match):
-    template = expand_SUBST(template, regex_match)
-    template = expand_P(template, regex_match)
-    template = expand_I(template, regex_match)
+    template = expand_subst(template, regex_match)
+    template = expand_p(template, regex_match)
+    template = expand_i(template, regex_match)
     template = expand_place(template, regex_match)
     return template
 
 
-class result:
+class Result:
+    """Version fields (template, product, CPE, ...) extracted from a matched signature."""
+
     def __init__(
         self,
         version_template=None,
@@ -185,16 +187,16 @@ def get_probes_for_sslport(self):
                 specific.append(p)
         return specific
 
-    def Match_response(self, response, signature):
+    def match_response(self, response, signature):
         if response is None:
-            return {"status": False, "result": result()}
+            return {"status": False, "result": Result()}
 
         if isinstance(response, str):
             response = response.encode("latin-1", errors="ignore")
         regex = signature.regex
         match = regex.search(response)
         if not match:
-            return {"status": False, "result": result()}
+            return {"status": False, "result": Result()}
 
         version_ = signature.version_details
         version_template = product = info = hostname = None
@@ -220,7 +222,7 @@ def Match_response(self, response, signature):
 
         return {
             "status": True,
-            "result": result(
+            "result": Result(
                 version_template=version_template,
                 product=product,
                 info=info,
@@ -233,8 +235,8 @@ def Match_response(self, response, signature):
             ),
         }
 
-    def check_match_service(self, Signatures, service) -> bool:
-        for sig_ in Signatures:
+    def check_match_service(self, signatures, service) -> bool:
+        for sig_ in signatures:
             if sig_.service == service:
                 return True
         return False
@@ -266,22 +268,22 @@ def probe_sequentially(self):
                         self.port,
                         probe.probe_string,
                         probe.totalwaits,
-                        probe.tcpwrapperdms,
+                        probe.tcpwrapped_ms,
                     )
                 else:
                     response = tcp_probe_ssl(
                         self.host,
                         self.port,
                         probe.probe_string,
                         probe.totalwaits,
-                        probe.tcpwrapperdms,
+                        probe.tcpwrapped_ms,
                     )
             else:
                 response = udp_probe(self.host, self.port, probe.probe_string, probe.totalwaits)
 
             if response is None:
                 response = tcp_probe(
-                    self.host, self.port, probe.probe_string, probe.totalwaits, probe.tcpwrapperdms
+                    self.host, self.port, probe.probe_string, probe.totalwaits, probe.tcpwrapped_ms
                 )
 
             if not response or response.get("raw_bytes") is None:
@@ -290,8 +292,8 @@ def probe_sequentially(self):
             raw_response = response.get("raw_bytes")
 
             # 1. Check Primary Signatures
-            for signature in probe.Signatures:
-                matched_data = self.Match_response(raw_response, signature)
+            for signature in probe.signatures:
+                matched_data = self.match_response(raw_response, signature)
                 res_ = matched_data["result"]
                 if matched_data["status"]:
                     if signature.sig_type == "match":
@@ -344,8 +346,8 @@ def probe_sequentially(self):
                 if not fallback_probe:
                     continue
 
-                for signature in fallback_probe.Signatures:
-                    matched_data = self.Match_response(raw_response, signature)
+                for signature in fallback_probe.signatures:
+                    matched_data = self.match_response(raw_response, signature)
                     res_ = matched_data["result"]
                     if matched_data["status"]:
                         if signature.sig_type == "match":

nettacker/probing/loader.py

@@ -8,7 +8,9 @@
 log = logger.get_logger()
 
 
-class version_details:
+class VersionDetails:
+    """Nmap-style version templates extracted from a matched signature."""
+
     def __init__(
         self,
         raw,
@@ -35,6 +37,8 @@ def __init__(
 
 
 class Signature:
+    """A single match/softmatch regex rule belonging to a probe."""
+
     def __init__(
         self,
         service,
@@ -53,6 +57,8 @@ def __init__(
 
 
 class Probe:
+    """An nmap-service-probes probe: the payload to send plus its signatures."""
+
     def __init__(
         self,
         name,
@@ -65,26 +71,27 @@ def __init__(
         fallbacks=None,
         probe_string="",
         no_payload=False,
-        Signatures=None,
+        signatures=None,
     ):
         self.name = name
         self.protocol = protocol
         self.totalwaits = totalwaits
-        self.tcpwrapperdms = tcpwrappedms
+        self.tcpwrapped_ms = tcpwrappedms
         self.rarity = rarity
         self.ports = ports or []
         self.sslports = sslports or []
         self.fallbacks = fallbacks or []
         self.probe_string = probe_string
         self.no_payload = no_payload
-        self.Signatures = Signatures or []
+        self.signatures = signatures or []
 
 
 _PROBES_CACHE = None
 _probes_by_name = {}
 
 
 def load_probes_from_yaml():
+    """Parse Config.path.probes_yaml_file into Probe objects, caching the result."""
     global _PROBES_CACHE
     global _probes_by_name
 
@@ -129,7 +136,7 @@ def load_probes_from_yaml():
                 log.verbose_info(f"Probe signature failed to compile: {pattern!r} ({e})")
                 continue
             v = s.get("version", {}) or {}
-            version = version_details(
+            version = VersionDetails(
                 raw=v.get("raw", ""),
                 version_template=v.get("version_template", ""),
                 product=v.get("product", ""),
@@ -163,7 +170,7 @@ def load_probes_from_yaml():
             fallbacks=fallbacks,
             probe_string=probe_string,
             no_payload=no_payload,
-            Signatures=signatures,
+            signatures=signatures,
         )
         _probes_by_name[name] = probe
 
@@ -172,6 +179,7 @@ def load_probes_from_yaml():
 
 
 def build_probes_from_yaml():
+    """Return the cached probes dict, loading it from YAML on first use."""
     if not _probes_by_name:
         load_probes_from_yaml()
     return _probes_by_name

tests/probing/test_loader.py

@@ -58,7 +58,7 @@ def test_loads_probes_from_configured_path(self, tmp_path, monkeypatch):
         assert "NULL" in probes
         assert probes["NULL"].protocol == "tcp"
         assert probes["NULL"].ports == [22]
-        assert len(probes["NULL"].Signatures) == 1
+        assert len(probes["NULL"].signatures) == 1
 
     def test_null_fallback_is_always_appended(self, tmp_path, monkeypatch):
         probes_file = tmp_path / "probes.yaml"
@@ -77,15 +77,15 @@ def test_broken_regex_signature_is_skipped_not_fatal(self, tmp_path, monkeypatch
         probes = load_probes_from_yaml()
 
         assert "BROKEN" in probes
-        assert probes["BROKEN"].Signatures == []
+        assert probes["BROKEN"].signatures == []
 
     def test_compiled_signature_regex_is_usable(self, tmp_path, monkeypatch):
         probes_file = tmp_path / "probes.yaml"
         probes_file.write_text(PROBE_YAML)
         monkeypatch.setattr(loader_module.Config.path, "probes_yaml_file", probes_file)
 
         probes = load_probes_from_yaml()
-        sig = probes["NULL"].Signatures[0]
+        sig = probes["NULL"].signatures[0]
         assert sig.regex.search(b"SSH-2.0\r\n")
 
     def test_second_call_uses_cache(self, tmp_path, monkeypatch):