HTTP 1.0: support HEAD in reconstruction (#4307)

Author: gpotter2

scapy/layers/http.py

@@ -19,7 +19,7 @@
 Note that this layer ISN'T loaded by default, as quite experimental for now.
 
 To follow HTTP packets streams = group packets together to get the
-whole request/answer, use ``TCPSession`` as:
+whole request/answer, use ``TCPSession`` as::
 
     >>> sniff(session=TCPSession)  # Live on-the-flow session
     >>> sniff(offline="./http_chunk.pcap", session=TCPSession)  # pcap
@@ -28,14 +28,14 @@
 and will also decompress the packets when needed.
 Note: on failure, decompression will be ignored.
 
-You can turn auto-decompression/auto-compression off with:
+You can turn auto-decompression/auto-compression off with::
 
     >>> conf.contribs["http"]["auto_compression"] = False
 
 (Defaults to True)
 """
 
-# This file is a modified version of the former scapy_http plugin.
+# This file is a rewritten version of the former scapy_http plugin.
 # It was reimplemented for scapy 2.4.3+ using sessions, stream handling.
 # Original Authors : Steeve Barbeau, Luca Invernizzi
 
@@ -66,6 +66,12 @@
 except ImportError:
     _is_brotli_available = False
 
+try:
+    import lzw
+    _is_lzw_available = True
+except ImportError:
+    _is_lzw_available = False
+
 try:
     import zstandard
     _is_zstd_available = True
@@ -312,8 +318,13 @@ def post_dissect(self, s):
             elif "gzip" in encodings:
                 s = gzip.decompress(s)
             elif "compress" in encodings:
-                import lzw
-                s = lzw.decompress(s)
+                if _is_lzw_available:
+                    s = lzw.decompress(s)
+                else:
+                    log_loading.info(
+                        "Can't import lzw. compress decompression "
+                        "will be ignored !"
+                    )
             elif "br" in encodings:
                 if _is_brotli_available:
                     s = brotli.decompress(s)
@@ -351,8 +362,13 @@ def post_build(self, pkt, pay):
         elif "gzip" in encodings:
             pay = gzip.compress(pay)
         elif "compress" in encodings:
-            import lzw
-            pay = lzw.compress(pay)
+            if _is_lzw_available:
+                pay = lzw.compress(pay)
+            else:
+                log_loading.info(
+                    "Can't import lzw. compress compression "
+                    "will be ignored !"
+                )
         elif "br" in encodings:
             if _is_brotli_available:
                 pay = brotli.compress(pay)
@@ -589,14 +605,22 @@ def dispatch_hook(cls, _pkt=None, *args, **kargs):
     def tcp_reassemble(cls, data, metadata, _):
         detect_end = metadata.get("detect_end", None)
         is_unknown = metadata.get("detect_unknown", True)
+        # General idea of the following is explained at
+        # https://datatracker.ietf.org/doc/html/rfc2616#section-4.4
         if not detect_end or is_unknown:
             metadata["detect_unknown"] = False
             http_packet = cls(data)
             # Detect packing method
             if not isinstance(http_packet.payload, _HTTPContent):
                 return http_packet
+            is_response = isinstance(http_packet.payload, cls.clsresp)
+            # Packets may have a Content-Length we must honnor
             length = http_packet.Content_Length
-            if length is not None:
+            # Heuristic to try and detect instant HEAD responses, as those include a
+            # Content-Length that must not be honored.
+            if is_response and data.endswith(b"\r\n\r\n"):
+                detect_end = lambda _: True
+            elif length is not None:
                 # The packet provides a Content-Length attribute: let's
                 # use it. When the total size of the frags is high enough,
                 # we have the packet
@@ -613,11 +637,10 @@ def tcp_reassemble(cls, data, metadata, _):
                 # It's not Content-Length based. It could be chunked
                 encodings = http_packet[cls].payload._get_encodings()
                 chunked = ("chunked" in encodings)
-                is_response = isinstance(http_packet.payload, cls.clsresp)
                 if chunked:
                     detect_end = lambda dat: dat.endswith(b"0\r\n\r\n")
                 # HTTP Requests that do not have any content,
-                # end with a double CRLF
+                # end with a double CRLF. Same for HEAD responses
                 elif isinstance(http_packet.payload, cls.clsreq):
                     detect_end = lambda dat: dat.endswith(b"\r\n\r\n")
                     # In case we are handling a HTTP Request,

test/pcaps/http_head.pcapng.gz

@@ -68,6 +68,23 @@ assert HTTPResponse in pkt
 print(pkt[Raw].load, expected_data)
 assert pkt[Raw].load == expected_data
 
+= TCPSession - dissect HTTP 1.0 HEAD response
+~ http
+
+load_layer("http")
+
+a = sniff(offline=scapy_path("/test/pcaps/http_head.pcapng.gz"), session=TCPSession)
+
+assert HTTPRequest in a[3]
+assert a[3].Method == b"HEAD"
+assert a[3].User_Agent == b'curl/7.88.1'
+
+assert HTTPResponse in a[5]
+assert a[5].Content_Type == b'text/html; charset=UTF-8'
+assert a[5].Expires == b'Mon, 01 Apr 2024 22:25:38 GMT'
+assert a[5].Reason_Phrase == b'Moved Permanently'
+assert a[5].X_Frame_Options == b"SAMEORIGIN"
+
 = HTTP decompression (gzip)
 
 conf.debug_dissector = True