mitmproxy.tcp

 1import time
 2
 3from mitmproxy import connection, flow
 4from mitmproxy.coretypes import serializable
 5
 6
 7class TCPMessage(serializable.Serializable):
 8    """
 9    An individual TCP "message".
10    Note that TCP is *stream-based* and not *message-based*.
11    For practical purposes the stream is chunked into messages here,
12    but you should not rely on message boundaries.
13    """
14
15    def __init__(self, from_client, content, timestamp=None):
16        self.from_client = from_client
17        self.content = content
18        self.timestamp = timestamp or time.time()
19
20    @classmethod
21    def from_state(cls, state):
22        return cls(*state)
23
24    def get_state(self):
25        return self.from_client, self.content, self.timestamp
26
27    def set_state(self, state):
28        self.from_client, self.content, self.timestamp = state
29
30    def __repr__(self):
31        return "{direction} {content}".format(
32            direction="->" if self.from_client else "<-", content=repr(self.content)
33        )
34
35
36class TCPFlow(flow.Flow):
37    """
38    A TCPFlow is a simplified representation of a TCP session.
39    """
40
41    messages: list[TCPMessage]
42    """
43    The messages transmitted over this connection.
44
45    The latest message can be accessed as `flow.messages[-1]` in event hooks.
46    """
47
48    def __init__(
49        self,
50        client_conn: connection.Client,
51        server_conn: connection.Server,
52        live: bool = False,
53    ):
54        super().__init__(client_conn, server_conn, live)
55        self.messages = []
56
57    _stateobject_attributes = flow.Flow._stateobject_attributes.copy()
58    _stateobject_attributes["messages"] = list[TCPMessage]
59
60    def __repr__(self):
61        return f"<TCPFlow ({len(self.messages)} messages)>"
62
63
64__all__ = [
65    "TCPFlow",
66    "TCPMessage",
67]