Fix ICMP false positives by validating response type

The socket_icmp() method treated any ICMP response as host detection,
causing false positives when firewalls sent ICMP Type 3 errors.

Code only validated packet_id match, not ICMP response type.
RFC 792 specifies only Type 0 (Echo Reply) indicates host is alive.

1. Initialize delay/type/code to None (prevents UnboundLocalError)
2. Capture ICMP type and code from response packet
3. Validate type before declaring host alive
4. Only ICMP Echo Reply (type 0) = host alive
5. ICMP Type 3 errors = host unreachable
6. No response (timeout) = filtered/blocked

- Modified socket_icmp() to extract ICMP type/code
- Added validation: status = 'alive' only if type == 0
- Updated return to include icmp_type, icmp_code, status fields
- Modified response_conditions_matched() to validate status field
- Added 4 comprehensive unit tests

- Unit tests: 4 new tests cover all scenarios
  * Echo Reply (type 0) = match
  * Destination Unreachable (type 3) = no match
  * Network Unreachable (type 3, code 0) = no match
  * No response (timeout) = no match
- Integration tests: Validated with real hosts
- RFC 792 compliance: Only type 0 marked as alive

- Eliminates false positives from firewall error responses
- More accurate ICMP scanning results
- Better error diagnostics with type/code fields

Fixes #1186

Author: webbsssss

nettacker/core/lib/socket.py

@@ -205,6 +205,10 @@ def socket_icmp(self, host, timeout):
             header + data, (socket.gethostbyname(host), 1)
         )  # Don't know about the 1
 
+        delay = None
+        icmp_response_type = None
+        icmp_response_code = None
+
         while True:
             started_select = time.time()
             what_ready = select.select([socket_connection], [], [], timeout)
@@ -221,17 +225,42 @@ def socket_icmp(self, host, timeout):
                 packet_id,
                 packet_sequence,
             ) = struct.unpack("bbHHh", icmp_header)
+
             if packet_id == random_integer:
                 packet_bytes = struct.calcsize("d")
                 time_sent = struct.unpack("d", received_packet[28 : 28 + packet_bytes])[0]
                 delay = time_received - time_sent
+                # Store ICMP type and code for validation
+                icmp_response_type = packet_type
+                icmp_response_code = packet_code
                 break
 
             timeout = timeout - how_long_in_select
             if timeout <= 0:
                 break
         socket_connection.close()
-        return {"host": host, "response_time": delay, "ssl_flag": False}
+        
+
+        # Return response with ICMP type validation
+        if delay is not None:
+            status = "alive" if icmp_response_type == 0 else "unreachable"
+            return {
+                "host": host,
+                "response_time": delay,
+                "ssl_flag": False,
+                "icmp_type": icmp_response_type,
+                "icmp_code": icmp_response_code,
+                "status": status,
+            }
+        else:
+            return {
+                "host": host,
+                "response_time": None,
+                "ssl_flag": False,
+                "icmp_type": None,
+                "icmp_code": None,
+                "status": "filtered",
+            }
 
 
 class SocketEngine(BaseEngine):
@@ -288,7 +317,22 @@ def response_conditions_matched(self, sub_step, response):
                     return condition_results if condition_results else []
                 return []
         if sub_step["method"] == "socket_icmp":
-            return response
+            # Only count ICMP Echo Reply (type 0) as success
+            if isinstance(response, dict) and response.get("status") == "alive":
+                return response
+            else:
+                # Log non-Echo replies for debugging
+                if isinstance(response, dict):
+                    log.debug(
+                        "ICMP response from {}: type={}, code={}, status={} - "
+                        "not counting as 'alive' (only ICMP Echo Reply type 0 counts)".format(
+                            response.get("host"),
+                            response.get("icmp_type"),
+                            response.get("icmp_code"),
+                            response.get("status"),
+                        )
+                    )
+                return []
         return []
 
     def apply_extra_data(self, sub_step, response):

tests/core/lib/test_socket.py

@@ -6,7 +6,46 @@
 
 
 class Responses:
-    tcp_connect_only = socket_icmp = {}
+    tcp_connect_only = {}
+
+    # ICMP responses with type/code validation
+    socket_icmp = {}
+    
+    socket_icmp_echo_reply = {
+        "host": "127.0.0.1",
+        "response_time": 0.001,
+        "ssl_flag": False,
+        "icmp_type": 0,       # Echo Reply - SUCCESS
+        "icmp_code": 0,
+        "status": "alive",
+    }
+    
+    socket_icmp_dest_unreachable = {
+        "host": "127.0.0.1",
+        "response_time": None,
+        "ssl_flag": False,
+        "icmp_type": 3,       # Destination Unreachable
+        "icmp_code": 3,       # Port Unreachable
+        "status": "unreachable",
+    }
+    
+    socket_icmp_network_unreachable = {
+        "host": "192.0.2.1",
+        "response_time": None,
+        "ssl_flag": False,
+        "icmp_type": 3,       # Destination Unreachable
+        "icmp_code": 0,       # Network Unreachable
+        "status": "unreachable",
+    }
+    
+    socket_icmp_no_response = {
+        "host": "10.255.255.1",
+        "response_time": None,
+        "ssl_flag": False,
+        "icmp_type": None,
+        "icmp_code": None,
+        "status": "filtered",
+    }
 
     tcp_connect_send_and_receive = {
         "response": 'HTTP/1.1 400 Bad Request\r\nServer: Apache/2.4.62 (Debian)\r\nContent-Length: 302\r\nConnection: close\r\nContent-Type: text/html; charset=iso-8859-1\r\n\r\n<!DOCTYPE HTML PUBLIC "-//IETF//DTD HTML 2.0//EN">\n<html><head>\n<title>400 Bad Request</title>\n</head><body>\n<h1>Bad Request</h1>\n<p>Your browser sent a request that this server could not understand.<br />\n</p>\n<hr>\n<address>Apache/2.4.62 (Debian)</address>\n</body></html>\n',
@@ -153,11 +192,31 @@ def test_create_tcp_socket(self, mock_wrap, mock_socket):
         socket_instance.connect.assert_called_with((HOST, PORT))
         mock_wrap.assert_called_with(socket_instance)
 
-    def test_response_conditions_matched_socket_icmp(self, socket_engine, substeps, responses):
-        result = socket_engine.response_conditions_matched(
-            substeps.socket_icmp, responses.socket_icmp
-        )
-        assert result == responses.socket_icmp
+    def test_socket_icmp_echo_reply_should_match(self, socket_engine, substeps):
+        """ICMP Echo Reply (type 0) should be treated as successful detection"""
+        response = Responses.socket_icmp_echo_reply
+        result = socket_engine.response_conditions_matched(substeps.socket_icmp, response)
+        assert result == response  # Should return the response (match)
+        assert result.get("status") == "alive"
+        assert result.get("icmp_type") == 0
+
+    def test_socket_icmp_dest_unreachable_should_not_match(self, socket_engine, substeps):
+        """ICMP Destination Unreachable (type 3) should NOT be treated as success"""
+        response = Responses.socket_icmp_dest_unreachable
+        result = socket_engine.response_conditions_matched(substeps.socket_icmp, response)
+        assert result == []  # Should return empty list (no match)
+
+    def test_socket_icmp_network_unreachable_should_not_match(self, socket_engine, substeps):
+        """ICMP Network Unreachable should NOT be treated as success"""
+        response = Responses.socket_icmp_network_unreachable
+        result = socket_engine.response_conditions_matched(substeps.socket_icmp, response)
+        assert result == []  # Should return empty list (no match)
+
+    def test_socket_icmp_no_response_should_not_match(self, socket_engine, substeps):
+        """No ICMP response (filtered/timeout) should NOT be treated as success"""
+        response = Responses.socket_icmp_no_response
+        result = socket_engine.response_conditions_matched(substeps.socket_icmp, response)
+        assert result == []  # Should return empty list (no match)
 
     def test_response_conditions_matched_tcp_connect_send_and_receive(
         self, socket_engine, substeps, responses