net

net

NIP-66 network metadata container with ASN lookup capabilities.

Resolves a relay's hostname to IPv4/IPv6 addresses and queries the GeoIP ASN database for autonomous system number, organization name, and CIDR network ranges as part of NIP-66 monitoring. Clearnet relays only.

Note

Hostname resolution uses resolve_host to obtain both IPv4 and IPv6 addresses. The GeoIP ASN database (GeoLite2-ASN) must be provided as an open geoip2.database.Reader -- the caller is responsible for database lifecycle management.

IPv4 ASN data takes priority; IPv6 ASN data is used only as a fallback when no IPv4 address is available. IPv6-specific network ranges are always recorded separately.

See Also

bigbrotr.nips.nip66.data.Nip66NetData: Data model for network/ASN fields. bigbrotr.nips.nip66.logs.Nip66NetLogs: Log model for network lookup results. bigbrotr.nips.nip66.geo.Nip66GeoMetadata: Geolocation test that also uses resolve_host for IP resolution. bigbrotr.utils.dns.resolve_host: DNS resolution utility used to obtain IP addresses.

Classes

Nip66NetMetadata

Bases: BaseNipMetadata

Container for network/ASN data and lookup logs.

Provides the execute() class method that resolves the relay hostname and performs GeoIP ASN lookups for both IPv4 and IPv6 addresses.

See Also

bigbrotr.nips.nip66.nip66.Nip66: Top-level model that orchestrates this alongside other tests. bigbrotr.models.metadata.MetadataType: The NIP66_NET variant used when storing these results. bigbrotr.nips.nip66.geo.Nip66GeoMetadata: Geolocation test that shares the IP resolution step.

Functions

execute async classmethod

execute(
    relay: Relay,
    asn_reader: Reader,
    timeout: float | None = None,
) -> Self

Perform network/ASN lookups for a clearnet relay.

Resolves the relay hostname to IPv4 and IPv6 addresses, then queries the GeoIP ASN database in a thread pool.

Parameters:

• relay (Relay) –

  Clearnet relay to look up.

• asn_reader (Reader) –

  Open GeoLite2-ASN database reader.

• timeout (float | None, default: None ) –

  Overall timeout in seconds (default: 10.0).

Returns:

• Self –

  An Nip66NetMetadata instance with network data and logs.

Source code in src/bigbrotr/nips/nip66/net.py

@classmethod
async def execute(
    cls,
    relay: Relay,
    asn_reader: geoip2.database.Reader,
    timeout: float | None = None,  # noqa: ASYNC109
) -> Self:
    """Perform network/ASN lookups for a clearnet relay.

    Resolves the relay hostname to IPv4 and IPv6 addresses, then
    queries the GeoIP ASN database in a thread pool.

    Args:
        relay: Clearnet relay to look up.
        asn_reader: Open GeoLite2-ASN database reader.
        timeout: Overall timeout in seconds (default: 10.0).

    Returns:
        An ``Nip66NetMetadata`` instance with network data and logs.
    """
    timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
    logger.debug("net_testing relay=%s timeout_s=%s", relay.url, timeout)

    if relay.network != NetworkType.CLEARNET:
        return cls(
            data=Nip66NetData(),
            logs=Nip66NetLogs(
                success=False, reason=f"requires clearnet, got {relay.network.value}"
            ),
        )

    logs: dict[str, Any] = {"success": False, "reason": None}

    try:
        resolved = await asyncio.wait_for(resolve_host(relay.host), timeout=timeout)
    except TimeoutError:
        logs["reason"] = "timeout resolving hostname"
        logger.debug("net_resolve_timeout relay=%s", relay.url)
        return cls(
            data=Nip66NetData.model_validate(Nip66NetData.parse({})),
            logs=Nip66NetLogs.model_validate(logs),
        )

    data: dict[str, Any] = {}
    if resolved.has_ip:
        try:
            data = await asyncio.wait_for(
                asyncio.to_thread(cls._net, resolved.ipv4, resolved.ipv6, asn_reader),
                timeout=timeout,
            )
            if data:
                logs["success"] = True
                logger.debug("net_completed relay=%s asn=%s", relay.url, data.get("net_asn"))
            else:
                logs["reason"] = "no ASN data found for IP"
                logger.debug("net_no_data relay=%s", relay.url)
        except (TimeoutError, geoip2.errors.GeoIP2Error, ValueError) as e:
            logs["reason"] = str(e) or type(e).__name__
            logger.debug("net_lookup_failed relay=%s error=%s", relay.url, logs["reason"])
    else:
        logs["reason"] = "could not resolve hostname to IP"
        logger.debug("net_no_ip relay=%s", relay.url)

    return cls(
        data=Nip66NetData.model_validate(Nip66NetData.parse(data)),
        logs=Nip66NetLogs.model_validate(logs),
    )

Functions