aboutsummaryrefslogtreecommitdiff
path: root/pyfingerd/binds.py
diff options
context:
space:
mode:
Diffstat (limited to 'pyfingerd/binds.py')
-rw-r--r--pyfingerd/binds.py69
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 (