Update session id logic and file stand info parsing

Updates the session id lookup logic to comply with the MS-SMB2 spec.
Also updates the parsing logic for FileStandardInformation to handle the
input data without the reserved fields that servers like Impacket may
return.

CHANGELOG.md

@@ -2,8 +2,9 @@
 
 ## 1.14.1 - TBD
 
-* Add workaround for Impackets logoff response not setting the session id for message verification
+* Update session id lookup logic to comply with MS-SMB2 spec
 * Remove connection from global connection cache even if failing to close it
+* Handle smaller `FileStandardInformation` result parsing on servers that don't return the `reserved` field like Impacket
 
 ## 1.14.0 - 2024-08-26
 

src/smbprotocol/connection.py

@@ -1356,16 +1356,14 @@ def _process_message_thread(self):
                     message_id = header["message_id"].get_value()
                     request = self.outstanding_requests[message_id]
 
-                    # Typically you want to get the Session Id from the first message in a compound request but that is
-                    # unreliable for async responses. Instead get the Session Id from the original request object if
-                    # the Session Id is 0xFFFFFFFFFFFFFFFF.
-                    # https://social.msdn.microsoft.com/Forums/en-US/a580f7bc-6746-4876-83db-6ac209b202c4/mssmb2-change-notify-response-sessionid?forum=os_fileservices
-                    # Impacket also sets session id to 0 on the logoff response
-                    # so fallback to the request for that one
-                    # https://github.com/jborean93/smbprotocol/issues/289#issuecomment-2396040117.
+                    # For SMB2 SESSION_SETUP, the client MUST retrieve SessionId
+                    # from SMB2 header of the response. For all other messages,
+                    # the client MUST retrieve SessionId from the corresponding
+                    # Request.Message.
                     command = header["command"].get_value()
-                    session_id = header["session_id"].get_value()
-                    if session_id == 0xFFFFFFFFFFFFFFFF or (session_id == 0 and command == Commands.SMB2_LOGOFF):
+                    if command == Commands.SMB2_SESSION_SETUP:
+                        session_id = header["session_id"].get_value()
+                    else:
                         session_id = request.session_id
 
                     # No need to waste CPU cycles to verify the signature if we already decrypted the header.

src/smbprotocol/file_info.py

@@ -927,6 +927,14 @@ def __init__(self):
         )
         super().__init__()
 
+    def unpack(self, data):
+        # Some servers may not return the reserved field so we need to add it.
+        # https://github.com/jborean93/smbprotocol/issues/289#issuecomment-2408184608
+        if len(data) == 22:
+            data += b"\x00\x00"
+
+        return super().unpack(data)
+
 
 class FileStreamInformation(Structure):
     """

src/smbprotocol/header.py

@@ -58,6 +58,7 @@ class NtStatus:
     STATUS_STOPPED_ON_SYMLINK = 0x8000002D
     STATUS_INVALID_INFO_CLASS = 0xC0000003
     STATUS_INFO_LENGTH_MISMATCH = 0xC0000004
+    STATUS_INVALID_HANDLE = 0xC0000008
     STATUS_INVALID_PARAMETER = 0xC000000D
     STATUS_NO_SUCH_FILE = 0xC000000F
     STATUS_INVALID_DEVICE_REQUEST = 0xC0000010

tests/test_file_info.py

@@ -736,6 +736,25 @@ def test_parse_message(self):
         assert actual["delete_pending"].get_value() is True
         assert actual["directory"].get_value() is False
 
+    def test_parse_message_no_reserved(self):
+        actual = FileStandardInformation()
+        data = (
+            b"\xc0\x02\x00\x00\x00\x00\x00\x00"
+            b"\xc0\x02\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00"
+            b"\x00"
+            b"\x01"
+        )
+
+        data = actual.unpack(data)
+        assert len(actual) == 24
+        assert data == b""
+        assert actual["allocation_size"].get_value() == 704
+        assert actual["end_of_file"].get_value() == 704
+        assert actual["number_of_links"].get_value() == 0
+        assert actual["delete_pending"].get_value() is False
+        assert actual["directory"].get_value() is True
+
 
 class TestFileFsObjectInformation:
     DATA = (