(
self,
message: str,
reply_code: int | None = None,
)
| 134 | buf: bytes = b"" |
| 135 | |
| 136 | def socks_err( |
| 137 | self, |
| 138 | message: str, |
| 139 | reply_code: int | None = None, |
| 140 | ) -> layer.CommandGenerator[None]: |
| 141 | if reply_code is not None: |
| 142 | yield commands.SendData( |
| 143 | self.context.client, |
| 144 | bytes([SOCKS5_VERSION, reply_code]) |
| 145 | + b"\x00\x01\x00\x00\x00\x00\x00\x00", |
| 146 | ) |
| 147 | yield commands.CloseConnection(self.context.client) |
| 148 | yield commands.Log(message) |
| 149 | self._handle_event = self.done |
| 150 | |
| 151 | @expect(events.Start, events.DataReceived, events.ConnectionClosed) |
| 152 | def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]: |
no outgoing calls
no test coverage detected