Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def payload(self):
# type: () -> binary_type
wbuf = WriteBuf()
self.write(wbuf)
return wbuf.write_flush()
def __init__(self, data=None):
# type: (Optional[binary_type]) -> None
super(WriteBuf, self).__init__()
self._wbuf = BytesIO(data) if data else BytesIO()
def read_packet(self, sshv=2):
# type: (int) -> Tuple[int, binary_type]
try:
header = WriteBuf()
self.ensure_read(4)
packet_length = self.read_int()
header.write_int(packet_length)
# XXX: validate length
if sshv == 1:
padding_length = (8 - packet_length % 8)
self.ensure_read(padding_length)
padding = self.read(padding_length)
header.write(padding)
payload_length = packet_length
check_size = padding_length + payload_length
else:
self.ensure_read(1)
padding_length = self.read_byte()
header.write_byte(padding_length)
payload_length = packet_length - padding_length - 1
def payload(self):
# type: () -> binary_type
wbuf = WriteBuf()
self.write(wbuf)
return wbuf.write_flush()