(self)
| 22 | return True |
| 23 | |
| 24 | def state_pre_connect(self): |
| 25 | # Get the response |
| 26 | if self.read_buf[0:1] != chr(0x00).encode(): |
| 27 | # bad data |
| 28 | self.close() |
| 29 | raise GeneralProxyError(1) |
| 30 | elif self.read_buf[1:2] != chr(0x5A).encode(): |
| 31 | # Connection failed |
| 32 | self.close() |
| 33 | if ord(self.read_buf[1:2]) in (91, 92, 93): |
| 34 | # socks 4 error |
| 35 | raise Socks4aError(ord(self.read_buf[1:2]) - 90) |
| 36 | else: |
| 37 | raise Socks4aError(4) |
| 38 | # Get the bound address/port |
| 39 | self.boundport = struct.unpack(">H", self.read_buf[2:4])[0] |
| 40 | self.boundaddr = self.read_buf[4:] |
| 41 | self.__proxysockname = (self.boundaddr, self.boundport) |
| 42 | if self.ipaddr: |
| 43 | self.__proxypeername = (socket.inet_ntoa(self.ipaddr), self.destination[1]) |
| 44 | else: |
| 45 | self.__proxypeername = (self.destination[0], self.destport) |
| 46 | self.set_state("proxy_handshake_done", length=8) |
| 47 | return True |
| 48 | |
| 49 | def proxy_sock_name(self): |
| 50 | return socket.inet_ntoa(self.__proxysockname[0]) |
no test coverage detected