diff options
Diffstat (limited to 'pyfingerd/binds.py')
-rw-r--r-- | pyfingerd/binds.py | 69 |
1 files changed, 26 insertions, 43 deletions
diff --git a/pyfingerd/binds.py b/pyfingerd/binds.py index 4d22c93..dd81e58 100644 --- a/pyfingerd/binds.py +++ b/pyfingerd/binds.py @@ -3,11 +3,10 @@ # Copyright (C) 2017-2022 Thomas Touhey <thomas@touhey.fr> # This file is part of the pyfingerd project, which is MIT-licensed. # ***************************************************************************** -""" Binds decoder for the finger server. """ +"""Binds decoder for the finger server.""" import socket as _socket - -from typing import Sequence as _Sequence +import typing as _t from .errors import InvalidBindError as _InvalidBindError @@ -18,19 +17,18 @@ __all__ = [ class FingerBind: - """ Bind address for pyfingerd. """ + """Bind address for pyfingerd.""" @property - def runserver_params(self): - """ Return the data as ``_runserver`` arguments. """ - - raise NotImplementedError + def runserver_params(self) -> _t.Tuple[_socket.AddressFamily, str, int]: + """Return the data as ``start_server`` arguments.""" + raise NotImplementedError() class FingerTCPv4Bind(FingerBind): - """ IPv4 TCP Address. """ + """IPv4 TCP Address.""" - def __init__(self, address, port): + def __init__(self, address: str, port: int): try: self._addr = _socket.inet_pton(_socket.AF_INET, address) except Exception: @@ -39,9 +37,8 @@ class FingerTCPv4Bind(FingerBind): self._port = port @property - def runserver_params(self): - """ Return the data as `_runserver` parameters. """ - + def runserver_params(self) -> _t.Tuple[_socket.AddressFamily, str, int]: + """Return the data as ``start_server`` parameters.""" return ( _socket.AF_INET, _socket.inet_ntop(_socket.AF_INET, self._addr), @@ -50,9 +47,9 @@ class FingerTCPv4Bind(FingerBind): class FingerTCPv6Bind(FingerBind): - """ IPv6 TCP Address. """ + """IPv6 TCP Address.""" - def __init__(self, address, port): + def __init__(self, address: str, port: int): try: self._addr = _socket.inet_pton(_socket.AF_INET6, address) except Exception: @@ -61,9 +58,8 @@ class FingerTCPv6Bind(FingerBind): self._port = port @property - def runserver_params(self): - """ Return the data as `_runserver` parameters. """ - + def runserver_params(self) -> _t.Tuple[_socket.AddressFamily, str, int]: + """Return the data as `start_server` parameters.""" return ( _socket.AF_INET6, _socket.inet_ntop(_socket.AF_INET6, self._addr), @@ -72,7 +68,7 @@ class FingerTCPv6Bind(FingerBind): class FingerBindsDecoder: - """ Binds decoder for pyfingerd. """ + """Binds decoder for pyfingerd.""" def __init__(self, proto: str = 'finger'): proto = proto.casefold() @@ -81,9 +77,8 @@ class FingerBindsDecoder: self._proto = proto - def decode(self, raw: str) -> _Sequence[FingerBind]: - """ Get binds for the server, using a given string. """ - + def decode(self, raw: str) -> _t.Sequence[FingerBind]: + """Get binds for the server, using a given string.""" binds = set() for addr in map(lambda x: x.strip(), raw.split(',')): @@ -91,16 +86,14 @@ class FingerBindsDecoder: continue # Try to find a scheme. - scheme, *rest = addr.split(':/') if not rest: # No scheme found, let's just guess the scheme based on # the situation. - raw = scheme scheme = {'finger': 'tcp'}[self._proto] else: - # just don't add the ':' of ':/' again + # Just don't add the ':' of ':/' again. raw = '/' + ':/'.join(rest) if ( @@ -110,11 +103,10 @@ class FingerBindsDecoder: raise _InvalidBindError( addr, f'Unsupported scheme {scheme!r} for ' - f'protocol {self._proto!r}', + + f'protocol {self._proto!r}', ) # Decode the address data. - if scheme == 'tcp': binds.update(self._decode_tcp_host(raw)) @@ -124,8 +116,7 @@ class FingerBindsDecoder: return f'{self._class__.__name__}()' def _decode_tcp_host(self, x): - """ Decode suitable hosts for a TCP bind. """ - + """Decode suitable hosts for a TCP bind.""" addrs = () addr = x @@ -133,15 +124,12 @@ class FingerBindsDecoder: # TODO: decode hosts without the default host. # Get the host part first, we'll decode it later. - if x[0] == '[': # The host part is an IPv6, look for the closing ']' and # decode it later. - to = x.find(']') if to < 0: - raise _InvalidBindError( - addr, "Expected closing ']'") + raise _InvalidBindError(addr, "Expected closing ']'") host = x[1:to] x = x[to + 1:] @@ -150,14 +138,12 @@ class FingerBindsDecoder: else: # The host part is either an IPv4 or a host name, look for # the ':' and decode it later. - host, *x = x.split(':') x = ':' + ':'.join(x) is_ipv6 = False # Decode the port part. - if x in ('', ':'): port = 79 elif x[0] == ':': @@ -167,30 +153,26 @@ class FingerBindsDecoder: try: if x[1:] != '': raise AssertionError('Expected a port number') + port = _socket.getservbyname(x[1:]) except Exception: raise _InvalidBindError( addr, 'Expected a valid port number or name ' - f'(got {x[1:]!r})', + + f'(got {x[1:]!r})', ) from None else: - raise _InvalidBindError( - addr, 'Garbage found after the host', - ) + raise _InvalidBindError(addr, 'Garbage found after the host') # Decode the host part and get the addresses. - addrs = () if is_ipv6: # Decode the IPv6 address (validate it using `_socket.inet_pton`). - ip6 = host _socket.inet_pton(_socket.AF_INET6, host) addrs += (FingerTCPv6Bind(ip6, port),) else: # Decode the host (try IPv4, otherwise, resolve domain). - try: ip = host.split('.') if len(ip) < 2 or len(ip) > 4: @@ -210,7 +192,8 @@ class FingerBindsDecoder: entries = _socket.getaddrinfo( host, port, proto=_socket.IPPROTO_TCP, - type=_socket.SOCK_STREAM) + type=_socket.SOCK_STREAM, + ) for ent in entries: if ( |