mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-30 03:14:22 +00:00
5192810ff6
We get little benefit from our mypy QA checks at the moment, because we skip imports. This patch is what's needed to make mypy succeed with imports on a single file: master.py It also updates mypy to the current version, and enables a QA check. Mypy bugs I encountered: dict.update with kwargs not supported: https://github.com/python/mypy/issues/1031 property setters and getters must be adjacent: https://github.com/python/mypy/issues/1465
48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
import time
|
|
|
|
from typing import List
|
|
|
|
from mitmproxy import flow
|
|
from mitmproxy.types import serializable
|
|
|
|
|
|
class TCPMessage(serializable.Serializable):
|
|
|
|
def __init__(self, from_client, content, timestamp=None):
|
|
self.from_client = from_client
|
|
self.content = content
|
|
self.timestamp = timestamp or time.time()
|
|
|
|
@classmethod
|
|
def from_state(cls, state):
|
|
return cls(*state)
|
|
|
|
def get_state(self):
|
|
return self.from_client, self.content, self.timestamp
|
|
|
|
def set_state(self, state):
|
|
self.from_client, self.content, self.timestamp = state
|
|
|
|
def __repr__(self):
|
|
return "{direction} {content}".format(
|
|
direction="->" if self.from_client else "<-",
|
|
content=repr(self.content)
|
|
)
|
|
|
|
|
|
class TCPFlow(flow.Flow):
|
|
|
|
"""
|
|
A TCPFlow is a simplified representation of a TCP session.
|
|
"""
|
|
|
|
def __init__(self, client_conn, server_conn, live=None):
|
|
super().__init__("tcp", client_conn, server_conn, live)
|
|
self.messages = [] # type: List[TCPMessage]
|
|
|
|
_stateobject_attributes = flow.Flow._stateobject_attributes.copy()
|
|
_stateobject_attributes["messages"] = List[TCPMessage]
|
|
|
|
def __repr__(self):
|
|
return "<TCPFlow ({} messages)>".format(len(self.messages))
|