Edit on GitHub

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]
class TCPFlow(mitmproxy.flow.Flow):
37class TCPFlow(flow.Flow):
38    """
39    A TCPFlow is a simplified representation of a TCP session.
40    """
41
42    messages: list[TCPMessage]
43    """
44    The messages transmitted over this connection.
45
46    The latest message can be accessed as `flow.messages[-1]` in event hooks.
47    """
48
49    def __init__(
50        self,
51        client_conn: connection.Client,
52        server_conn: connection.Server,
53        live: bool = False,
54    ):
55        super().__init__(client_conn, server_conn, live)
56        self.messages = []
57
58    _stateobject_attributes = flow.Flow._stateobject_attributes.copy()
59    _stateobject_attributes["messages"] = list[TCPMessage]
60
61    def __repr__(self):
62        return f"<TCPFlow ({len(self.messages)} messages)>"

A TCPFlow is a simplified representation of a TCP session.

TCPFlow( client_conn: mitmproxy.connection.Client, server_conn: mitmproxy.connection.Server, live: bool = False)
49    def __init__(
50        self,
51        client_conn: connection.Client,
52        server_conn: connection.Server,
53        live: bool = False,
54    ):
55        super().__init__(client_conn, server_conn, live)
56        self.messages = []
messages: list[mitmproxy.tcp.TCPMessage]

The messages transmitted over this connection.

The latest message can be accessed as flow.messages[-1] in event hooks.

class TCPMessage(mitmproxy.coretypes.serializable.Serializable):
 8class TCPMessage(serializable.Serializable):
 9    """
10    An individual TCP "message".
11    Note that TCP is *stream-based* and not *message-based*.
12    For practical purposes the stream is chunked into messages here,
13    but you should not rely on message boundaries.
14    """
15
16    def __init__(self, from_client, content, timestamp=None):
17        self.from_client = from_client
18        self.content = content
19        self.timestamp = timestamp or time.time()
20
21    @classmethod
22    def from_state(cls, state):
23        return cls(*state)
24
25    def get_state(self):
26        return self.from_client, self.content, self.timestamp
27
28    def set_state(self, state):
29        self.from_client, self.content, self.timestamp = state
30
31    def __repr__(self):
32        return "{direction} {content}".format(
33            direction="->" if self.from_client else "<-", content=repr(self.content)
34        )

An individual TCP "message". Note that TCP is stream-based and not message-based. For practical purposes the stream is chunked into messages here, but you should not rely on message boundaries.

TCPMessage(from_client, content, timestamp=None)
16    def __init__(self, from_client, content, timestamp=None):
17        self.from_client = from_client
18        self.content = content
19        self.timestamp = timestamp or time.time()
Inherited Members
mitmproxy.coretypes.serializable.Serializable
copy