(self, ssp_tuple)
| 487 | |
| 488 | @ATMT.action(should_send_session_setup_request) |
| 489 | def send_setup_session_request(self, ssp_tuple): |
| 490 | self.session.sspcontext, token, status = ssp_tuple |
| 491 | if self.SMB2 and status == GSS_S_CONTINUE_NEEDED: |
| 492 | # New session: force 0 |
| 493 | self.SessionId = 0 |
| 494 | if self.SMB2 or self.EXTENDED_SECURITY: |
| 495 | # SMB1 extended / SMB2 |
| 496 | if self.SMB2: |
| 497 | # SMB2 |
| 498 | pkt = self.smb_header.copy() / SMB2_Session_Setup_Request( |
| 499 | Capabilities="DFS", |
| 500 | SecurityMode=( |
| 501 | "SIGNING_ENABLED+SIGNING_REQUIRED" |
| 502 | if self.session.SigningRequired |
| 503 | else "SIGNING_ENABLED" |
| 504 | ), |
| 505 | ) |
| 506 | else: |
| 507 | # SMB1 extended |
| 508 | pkt = ( |
| 509 | self.smb_header.copy() |
| 510 | / SMBSession_Setup_AndX_Request_Extended_Security( |
| 511 | ServerCapabilities=( |
| 512 | "UNICODE+NT_SMBS+STATUS32+LEVEL_II_OPLOCKS+" |
| 513 | "DYNAMIC_REAUTH+EXTENDED_SECURITY" |
| 514 | ), |
| 515 | NativeOS=b"", |
| 516 | NativeLanMan=b"", |
| 517 | ) |
| 518 | ) |
| 519 | pkt.SecurityBlob = token |
| 520 | else: |
| 521 | # Non-extended security. |
| 522 | pkt = self.smb_header.copy() / SMBSession_Setup_AndX_Request( |
| 523 | ServerCapabilities="UNICODE+NT_SMBS+STATUS32+LEVEL_II_OPLOCKS", |
| 524 | NativeOS=b"", |
| 525 | NativeLanMan=b"", |
| 526 | OEMPassword=b"\0" * 24, |
| 527 | UnicodePassword=token, |
| 528 | ) |
| 529 | self.send(pkt) |
| 530 | if self.SMB2: |
| 531 | # If required, compute sessions |
| 532 | self.session.computeSMBSessionPreauth( |
| 533 | bytes(pkt[SMB2_Header]), # session request |
| 534 | ) |
| 535 | |
| 536 | @ATMT.receive_condition(SENT_SESSION_REQUEST) |
| 537 | def receive_session_setup_response(self, pkt): |
nothing calls this directly
no test coverage detected