(self)
| 207 | state: Callable[..., layer.CommandGenerator[None]] = state_greet |
| 208 | |
| 209 | def state_auth(self) -> layer.CommandGenerator[None]: |
| 210 | if len(self.buf) < 3: |
| 211 | return |
| 212 | |
| 213 | # Parsing username and password, which is somewhat atrocious |
| 214 | user_len = self.buf[1] |
| 215 | if len(self.buf) < 3 + user_len: |
| 216 | return |
| 217 | pass_len = self.buf[2 + user_len] |
| 218 | if len(self.buf) < 3 + user_len + pass_len: |
| 219 | return |
| 220 | user = self.buf[2 : (2 + user_len)].decode("utf-8", "backslashreplace") |
| 221 | password = self.buf[(3 + user_len) : (3 + user_len + pass_len)].decode( |
| 222 | "utf-8", "backslashreplace" |
| 223 | ) |
| 224 | |
| 225 | data = Socks5AuthData(self.context.client, user, password) |
| 226 | yield Socks5AuthHook(data) |
| 227 | if not data.valid: |
| 228 | # The VER field contains the current **version of the subnegotiation**, which is X'01'. |
| 229 | yield commands.SendData(self.context.client, b"\x01\x01") |
| 230 | yield from self.socks_err("authentication failed") |
| 231 | return |
| 232 | |
| 233 | yield commands.SendData(self.context.client, b"\x01\x00") |
| 234 | self.buf = self.buf[3 + user_len + pass_len :] |
| 235 | self.state = self.state_connect |
| 236 | yield from self.state() |
| 237 | |
| 238 | def state_connect(self) -> layer.CommandGenerator[None]: |
| 239 | # Parse Connect Request |
nothing calls this directly
no test coverage detected