| 102 | |
| 103 | |
| 104 | class SMBMessage: |
| 105 | |
| 106 | HEADER_STRUCT_FORMAT = "<4sBIBHHQxxHHHHB" |
| 107 | HEADER_STRUCT_SIZE = struct.calcsize(HEADER_STRUCT_FORMAT) |
| 108 | |
| 109 | log = logging.getLogger('SMB.SMBMessage') |
| 110 | protocol = 1 |
| 111 | |
| 112 | def __init__(self, payload = None): |
| 113 | self.reset() |
| 114 | if payload: |
| 115 | self.payload = payload |
| 116 | self.payload.initMessage(self) |
| 117 | |
| 118 | def __str__(self): |
| 119 | b = StringIO() |
| 120 | b.write('Command: 0x%02X (%s) %s' % ( self.command, SMB_COMMAND_NAMES.get(self.command, '<unknown>'), os.linesep )) |
| 121 | b.write('Status: %s %s' % ( str(self.status), os.linesep )) |
| 122 | b.write('Flags: 0x%02X %s' % ( self.flags, os.linesep )) |
| 123 | b.write('Flags2: 0x%04X %s' % ( self.flags2, os.linesep )) |
| 124 | b.write('PID: %d %s' % ( self.pid, os.linesep )) |
| 125 | b.write('UID: %d %s' % ( self.uid, os.linesep )) |
| 126 | b.write('MID: %d %s' % ( self.mid, os.linesep )) |
| 127 | b.write('TID: %d %s' % ( self.tid, os.linesep )) |
| 128 | b.write('Security: 0x%016X %s' % ( self.security, os.linesep )) |
| 129 | b.write('Parameters: %d bytes %s%s %s' % ( len(self.parameters_data), os.linesep, str(binascii.hexlify(self.parameters_data)), os.linesep )) |
| 130 | b.write('Data: %d bytes %s%s %s' % ( len(self.data), os.linesep, str(binascii.hexlify(self.data)), os.linesep )) |
| 131 | return b.getvalue() |
| 132 | |
| 133 | def reset(self): |
| 134 | self.raw_data = b'' |
| 135 | self.command = 0 |
| 136 | self.status = SMBError() |
| 137 | self.flags = 0 |
| 138 | self.flags2 = 0 |
| 139 | self.pid = 0 |
| 140 | self.tid = 0 |
| 141 | self.uid = 0 |
| 142 | self.mid = 0 |
| 143 | self.security = 0 |
| 144 | self.parameters_data = b'' |
| 145 | self.data = b'' |
| 146 | self.payload = None |
| 147 | |
| 148 | @property |
| 149 | def isAsync(self): |
| 150 | return bool(self.flags & SMB2_FLAGS_ASYNC_COMMAND) |
| 151 | |
| 152 | @property |
| 153 | def isReply(self): |
| 154 | return bool(self.flags & SMB_FLAGS_REPLY) |
| 155 | |
| 156 | @property |
| 157 | def hasExtendedSecurity(self): |
| 158 | return bool(self.flags2 & SMB_FLAGS2_EXTENDED_SECURITY) |
| 159 | |
| 160 | def encode(self): |
| 161 | """ |
no outgoing calls
no test coverage detected