networking.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. import socket
  2. from contextlib import closing
  3. from ipaddress import ip_address
  4. from typing import Optional, Sequence
  5. from multiaddr import Multiaddr
  6. Hostname, Port = str, int # flavour types
  7. Endpoint = str # e.g. 1.2.3.4:1337 or [2a21:6с8:b192:2105]:8888, https://networkengineering.stackexchange.com/a/9435
  8. LOCALHOST = "127.0.0.1"
  9. def get_port(endpoint: Endpoint) -> Optional[Port]:
  10. """get port or None if port is undefined"""
  11. # TODO: find a standard way to get port, make sure it works in malformed ports
  12. try:
  13. return int(endpoint[endpoint.rindex(":") + 1 :], base=10)
  14. except ValueError: # :* or not specified
  15. return None
  16. def replace_port(endpoint: Endpoint, new_port: Port) -> Endpoint:
  17. assert endpoint.endswith(":*") or get_port(endpoint) is not None, endpoint
  18. return f"{endpoint[:endpoint.rindex(':')]}:{new_port}"
  19. def strip_port(endpoint: Endpoint) -> Hostname:
  20. """Removes port from the end of endpoint. If port is not specified, does nothing"""
  21. maybe_port = endpoint[endpoint.rindex(":") + 1 :]
  22. return endpoint[: endpoint.rindex(":")] if maybe_port.isdigit() or maybe_port == "*" else endpoint
  23. def find_open_port(params=(socket.AF_INET, socket.SOCK_STREAM), opt=(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)):
  24. """Finds a tcp port that can be occupied with a socket with *params and use *opt options"""
  25. try:
  26. with closing(socket.socket(*params)) as sock:
  27. sock.bind(("", 0))
  28. sock.setsockopt(*opt)
  29. return sock.getsockname()[1]
  30. except Exception as e:
  31. raise e
  32. def choose_ip_address(
  33. maddrs: Sequence[Multiaddr], prefer_global: bool = True, protocol_priority: Sequence[str] = ("ip4", "ip6")
  34. ) -> Hostname:
  35. """
  36. Currently, some components of hivemind are not converted to work over libp2p and use classical networking.
  37. To allow other peers reach a server when needed, these components announce a machine's IP address.
  38. This function automatically selects the best IP address to announce among publicly visible multiaddrs
  39. of this machine identified by libp2p (typically, using the ``P2P.get_visible_maddrs()`` method),
  40. so a user does not need to define this address manually (unless the user wants to).
  41. The best IP address is chosen using the following logic:
  42. - Prefer IP addresses from global address blocks
  43. (in terms of https://docs.python.org/3/library/ipaddress.html#ipaddress.IPv4Address.is_global)
  44. - Among the IP addresses of the same globality status, prefer IPv4 addresses over IPv6
  45. If the default logic does not suit you, it is recommended to set the announced IP address manually.
  46. """
  47. for need_global in [prefer_global, not prefer_global]:
  48. for protocol in protocol_priority:
  49. for addr in maddrs:
  50. if protocol in addr.protocols():
  51. value_for_protocol = addr[protocol]
  52. if ip_address(value_for_protocol).is_global == need_global:
  53. return value_for_protocol
  54. raise ValueError(f"No IP address found among given multiaddrs: {maddrs}")