SOCKS5: support looking up names remotely #2666
Conversation
jschwartzenberg
force-pushed
the
feature/socks5h
branch
2 times, most recently
from
5e712e4 to
8771bee
Compare
jschwartzenberg
changed the title
jschwartzenberg
force-pushed
the
feature/socks5h
branch
from
8771bee to
d83a269
Compare
jschwartzenberg
changed the title
kafka/conn.py
Outdated
| else: | ||
| self._sock = self._socks5_proxy.socket(None, socket.SOCK_STREAM) | ||
| self._sock_afi = None | ||
| self._sock_addr = [self.host.encode('ascii'), self.port] |
There was a problem hiding this comment.
recommend you defer str.encode() until needed by socks code.
There was a problem hiding this comment.
I moved this as you suggested. Let me know what you think!
kafka/conn.py
Outdated
| return self.state | ||
| else: | ||
| if self.config["socks5_proxy"] is None or not self._socks5_proxy.use_remote_lookup(): | ||
| next_lookup = self._next_afi_sockaddr() |
There was a problem hiding this comment.
Would it be simpler to move the self._socks5_proxy.use_remote_lookup() inside of _next_afi_sockaddr() ? For socks proxy w/ remote_lookup, it should return (socket.AF_UNSPEC, (self.host, self.port)). With that change I think most of these other changes would be unnecessary.
There was a problem hiding this comment.
Thanks, that indeed reduces the changes quite a bit! Let me know what you think of the current version!
kafka/conn.py
Outdated
| @@ -862,7 +873,7 @@ def connection_delay(self): | |||
| large number to handle slow/stalled connections. | |||
| """ | |||
| if self.disconnected() or self.connecting(): | |||
| if len(self._gai) > 0: | |||
| if len(self._gai) > 0 or self._socks5_proxy.use_remote_lookup(): | |||
There was a problem hiding this comment.
won't this crash if/when self._socks5_proxy is None and self._gai is empty?
There was a problem hiding this comment.
You're completely right. I added some extra logic now to check whether self._socks5_proxy is set.
kafka/conn.py
Outdated
|
|
||
| for option in self.config['socket_options']: | ||
| log.debug('%s: setting socket option %s', self, option) | ||
| self._sock.setsockopt(*option) | ||
|
|
||
| self._sock.setblocking(False) | ||
| self.config['state_change_callback'](self.node_id, self._sock, self) | ||
| if self._sock_afi != None: |
There was a problem hiding this comment.
generally should check is or is not for None, not equality.
There was a problem hiding this comment.
I got rid of the None variant for the socket family now.
kafka/conn.py
Outdated
| @@ -1291,9 +1302,13 @@ def check_version(self, timeout=2, **kwargs): | |||
| return self._api_version | |||
|
|
|||
| def __str__(self): | |||
| if self._sock_afi != None: | |||
There was a problem hiding this comment.
implied suggestion above is to use socket.AF_UNSPEC instead of None
|
Many thanks for your review! I'll rework and push a better commit! |
jschwartzenberg
force-pushed
the
feature/socks5h
branch
from
d83a269 to
f3d9fb2
Compare
Hi all, I'm dealing with a Kafka cluster behind a SOCKS proxy where the broker names cannot locally be resolved.
This PR implements resolving the name remotely. It is probably rather hacky though, so I appreciate any feedback to improve it and will make changes accordingly.