Update session id logic

Updates the session id lookup logic to comply with the MS-SMB2 spec.

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## 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
 
 ## 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/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

@@ -6,6 +6,7 @@
 
 from smbprotocol.file_info import (
     FileAllInformation,
+    FileBasicInformation,
     FileBothDirectoryInformation,
     FileDirectoryInformation,
     FileDispositionInformation,
@@ -27,6 +28,31 @@
 from smbprotocol.structure import DateTimeField
 
 
+class TestFileBasicInformation:
+
+    def test_parse_message(self):
+        data = (
+            b"\x00\xf2\xc4\x22\x2d\x1c\xdb\x01"
+            b"\x00\xf2\xc4\x22\x2d\x1c\xdb\x01"
+            b"\x00\xf2\xc4\x22\x2d\x1c\xdb\x01"
+            b"\x00\xf2\xc4\x22\x2d\x1c\xdb\x01"
+            b"\x10\x00\x00\x00"
+            b"\x00\x00\x00\x00"
+        )
+
+        actual = FileBasicInformation()
+        data = actual.unpack(data)
+        assert data == b""
+        assert len(actual) == 40
+
+        assert actual["creation_time"].get_value() == 133731594120000000
+        assert actual["last_access_time"].get_value() == 133731594120000000
+        assert actual["last_write_time"].get_value() == 133731594120000000
+        assert actual["change_time"].get_value() == 133731594120000000
+        assert actual["file_attributes"].get_value() == 0x10
+        assert actual["reserved"].get_value() == 0
+
+
 class TestFileNameInformation:
     DATA = b"\x08\x00\x00\x00" b"\x63\x00\x61\x00\x66\x00\xe9\x00"