For an assignment (not school), I was supposed to make a "web browser" using Scapy.
While storing the response, I initially thought of
# got an array of packets using sniff() named responses
ans = ""
for pack in responses:
ans += pack[Raw].load
But that clearly doesn't work if one of the ACKs gets lost or with any of that kind of edge cases where I receive more than one of the same packet or where I receive them in the wrong order. So I wrote a "buffer" class to help handle that.
Code:
class MessageBuffer:
"""
Used to store a buffer or message
"""
def __init__(self, init_key):
"""
Creates a MessageBuffer object that will hold a buffer
containing all the message that will be received.
:param init_key: Initial sequence number
"""
self._buffer = []
self._init_key = init_key
self.unsafe = False # Can be set to be able to convert incomplete buffer
def __setitem__(self, key, value):
"""
Adds a message into the buffer, automatically extending the storage if needed.
:param key: the starting point
:param value: the string to insert
"""
key -= self._init_key
if key < 0:
raise KeyError("Key cannot be less than initial key")
if key > len(self._buffer): # Missing values
while key > len(self._buffer):
self._buffer.append("") # Extend till Ok
self._buffer.extend(value) # Add all string at the end.
else:
try:
while value:
self._buffer[key] = value[0]
key += 1
value = value[1:]
except IndexError: # End of self._buffer
self._buffer.extend(value)
def is_complete(self):
"""
:return: True if buffer is complete, otherwise False
"""
return "" not in self._buffer
def __str__(self):
"""
Converts the buffer into a string.
"""
if not self.unsafe and not self.is_complete():
raise Exception("Buffer isn't complete")
return "".join(self._buffer)
I don't have particular concerns, but maybe I have some kind of bad practice or something like that in the code? Also, no usage of the class but in pseudo-python-code (since copying the whole code is pretty long and not what I'm asking about)
def handler_for_packet(packet):
buffer[packet[TCP].seq] = packet[Raw].load
if buffer.is_complete():
send_ack_for_packet(packet)