Created
February 25, 2021 14:37
-
-
Save adiroiban/151b04c0bf2b5d59240cd253ee446c10 to your computer and use it in GitHub Desktop.
pyradius - no timeout error for invalid shared secret
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class ChevahRadius(radius.Radius): | |
""" | |
Patched version while the upstream project is re-enabled. | |
""" | |
def send_message(self, message): | |
""" | |
Raise VerificationError if we got a response but couldn't be validated, | |
instead of raising the same error as timeout. | |
""" | |
send = message.pack() | |
addrs = socket.getaddrinfo( | |
self.host, | |
self.port, | |
0, | |
socket.SOCK_DGRAM, | |
) | |
@contextlib.contextmanager | |
def connect(res): | |
af, socktype, proto, canonname, sa = res | |
sock = None | |
try: | |
sock = socket.socket(af, socktype, proto) | |
sock.settimeout(self.timeout) | |
sock.connect(sa) | |
yield sock | |
finally: | |
if sock is not None: | |
sock.close() | |
def attempt(res): | |
with connect(res) as c: | |
c.send(send) | |
recv = c.recv(radius.PACKET_MAX) | |
return message.verify(recv) | |
err = None | |
for i in range(1, self.retries + 1): | |
for res in addrs: | |
try: | |
return attempt(res) | |
except socket.timeout: | |
# Just try again on timeout. | |
err = None | |
except radius.VerificationError as e: | |
if isinstance(err, radius.VerificationError): | |
# No need to retry, as most probably we have an | |
# invalid shared secret. | |
break | |
# Silently discard invalid replies (as RFC states). | |
err = e | |
except socket.error as e: | |
err = radius.SocketError(e) | |
else: | |
# Inner loop not broken. | |
continue | |
break | |
if err is not None: | |
raise err | |
raise radius.NoResponse() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment