(self, *, listening_addr: str='127.0.0.1',
listening_port: typing.Optional[int]=None,
backend_host: str, backend_port: int,
settings: typing.Optional[dict]=None)
| 19 | |
| 20 | class TCPFuzzingProxy: |
| 21 | def __init__(self, *, listening_addr: str='127.0.0.1', |
| 22 | listening_port: typing.Optional[int]=None, |
| 23 | backend_host: str, backend_port: int, |
| 24 | settings: typing.Optional[dict]=None) -> None: |
| 25 | self.listening_addr = listening_addr |
| 26 | self.listening_port = listening_port |
| 27 | self.backend_host = backend_host |
| 28 | self.backend_port = backend_port |
| 29 | self.settings = settings or {} |
| 30 | self.loop = None |
| 31 | self.connectivity = None |
| 32 | self.connectivity_loss = None |
| 33 | self.stop_event = None |
| 34 | self.connections = {} |
| 35 | self.sock = None |
| 36 | self.listen_task = None |
| 37 | |
| 38 | async def _wait(self, work): |
| 39 | work_task = asyncio.ensure_future(work) |
nothing calls this directly
no outgoing calls
no test coverage detected