mitmproxy.tcp

 1import time
 2
 3from mitmproxy import connection
 4from mitmproxy import flow
 5from mitmproxy.coretypes import serializable
 6
 7
 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        )
35
36
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    def get_state(self) -> serializable.State:
59        return {
60            **super().get_state(),
61            "messages": [m.get_state() for m in self.messages],
62        }
63
64    def set_state(self, state: serializable.State) -> None:
65        self.messages = [TCPMessage.from_state(m) for m in state.pop("messages")]
66        super().set_state(state)
67
68    def __repr__(self):
69        return f"<TCPFlow ({len(self.messages)} messages)>"
70
71
72__all__ = [
73    "TCPFlow",
74    "TCPMessage",
75]