API Reference

Every public name, generated from the package's own docstrings.

For the contracts and gotchas in prose form — including the platform differences that are easy to get wrong — see src/netimps/AGENTS.md in the repository.

netimps -- small, self-contained network utilities.

A thin, typed convenience layer over the standard library's :mod:ipaddress plus a handful of host helpers (DNS lookup, ping, interface discovery). One flat import surface; the only runtime dependency is dnspython, used solely inside :func:resolve.

::

import netimps

netimps.parse("10.0.0.5")                           # -> IPv4Address
netimps.MACAddress("AA:BB:CC:DD:EE:FF").as_str("-")
netimps.resolve("example.com", "aaaa")
for iface in netimps.get_interfaces():
    print(iface.name, iface.mac, iface.ips)

Types and parsing

IPAddress/IPInterface/IPNetwork are the v4/v6 unions you annotate with, reading the way :class:ipaddress.IPv4Address does::

def get_route(dst: netimps.IPAddress, via: netimps.IPNetwork) -> None: ...

The same names are what you parse into, via one entry point::

parse(value, IPNetwork)              # raises on bad input
try_parse(value, IPNetwork)          # None instead
is_valid(value, IPNetwork)           # bool convertibility check

All IP/network values are the concrete :mod:ipaddress classes, so .exploded, .network_address, .netmask and addr in network membership all behave exactly as the stdlib does.

Host(value)

A host named by either an address or a hostname.

Config files and URLs hold "the host" as a string that may be either, and the useful operations differ. This keeps the original text and resolves on demand::

host = Host("db.internal")
host.ip()                  # IPv4Address(...) once DNS answers
str(host)                  # 'db.internal' -- always the original

Host("10.0.0.5").is_address    # True, no DNS involved

The point is that str(host) is always what was given, so a URL can still be rebuilt when resolution fails -- which is the case a bare get_ip() handles badly, since it returns None and loses the name.

Source code in src/netimps/_ip.py

def __init__(self, value) -> None:
    if isinstance(value, Host):
        value = value.value
    self.value = "" if value is None else str(value).strip()
    self._resolved = None
    self._attempted = False

is_address property

True if the value is already an IP literal -- no DNS needed.

ip(refresh=False)

Resolve to an address, or None.

A literal is parsed directly; a hostname goes to DNS. The result is cached, including a failure, because the common use is several lookups in a row on the same object. Pass refresh=True to retry -- a name that failed once may resolve later.

Source code in src/netimps/_ip.py

def ip(self, refresh: bool = False):
    """Resolve to an address, or ``None``.

    A literal is parsed directly; a hostname goes to DNS. **The result is
    cached**, including a failure, because the common use is several
    lookups in a row on the same object. Pass ``refresh=True`` to retry --
    a name that failed once may resolve later.
    """
    if refresh:
        self._attempted = False
        self._resolved = None
    if self._attempted:
        return self._resolved

    self._attempted = True
    if not self.value:
        self._resolved = None
        return None

    from . import get_ip, try_parse

    literal = try_parse(self.value, IPAddress)
    self._resolved = literal if literal is not None else get_ip(self.value)
    return self._resolved

MACAddress(value)

An IEEE 802 MAC address.

Accepts the common textual forms on construction -- colon (AA:BB:CC:DD:EE:FF), hyphen (AA-BB-CC-DD-EE-FF), dot/Cisco (aabb.ccdd.eeff) or bare (AABBCCDDEEFF) -- as well as an int or another MACAddress. The value is normalised to lowercase and compared/hashed by its canonical bytes, so instances are usable as dict keys and set members.

as_str(sep) renders the address with an arbitrary separator between octets; sep="" produces the bare form.

Source code in src/netimps/_mac.py

def __init__(self, value: MACLike) -> None:
    if isinstance(value, MACAddress):
        self._octets = value._octets
        return
    if isinstance(value, (bytes, bytearray)):
        octets = bytes(value)
        if len(octets) != 6:
            raise ValueError("MAC address must be 6 bytes, got %d" % len(octets))
        self._octets = octets
        return
    if isinstance(value, int):
        if value < 0 or value > 0xFFFFFFFFFFFF:
            raise ValueError("MAC integer out of range: %r" % (value,))
        self._octets = value.to_bytes(6, "big")
        return
    if isinstance(value, str):
        text = value.strip()
        if not self._VALID_MAC.match(text):
            raise ValueError("Invalid MAC address: %r" % (value,))
        hexdigits = _re.sub(r"[.:-]", "", text)
        self._octets = bytes.fromhex(hexdigits)
        return
    raise TypeError("Cannot build MACAddress from %r" % (type(value).__name__,))

packed property

The 6 raw bytes of the address.

The escape hatch for wire formats and syscalls, mirroring :attr:ipaddress.IPv4Address.packed. MACAddress deliberately is not a :class:bytes subclass -- see the class docstring.

oui property

The 3-byte Organisationally Unique Identifier (vendor prefix).

is_multicast property

True if the group bit (low bit of the first octet) is set.

Multicast MACs are destinations only -- a NIC never has one -- so this is the check for "did I mistake a group address for a host?".

is_local property

True if locally administered (the U/L bit is set).

Locally administered addresses are assigned by software -- VMs, containers, and MAC-randomising clients -- rather than burned in by the vendor, so they are not stable identifiers.

is_universal property

True if universally administered (vendor-assigned). Inverse of :attr:is_local.

is_valid(value) classmethod

Return True if value can be parsed as a MAC. Never raises.

The type-local spelling of netimps.is_valid(value, MACAddress), which is often what reads best at a call site::

if MACAddress.is_valid(user_input):
    ...

A classmethod rather than a staticmethod so a subclass validates against itself. Declared as a :data:typing.TypeGuard, so a checker narrows value in the True branch.

Source code in src/netimps/_mac.py

@classmethod
def is_valid(cls, value: object) -> "TypeGuard[MACAddress]":
    """Return True if ``value`` can be parsed as a MAC. Never raises.

    The type-local spelling of ``netimps.is_valid(value, MACAddress)``,
    which is often what reads best at a call site::

        if MACAddress.is_valid(user_input):
            ...

    A classmethod rather than a staticmethod so a subclass validates
    against itself. Declared as a :data:`typing.TypeGuard`, so a checker
    narrows ``value`` in the ``True`` branch.
    """
    try:
        cls(value)  # type: ignore[arg-type]
        return True
    except (ValueError, TypeError):
        return False

try_parse(value) classmethod

Return a MACAddress, or None if value is not one.

The type-local spelling of netimps.try_parse(value, MACAddress). Prefer it to :meth:is_valid followed by construction -- one call, and no window in which the two disagree.

Source code in src/netimps/_mac.py

@classmethod
def try_parse(cls, value: object) -> "Optional[MACAddress]":
    """Return a ``MACAddress``, or ``None`` if ``value`` is not one.

    The type-local spelling of ``netimps.try_parse(value, MACAddress)``.
    Prefer it to :meth:`is_valid` followed by construction -- one call, and
    no window in which the two disagree.
    """
    try:
        return cls(value)  # type: ignore[arg-type]
    except (ValueError, TypeError):
        return None

as_str(sep=':', upper=False)

Return the MAC as a string with sep between octets.

Lowercase by default (the canonical form used by str(mac) and by equality/hashing); pass upper=True for the uppercase rendering favoured by Windows tooling and much vendor output::

mac.as_str("-")               # 'aa-bb-cc-dd-ee-ff'
mac.as_str("-", upper=True)   # 'AA-BB-CC-DD-EE-FF'

Case affects only this rendering -- two MACAddress values that differ solely in the case they were parsed from remain equal.

Source code in src/netimps/_mac.py

def as_str(self, sep: str = ":", upper: bool = False) -> str:
    """Return the MAC as a string with ``sep`` between octets.

    Lowercase by default (the canonical form used by ``str(mac)`` and by
    equality/hashing); pass ``upper=True`` for the uppercase rendering
    favoured by Windows tooling and much vendor output::

        mac.as_str("-")               # 'aa-bb-cc-dd-ee-ff'
        mac.as_str("-", upper=True)   # 'AA-BB-CC-DD-EE-FF'

    Case affects only this rendering -- two ``MACAddress`` values that
    differ solely in the case they were parsed from remain equal.
    """
    fmt = "%02X" if upper else "%02x"
    return sep.join(fmt % b for b in self._octets)

Interface(name, index=0, mac=None, ips=None, mtu=None, raw=None)

One network interface, normalised to be identical across platforms.

Attributes:

Name	Type	Description
name		Human-usable adapter name ("eth0", "en0", or the Windows friendly name -- never the raw GUID).
index		:func:socket.if_nametoindex value, or 0 when unknown.
mac		The hardware address, or None for interfaces without one (loopback, tunnels).
ips		Every address bound to the interface, each as an IPv4Interface/IPv6Interface carrying its real prefix.
mtu		Link MTU in bytes, or None when the platform does not report it. This is the local link MTU -- for a bottleneck further along a path see :func:netimps.discover_mtu.
raw		None unless enumerated with get_interfaces(raw=True), in which case a platform-specific dict of leftovers. Not portable and explicitly outside the stability guarantee.

Source code in src/netimps/_ifaddrs.py

def __init__(
    self,
    name: str,
    index: int = 0,
    mac: "Optional[MACAddress]" = None,
    ips: "Optional[List[_IPInterface]]" = None,
    mtu: "Optional[int]" = None,
    raw: "Optional[Dict[str, Any]]" = None,
) -> None:
    self.name = name
    self.index = index
    self.mac = mac
    self.ips = ips if ips is not None else []
    self.mtu = mtu
    self.raw = raw

is_loopback property

True when this is the loopback interface.

Derived from the addresses rather than the name: lo (Linux), lo0 (macOS) and Loopback Pseudo-Interface 1 (Windows) share no common spelling, but 127.0.0.0/8 and ::1 do.

Requires a loopback address and no routable one. Not "every address is loopback": macOS's lo0 also carries fe80::1/64, so an all-must-match test reports it as non-loopback. Link-local addresses are ignored here for that reason -- they are not routable, so they do not make an interface non-loopback.

ipv4 property

Just the IPv4 addresses.

ipv6 property

Just the IPv6 addresses.

primary_ip(ipv6=False, loopback_ok=True)

Pick the one entry that best represents this interface, or None.

Answers "which of this adapter's addresses do I use?" -- the question IP_MULTICAST_IF, a bind target and ping -S all ask. A non-loopback entry wins; a loopback one is returned only when that is genuinely all the interface has::

iface.primary_ip()               # IPv4Interface('10.0.0.5/24')
iface.primary_ip().ip            # IPv4Address('10.0.0.5')
iface.primary_ip(ipv6=True)      # its IPv6 entry instead

Named primary rather than ip because this is a selection, not "the" address: an interface routinely has several, and the full lists remain on :attr:ips / :attr:ipv4 / :attr:ipv6.

:param ipv6: pick from :attr:ipv6 rather than :attr:ipv4. :param loopback_ok: when False, an interface holding only loopback addresses yields None instead -- for callers that need a routable address specifically.

Returns an IPv4Interface/IPv6Interface, the same element type as :attr:ips -- one of them, not a different shape. Use .ip for the bare address that socket options take.

Source code in src/netimps/_ifaddrs.py

def primary_ip(
    self, ipv6: bool = False, loopback_ok: bool = True
) -> "Optional[_IPInterface]":
    """Pick the one entry that best represents this interface, or ``None``.

    Answers "which of this adapter's addresses do I use?" -- the question
    ``IP_MULTICAST_IF``, a bind target and ``ping -S`` all ask. A
    **non-loopback** entry wins; a loopback one is returned only when that
    is genuinely all the interface has::

        iface.primary_ip()               # IPv4Interface('10.0.0.5/24')
        iface.primary_ip().ip            # IPv4Address('10.0.0.5')
        iface.primary_ip(ipv6=True)      # its IPv6 entry instead

    Named *primary* rather than *ip* because this is a **selection**, not
    "the" address: an interface routinely has several, and the full lists
    remain on :attr:`ips` / :attr:`ipv4` / :attr:`ipv6`.

    :param ipv6: pick from :attr:`ipv6` rather than :attr:`ipv4`.
    :param loopback_ok: when False, an interface holding only loopback
        addresses yields ``None`` instead -- for callers that need a
        routable address specifically.

    Returns an ``IPv4Interface``/``IPv6Interface``, the **same element type
    as** :attr:`ips` -- one of them, not a different shape. Use ``.ip`` for
    the bare address that socket options take.
    """
    candidates = self.ipv6 if ipv6 else self.ipv4
    for entry in candidates:
        if not entry.ip.is_loopback:
            return entry
    if candidates and loopback_ok:
        return candidates[0]
    return None

PingResult(ok, host, rtt_ms=None, ttl=None, src=None, attempts=1)

Outcome of a :func:ping, usable directly as a boolean.

ping() has always answered "did it reply?", so this stays truthy on success and falsy on failure -- if ping(host): keeps working -- while carrying the details a caller would otherwise re-run ping to scrape.

Attributes:

Name	Type	Description
ok		whether the destination replied.
host		the destination as given.
rtt_ms		round-trip time in milliseconds, or None if not reported. Sub-millisecond replies (time<1ms) are recorded as 0.0, which is falsy -- test is None rather than truthiness.
ttl		TTL/hop-limit of the reply, or None. Counts down from the sender's initial value, so a smaller number means more hops.
src		address that answered, which on success is the destination.
attempts		how many probes were sent before this outcome.

Source code in src/netimps/_ping.py

def __init__(self, ok, host, rtt_ms=None, ttl=None, src=None, attempts=1):
    self.ok = ok
    self.host = host
    self.rtt_ms = rtt_ms
    self.ttl = ttl
    self.src = src
    self.attempts = attempts

Datagram

Bases: NamedTuple

One received datagram and where it came from.

Attributes:

Name	Type	Description
data	bytes	the payload.
sender	Any	(address, port) of the peer, as recvfrom reports it.
local_address	Optional[Any]	the address the datagram was sent to, or None. For a broadcast this is the broadcast address, not the interface's own address -- use interface to identify the adapter.
interface_index	int	receiving interface index, or 0 when unknown.
interface	Optional[Any]	the resolved :class:Interface, or None when unavailable (no IP_PKTINFO, or no matching adapter).

UdpEndpoint(sock, pktinfo=True)

A UDP socket that can report which interface each datagram arrived on.

::

endpoint = UdpEndpoint(netimps.bind("", 67, broadcast=True))
while True:
    packet = endpoint.recv(2048)
    if packet.interface is not None:
        reply_on(packet.interface, packet.data)

Wraps rather than subclasses socket.socket: the raw socket stays reachable as :attr:socket for anything this does not cover.

:param sock: an already-bound UDP socket -- build it with :func:netimps.bind. :param pktinfo: request arrival-interface data. True (the default) enables it where supported and is a no-op elsewhere.

Source code in src/netimps/_udp.py

def __init__(self, sock, pktinfo: bool = True) -> None:
    self.socket = sock
    self.supports_pktinfo = False
    self._cmsg_size = 0

    if not pktinfo or _IP_PKTINFO is None or _CMSG_SPACE is None:
        return
    if not hasattr(sock, "recvmsg"):
        return  # Windows
    try:
        sock.setsockopt(_socket.IPPROTO_IP, _IP_PKTINFO, 1)
    except OSError:
        return  # option exists but this socket/family refuses it
    self.supports_pktinfo = True
    self._cmsg_size = _CMSG_SPACE(_struct.calcsize(_PKTINFO))

recv(bufsize=65535, resolve_interface=True)

Receive one datagram.

:param resolve_interface: look the arrival index up in :func:netimps.get_interfaces to populate .interface. Pass False in a hot loop and use .interface_index directly -- enumeration is not free.

When IP_PKTINFO is unavailable this still works; the interface fields are simply empty.

Source code in src/netimps/_udp.py

def recv(self, bufsize: int = 65535, resolve_interface: bool = True) -> Datagram:
    """Receive one datagram.

    :param resolve_interface: look the arrival index up in
        :func:`netimps.get_interfaces` to populate ``.interface``. Pass
        ``False`` in a hot loop and use ``.interface_index`` directly --
        enumeration is not free.

    When ``IP_PKTINFO`` is unavailable this still works; the interface
    fields are simply empty.
    """
    if not self.supports_pktinfo:
        data, sender = self.socket.recvfrom(bufsize)
        return Datagram(data=data, sender=sender)

    data, ancdata, _flags, sender = self.socket.recvmsg(bufsize, self._cmsg_size)

    index = 0
    local = None
    size = _struct.calcsize(_PKTINFO)
    for level, ctype, cdata in ancdata:
        if level != _socket.IPPROTO_IP or ctype != _IP_PKTINFO:
            continue
        if len(cdata) < size:
            break  # truncated -- treat as absent rather than guessing
        index, _local_if, destination = _struct.unpack(_PKTINFO, cdata[:size])
        from . import try_parse

        local = try_parse(_socket.inet_ntoa(destination))
        break

    interface = None
    if resolve_interface and index:
        from ._ifaddrs import get_interfaces

        interface = next((i for i in get_interfaces() if i.index == index), None)

    return Datagram(
        data=data,
        sender=sender,
        local_address=local,
        interface_index=int(index),
        interface=interface,
    )

send(data, address, port, src=None)

Send a datagram, optionally forcing the src interface.

src accepts the usual union (an :class:Interface, a MAC, an adapter name or an address). Where IP_PKTINFO is available this pins the outgoing interface via sendmsg -- which matters when replying to a broadcast on a multi-homed host, since the routing table would otherwise pick for you.

Falls back to plain sendto when unsupported, so the call works everywhere; the src is then whatever the kernel chooses.

Source code in src/netimps/_udp.py

def send(self, data, address, port: int, src=None) -> int:
    """Send a datagram, optionally forcing the *src* interface.

    ``src`` accepts the usual union (an :class:`Interface`, a MAC, an
    adapter name or an address). Where ``IP_PKTINFO`` is available this
    pins the outgoing interface via ``sendmsg`` -- which matters when
    replying to a broadcast on a multi-homed host, since the routing table
    would otherwise pick for you.

    Falls back to plain ``sendto`` when unsupported, so the call works
    everywhere; the src is then whatever the kernel chooses.
    """
    target = (str(address), int(port))

    if src is None or not self.supports_pktinfo:
        return self.socket.sendto(data, target)

    from ._iface_spec import interface_address

    local = interface_address(src, strict=False)
    if local is None or not hasattr(self.socket, "sendmsg"):
        return self.socket.sendto(data, target)

    packed = _socket.inet_aton(str(local))
    pktinfo = _struct.pack(_PKTINFO, 0, packed, packed)
    return int(
        self.socket.sendmsg(
            [data], [(_socket.IPPROTO_IP, _IP_PKTINFO, pktinfo)], 0, target
        )
    )

Route(dst, src=None, gateway=None, interface_index=0)

How traffic to a destination leaves this host.

Attributes:

Name	Type	Description
dst		the destination this route was computed for.
src		local address the kernel would use (see :func:get_source_ip).
gateway		next-hop router, or None when the destination is on-link (same subnet, or loopback) and no router is involved.
interface_index		index of the outgoing interface, 0 if unknown.
on_link	bool	True when no gateway is needed.

Source code in src/netimps/_sockets.py

def __init__(self, dst, src=None, gateway=None, interface_index=0):
    self.dst = dst
    self.src = src
    self.gateway = gateway
    self.interface_index = interface_index

on_link property

True when the destination is reachable without a router.

collapse(networks)

Merge an iterable of networks into the smallest equivalent list.

Adjacent and overlapping networks are combined; the result is sorted and covers exactly the same addresses::

collapse(["10.0.0.0/25", "10.0.0.128/25"])   # [IPv4Network('10.0.0.0/24')]
collapse(["10.0.0.0/24", "10.0.0.8/29"])     # [IPv4Network('10.0.0.0/24')]

Accepts anything :func:parse does, mixed v4 and v6 -- the families are collapsed independently and returned v4 first. Raises :class:ValueError on malformed input.

Source code in src/netimps/_ip.py

def collapse(networks) -> "List[IPNetwork]":
    """Merge an iterable of networks into the smallest equivalent list.

    Adjacent and overlapping networks are combined; the result is sorted and
    covers exactly the same addresses::

        collapse(["10.0.0.0/25", "10.0.0.128/25"])   # [IPv4Network('10.0.0.0/24')]
        collapse(["10.0.0.0/24", "10.0.0.8/29"])     # [IPv4Network('10.0.0.0/24')]

    Accepts anything :func:`parse` does, mixed v4 and v6 -- the families are
    collapsed independently and returned v4 first. Raises :class:`ValueError`
    on malformed input.
    """
    from . import parse as _parse

    v4, v6 = [], []
    for item in networks:
        net = _parse(item, IPNetwork)
        (v4 if net.version == 4 else v6).append(net)
    out = []
    for group in (v4, v6):
        if group:
            out.extend(_ipaddress.collapse_addresses(group))
    return out

get_ip(address)

Resolve a hostname or literal address to an address object, or None.

Tries to parse address as a literal first and falls back to a DNS lookup, returning None if both fail::

get_ip("10.0.0.5")        # IPv4Address('10.0.0.5')   -- no DNS traffic
get_ip("example.com")     # IPv4Address('93.184.216.34')
get_ip("nonexistent.")    # None

.. note:: The difference from try_parse(address) matters: that never touches the network, while this may block on DNS. Use try_parse to validate user input; use get_ip when you genuinely want a name resolved.

Source code in src/netimps/_ip.py

def get_ip(address: str) -> Optional[IPAddress]:
    """Resolve a hostname *or* literal address to an address object, or ``None``.

    Tries to parse ``address`` as a literal first and falls back to a DNS
    lookup, returning ``None`` if both fail::

        get_ip("10.0.0.5")        # IPv4Address('10.0.0.5')   -- no DNS traffic
        get_ip("example.com")     # IPv4Address('93.184.216.34')
        get_ip("nonexistent.")    # None

    .. note::
       The difference from ``try_parse(address)`` matters: that never
       touches the network, while this **may block on DNS**. Use ``try_parse``
       to validate user input; use ``get_ip`` when you genuinely want a name
       resolved.
    """
    try:
        try:
            return _ipaddress.ip_address(address)
        except ValueError:
            return _ipaddress.ip_address(_socket.gethostbyname(address))
    except (ValueError, OSError):
        return None

is_link_scoped(ip)

True if ip is confined to link scope or narrower.

Covers loopback (127/8, ::1 -- host scope) and link-local (169.254/16, fe80::/10 -- link scope), borrowing IPv6's scope vocabulary for both families::

is_link_scoped(parse("127.0.0.1"))      # True  -- host scope
is_link_scoped(parse("169.254.1.1"))    # True  -- link scope
is_link_scoped(parse("10.0.0.5"))       # False -- private, global scope

The shared practical property is that neither can usefully be routed off the local host or link, so proxying, forwarding or advertising such an address is always wrong. Keeping the definition in one place stops each caller from writing a subtly different version.

.. note:: This is not "is private". RFC 1918 ranges (10/8, 192.168/16) are globally scoped and routable within a site, so they return False -- use ip.is_private for that question.

Source code in src/netimps/_ip.py

def is_link_scoped(ip: IPAddress) -> bool:
    """True if ``ip`` is confined to link scope or narrower.

    Covers loopback (``127/8``, ``::1`` -- host scope) and link-local
    (``169.254/16``, ``fe80::/10`` -- link scope), borrowing IPv6's scope
    vocabulary for both families::

        is_link_scoped(parse("127.0.0.1"))      # True  -- host scope
        is_link_scoped(parse("169.254.1.1"))    # True  -- link scope
        is_link_scoped(parse("10.0.0.5"))       # False -- private, global scope

    The shared practical property is that neither can usefully be routed off
    the local host or link, so proxying, forwarding or advertising such an
    address is always wrong. Keeping the definition in one place stops each
    caller from writing a subtly different version.

    .. note::
       This is **not** "is private". RFC 1918 ranges (``10/8``,
       ``192.168/16``) are globally *scoped* and routable within a site, so
       they return ``False`` -- use ``ip.is_private`` for that question.
    """
    return ip.is_loopback or ip.is_link_local

normalize_host(text, default_port=None)

Split "host:port" into (host, port), handling IPv6 brackets.

The parsing that looks trivial until IPv6 arrives, because a bare v6 address is full of colons::

normalize_host("example.com:8080")     # ('example.com', 8080)
normalize_host("10.0.0.5")             # ('10.0.0.5', None)
normalize_host("[::1]:8080")           # ('::1', 8080)
normalize_host("::1")                  # ('::1', None)   -- not port 1
normalize_host("example.com", 443)     # ('example.com', 443)

The rule this implements: a bare IPv6 address must not be split on its last colon, and only a bracketed one may carry a port. "::1" is the address, never host "::" port 1 -- the mistake hand-rolled splitters almost always make.

Brackets are stripped from the returned host, and a scope id is preserved ("[fe80::1%eth0]:80" -> ("fe80::1%eth0", 80)). default_port is used when no port is present.

Raises :class:ValueError on empty input, an unclosed bracket, or a port that is not an integer in 0-65535.

Source code in src/netimps/_ip.py

def normalize_host(
    text: str, default_port: Optional[int] = None
) -> "Tuple[str, Optional[int]]":
    """Split ``"host:port"`` into ``(host, port)``, handling IPv6 brackets.

    The parsing that looks trivial until IPv6 arrives, because a bare v6
    address is *full of colons*::

        normalize_host("example.com:8080")     # ('example.com', 8080)
        normalize_host("10.0.0.5")             # ('10.0.0.5', None)
        normalize_host("[::1]:8080")           # ('::1', 8080)
        normalize_host("::1")                  # ('::1', None)   -- not port 1
        normalize_host("example.com", 443)     # ('example.com', 443)

    The rule this implements: a bare IPv6 address must **not** be split on its
    last colon, and only a bracketed one may carry a port. ``"::1"`` is the
    address, never host ``"::"`` port ``1`` -- the mistake hand-rolled splitters
    almost always make.

    Brackets are stripped from the returned host, and a scope id is preserved
    (``"[fe80::1%eth0]:80"`` -> ``("fe80::1%eth0", 80)``). ``default_port`` is
    used when no port is present.

    Raises :class:`ValueError` on empty input, an unclosed bracket, or a port
    that is not an integer in 0-65535.
    """
    if not isinstance(text, str) or not text.strip():
        raise ValueError("host must be a non-empty string, got %r" % (text,))
    text = text.strip()

    if text.startswith("["):
        end = text.find("]")
        if end == -1:
            raise ValueError("unclosed '[' in %r" % (text,))
        host = text[1:end]
        rest = text[end + 1 :]
        if not rest:
            port = default_port
        elif rest.startswith(":"):
            port = _parse_port(rest[1:], text)
        else:
            raise ValueError("unexpected %r after ']' in %r" % (rest, text))
    elif text.count(":") > 1:
        # More than one colon and no brackets: a bare IPv6 address. Splitting
        # here would turn "::1" into host "::" port 1.
        host, port = text, default_port
    elif ":" in text:
        host, _, raw_port = text.partition(":")
        port = _parse_port(raw_port, text)
    else:
        host, port = text, default_port

    if not host:
        raise ValueError("empty host in %r" % (text,))
    return host, port

subtract(networks, remove)

Return networks minus every address in remove.

The set difference :mod:ipaddress leaves out -- it ships collapse_addresses but nothing to punch holes::

subtract(["10.0.0.0/24"], ["10.0.0.64/26"])
# [IPv4Network('10.0.0.0/26'), IPv4Network('10.0.0.128/25')]

subtract(["0.0.0.0/0"], ["10.0.0.0/8", "192.168.0.0/16"])  # public v4

The result is collapsed, so it is the minimal set of networks covering what is left. Removing something absent is a no-op, and removing a superset yields []. Mixed families are handled independently: an IPv6 exclusion never affects IPv4 output.

Source code in src/netimps/_ip.py

def subtract(networks, remove) -> "List[IPNetwork]":
    """Return ``networks`` minus every address in ``remove``.

    The set difference :mod:`ipaddress` leaves out -- it ships
    ``collapse_addresses`` but nothing to punch holes::

        subtract(["10.0.0.0/24"], ["10.0.0.64/26"])
        # [IPv4Network('10.0.0.0/26'), IPv4Network('10.0.0.128/25')]

        subtract(["0.0.0.0/0"], ["10.0.0.0/8", "192.168.0.0/16"])  # public v4

    The result is collapsed, so it is the minimal set of networks covering
    what is left. Removing something absent is a no-op, and removing a
    superset yields ``[]``. Mixed families are handled independently: an IPv6
    exclusion never affects IPv4 output.
    """
    from . import parse as _parse

    remaining = collapse(networks)
    for item in remove:
        excluded = _parse(item, IPNetwork)
        next_round = []
        for net in remaining:
            if net.version != excluded.version:
                next_round.append(net)  # different family: untouched
                continue
            if not (
                net.subnet_of(excluded)
                or excluded.subnet_of(net)
                or net.overlaps(excluded)
            ):
                next_round.append(net)
                continue
            if net.subnet_of(excluded):
                continue  # fully removed
            next_round.extend(net.address_exclude(excluded))
        remaining = next_round
    return collapse(remaining)

parse(value, type=IPAddress, **kwargs)

parse(
    value: object, type: TypeForm[_T], **kwargs: Any
) -> _T

parse(
    value: object, type: Callable[..., _T], **kwargs: Any
) -> _T

parse(value: object, **kwargs: Any) -> IPAddress

Build type from value, raising on bad input.

The single parsing entry point. type is a result type -- one of the :data:IPAddress/:data:IPInterface/:data:IPNetwork unions, a concrete IPv4Address &co, or any callable::

parse("10.0.0.5")                        # IPv4Address  (the default)
parse("10.0.0.5/24", IPInterface)        # IPv4Interface
parse("10.0.0.5/24", IPNetwork)          # IPv4Network('10.0.0.0/24')
parse("10.0.0.5/24", IPNetwork, strict=True)   # raises: host bits set
parse("aa:bb:cc:dd:ee:ff", MACAddress)   # MACAddress

Every type accepts the full range of stdlib inputs -- str, int, packed bytes, or an existing object -- because the builders are the ipaddress.ip_* functions rather than the concrete constructors.

A union accepts either family; a concrete type enforces its own, so parse("::1", IPv4Address) raises rather than quietly returning an IPv6Address.

Networks are parsed non-strict by default (unlike the stdlib), so a host address with a prefix normalises to its network instead of raising. Extra kwargs pass through to the underlying builder.

Raises :class:ValueError on malformed input or a family mismatch, and :class:TypeError for an unusable type. Use :func:try_parse for the non-raising form.

Source code in src/netimps/__init__.py

def parse(value: object, type: "Any" = IPAddress, **kwargs) -> "Any":
    """Build ``type`` from ``value``, raising on bad input.

    The single parsing entry point. ``type`` is a result type -- one of the
    :data:`IPAddress`/:data:`IPInterface`/:data:`IPNetwork` unions, a concrete
    ``IPv4Address`` &co, or any callable::

        parse("10.0.0.5")                        # IPv4Address  (the default)
        parse("10.0.0.5/24", IPInterface)        # IPv4Interface
        parse("10.0.0.5/24", IPNetwork)          # IPv4Network('10.0.0.0/24')
        parse("10.0.0.5/24", IPNetwork, strict=True)   # raises: host bits set
        parse("aa:bb:cc:dd:ee:ff", MACAddress)   # MACAddress

    Every type accepts the full range of stdlib inputs -- ``str``, ``int``,
    packed ``bytes``, or an existing object -- because the builders are the
    ``ipaddress.ip_*`` functions rather than the concrete constructors.

    A **union** accepts either family; a **concrete** type enforces its own, so
    ``parse("::1", IPv4Address)`` raises rather than quietly returning an
    ``IPv6Address``.

    Networks are parsed **non-strict** by default (unlike the stdlib), so a host
    address with a prefix normalises to its network instead of raising. Extra
    ``kwargs`` pass through to the underlying builder.

    Raises :class:`ValueError` on malformed input or a family mismatch, and
    :class:`TypeError` for an unusable ``type``. Use :func:`try_parse` for the
    non-raising form.
    """
    # Guarded: an unhashable ``type`` would make these lookups raise TypeError,
    # which try_parse would then swallow into `default` -- turning a caller bug
    # into a silent "invalid value". Fall through to the explicit checks below.
    try:
        wanted = _CONCRETE.get(type)
        builder = _BUILDERS.get(wanted if wanted is not None else type)
    except TypeError:
        wanted = builder = None

    if builder is None:
        _check_parser(type)  # raises for anything unusable
        return type(value, **kwargs)

    options = dict(_BUILDER_DEFAULTS.get(builder, ()))
    options.update(kwargs)
    result = builder(value, **options)

    if wanted is not None and not isinstance(result, type):
        raise ValueError("%r is not a %s" % (value, type.__name__))
    return result

try_parse(value, type=IPAddress, default=None, **kwargs)

try_parse(
    value: object,
    type: TypeForm[_T],
    default: None = ...,
    **kwargs: Any
) -> Optional[_T]

try_parse(
    value: object,
    type: TypeForm[_T],
    default: _D,
    **kwargs: Any
) -> Union[_T, _D]

try_parse(
    value: object,
    type: Callable[..., _T],
    default: None = ...,
    **kwargs: Any
) -> Optional[_T]

try_parse(
    value: object,
    type: Callable[..., _T],
    default: _D,
    **kwargs: Any
) -> Union[_T, _D]

try_parse(
    value: object, *, default: None = ..., **kwargs: Any
) -> Optional[IPAddress]

try_parse(
    value: object, *, default: _D, **kwargs: Any
) -> Union[IPAddress, _D]

Return type(value), or default if it rejects the input. Never raises.

The one non-raising parse for the whole package. type is either a type -- including the union aliases, which are not themselves callable -- or any callable that signals bad input with ValueError/TypeError::

try_parse("10.0.0.5", IPAddress)     # IPv4Address('10.0.0.5')
try_parse("10.0.0.5", IPv4Address)   # concrete: v6 input rejected
try_parse("nonsense", IPAddress)     # None
try_parse(user_input, MACAddress) or DEFAULT_MAC
try_parse(raw, IPAddress, default=LOCALHOST)   # explicit fallback

The union aliases IPAddress/IPInterface/IPNetwork accept either family. A concrete type stays strict, so asking for one family and getting the other is impossible::

try_parse("::1", IPAddress)      # IPv6Address('::1')  -- either family
try_parse("::1", IPv4Address)    # None                -- v4 was asked for
try_parse("10.0.0.5", IPv4Address)   # IPv4Address('10.0.0.5')

Prefer this to is_valid followed by a parse: that pattern does the work twice and leaves a window where the two disagree.

Generic in the type: try_parse(x, MACAddress) is typed Optional[MACAddress], so a checker knows the result without a cast.

Only ValueError and TypeError are swallowed -- the two exceptions that mean "bad input". Anything else (an OSError from a builder that touches the network, a bug in it) propagates, because turning it into None would disguise a real failure as a rejected value. A type that is neither callable nor a known type raises TypeError: that is a caller bug, not a rejected value.

:param default: returned instead of None when the input is rejected. Also the seam :func:is_valid uses -- passing a sentinel is the only way to tell "the parse returned None" from "it rejected the input".

Source code in src/netimps/__init__.py

def try_parse(
    value: object,
    type: "Any" = IPAddress,
    default: "Any" = None,
    **kwargs,
) -> "Any":
    """Return ``type(value)``, or ``default`` if it rejects the input. Never raises.

    The one non-raising parse for the whole package. ``type`` is either a
    **type** -- including the union aliases, which are not themselves callable
    -- or any callable that signals bad input with ``ValueError``/``TypeError``::

        try_parse("10.0.0.5", IPAddress)     # IPv4Address('10.0.0.5')
        try_parse("10.0.0.5", IPv4Address)   # concrete: v6 input rejected
        try_parse("nonsense", IPAddress)     # None
        try_parse(user_input, MACAddress) or DEFAULT_MAC
        try_parse(raw, IPAddress, default=LOCALHOST)   # explicit fallback

    The union aliases ``IPAddress``/``IPInterface``/``IPNetwork`` accept either
    family. A **concrete** type stays strict, so asking for one family and
    getting the other is impossible::

        try_parse("::1", IPAddress)      # IPv6Address('::1')  -- either family
        try_parse("::1", IPv4Address)    # None                -- v4 was asked for
        try_parse("10.0.0.5", IPv4Address)   # IPv4Address('10.0.0.5')

    Prefer this to ``is_valid`` followed by a parse: that pattern does the work
    twice and leaves a window where the two disagree.

    Generic in the type: ``try_parse(x, MACAddress)`` is typed
    ``Optional[MACAddress]``, so a checker knows the result without a cast.

    Only ``ValueError`` and ``TypeError`` are swallowed -- the two exceptions
    that mean "bad input". Anything else (an ``OSError`` from a builder that
    touches the network, a bug in it) propagates, because turning it
    into ``None`` would disguise a real failure as a rejected value. A
    ``type`` that is neither callable nor a known type raises ``TypeError``:
    that is a caller bug, not a rejected value.

    :param default: returned instead of ``None`` when the input is rejected.
        Also the seam :func:`is_valid` uses -- passing a sentinel is the only
        way to tell "the parse returned ``None``" from "it rejected the input".
    """
    # Validate the type *before* the try, so the TypeError raised for an
    # unusable one is not swallowed as if the value had been rejected. Only the
    # parse itself is guarded.
    _check_parser(type)
    try:
        return parse(value, type, **kwargs)
    except (ValueError, TypeError):
        return default

is_valid(value, type=IPAddress, **kwargs)

is_valid(
    value: object, type: TypeForm[_T], **kwargs: Any
) -> bool

is_valid(
    value: object, type: Callable[..., _T], **kwargs: Any
) -> bool

is_valid(value: object, **kwargs: Any) -> bool

Return True if value parses as type. Never raises.

Accepts the same type forms as :func:try_parse -- a type, a union alias, or any callable::

is_valid("10.0.0.5", IPAddress)      # True  (the type alias)
is_valid("10.0.0.0/24", IPNetwork)   # True
is_valid("aa:bb:cc:dd:ee:ff", MACAddress)
is_valid("nonsense", IPAddress)      # False

When you want the parsed value too, use :func:try_parse instead of calling this first -- one call, no double work. Same exception policy: only ValueError/TypeError count as "invalid".

.. note:: A parser that legitimately returns None for valid input still counts as valid here -- the parse succeeded. That is why this delegates via a sentinel rather than testing try_parse(...) is not None, which cannot tell "returned None" from "rejected the input".

Source code in src/netimps/__init__.py

def is_valid(
    value: object,
    type: "Any" = IPAddress,
    **kwargs,
) -> "bool":
    """Return ``True`` if ``value`` parses as ``type``. Never raises.

    Accepts the same ``type`` forms as :func:`try_parse` -- a type, a union
    alias, or any callable::

        is_valid("10.0.0.5", IPAddress)      # True  (the type alias)
        is_valid("10.0.0.0/24", IPNetwork)   # True
        is_valid("aa:bb:cc:dd:ee:ff", MACAddress)
        is_valid("nonsense", IPAddress)      # False

    When you want the parsed value too, use :func:`try_parse` instead of
    calling this first -- one call, no double work. Same exception policy: only
    ``ValueError``/``TypeError`` count as "invalid".

    .. note::
       A parser that legitimately returns ``None`` for valid input still counts
       as valid here -- the parse *succeeded*. That is why this delegates via a
       sentinel rather than testing ``try_parse(...) is not None``, which cannot
       tell "returned None" from "rejected the input".
    """
    return try_parse(value, type, _MISSING, **kwargs) is not _MISSING

get_default_port(scheme)

Return the conventional port for a URL scheme, or None if unknown.

Checks the built-in/registered table first, then falls back to the system services database via :func:socket.getservbyname::

get_default_port("https")    # 443
get_default_port("socks5")   # 1080  (absent from /etc/services)
get_default_port("nope")     # None

Case-insensitive. Extend the table with :func:register_port.

Source code in src/netimps/_scheme.py

def get_default_port(scheme: str) -> Optional[int]:
    """Return the conventional port for a URL scheme, or ``None`` if unknown.

    Checks the built-in/registered table first, then falls back to the system
    services database via :func:`socket.getservbyname`::

        get_default_port("https")    # 443
        get_default_port("socks5")   # 1080  (absent from /etc/services)
        get_default_port("nope")     # None

    Case-insensitive. Extend the table with :func:`register_port`.
    """
    scheme = scheme.lower()
    if scheme in _DEFAULT_PORTS:
        return _DEFAULT_PORTS[scheme]
    try:
        return _socket.getservbyname(scheme)
    except OSError:
        return None

get_default_scheme(port)

Return the conventional scheme for a port, or None if unknown.

The inverse of :func:get_default_port::

get_default_scheme(443)     # 'https'
get_default_scheme(1080)    # 'socks'   (canonical, not an alias)
get_default_scheme(9999)    # None

Falls back to the system services database via :func:socket.getservbyport. Where several schemes share a port, the canonical one is returned -- see :func:register_port.

Source code in src/netimps/_scheme.py

def get_default_scheme(port: int) -> Optional[str]:
    """Return the conventional scheme for a port, or ``None`` if unknown.

    The inverse of :func:`get_default_port`::

        get_default_scheme(443)     # 'https'
        get_default_scheme(1080)    # 'socks'   (canonical, not an alias)
        get_default_scheme(9999)    # None

    Falls back to the system services database via
    :func:`socket.getservbyport`. Where several schemes share a port, the
    canonical one is returned -- see :func:`register_port`.
    """
    if port in _PORT_SCHEMES:
        return _PORT_SCHEMES[port]
    try:
        return _socket.getservbyport(port)
    except (OSError, OverflowError, TypeError):
        return None

register_port(scheme, port, canonical=False)

Register (or override) a scheme's conventional port.

The built-in table covers the common cases, but every consumer eventually has a protocol of its own::

register_port("myproto", 9999)
get_default_port("myproto")     # 9999
get_default_scheme(9999)        # 'myproto'

:param scheme: scheme name; matched case-insensitively. :param port: TCP/UDP port number, 0-65535. :param canonical: make scheme the name :func:get_default_scheme returns for port, displacing any existing one. By default the first registration for a port keeps that slot, so adding an alias does not silently change what an existing port maps back to.

Raises :class:ValueError on an out-of-range port or empty scheme.

Source code in src/netimps/_scheme.py

def register_port(scheme: str, port: int, canonical: bool = False) -> None:
    """Register (or override) a scheme's conventional port.

    The built-in table covers the common cases, but every consumer eventually
    has a protocol of its own::

        register_port("myproto", 9999)
        get_default_port("myproto")     # 9999
        get_default_scheme(9999)        # 'myproto'

    :param scheme: scheme name; matched case-insensitively.
    :param port: TCP/UDP port number, 0-65535.
    :param canonical: make ``scheme`` the name :func:`get_default_scheme` returns for
        ``port``, displacing any existing one. By default the first registration
        for a port keeps that slot, so adding an alias does not silently change
        what an existing port maps back to.

    Raises :class:`ValueError` on an out-of-range port or empty scheme.
    """
    if not scheme or not scheme.strip():
        raise ValueError("scheme must be a non-empty string")
    if not isinstance(port, int) or isinstance(port, bool):
        raise TypeError("port must be an int, got %r" % (type(port).__name__,))
    if not 0 <= port <= 65535:
        raise ValueError("port out of range: %r" % (port,))

    scheme = scheme.strip().lower()
    _DEFAULT_PORTS[scheme] = port
    if canonical or port not in _PORT_SCHEMES:
        _PORT_SCHEMES[port] = scheme

get_interfaces(raw=False)

Return this host's network interfaces.

Uses getifaddrs(3) on POSIX and GetAdaptersAddresses on Windows via :mod:ctypes -- no third-party dependency -- so adapter names, MACs and real prefix lengths are all available::

for iface in get_interfaces():
    print(iface.name, iface.mac, [str(ip) for ip in iface.ips])

:param raw: when True, populate :attr:Interface.raw with the untouched platform data (Linux/BSD flags; Windows adapter guid, if_type, ...). Not portable -- outside the stability guarantee.

Never raises for enumeration failure: if the native call is unavailable it degrades to a hostname-resolution fallback in which prefixes are not real (every address becomes a /32//128 under an interface named "<unknown>").

Source code in src/netimps/_ifaddrs.py

def get_interfaces(raw: bool = False) -> "List[Interface]":
    """Return this host's network interfaces.

    Uses ``getifaddrs(3)`` on POSIX and ``GetAdaptersAddresses`` on Windows via
    :mod:`ctypes` -- no third-party dependency -- so adapter names, MACs and
    real prefix lengths are all available::

        for iface in get_interfaces():
            print(iface.name, iface.mac, [str(ip) for ip in iface.ips])

    :param raw: when True, populate :attr:`Interface.raw` with the untouched
        platform data (Linux/BSD ``flags``; Windows adapter ``guid``,
        ``if_type``, ...). **Not portable** -- outside the stability guarantee.

    Never raises for enumeration failure: if the native call is unavailable it
    degrades to a hostname-resolution fallback in which prefixes are *not*
    real (every address becomes a ``/32``/``/128`` under an interface named
    ``"<unknown>"``).
    """
    try:
        if _IS_WINDOWS:
            return _windows_interfaces(raw)
        return _posix_interfaces(raw)
    except (OSError, AttributeError, ValueError):
        return _fallback_interfaces(raw)

iter_addresses(interfaces=None, family=None)

Yield (interface, address) once per address, not once per adapter.

:func:get_interfaces groups every address under its adapter, which is the right shape for "describe this host". Consumers that filter or act per address -- picking a bind target, excluding link-local, matching a subnet -- want the flattened view instead, and would otherwise write the same nested loop each time::

for iface, addr in iter_addresses():
    if addr.ip in some_network:
        bind_to(addr.ip)

:param interfaces: reuse an existing enumeration instead of calling :func:get_interfaces again. Worth passing in a loop, since enumeration is a syscall. :param family: 4 or 6 to yield only that family; None for both.

The interface is the full :class:Interface, so its name, MAC and MTU stay reachable -- the flattening loses no information.

Source code in src/netimps/_ifaddrs.py

def iter_addresses(
    interfaces: "Optional[Iterable[Interface]]" = None,
    family: "Optional[int]" = None,
) -> "Iterator[Tuple[Interface, _IPInterface]]":
    """Yield ``(interface, address)`` once per address, not once per adapter.

    :func:`get_interfaces` groups every address under its adapter, which is the
    right shape for "describe this host". Consumers that filter or act *per
    address* -- picking a bind target, excluding link-local, matching a subnet
    -- want the flattened view instead, and would otherwise write the same
    nested loop each time::

        for iface, addr in iter_addresses():
            if addr.ip in some_network:
                bind_to(addr.ip)

    :param interfaces: reuse an existing enumeration instead of calling
        :func:`get_interfaces` again. Worth passing in a loop, since
        enumeration is a syscall.
    :param family: ``4`` or ``6`` to yield only that family; ``None`` for both.

    The ``interface`` is the full :class:`Interface`, so its name, MAC and MTU
    stay reachable -- the flattening loses no information.
    """
    if interfaces is None:
        interfaces = get_interfaces()
    for iface in interfaces:
        if family == 4:
            entries = iface.ipv4
        elif family == 6:
            entries = iface.ipv6
        elif family is None:
            entries = iface.ips
        else:
            raise ValueError("family must be 4, 6 or None, got %r" % (family,))
        for entry in entries:
            yield iface, entry

resolve(query, rdtype='a', ns=None, timeout=5.0, port=53, tcp=False)

Resolve query via DNS and return the answers as a list of strings.

::

resolve("example.com")                    # ['93.184.216.34']
resolve("example.com", "aaaa")
resolve("example.com", "mx", ns="1.1.1.1")

Contract: always a list, empty when the name does not resolve -- never None. Callers can therefore write if result: and index result[0] safely.

Records come back as native types: address records (A/AAAA) are :class:ipaddress objects, everything else is a str::

resolve("example.com")[0].is_private     # an IPv4Address, not "1.2.3.4"
resolve("example.com", "mx")             # ['10 mail.example.com']
resolve("example.com", "txt")            # ['v=spf1 -all']  -- unquoted

Names lose their trailing root dot and TXT strings lose their surrounding quotes, since neither is wanted in practice.

:param query: the name (or address, for reverse types) to look up. :param rdtype: DNS record type ("a", "aaaa", "mx" ...). Second because it is the argument callers actually vary. :param ns: optional nameserver, or list of nameservers, to query instead of the system resolver. :param timeout: seconds to spend on the whole resolution, retries included (None for dnspython's default). Bounds total time, not each query -- a list of unreachable nameservers cannot stretch past it. :param port: nameserver port, for resolvers not on 53. :param tcp: query over TCP instead of UDP. Useful for large responses that would otherwise be truncated.

A genuine lookup failure (NXDOMAIN, no answer, timeout, all servers failed) yields []; a malformed query or unknown record type raises :class:ValueError, since that is a caller bug rather than a DNS result.

Requires the dnspython package (installed with netimps).

Source code in src/netimps/_dns.py

def resolve(
    query: str,
    rdtype: str = "a",
    ns: Optional[Union[str, List[str]]] = None,
    timeout: Optional[float] = 5.0,
    port: int = 53,
    tcp: bool = False,
) -> "List[Any]":
    """Resolve ``query`` via DNS and return the answers as a list of strings.

    ::

        resolve("example.com")                    # ['93.184.216.34']
        resolve("example.com", "aaaa")
        resolve("example.com", "mx", ns="1.1.1.1")

    Contract: always a ``list``, **empty** when the name does not resolve --
    never ``None``. Callers can therefore write ``if result:`` and index
    ``result[0]`` safely.

    Records come back as **native types**: address records (``A``/``AAAA``) are
    :class:`ipaddress` objects, everything else is a ``str``::

        resolve("example.com")[0].is_private     # an IPv4Address, not "1.2.3.4"
        resolve("example.com", "mx")             # ['10 mail.example.com']
        resolve("example.com", "txt")            # ['v=spf1 -all']  -- unquoted

    Names lose their trailing root dot and TXT strings lose their surrounding
    quotes, since neither is wanted in practice.

    :param query: the name (or address, for reverse types) to look up.
    :param rdtype: DNS record type (``"a"``, ``"aaaa"``, ``"mx"`` ...). Second
        because it is the argument callers actually vary.
    :param ns: optional nameserver, or list of nameservers, to query instead of
        the system resolver.
    :param timeout: seconds to spend on the whole resolution, retries included
        (``None`` for dnspython's default). Bounds *total* time, not each query
        -- a list of unreachable nameservers cannot stretch past it.
    :param port: nameserver port, for resolvers not on 53.
    :param tcp: query over TCP instead of UDP. Useful for large responses that
        would otherwise be truncated.

    A genuine lookup failure (NXDOMAIN, no answer, timeout, all servers failed)
    yields ``[]``; a malformed query or unknown record type raises
    :class:`ValueError`, since that is a caller bug rather than a DNS result.

    Requires the ``dnspython`` package (installed with ``netimps``).
    """
    from dns import resolver as _resolver

    r = _resolver.Resolver(configure=not ns)
    if isinstance(ns, str):
        ns = [ns]
    if ns:
        r.nameservers = list(ns)
    if port != 53:
        r.port = port
    if timeout is not None:
        # `timeout` bounds a single query; `lifetime` bounds the whole
        # resolution including retries against every nameserver. Without the
        # lifetime, a list of dead servers blocks for far longer than asked.
        r.timeout = timeout
        r.lifetime = timeout

    # Looked up by name rather than referenced directly: LifetimeTimeout only
    # exists in dnspython >= 2.0, and the set has shifted between releases, so
    # a hard reference would break on older versions. Anything missing simply
    # drops out of the tuple.
    _lookup_failures = tuple(
        exc
        for exc in (
            getattr(_resolver, name, None)
            for name in (
                "NXDOMAIN",  # name definitively does not exist
                "NoAnswer",  # name exists, no record of this type
                "NoNameservers",  # every nameserver refused or failed
                "LifetimeTimeout",  # ran out of time
                "Timeout",
                "NoResolverConfiguration",  # no system resolver to use
            )
        )
        if isinstance(exc, type) and issubclass(exc, Exception)
    )

    try:
        answer = r.resolve(query, rdtype, tcp=tcp)
    except _lookup_failures:
        # A genuine "no result" -- the documented [] contract.
        return []
    except Exception as exc:
        # Everything else (malformed name, unknown rdtype) is a caller bug
        # rather than a lookup outcome. The old code swallowed these into [],
        # which turned a typo'd record type into a silent empty result.
        raise ValueError("invalid DNS query %r (%s): %s" % (query, rdtype, exc))
    return [_native_record(record) for record in answer]

ping(dst, tries=1, timeout=1.0, ipv6=None, src=None, size=None, ttl=None, dont_fragment=False, method='icmp', port=None)

Ping dst; the result is truthy if it answered.

Returns a :class:PingResult rather than a bare bool, so the reply details are available without re-running and re-parsing ping::

ping("8.8.8.8")                          # ICMP echo
ping("8.8.8.8", method="tcp", port=53)   # time a TCP handshake
ping("10.0.0.5", method="udp", port=53)  # datagram + reply or ICMP

if ping("8.8.8.8"):                  # still reads as a boolean
    ...
result = ping("8.8.8.8")
result.rtt_ms                        # 5.0
result.ttl                           # 119

Shells out to the platform ping binary, translating timeout into the right per-platform flags (Windows -n/-w in milliseconds; POSIX -c/-W in whole seconds), and returns on the first success::

ping("10.0.0.1")                      # one attempt, 1s timeout
ping("example.com", tries=3, timeout=2.5)
ping("2001:db8::1", ipv6=True)        # force ping6 semantics

:param tries: attempts before giving up. Values below 1 are treated as 1. :param timeout: seconds to wait per attempt. POSIX ping only accepts a whole number of seconds, so sub-second values are rounded up to 1 -- never down to 0, which some implementations read as "wait forever". :param ipv6: force the IPv6 (-6) or IPv4 (-4) binary. None (default) lets the system resolver decide. :param src: send from this local address, choosing which interface the echo leaves by. Accepts an :class:Interface, an address object, or a string::

    ping("8.8.8.8", src=get_source_ip())            # address object
    ping("8.8.8.8", src=get_interfaces()[0])        # Interface
    ping("8.8.8.8", src="192.0.2.10")               # literal
    ping("8.8.8.8", src=MACAddress("00:00:5e:00:53:01"))  # by MAC

A **MAC address** (object or string) is resolved to the interface
holding it -- convenient when the adapter is known by hardware address
rather than by a possibly-changing IP. An unknown MAC yields a falsy
result rather than falling back.

An ``Interface`` contributes its first non-loopback IPv4 address (its
IPv6 address when ``ipv6=True``), because Windows ``-S`` requires an
*address* -- passing an adapter name there fails. POSIX ``-I`` would
accept a name, but resolving it here keeps behaviour identical on both.
An interface holding no usable address yields a falsy result rather
than falling back to the default route.

Likewise an address not held by any local interface makes ``ping``
fail, so the result is falsy -- this never silently reroutes.

:param size: ICMP payload bytes -- Windows -l, POSIX -s. Both flags mean the same thing: neither counts headers, so the wire packet is 28 bytes larger (20 IP + 8 ICMP). Payload 1472 is exactly 1500 on the wire, which is why that is the number a 1500-MTU link tops out at. Verified on Windows against the DF boundary (1472 passes, 1473 does not); ping(8) documents -s as "data bytes" identically. :param ttl: initial hop limit (-i on Windows, -t on POSIX -- the letters are swapped between platforms, a classic src of scripts that silently do the wrong thing).

A ``ttl`` too small to reach the target yields ``False`` on every
platform. That takes explicit work on Windows, whose ``ping`` exits
``0`` for "TTL expired in transit" -- counting a router's error as a
received reply -- so the raw exit code would report success although
the target was never reached. The reply address is verified instead
(see below), which is locale-independent.

:param method: how to probe -- "icmp" (default), "tcp" or "udp". ICMP is the classic echo; the other two reach a host through firewalls that drop echo but permit ordinary traffic.

All three answer **"is the host up?"**. A TCP *refusal* therefore counts
as success -- the RST proves something answered -- and so does an ICMP
port-unreachable for UDP. Use :func:`netimps.tcp_check` when the
question is "is the *service* up?", where a refusal is a failure.

:param port: destination port for tcp/udp. Required for those, and ignored for ICMP. :param dont_fragment: set the DF bit (Windows -f, Linux -M do). Combined with size, the standard manual MTU probe: the largest size that still succeeds is the path MTU minus 28. Unsupported on macOS/BSD ping, where it is ignored.

An empty dst gives a falsy result. Never raises: a missing ping binary or a non-zero exit both yield a falsy :class:PingResult.

.. note:: This measures whether ICMP echo is answered, which is not the same as whether a host is up -- plenty of hosts and most cloud firewalls drop echo requests while serving traffic normally. Prefer a TCP connect to the port you actually care about when you can.

Source code in src/netimps/_ping.py

def ping(
    dst: str,
    tries: int = 1,
    timeout: float = 1.0,
    ipv6: Optional[bool] = None,
    src=None,
    size: Optional[int] = None,
    ttl: Optional[int] = None,
    dont_fragment: bool = False,
    method: str = "icmp",
    port: "Optional[int]" = None,
) -> "PingResult":
    """Ping ``dst``; the result is truthy if it answered.

    Returns a :class:`PingResult` rather than a bare bool, so the reply details
    are available without re-running and re-parsing ``ping``::

        ping("8.8.8.8")                          # ICMP echo
        ping("8.8.8.8", method="tcp", port=53)   # time a TCP handshake
        ping("10.0.0.5", method="udp", port=53)  # datagram + reply or ICMP

        if ping("8.8.8.8"):                  # still reads as a boolean
            ...
        result = ping("8.8.8.8")
        result.rtt_ms                        # 5.0
        result.ttl                           # 119

    Shells out to the platform ``ping`` binary, translating ``timeout`` into the
    right per-platform flags (Windows ``-n``/``-w`` in milliseconds; POSIX
    ``-c``/``-W`` in whole seconds), and returns on the first success::

        ping("10.0.0.1")                      # one attempt, 1s timeout
        ping("example.com", tries=3, timeout=2.5)
        ping("2001:db8::1", ipv6=True)        # force ping6 semantics

    :param tries: attempts before giving up. Values below 1 are treated as 1.
    :param timeout: seconds to wait per attempt. POSIX ``ping`` only accepts a
        whole number of seconds, so sub-second values are rounded **up** to 1 --
        never down to 0, which some implementations read as "wait forever".
    :param ipv6: force the IPv6 (``-6``) or IPv4 (``-4``) binary. ``None``
        (default) lets the system resolver decide.
    :param src: send from this local address, choosing which interface the
        echo leaves by. Accepts an :class:`Interface`, an address object, or a
        string::

            ping("8.8.8.8", src=get_source_ip())            # address object
            ping("8.8.8.8", src=get_interfaces()[0])        # Interface
            ping("8.8.8.8", src="192.0.2.10")               # literal
            ping("8.8.8.8", src=MACAddress("00:00:5e:00:53:01"))  # by MAC

        A **MAC address** (object or string) is resolved to the interface
        holding it -- convenient when the adapter is known by hardware address
        rather than by a possibly-changing IP. An unknown MAC yields a falsy
        result rather than falling back.

        An ``Interface`` contributes its first non-loopback IPv4 address (its
        IPv6 address when ``ipv6=True``), because Windows ``-S`` requires an
        *address* -- passing an adapter name there fails. POSIX ``-I`` would
        accept a name, but resolving it here keeps behaviour identical on both.
        An interface holding no usable address yields a falsy result rather
        than falling back to the default route.

        Likewise an address not held by any local interface makes ``ping``
        fail, so the result is falsy -- this never silently reroutes.
    :param size: ICMP **payload** bytes -- Windows ``-l``, POSIX ``-s``. Both
        flags mean the same thing: neither counts headers, so the wire packet is
        28 bytes larger (20 IP + 8 ICMP). Payload 1472 is exactly 1500 on the
        wire, which is why that is the number a 1500-MTU link tops out at.
        Verified on Windows against the DF boundary (1472 passes, 1473 does
        not); ``ping(8)`` documents ``-s`` as "data bytes" identically.
    :param ttl: initial hop limit (``-i`` on Windows, ``-t`` on POSIX -- the
        letters are **swapped** between platforms, a classic src of scripts
        that silently do the wrong thing).

        A ``ttl`` too small to reach the target yields ``False`` on every
        platform. That takes explicit work on Windows, whose ``ping`` exits
        ``0`` for "TTL expired in transit" -- counting a router's error as a
        received reply -- so the raw exit code would report success although
        the target was never reached. The reply address is verified instead
        (see below), which is locale-independent.
    :param method: how to probe -- ``"icmp"`` (default), ``"tcp"`` or
        ``"udp"``. ICMP is the classic echo; the other two reach a host through
        firewalls that drop echo but permit ordinary traffic.

        All three answer **"is the host up?"**. A TCP *refusal* therefore counts
        as success -- the RST proves something answered -- and so does an ICMP
        port-unreachable for UDP. Use :func:`netimps.tcp_check` when the
        question is "is the *service* up?", where a refusal is a failure.
    :param port: destination port for ``tcp``/``udp``. Required for those, and
        ignored for ICMP.
    :param dont_fragment: set the DF bit (Windows ``-f``, Linux ``-M do``).
        Combined with ``size``, the standard manual MTU probe: the largest
        ``size`` that still succeeds is the path MTU minus 28. Unsupported on
        macOS/BSD ping, where it is ignored.

    An empty ``dst`` gives a falsy result. Never raises: a missing ``ping``
    binary or a non-zero exit both yield a falsy :class:`PingResult`.

    .. note::
       This measures whether *ICMP echo* is answered, which is not the same as
       whether a host is up -- plenty of hosts and most cloud firewalls drop
       echo requests while serving traffic normally. Prefer a TCP connect to
       the port you actually care about when you can.
    """
    if not dst:
        return PingResult(False, dst, attempts=0)

    from . import try_parse as _try_parse

    method = (method or "icmp").lower()
    if method not in ("icmp", "tcp", "udp"):
        raise ValueError("method must be 'icmp', 'tcp' or 'udp', got %r" % (method,))
    if method != "icmp":
        if port is None:
            raise ValueError("method=%r needs a port" % (method,))
        prober = _tcp_ping if method == "tcp" else _udp_ping
        probe_size = size or 0
        last = None
        for attempt in range(1, max(1, tries) + 1):
            ok, rtt, note = prober(dst, port, timeout, probe_size)
            if ok:
                return PingResult(
                    True,
                    dst,
                    rtt_ms=rtt,
                    src=_try_parse(dst),
                    attempts=attempt,
                )
            last = attempt
        return PingResult(False, dst, attempts=last or 1)

    tries = max(1, tries)
    if _os.name == "nt":
        # Windows takes milliseconds and counts with -n.
        options = ["-n", "1", "-w", str(max(1, int(timeout * 1000)))]
    else:
        # POSIX -W is whole seconds; round up so a sub-second timeout never
        # becomes 0 (read as "no timeout" by some implementations).
        options = ["-c", "1", "-W", str(max(1, int(_math.ceil(timeout)))), "-n"]

    if ipv6 is True:
        options.append("-6")
    elif ipv6 is False:
        options.append("-4")

    if src is not None:
        resolved_source = _interface_address(src, want_ipv6=bool(ipv6), strict=False)
        if resolved_source is None:
            # An interface with no usable address cannot be a src.
            return PingResult(False, hostname, attempts=0)
        # Windows spells it -S <addr>; POSIX uses -I <addr-or-ifname>.
        options.extend(["-S" if _os.name == "nt" else "-I", str(resolved_source)])

    if size is not None:
        if size < 0:
            raise ValueError("size must be non-negative, got %r" % (size,))
        options.extend(["-l" if _os.name == "nt" else "-s", str(size)])

    if ttl is not None:
        if not 1 <= ttl <= 255:
            raise ValueError("ttl must be 1-255, got %r" % (ttl,))
        # -i on Windows is TTL; on POSIX -i is the *interval* and -t is TTL.
        options.extend(["-i" if _os.name == "nt" else "-t", str(ttl)])

    if dont_fragment:
        if _os.name == "nt":
            options.append("-f")
        elif _sys.platform.startswith("linux"):
            options.extend(["-M", "do"])
        # macOS/BSD ping has no portable DF flag; silently omitted.

    # A hard cap on the subprocess itself: -W bounds how long ping waits for a
    # reply, but not how long name resolution can hang beforehand.
    wall_timeout = max(timeout, 1.0) + 5.0

    # Windows exits 0 for "TTL expired in transit", so a zero exit alone does
    # not mean the target answered. Confirm the reply came from the target
    # itself by matching its address in the output -- an address comparison,
    # never the localised prose around it.
    expect_address = _try_parse(dst)
    if expect_address is None:
        try:
            expect_address = _try_parse(_socket.gethostbyname(dst))
        except OSError:
            expect_address = None

    for attempt in range(1, tries + 1):
        try:
            response = _run(
                ["ping", *options, dst],
                capture_output=True,
                timeout=wall_timeout,
            )
        except (OSError, _SubprocessTimeout):
            # No ping binary, or it hung past the wall clock.
            return PingResult(False, dst, attempts=attempt)
        if response.returncode != 0:
            continue

        text = (response.stdout or b"").decode("utf-8", "replace")

        # A zero exit is not proof the *target* answered: Windows also exits 0
        # for "TTL expired in transit", where a router replied instead. Confirm
        # by address, which is locale-independent -- but only when the target's
        # address is known, since a bare exit code is all we have otherwise.
        if expect_address is not None:
            needle = "%s:" % expect_address
            answered = False
            for line in text.splitlines():
                if needle not in line:
                    continue
                lowered = line.lower()
                if "expired" in lowered or "unreachable" in lowered:
                    continue  # a router's error, not the destination
                if "bytes=" in lowered or "time" in lowered or "ttl=" in lowered:
                    answered = True
                    break
            if not answered:
                continue

        rtt, reply_ttl, src = _parse_ping_output(text, expect_address)
        return PingResult(
            True,
            dst,
            rtt_ms=rtt,
            ttl=reply_ttl,
            src=src if src is not None else expect_address,
            attempts=attempt,
        )
    return PingResult(False, dst, attempts=tries)

scan_hosts(network, port=None, ports=None, timeout=1.0, workers=_DEFAULT_WORKERS)

Find responsive hosts on network, with the ports each answers on.

::

scan_hosts("192.168.1.0/24", port=22)        # who has SSH open
scan_hosts("10.0.0.0/28", ports=[80, 443])
scan_hosts("192.168.1.0/24")                 # the 'common' set each
scan_hosts("192.168.1.0/24", port="https")   # scheme name works too

:param network: anything :func:netimps.parse accepts as a network. Only usable host addresses are probed -- the network and broadcast addresses are skipped. :param port: shorthand for ports=[port]. Accepts a scheme name too, so port="ssh" is port=22. :param ports: ports to probe per host; defaults to the "common" set. Same forms as :func:scan_ports. :param timeout: per-connection timeout. :param workers: total concurrent connections across all hosts.

Returns [(address, [open_ports]), ...] sorted by address, including only hosts with at least one open port.

.. note:: This is a TCP sweep, so a host that is up but answers on none of the probed ports does not appear. That is the honest result for the question asked -- it is not an ARP or ICMP discovery scan, and a firewalled host is indistinguishable from an absent one. Widen ports if you need better coverage.

Refuses networks larger than /16 (or IPv6 /112): a /8 sweep is 16 million hosts, which is a mistake rather than an intention.

Source code in src/netimps/_scan.py

def scan_hosts(
    network,
    port: "Optional[int]" = None,
    ports=None,
    timeout: float = 1.0,
    workers: int = _DEFAULT_WORKERS,
) -> "List[Tuple[object, List[int]]]":
    """Find responsive hosts on ``network``, with the ports each answers on.

    ::

        scan_hosts("192.168.1.0/24", port=22)        # who has SSH open
        scan_hosts("10.0.0.0/28", ports=[80, 443])
        scan_hosts("192.168.1.0/24")                 # the 'common' set each
        scan_hosts("192.168.1.0/24", port="https")   # scheme name works too

    :param network: anything :func:`netimps.parse` accepts as a network. Only
        usable host addresses are probed -- the network and broadcast addresses
        are skipped.
    :param port: shorthand for ``ports=[port]``. Accepts a scheme name too, so
        ``port="ssh"`` is ``port=22``.
    :param ports: ports to probe per host; defaults to the ``"common"`` set.
        Same forms as :func:`scan_ports`.
    :param timeout: per-connection timeout.
    :param workers: total concurrent connections across all hosts.

    Returns ``[(address, [open_ports]), ...]`` sorted by address, including
    only hosts with at least one open port.

    .. note::
       This is a **TCP** sweep, so a host that is up but answers on none of the
       probed ports does not appear. That is the honest result for the question
       asked -- it is not an ARP or ICMP discovery scan, and a firewalled host
       is indistinguishable from an absent one. Widen ``ports`` if you need
       better coverage.

    Refuses networks larger than /16 (or IPv6 /112): a /8 sweep is 16 million
    hosts, which is a mistake rather than an intention.
    """
    from . import IPNetwork, parse, tcp_check

    net = parse(network, IPNetwork)
    if net.version == 4 and net.prefixlen < 16:
        raise ValueError(
            "%s has %d addresses; scan a /16 or smaller" % (net, net.num_addresses)
        )
    if net.version == 6 and net.prefixlen < 112:
        raise ValueError("%s is too large to sweep; scan a /112 or smaller" % (net,))

    if port is not None and ports is not None:
        raise ValueError("pass either port or ports, not both")
    targets = _resolve_ports([port] if port is not None else (ports or "common"))
    if not targets:
        return []

    # A /31 or /32 has no separate network/broadcast address, and .hosts()
    # already accounts for that.
    addresses = list(net.hosts()) or [net.network_address]
    work = [(address, probe) for address in addresses for probe in targets]

    found = {}
    with _ThreadPool(max_workers=min(workers, len(work))) as pool:
        results = pool.map(
            lambda item: (item[0], item[1], tcp_check(str(item[0]), item[1], timeout)),
            work,
        )
        for address, probe, is_open in results:
            if is_open:
                found.setdefault(address, []).append(probe)

    return [(address, sorted(found[address])) for address in sorted(found)]

scan_ports(host, ports='common', timeout=1.0, workers=_DEFAULT_WORKERS)

Return the sorted open TCP ports on host.

::

scan_ports("192.168.1.1")                    # the 'common' set
scan_ports("localhost", "well-known")        # ports 1-1023
scan_ports("10.0.0.5", range(8000, 8100))
scan_ports("10.0.0.5", [22, 80, 443])
scan_ports("10.0.0.5", "https")              # scheme name -> 443
scan_ports("10.0.0.5", ["ssh", "https"])     # -> 22, 443

:param ports: a :data:PORT_RANGES name ("common", "well-known", "all"), a scheme name resolved via :func:get_default_port, a port number, or any iterable mixing those. A range name wins over a scheme name where the two collide. :param timeout: per-port connect timeout. This bounds the whole scan (timeout x rounds), so keep it small on a large range -- but not so small that a slow host reads as closed. :param workers: concurrent connections. These tasks are I/O-bound, so the useful number is far above the CPU count; very high values can exhaust file descriptors or trip rate limiting.

Open means "the TCP handshake completed" -- not that the service is healthy, and not that a filtered port is distinguishable from a closed one (both simply fail to connect).

Source code in src/netimps/_scan.py

def scan_ports(
    host: str,
    ports="common",
    timeout: float = 1.0,
    workers: int = _DEFAULT_WORKERS,
) -> "List[int]":
    """Return the sorted open TCP ports on ``host``.

    ::

        scan_ports("192.168.1.1")                    # the 'common' set
        scan_ports("localhost", "well-known")        # ports 1-1023
        scan_ports("10.0.0.5", range(8000, 8100))
        scan_ports("10.0.0.5", [22, 80, 443])
        scan_ports("10.0.0.5", "https")              # scheme name -> 443
        scan_ports("10.0.0.5", ["ssh", "https"])     # -> 22, 443

    :param ports: a :data:`PORT_RANGES` name (``"common"``, ``"well-known"``,
        ``"all"``), a scheme name resolved via :func:`get_default_port`, a port
        number, or any iterable mixing those. A range name wins over a scheme
        name where the two collide.
    :param timeout: per-port connect timeout. This bounds the whole scan
        (``timeout`` x rounds), so keep it small on a large range -- but not so
        small that a slow host reads as closed.
    :param workers: concurrent connections. These tasks are I/O-bound, so the
        useful number is far above the CPU count; very high values can exhaust
        file descriptors or trip rate limiting.

    Open means "the TCP handshake completed" -- not that the service is
    healthy, and not that a filtered port is distinguishable from a closed one
    (both simply fail to connect).
    """
    from . import tcp_check

    targets = _resolve_ports(ports)
    if not targets:
        return []

    open_ports = []
    with _ThreadPool(max_workers=min(workers, len(targets))) as pool:
        results = pool.map(lambda p: (p, tcp_check(host, p, timeout)), targets)
        open_ports = [port for port, is_open in results if is_open]
    return sorted(open_ports)

is_multicast(address)

True if address is a multicast group (224.0.0.0/4 or ff00::/8).

::

is_multicast("224.0.0.251")   # True  -- mDNS
is_multicast("ff02::fb")      # True
is_multicast("10.0.0.1")      # False

Never raises: anything unparseable is False.

Source code in src/netimps/_multicast.py

def is_multicast(address) -> bool:
    """True if ``address`` is a multicast group (``224.0.0.0/4`` or ``ff00::/8``).

    ::

        is_multicast("224.0.0.251")   # True  -- mDNS
        is_multicast("ff02::fb")      # True
        is_multicast("10.0.0.1")      # False

    Never raises: anything unparseable is ``False``.
    """
    from . import IPAddress, try_parse

    parsed = try_parse(address, IPAddress)
    return bool(parsed is not None and parsed.is_multicast)

join_group(sock, group, interface=None)

Join group on sock, optionally via a specific interface.

interface accepts an :class:Interface, a MAC, an adapter name or a local address. Without one the kernel chooses by routing table, which on a host with VMs or a VPN is often the wrong adapter -- and the failure is silent: the socket simply never receives.

Raises :class:ValueError for a non-multicast group or an interface with no usable address, and :class:OSError if the kernel rejects the join.

Source code in src/netimps/_multicast.py

def join_group(sock, group: str, interface=None) -> None:
    """Join ``group`` on ``sock``, optionally via a specific ``interface``.

    ``interface`` accepts an :class:`Interface`, a MAC, an adapter name or a
    local address. Without one the kernel chooses by routing table, which on a
    host with VMs or a VPN is often the wrong adapter -- and the failure is
    silent: the socket simply never receives.

    Raises :class:`ValueError` for a non-multicast group or an interface with no
    usable address, and :class:`OSError` if the kernel rejects the join.
    """
    if not is_multicast(group):
        raise ValueError("%r is not a multicast group" % (group,))

    ipv6 = ":" in group
    address = _interface_address(interface, want_ipv6=ipv6)
    request = _membership_request(
        group, None if address is None else str(address), ipv6
    )
    if ipv6:
        sock.setsockopt(_socket.IPPROTO_IPV6, _socket.IPV6_JOIN_GROUP, request)
    else:
        sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_ADD_MEMBERSHIP, request)

leave_group(sock, group, interface=None)

Leave group on sock. The inverse of :func:join_group.

Closing the socket drops membership too, so this is only needed to leave a group while keeping the socket open.

Source code in src/netimps/_multicast.py

def leave_group(sock, group: str, interface=None) -> None:
    """Leave ``group`` on ``sock``. The inverse of :func:`join_group`.

    Closing the socket drops membership too, so this is only needed to leave a
    group while keeping the socket open.
    """
    if not is_multicast(group):
        raise ValueError("%r is not a multicast group" % (group,))

    ipv6 = ":" in group
    address = _interface_address(interface, want_ipv6=ipv6)
    request = _membership_request(
        group, None if address is None else str(address), ipv6
    )
    if ipv6:
        sock.setsockopt(_socket.IPPROTO_IPV6, _socket.IPV6_LEAVE_GROUP, request)
    else:
        sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_DROP_MEMBERSHIP, request)

multicast_socket(group=None, port=0, interface=None, ttl=1, loop=True, bind=True, reuse=True)

Return a UDP socket configured for multicast, joined to group.

The whole setup in one call::

sock = multicast_socket("224.0.0.251", 5353)     # mDNS listener
data, sender = sock.recvfrom(2048)

sender = multicast_socket(ttl=32, bind=False)    # send-only, off-link
sender.sendto(b"hello", ("239.1.2.3", 9999))

:param group: group to join, or a list of them. None configures the socket for sending without joining anything. :param port: local port to bind. 0 picks a free one; a listener must pass the port the senders use. :param interface: :class:Interface, MAC, adapter name or local address to join through. Strongly recommended on multi-homed hosts -- the routing-table default is frequently the wrong adapter. :param ttl: hop limit for outgoing datagrams. The default of 1 keeps traffic on the local link; raise it deliberately to cross routers. :param loop: whether this host receives its own transmissions. True (the default) is what you want when sender and listener share a host. :param bind: bind to port. Binding to "" rather than the group address, because binding to the group fails on Windows. :param reuse: set SO_REUSEADDR (plus SO_REUSEPORT where it exists), so several listeners can share the port. SO_REUSEPORT is absent on Windows and is skipped there rather than raising.

The caller owns the socket and should close it; closing drops membership. Raises :class:ValueError for a non-multicast group, :class:OSError if binding or joining fails.

Source code in src/netimps/_multicast.py

def multicast_socket(
    group: "Union[str, List[str], None]" = None,
    port: int = 0,
    interface=None,
    ttl: int = 1,
    loop: bool = True,
    bind: bool = True,
    reuse: bool = True,
) -> "_socket.socket":
    """Return a UDP socket configured for multicast, joined to ``group``.

    The whole setup in one call::

        sock = multicast_socket("224.0.0.251", 5353)     # mDNS listener
        data, sender = sock.recvfrom(2048)

        sender = multicast_socket(ttl=32, bind=False)    # send-only, off-link
        sender.sendto(b"hello", ("239.1.2.3", 9999))

    :param group: group to join, or a list of them. ``None`` configures the
        socket for sending without joining anything.
    :param port: local port to bind. ``0`` picks a free one; a listener must
        pass the port the senders use.
    :param interface: :class:`Interface`, MAC, adapter name or local address to
        join through. **Strongly recommended on multi-homed hosts** -- the
        routing-table default is frequently the wrong adapter.
    :param ttl: hop limit for outgoing datagrams. The default of **1 keeps
        traffic on the local link**; raise it deliberately to cross routers.
    :param loop: whether this host receives its own transmissions. ``True``
        (the default) is what you want when sender and listener share a host.
    :param bind: bind to ``port``. Binding to ``""`` rather than the group
        address, because binding to the group fails on Windows.
    :param reuse: set ``SO_REUSEADDR`` (plus ``SO_REUSEPORT`` where it exists),
        so several listeners can share the port. ``SO_REUSEPORT`` is absent on
        Windows and is skipped there rather than raising.

    The caller owns the socket and should close it; closing drops membership.
    Raises :class:`ValueError` for a non-multicast group, :class:`OSError` if
    binding or joining fails.
    """
    groups = [group] if isinstance(group, str) else list(group or [])
    ipv6 = any(":" in entry for entry in groups)
    family = _socket.AF_INET6 if ipv6 else _socket.AF_INET

    sock = _socket.socket(family, _socket.SOCK_DGRAM)
    try:
        if reuse:
            sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
            # Absent on Windows; setting it unconditionally would raise there.
            reuse_port = getattr(_socket, "SO_REUSEPORT", None)
            if reuse_port is not None:
                try:
                    sock.setsockopt(_socket.SOL_SOCKET, reuse_port, 1)
                except OSError:
                    pass  # present but refused (some kernels) -- not fatal

        if ipv6:
            sock.setsockopt(_socket.IPPROTO_IPV6, _socket.IPV6_MULTICAST_HOPS, ttl)
            sock.setsockopt(
                _socket.IPPROTO_IPV6, _socket.IPV6_MULTICAST_LOOP, int(loop)
            )
        else:
            sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_MULTICAST_TTL, ttl)
            sock.setsockopt(_socket.IPPROTO_IP, _socket.IP_MULTICAST_LOOP, int(loop))

        # Pin outgoing traffic to the chosen adapter as well as incoming, or
        # sends leave by the default route while joins listen elsewhere.
        outgoing = _interface_address(interface, want_ipv6=ipv6)
        if outgoing is not None and not ipv6:
            sock.setsockopt(
                _socket.IPPROTO_IP,
                _socket.IP_MULTICAST_IF,
                _socket.inet_aton(str(outgoing)),
            )

        if bind:
            # "" rather than the group address: binding to the group works on
            # Linux but fails on Windows.
            sock.bind(("", port))

        for entry in groups:
            join_group(sock, entry, interface)
    except BaseException:
        sock.close()
        raise
    return sock

backoff_delays(attempts=3, delay=0.5, multiplier=2.0, max_delay=30.0, jitter=0.1, _random=None)

Yield the delay before each retry -- attempts - 1 values.

Exposed separately so a caller driving its own loop (async, or with progress reporting) gets the same schedule without reimplementing it::

for wait in backoff_delays(attempts=5):
    ...

:param jitter: fraction of each delay to randomise, spreading retries so that many clients failing together do not resynchronise into a thundering herd. 0 disables it and makes the schedule exact.

Delays are capped at max_delay. Jitter is applied after the cap and only ever reduces the wait, so max_delay is a genuine ceiling.

Source code in src/netimps/_retry.py

def backoff_delays(
    attempts: int = 3,
    delay: float = 0.5,
    multiplier: float = 2.0,
    max_delay: float = 30.0,
    jitter: float = 0.1,
    _random=None,
) -> "Iterator[float]":
    """Yield the delay before each retry -- ``attempts - 1`` values.

    Exposed separately so a caller driving its own loop (async, or with
    progress reporting) gets the same schedule without reimplementing it::

        for wait in backoff_delays(attempts=5):
            ...

    :param jitter: fraction of each delay to randomise, spreading retries so
        that many clients failing together do not resynchronise into a
        thundering herd. ``0`` disables it and makes the schedule exact.

    Delays are capped at ``max_delay``. Jitter is applied *after* the cap and
    only ever reduces the wait, so ``max_delay`` is a genuine ceiling.
    """
    if attempts < 1:
        raise ValueError("attempts must be at least 1, got %r" % (attempts,))
    if delay < 0:
        raise ValueError("delay must be non-negative, got %r" % (delay,))
    if not 0 <= jitter <= 1:
        raise ValueError("jitter must be between 0 and 1, got %r" % (jitter,))

    if _random is None:
        import random as _random_module

        _random = _random_module.random

    current = delay
    for _ in range(attempts - 1):
        capped = min(current, max_delay)
        if jitter:
            capped -= capped * jitter * _random()
        yield capped
        current *= multiplier

retry(func, attempts=3, delay=0.5, multiplier=2.0, max_delay=30.0, jitter=0.1, retryable=DEFAULT_RETRYABLE, on_retry=None, _sleep=_time.sleep, _random=None)

Call func(), retrying transient failures with exponential backoff.

::

result = retry(lambda: client.fetch(url))
result = retry(fetch, attempts=5, delay=1.0)

Returns whatever func returns. If every attempt fails, the last exception is re-raised -- not wrapped -- so the traceback still points at the real problem.

:param retryable: exception types worth another attempt. Defaults to OSError, which covers the socket family. Anything else propagates immediately: a ValueError means the call is malformed and will fail identically next time. :param on_retry: called as (attempt, exception, next_delay) before each wait -- the hook for logging, since this deliberately does no logging of its own.

attempts counts total calls, not retries: attempts=1 calls once and never sleeps. Delay grows by multiplier each round, capped at max_delay, with jitter applied to avoid a thundering herd.

Synchronous by design -- it blocks. For async, drive :func:backoff_delays from your own loop.

Source code in src/netimps/_retry.py

def retry(
    func: "Callable[[], object]",
    attempts: int = 3,
    delay: float = 0.5,
    multiplier: float = 2.0,
    max_delay: float = 30.0,
    jitter: float = 0.1,
    retryable: "Tuple[Type[BaseException], ...]" = DEFAULT_RETRYABLE,
    on_retry: "Optional[Callable[[int, BaseException, float], None]]" = None,
    _sleep=_time.sleep,
    _random=None,
) -> "Any":
    """Call ``func()``, retrying transient failures with exponential backoff.

    ::

        result = retry(lambda: client.fetch(url))
        result = retry(fetch, attempts=5, delay=1.0)

    Returns whatever ``func`` returns. If every attempt fails, **the last
    exception is re-raised** -- not wrapped -- so the traceback still points at
    the real problem.

    :param retryable: exception types worth another attempt. Defaults to
        ``OSError``, which covers the socket family. **Anything else
        propagates immediately**: a ``ValueError`` means the call is malformed
        and will fail identically next time.
    :param on_retry: called as ``(attempt, exception, next_delay)`` before each
        wait -- the hook for logging, since this deliberately does no logging
        of its own.

    ``attempts`` counts *total* calls, not retries: ``attempts=1`` calls once
    and never sleeps. Delay grows by ``multiplier`` each round, capped at
    ``max_delay``, with ``jitter`` applied to avoid a thundering herd.

    Synchronous by design -- it blocks. For async, drive :func:`backoff_delays`
    from your own loop.
    """
    delays = list(
        backoff_delays(
            attempts=attempts,
            delay=delay,
            multiplier=multiplier,
            max_delay=max_delay,
            jitter=jitter,
            _random=_random,
        )
    )

    for index in range(attempts):
        try:
            return func()
        except retryable as exc:
            if index >= len(delays):
                raise  # last attempt: surface the real error, unwrapped
            wait = delays[index]
            if on_retry is not None:
                on_retry(index + 1, exc, wait)
            if wait:
                _sleep(wait)

    # Unreachable: the loop either returns or raises.
    raise AssertionError("retry loop exited without result")

bind(address='', port=0, *, family=_socket.AF_INET, kind=_socket.SOCK_DGRAM, reuse_address=True, reuse_port=False, broadcast=False, interface=None, options=(), listen=None)

Create, configure and bind a socket in one call.

The setup every server repeats, with the options that are easy to get wrong handled once::

sock = bind("", 67, broadcast=True)               # DHCP-style listener
sock = bind("127.0.0.1", 0, kind=SOCK_STREAM, listen=5)
sock = bind(port=5353, interface="eth0")          # pin to one adapter

:param address: local address to bind. "" (the default) is the wildcard, which is what a server almost always wants. :param port: local port; 0 lets the OS choose. :param interface: bind to this adapter's address instead of address. Accepts an :class:Interface, a MAC, an adapter name or an address -- the same union as ping(src=). Raises :class:ValueError if it cannot be resolved, rather than silently binding the wildcard. :param reuse_port: sets SO_REUSEPORT. A no-op where the option does not exist (Windows) rather than an error, so the same call works everywhere. :param options: extra (level, name, value) triples for anything not covered by the named arguments. :param listen: call listen(backlog) after binding. Ignored for datagram sockets, where it is meaningless.

Raises :class:OSError if the bind fails -- see :func:bind_error_hint for turning that into something a user can act on. The socket is closed before the exception propagates, so a failed call leaks nothing.

Source code in src/netimps/_sockets.py

def bind(
    address: str = "",
    port: int = 0,
    *,
    family: int = _socket.AF_INET,
    kind: int = _socket.SOCK_DGRAM,
    reuse_address: bool = True,
    reuse_port: bool = False,
    broadcast: bool = False,
    interface=None,
    options=(),
    listen: "Optional[int]" = None,
) -> "_socket.socket":
    """Create, configure and bind a socket in one call.

    The setup every server repeats, with the options that are easy to get
    wrong handled once::

        sock = bind("", 67, broadcast=True)               # DHCP-style listener
        sock = bind("127.0.0.1", 0, kind=SOCK_STREAM, listen=5)
        sock = bind(port=5353, interface="eth0")          # pin to one adapter

    :param address: local address to bind. ``""`` (the default) is the
        wildcard, which is what a server almost always wants.
    :param port: local port; ``0`` lets the OS choose.
    :param interface: bind to this adapter's address instead of ``address``.
        Accepts an :class:`Interface`, a MAC, an adapter name or an address --
        the same union as ``ping(src=)``. Raises :class:`ValueError` if it
        cannot be resolved, rather than silently binding the wildcard.
    :param reuse_port: sets ``SO_REUSEPORT``. **A no-op where the option does
        not exist** (Windows) rather than an error, so the same call works
        everywhere.
    :param options: extra ``(level, name, value)`` triples for anything not
        covered by the named arguments.
    :param listen: call ``listen(backlog)`` after binding. Ignored for
        datagram sockets, where it is meaningless.

    Raises :class:`OSError` if the bind fails -- see :func:`bind_error_hint`
    for turning that into something a user can act on. The socket is closed
    before the exception propagates, so a failed call leaks nothing.
    """
    if interface is not None:
        resolved = _interface_address(interface, want_ipv6=(family == _socket.AF_INET6))
        if resolved is None:
            raise ValueError("cannot resolve interface %r to an address" % (interface,))
        address = str(resolved)

    sock = _socket.socket(family, kind)
    try:
        if reuse_address:
            sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
        if reuse_port:
            # Absent on Windows; setting it unconditionally would raise there.
            option = getattr(_socket, "SO_REUSEPORT", None)
            if option is not None:
                try:
                    sock.setsockopt(_socket.SOL_SOCKET, option, 1)
                except OSError:
                    pass  # present but refused by this kernel -- not fatal
        if broadcast:
            sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_BROADCAST, 1)
        for level, name, value in options:
            sock.setsockopt(level, name, value)

        sock.bind((address, port))
        if listen is not None and kind == _socket.SOCK_STREAM:
            sock.listen(listen)
    except BaseException:
        sock.close()
        raise
    return sock

discover_mtu(dst, low=576, high=9000, timeout=1.0, src=None, port=80, probe=True, method='icmp', **ping_kwargs)

Measure the path MTU to dst in bytes, or None if undiscoverable.

Sends DF-flagged pings of growing size, binary-searching for the largest packet that survives the whole path unfragmented::

discover_mtu("example.com")      # 1500, or 1420 through a VPN

This actually traverses the path, which is the difference from :func:get_pmtu: that reports what the kernel already knows (often nothing), while this goes and finds out. Packets really reach dst and come back, so the answer reflects every hop in between -- including a router that silently drops oversized DF packets without sending "fragmentation needed", which nothing else will reveal.

Measured on one host: the local link was 9000 and get_pmtu returned None, while this reported the true 1500.

The cost is a dozen or so probes and a destination willing to answer ICMP.

:param low: smallest MTU to consider. 576 is the IPv4 minimum every host must accept, so anything smaller means the host is simply unreachable. :param high: largest to consider. 9000 covers jumbo frames; the search confirms the ceiling first, so a generous value costs one probe. :param src: send from this interface -- same union as ping(src=). :param port: destination port passed through to :func:get_pmtu. :param method: how to probe. "icmp" (default) uses DF-flagged echo; "udp" sends datagrams of growing size to port and needs something there that replies. Use "udp" when ICMP is filtered but a UDP service answers, or to measure what a UDP application can actually push -- a middlebox may cap that below the ICMP-derived MTU.

``"tcp"`` **does not probe** -- it cannot: TCP is a stream and the
kernel segments it transparently, so a large ``send()`` silently
becomes many packets. It instead reads the negotiated MSS
(:func:`get_tcp_mss`) and adds the 40-byte IPv4+TCP header back, which
is the closest true equivalent. That is what the two *kernels agreed*,
not necessarily what a middlebox further along will pass -- use
``"icmp"`` or ``"udp"`` when the answer must be measured.

:param probe: set False to skip probing entirely and just return :func:get_pmtu -- the kernel's cached answer, usually None. :param ping_kwargs: passed straight to :func:ping for method="icmp", so anything it accepts works here -- ipv6=True to force the family, tries=3 to tolerate a lossy path. size and dont_fragment are set by the search itself and cannot be overridden.

Returns the MTU including headers (payload + 28 for IPv4 + ICMP), so it is directly comparable with :attr:Interface.mtu. Returns None when the destination never answers -- common, since many hosts and most cloud firewalls drop echo entirely, and that is indistinguishable from "every size was too big".

.. note:: The result can be lower than any local Interface.mtu, and that is the useful case: the bottleneck is somewhere along the path, not on this host.

Source code in src/netimps/_sockets.py

def discover_mtu(
    dst: str,
    low: int = 576,
    high: int = 9000,
    timeout: float = 1.0,
    src=None,
    port: int = 80,
    probe: bool = True,
    method: str = "icmp",
    **ping_kwargs,
) -> "Optional[int]":
    """Measure the path MTU to ``dst`` in bytes, or ``None`` if undiscoverable.

    Sends DF-flagged pings of growing size, binary-searching for the largest
    packet that survives the whole path unfragmented::

        discover_mtu("example.com")      # 1500, or 1420 through a VPN

    **This actually traverses the path**, which is the difference from
    :func:`get_pmtu`: that reports what the kernel already knows (often
    nothing), while this goes and finds out. Packets really reach ``dst`` and
    come back, so the answer reflects every hop in between -- including a
    router that silently drops oversized DF packets without sending
    "fragmentation needed", which nothing else will reveal.

    Measured on one host: the local link was 9000 and ``get_pmtu`` returned
    ``None``, while this reported the true 1500.

    The cost is a dozen or so probes and a destination willing to answer ICMP.

    :param low: smallest MTU to consider. 576 is the IPv4 minimum every host
        must accept, so anything smaller means the host is simply unreachable.
    :param high: largest to consider. 9000 covers jumbo frames; the search
        confirms the ceiling first, so a generous value costs one probe.
    :param src: send from this interface -- same union as ``ping(src=)``.
    :param port: destination port passed through to :func:`get_pmtu`.
    :param method: how to probe. ``"icmp"`` (default) uses DF-flagged echo;
        ``"udp"`` sends datagrams of growing size to ``port`` and needs
        something there that replies. Use ``"udp"`` when ICMP is filtered but a
        UDP service answers, or to measure what a **UDP application** can
        actually push -- a middlebox may cap that below the ICMP-derived MTU.

        ``"tcp"`` **does not probe** -- it cannot: TCP is a stream and the
        kernel segments it transparently, so a large ``send()`` silently
        becomes many packets. It instead reads the negotiated MSS
        (:func:`get_tcp_mss`) and adds the 40-byte IPv4+TCP header back, which
        is the closest true equivalent. That is what the two *kernels agreed*,
        not necessarily what a middlebox further along will pass -- use
        ``"icmp"`` or ``"udp"`` when the answer must be measured.
    :param probe: set ``False`` to skip probing entirely and just return
        :func:`get_pmtu` -- the kernel's cached answer, usually ``None``.
    :param ping_kwargs: passed straight to :func:`ping` for ``method="icmp"``,
        so anything it accepts works here -- ``ipv6=True`` to force the family,
        ``tries=3`` to tolerate a lossy path. ``size`` and ``dont_fragment``
        are set by the search itself and cannot be overridden.

    Returns the MTU **including headers** (payload + 28 for IPv4 + ICMP), so it
    is directly comparable with :attr:`Interface.mtu`. Returns ``None`` when
    the destination never answers -- common, since many hosts and most cloud
    firewalls drop echo entirely, and that is indistinguishable from "every
    size was too big".

    .. note::
       The result can be **lower than any local** ``Interface.mtu``, and that
       is the useful case: the bottleneck is somewhere along the path, not on
       this host.
    """
    if not probe:
        # Explicitly asked for the kernel's cached answer only.
        return get_pmtu(dst, port)

    method = (method or "icmp").lower()
    if method not in ("icmp", "udp", "tcp"):
        raise ValueError("method must be 'icmp', 'udp' or 'tcp', got %r" % (method,))

    if method == "tcp":
        # TCP cannot probe: the kernel segments the stream, so a large send()
        # silently becomes many packets and measures nothing. The negotiated
        # MSS is the closest true equivalent -- derive the MTU from it rather
        # than refusing to answer.
        mss = get_tcp_mss(dst, port, timeout)
        if mss is None:
            return None
        # Header size differs by family: IPv4 is 20 + 20 TCP, IPv6 is 40 + 20.
        # Using the v4 figure for a v6 path would under-report by 20 bytes.
        return mss + _tcp_header_overhead(dst)

    for owned in ("size", "dont_fragment"):
        if owned in ping_kwargs:
            raise TypeError(
                "discover_mtu sets %r itself -- it is what the search varies" % (owned,)
            )

    if method == "udp":
        return _discover_mtu_udp(dst, port, low, high, timeout)

    from . import ping

    # ping's size= is the ICMP *payload* on both Windows (-l) and POSIX (-s) --
    # neither counts headers -- so the wire packet is larger by the IP header
    # plus 8 (ICMP). IPv4 gives 28, IPv6 gives 48: applying the v4 figure to a
    # v6 path would under-report by 20 bytes.
    overhead = _ip_header_bytes(dst) + 8

    def survives(mtu: int) -> bool:
        payload = mtu - overhead
        if payload < 0:
            return False
        return bool(
            ping(
                dst,
                size=payload,
                dont_fragment=True,
                timeout=timeout,
                src=src,
                **ping_kwargs,
            )
        )

    # Establish that the host answers at all, at the smallest size worth
    # trying. Without this a firewalled host looks like a tiny MTU.
    if not survives(low):
        return None

    if survives(high):
        return high  # nothing between low and high to find

    # Invariant: `low` survives, `high` does not. Narrow until they meet.
    while high - low > 1:
        middle = (low + high) // 2
        if survives(middle):
            low = middle
        else:
            high = middle
    return low

get_pmtu(dst, port=80)

Return the path MTU the kernel has already learned, or None.

A lookup, not a measurement -- it reads IP_MTU on a connected socket and sends nothing::

get_pmtu("example.com")      # 1420, or None if nothing is cached

Instant and silent, but it answers a weaker question than :func:discover_mtu:

• None is the common answer. The kernel only knows a path MTU once its own discovery has learned one, which needs prior traffic that actually hit the limit. A fresh destination reports nothing.
• Windows has no IP_MTU (nor IP_MTU_DISCOVER / IP_DONTFRAG), so this always returns None there. Nor is there another route: the dwForwardMtu field of MIB_IPFORWARDROW reads 0 (verified via GetBestRoute; Microsoft lists it as unsupported), and the newer MIB_IPFORWARD_ROW2 dropped the field entirely. Route MTU lives at the interface level on Windows, which is :attr:Interface.mtu. Probing with :func:discover_mtu is the only way to learn a path MTU there.
• When the kernel has an answer it can still be the local link MTU rather than the path minimum, if nothing has yet forced it lower.

Use it as a free first guess; use :func:discover_mtu when the answer has to be right.

Source code in src/netimps/_sockets.py

def get_pmtu(dst: str, port: int = 80) -> "Optional[int]":
    """Return the path MTU the kernel has **already learned**, or ``None``.

    A lookup, not a measurement -- it reads ``IP_MTU`` on a connected socket
    and sends nothing::

        get_pmtu("example.com")      # 1420, or None if nothing is cached

    Instant and silent, but it answers a weaker question than
    :func:`discover_mtu`:

    * **``None`` is the common answer.** The kernel only knows a path MTU once
      its own discovery has learned one, which needs prior traffic that
      actually hit the limit. A fresh destination reports nothing.
    * **Windows has no ``IP_MTU``** (nor ``IP_MTU_DISCOVER`` / ``IP_DONTFRAG``),
      so this always returns ``None`` there. Nor is there another route: the
      ``dwForwardMtu`` field of ``MIB_IPFORWARDROW`` reads **0** (verified via
      ``GetBestRoute``; Microsoft lists it as unsupported), and the newer
      ``MIB_IPFORWARD_ROW2`` dropped the field entirely. Route MTU lives at the
      interface level on Windows, which is :attr:`Interface.mtu`. Probing with
      :func:`discover_mtu` is the only way to learn a *path* MTU there.
    * When the kernel *has* an answer it can still be the **local link** MTU
      rather than the path minimum, if nothing has yet forced it lower.

    Use it as a free first guess; use :func:`discover_mtu` when the answer has
    to be right.
    """
    ip_mtu = getattr(_socket, "IP_MTU", None)
    if ip_mtu is None:
        return None

    try:
        sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
    except OSError:
        return None
    try:
        # PMTU is only maintained for connected sockets.
        discover = getattr(_socket, "IP_MTU_DISCOVER", None)
        do_want = getattr(_socket, "IP_PMTUDISC_DO", None)
        if discover is not None and do_want is not None:
            try:
                sock.setsockopt(_socket.IPPROTO_IP, discover, do_want)
            except OSError:
                pass
        sock.connect((dst, port))
        value = int(sock.getsockopt(_socket.IPPROTO_IP, ip_mtu))
        return value if value > 0 else None
    except (OSError, OverflowError):
        return None
    finally:
        sock.close()

get_tcp_mss(dst, port, timeout=3.0)

Return the TCP maximum segment size negotiated with dst, or None.

The TCP counterpart to an MTU: the largest payload a single segment may carry, agreed during the handshake::

get_tcp_mss("example.com", 443)     # 1460 on a 1500-MTU path

This opens a real connection to read the value, then closes it.

MSS is normally the path MTU minus 40 (20 IPv4 + 20 TCP), so a reduced value is a useful signal: a VPN or tunnel is shrinking the path. Measured on one host: 1412 over a VPN where the link MTU was 1500, and 32741 on loopback.

Returns None where the platform does not expose TCP_MAXSEG or the connection fails. Note this is what the kernels agreed, not what a middlebox further along will actually pass -- for that, measure with :func:discover_mtu.

Source code in src/netimps/_sockets.py

def get_tcp_mss(dst: str, port: int, timeout: float = 3.0) -> "Optional[int]":
    """Return the TCP maximum segment size negotiated with ``dst``, or ``None``.

    The TCP counterpart to an MTU: the largest payload a single segment may
    carry, agreed during the handshake::

        get_tcp_mss("example.com", 443)     # 1460 on a 1500-MTU path

    This **opens a real connection** to read the value, then closes it.

    MSS is normally the path MTU minus 40 (20 IPv4 + 20 TCP), so a reduced
    value is a useful signal: a VPN or tunnel is shrinking the path. Measured
    on one host: 1412 over a VPN where the link MTU was 1500, and 32741 on
    loopback.

    Returns ``None`` where the platform does not expose ``TCP_MAXSEG`` or the
    connection fails. Note this is what the *kernels agreed*, not what a
    middlebox further along will actually pass -- for that, measure with
    :func:`discover_mtu`.
    """
    option = getattr(_socket, "TCP_MAXSEG", None)
    if option is None:
        return None

    # Let getaddrinfo pick the family so a v6-only destination works.
    try:
        infos = _socket.getaddrinfo(dst, int(port), 0, _socket.SOCK_STREAM)
    except (OSError, OverflowError, ValueError):
        return None

    for family, kind, proto, _canon, addr in infos:
        sock = _socket.socket(family, kind, proto)
        sock.settimeout(timeout)
        try:
            sock.connect(addr)
            value = int(sock.getsockopt(_socket.IPPROTO_TCP, option))
            return value if value > 0 else None
        except (OSError, OverflowError, ValueError):
            continue
        finally:
            sock.close()
    return None

bind_error_hint(exc, port=None)

Turn a bind failure into a sentence a user can act on, or None.

The raw OSError from a failed bind is famously unhelpful, and the errno differs per platform -- Windows reports WinError 10013/10048 where POSIX reports EACCES/EADDRINUSE::

try:
    sock = bind("", 67)
except OSError as exc:
    raise OSError(bind_error_hint(exc, 67) or str(exc)) from exc

Returns None for anything unrecognised, so the caller keeps the original error rather than a worse paraphrase. This does not raise -- deciding what to do with a failure belongs to the caller.

Source code in src/netimps/_sockets.py

def bind_error_hint(
    exc: BaseException, port: "Optional[int]" = None
) -> "Optional[str]":
    """Turn a bind failure into a sentence a user can act on, or ``None``.

    The raw ``OSError`` from a failed bind is famously unhelpful, and the errno
    differs per platform -- Windows reports ``WinError 10013``/``10048`` where
    POSIX reports ``EACCES``/``EADDRINUSE``::

        try:
            sock = bind("", 67)
        except OSError as exc:
            raise OSError(bind_error_hint(exc, 67) or str(exc)) from exc

    Returns ``None`` for anything unrecognised, so the caller keeps the
    original error rather than a worse paraphrase. This **does not raise** --
    deciding what to do with a failure belongs to the caller.
    """
    import errno as _errno

    if not isinstance(exc, OSError):
        return None

    winerror = getattr(exc, "winerror", None)
    where = "port %d" % port if port is not None else "that port"

    if (
        isinstance(exc, PermissionError)
        or exc.errno == _errno.EACCES
        or winerror == 10013
    ):
        hint = "permission denied binding %s" % where
        if port is not None and port < 1024:
            hint += "; ports below 1024 need root/Administrator"
        return hint

    if exc.errno == _errno.EADDRINUSE or winerror == 10048:
        return "%s is already in use" % where.capitalize()

    if exc.errno == _errno.EADDRNOTAVAIL or winerror == 10049:
        return (
            "that address is not available on this host; "
            "it must belong to a local interface"
        )

    return None

interface_for(query, strict=True)

Return the first local interface matching query, or None.

The reverse of interface enumeration -- "a socket is bound here, which adapter is that?"::

interface_for(sock.getsockname()[0])

Accepts the same query forms as :func:interfaces_for; singular lookup is exactly the first plural result. Since addresses can appear on more than one adapter (especially unscoped IPv6 link-local addresses), use the plural form when every match matters.

:param strict: when True (the default), a miss returns None. When False, an address or IPInterface miss produces a synthetic single-address Interface so legacy callers can still attribute traffic. Network and MAC misses cannot be synthesized honestly and remain None.

The synthetic interface is named "<unknown>" and carries a host route (/32 or /128), matching how degraded enumeration reports itself.

Source code in src/netimps/_sockets.py

def interface_for(query: _InterfaceQuery, strict: bool = True) -> "Optional[Interface]":
    """Return the first local interface matching ``query``, or ``None``.

    The reverse of interface enumeration -- "a socket is bound here, which
    adapter is that?"::

        interface_for(sock.getsockname()[0])

    Accepts the same query forms as :func:`interfaces_for`; singular lookup is
    exactly the first plural result. Since addresses can appear on more than
    one adapter (especially unscoped IPv6 link-local addresses), use the plural
    form when every match matters.

    :param strict: when True (the default), a miss returns ``None``. When
        False, an address or ``IPInterface`` miss produces a synthetic
        single-address ``Interface`` so legacy callers can still attribute
        traffic. Network and MAC misses cannot be synthesized honestly and
        remain ``None``.

    The synthetic interface is named ``"<unknown>"`` and carries a host route
    (``/32`` or ``/128``), matching how degraded enumeration reports itself.
    """
    from ._ifaddrs import Interface

    kind, wanted = _classify_interface_query(query)
    match = next(_interfaces_for_query(kind, wanted), None)
    if match is not None or strict or kind != "address":
        return match

    built = _make_host_route(wanted)
    return Interface(name="<unknown>", ips=[built] if built else [])

interfaces_for(query)

Yield every local interface matching query, in OS order.

query may be an :class:Interface (yielded directly), an address, an IPInterface (matched by its exact .ip), an IPNetwork (matched when it contains any assigned address), or a :class:MACAddress. Address- like strings, integers and packed bytes are accepted too; a slash-bearing string is interpreted as a network when it is not an address.

MAC text and 6-byte packed values are recognised after IP parsing. Integer MACs must be wrapped in MACAddress because an integer is also a valid IP-address representation. Invalid queries and misses yield nothing. Each matching interface is yielded once even if several of its assigned addresses fall within a requested network.

Source code in src/netimps/_sockets.py

def interfaces_for(query: _InterfaceQuery) -> "Iterator[Interface]":
    """Yield every local interface matching ``query``, in OS order.

    ``query`` may be an :class:`Interface` (yielded directly), an address, an
    ``IPInterface`` (matched by its exact ``.ip``), an ``IPNetwork`` (matched
    when it contains any assigned address), or a :class:`MACAddress`. Address-
    like strings, integers and packed bytes are accepted too; a slash-bearing
    string is interpreted as a network when it is not an address.

    MAC text and 6-byte packed values are recognised after IP parsing.
    Integer MACs must be wrapped in ``MACAddress`` because an integer is also
    a valid IP-address representation. Invalid queries and misses yield
    nothing. Each matching interface is yielded once even if several of its
    assigned addresses fall within a requested network.
    """
    kind, wanted = _classify_interface_query(query)
    yield from _interfaces_for_query(kind, wanted)

is_local_address(address)

Return whether address is loopback or assigned on this host.

This is deliberately narrower than private, link-local, on-link, routable or reachable: those properties do not mean an address belongs to this machine. Malformed input raises exactly as :func:parse does.

Source code in src/netimps/_sockets.py

def is_local_address(address: "IPAddressLike") -> bool:
    """Return whether ``address`` is loopback or assigned on this host.

    This is deliberately narrower than private, link-local, on-link, routable
    or reachable: those properties do not mean an address belongs to this
    machine. Malformed input raises exactly as :func:`parse` does.
    """
    from . import IPAddress, parse

    wanted = parse(address, IPAddress)
    if wanted.is_loopback:
        return True
    return next(interfaces_for(wanted), None) is not None

get_free_port(src='127.0.0.1', family=_socket.AF_INET)

Return a port number that was free a moment ago.

Binds port 0, reads back whatever the OS assigned, and closes::

port = get_free_port()
server = start_my_server(port=port)

A getter, despite "free" in the name -- it acquires a number, it does not release anything. The port is not held open for you.

.. warning:: Inherently racy. The port is released the instant this returns, so another process can take it before you bind. There is no way around that with a returned port number -- if you can, bind port 0 in the server itself and read back getsockname() instead of calling this.

SO_REUSEADDR is deliberately not set: it would let the OS hand back a port still in TIME_WAIT, which then fails or steals traffic when the caller binds it for real.

Source code in src/netimps/_sockets.py

def get_free_port(src: str = "127.0.0.1", family: int = _socket.AF_INET) -> int:
    """Return a port number that was free a moment ago.

    Binds port 0, reads back whatever the OS assigned, and closes::

        port = get_free_port()
        server = start_my_server(port=port)

    A *getter*, despite "free" in the name -- it acquires a number, it does not
    release anything. The port is **not** held open for you.

    .. warning::
       **Inherently racy.** The port is released the instant this returns, so
       another process can take it before you bind. There is no way around that
       with a returned port number -- if you can, bind port 0 in the server
       itself and read back ``getsockname()`` instead of calling this.

    ``SO_REUSEADDR`` is deliberately **not** set: it would let the OS hand back
    a port still in ``TIME_WAIT``, which then fails or steals traffic when the
    caller binds it for real.
    """
    sock = _socket.socket(family, _socket.SOCK_STREAM)
    try:
        sock.bind((src, 0))
        return int(sock.getsockname()[1])
    finally:
        sock.close()

get_source_ip(dst=_DEFAULT_PROBE, port=80)

Return the local address the kernel would use to reach dst.

Answers "which of my addresses is the real one for this destination?" -- the question a hostname lookup gets wrong on any host with VMs, containers or a VPN::

get_source_ip()                  # IPv4Address('192.0.2.10')
get_source_ip("192.168.1.1")     # the LAN-facing address
get_source_ip("2001:4860::8888") # an IPv6 src address

No packets are sent. connect() on a UDP socket only fixes the socket's local endpoint by consulting the routing table, so this is immediate and invisible to dst.

The answer depends on dst: with a VPN up, a public probe returns the tunnel address while a LAN probe returns the physical one. Pass the address you actually intend to talk to rather than trusting the default.

Returns None if no route exists (e.g. IPv6 probe on an IPv4-only host).

Source code in src/netimps/_sockets.py

def get_source_ip(dst: str = _DEFAULT_PROBE, port: int = 80) -> "Optional[Any]":
    """Return the local address the kernel would use to reach ``dst``.

    Answers "which of my addresses is the *real* one for this destination?" --
    the question a hostname lookup gets wrong on any host with VMs, containers
    or a VPN::

        get_source_ip()                  # IPv4Address('192.0.2.10')
        get_source_ip("192.168.1.1")     # the LAN-facing address
        get_source_ip("2001:4860::8888") # an IPv6 src address

    **No packets are sent.** ``connect()`` on a UDP socket only fixes the
    socket's local endpoint by consulting the routing table, so this is
    immediate and invisible to ``dst``.

    The answer depends on ``dst``: with a VPN up, a public probe returns the
    tunnel address while a LAN probe returns the physical one. Pass the address
    you actually intend to talk to rather than trusting the default.

    Returns ``None`` if no route exists (e.g. IPv6 probe on an IPv4-only host).
    """
    from . import parse

    try:
        family = _socket.AF_INET6 if ":" in dst else _socket.AF_INET
        sock = _socket.socket(family, _socket.SOCK_DGRAM)
    except OSError:
        return None
    try:
        sock.connect((dst, port))
        return parse(sock.getsockname()[0].split("%")[0])
    except (OSError, ValueError):
        return None
    finally:
        sock.close()

hop_count(dst, max_hops=30, timeout=1.0, allow_traceroute=True)

Return the number of hops to dst, or None if it never answers.

Sends TTL-limited probes and counts the routers that reply, the same technique traceroute uses::

hop_count("8.8.8.8")     # 12

Uses raw-socket probes when available (root/Administrator), and otherwise falls back to driving the system traceroute/tracert, so this works unprivileged on a normal desktop. Pass allow_traceroute=False to require the in-process path, which then raises :class:PermissionError instead of shelling out.

The fallback is slower (seconds) because it runs a whole trace. Only the hop number and the destination address are read from its output, never the localised prose, so it is not locale-dependent.

Returns None when the destination never responds within max_hops. That is common and usually not a missing route: host firewalls (Windows Firewall in particular) routinely drop inbound ICMP even for an elevated process, so None here means "no answer", never "unreachable". Treat it as unknown rather than as a negative result.

Source code in src/netimps/_sockets.py

def hop_count(
    dst: str,
    max_hops: int = 30,
    timeout: float = 1.0,
    allow_traceroute: bool = True,
) -> Optional[int]:
    """Return the number of hops to ``dst``, or ``None`` if it never answers.

    Sends TTL-limited probes and counts the routers that reply, the same
    technique ``traceroute`` uses::

        hop_count("8.8.8.8")     # 12

    Uses raw-socket probes when available (root/Administrator), and otherwise
    falls back to driving the system ``traceroute``/``tracert``, so this works
    unprivileged on a normal desktop. Pass ``allow_traceroute=False`` to require
    the in-process path, which then raises :class:`PermissionError` instead of
    shelling out.

    The fallback is slower (seconds) because it runs a whole trace. Only the
    hop number and the destination address are read from its output, never the
    localised prose, so it is not locale-dependent.

    Returns ``None`` when the destination never responds within ``max_hops``.
    That is common and usually **not** a missing route: host firewalls (Windows
    Firewall in particular) routinely drop inbound ICMP even for an elevated
    process, so ``None`` here means "no answer", never "unreachable". Treat it
    as unknown rather than as a negative result.
    """
    try:
        target = _socket.gethostbyname(dst)
    except OSError:
        return None

    try:
        icmp = _socket.socket(_socket.AF_INET, _socket.SOCK_RAW, _socket.IPPROTO_ICMP)
    except (OSError, AttributeError) as exc:
        if allow_traceroute:
            return _hop_count_traceroute(target, max_hops, timeout)
        raise PermissionError(
            "hop_count needs a raw socket (root/Administrator); "
            "pass allow_traceroute=True, or use get_route() for the first hop"
        ) from exc

    try:
        icmp.settimeout(timeout)
        # Windows will not deliver ICMP to a raw socket bound to INADDR_ANY --
        # it must be bound to a real local address, and put into promiscuous
        # mode with SIO_RCVALL. On POSIX, binding to "" is both sufficient and
        # correct.
        bind_to = ""
        if _IS_WINDOWS:
            src = get_source_ip(target)
            bind_to = str(src) if src is not None else ""
        try:
            icmp.bind((bind_to, 0))
        except OSError:
            pass
        if _IS_WINDOWS:
            try:
                icmp.ioctl(_socket.SIO_RCVALL, _socket.RCVALL_ON)
            except (OSError, AttributeError):
                pass
        for ttl in range(1, max_hops + 1):
            probe = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
            try:
                probe.setsockopt(_socket.IPPROTO_IP, _socket.IP_TTL, ttl)
                probe.sendto(b"", (target, 33434 + ttl))
            except OSError:
                continue
            finally:
                probe.close()

            # With SIO_RCVALL the socket sees unrelated traffic too, so keep
            # reading (within this hop's budget) until an ICMP packet that is
            # actually a reply shows up, rather than trusting the first one.
            deadline = _time.monotonic() + timeout
            while _time.monotonic() < deadline:
                try:
                    icmp.settimeout(max(0.01, deadline - _time.monotonic()))
                    packet, addr = icmp.recvfrom(1024)
                except (_socket.timeout, OSError):
                    break
                if not _is_icmp_reply(packet):
                    continue
                if addr[0] == target:
                    return ttl  # destination itself answered: distance found
                break  # a router replied: this hop is done, try the next TTL

        # Raw probes got no answer -- commonly a host firewall dropping inbound
        # ICMP even for an elevated process. Try the system tool before giving
        # up, since it often succeeds where the raw socket does not.
        return (
            _hop_count_traceroute(target, max_hops, timeout)
            if allow_traceroute
            else None
        )
    finally:
        icmp.close()

get_route(dst=_DEFAULT_PROBE)

Return how traffic to dst leaves this host.

Reports the src address and the first hop -- the gateway a packet is handed to, or None when the destination is on-link::

r = get_route("8.8.8.8")
r.src        # IPv4Address('192.0.2.10')
r.gateway       # IPv4Address('192.0.2.1')
r.on_link       # False

get_route("127.0.0.1").on_link      # True -- no router involved

First hop only, deliberately: it is available unprivileged on every supported platform, whereas the full path requires raw sockets. See :func:hop_count for distance, which does not.

Never raises: unknown pieces come back as None/0 rather than an error. The gateway is only resolvable on Windows and Linux; elsewhere it stays None, so use .gateway is None to mean "on-link" only when .src is also set.

Source code in src/netimps/_sockets.py

def get_route(dst: str = _DEFAULT_PROBE) -> Route:
    """Return how traffic to ``dst`` leaves this host.

    Reports the src address and the **first hop** -- the gateway a packet is
    handed to, or ``None`` when the destination is on-link::

        r = get_route("8.8.8.8")
        r.src        # IPv4Address('192.0.2.10')
        r.gateway       # IPv4Address('192.0.2.1')
        r.on_link       # False

        get_route("127.0.0.1").on_link      # True -- no router involved

    First hop only, deliberately: it is available **unprivileged** on every
    supported platform, whereas the full path requires raw sockets. See
    :func:`hop_count` for distance, which does not.

    Never raises: unknown pieces come back as ``None``/``0`` rather than an
    error. The gateway is only resolvable on Windows and Linux; elsewhere it
    stays ``None``, so use ``.gateway is None`` to mean "on-link" only when
    ``.src`` is also set.
    """
    from . import parse, try_parse

    src = get_source_ip(dst)
    resolved = dst
    if try_parse(dst) is None:
        try:
            resolved = _socket.gethostbyname(dst)
        except OSError:
            resolved = dst

    gateway_text = None
    index = 0
    parsed_dest = try_parse(resolved)
    if parsed_dest is not None and parsed_dest.version == 4:
        try:
            if _IS_WINDOWS:
                gateway_text, index = _windows_next_hop(resolved)
            else:
                gateway_text, index = _posix_next_hop(resolved)
        except (OSError, AttributeError, ValueError, _struct.error):
            gateway_text, index = None, 0

    return Route(
        dst=parsed_dest if parsed_dest is not None else dst,
        src=src,
        gateway=try_parse(gateway_text) if gateway_text else None,
        interface_index=index,
    )

tcp_check(dst, port, timeout=3.0)

Return True if a TCP connection to dst:port is accepted.

The honest reachability test, and what you almost always want instead of :func:netimps.ping: it proves the service answers, not merely that the host replies to ICMP echo (which most cloud firewalls drop anyway)::

tcp_check("example.com", 443)
tcp_check("db.internal", 5432, timeout=1.0)

Never raises: refused, timed out, unresolvable and unreachable all yield False. Only TCP handshake completion is checked -- not that the service behind the port is healthy.

.. note:: Not the same question as ping(dst, method="tcp", port=...). This asks "is the service up?", so a refused connection is False. That asks "is the host up?", and counts a refusal as success -- the RST proves something answered. Same distinction as a service check versus an ICMP echo. :func:wait_for_port and the scanners build on this one, because they care about the service.

Source code in src/netimps/_sockets.py

def tcp_check(dst: str, port: int, timeout: float = 3.0) -> bool:
    """Return True if a TCP connection to ``dst``:``port`` is accepted.

    The honest reachability test, and what you almost always want instead of
    :func:`netimps.ping`: it proves the *service* answers, not merely that the
    host replies to ICMP echo (which most cloud firewalls drop anyway)::

        tcp_check("example.com", 443)
        tcp_check("db.internal", 5432, timeout=1.0)

    Never raises: refused, timed out, unresolvable and unreachable all yield
    ``False``. Only TCP handshake completion is checked -- not that the service
    behind the port is healthy.

    .. note::
       **Not the same question as** ``ping(dst, method="tcp", port=...)``. This
       asks "is the *service* up?", so a refused connection is ``False``. That
       asks "is the *host* up?", and counts a refusal as success -- the RST
       proves something answered. Same distinction as a service check versus an
       ICMP echo. :func:`wait_for_port` and the scanners build on this one,
       because they care about the service.
    """
    try:
        sock = _socket.create_connection((dst, port), timeout=timeout)
    except (OSError, ValueError, OverflowError):
        return False
    sock.close()
    return True

wait_for_port(dst, port, timeout=30.0, interval=0.1, connect_timeout=None)

Poll until dst:port accepts a connection, or timeout elapses.

The "wait for the service to come up" loop every deploy and container script contains::

if not wait_for_port("localhost", 5432, timeout=60):
    raise RuntimeError("database never started")

:param interval: delay between attempts. Backs off up to 1s so a long wait does not spin. :param connect_timeout: per-attempt connect timeout; defaults to interval bounded to at least 1s.

Returns True as soon as the port answers, False on timeout. The deadline is honoured overall, so this cannot overrun by more than one attempt regardless of how long individual connects block.

Source code in src/netimps/_sockets.py

def wait_for_port(
    dst: str,
    port: int,
    timeout: float = 30.0,
    interval: float = 0.1,
    connect_timeout: Optional[float] = None,
) -> bool:
    """Poll until ``dst``:``port`` accepts a connection, or ``timeout`` elapses.

    The "wait for the service to come up" loop every deploy and container
    script contains::

        if not wait_for_port("localhost", 5432, timeout=60):
            raise RuntimeError("database never started")

    :param interval: delay between attempts. Backs off up to 1s so a long wait
        does not spin.
    :param connect_timeout: per-attempt connect timeout; defaults to
        ``interval`` bounded to at least 1s.

    Returns ``True`` as soon as the port answers, ``False`` on timeout. The
    deadline is honoured overall, so this cannot overrun by more than one
    attempt regardless of how long individual connects block.
    """
    deadline = _time.monotonic() + timeout
    per_try = connect_timeout if connect_timeout is not None else max(interval, 1.0)
    delay = interval

    while True:
        remaining = deadline - _time.monotonic()
        if remaining <= 0:
            return False
        if tcp_check(dst, port, timeout=min(per_try, remaining)):
            return True
        remaining = deadline - _time.monotonic()
        if remaining <= 0:
            return False
        _time.sleep(min(delay, remaining))
        delay = min(delay * 1.5, 1.0)  # gentle backoff, capped