relay

relay ¶

Validated Nostr relay URL with network type detection.

Parses, normalizes, and validates WebSocket relay URLs (ws:// or wss://), automatically detecting the NetworkType (clearnet, Tor, I2P, Lokinet) and enforcing the correct scheme for each network. Local and private IP addresses are rejected.

Classes¶

RelayDbParams ¶

Bases: NamedTuple

Positional parameters for the relay database insert procedure.

Produced by Relay.to_db_params() and consumed by the relay_insert stored procedure in PostgreSQL.

Attributes:

url (str) –

Fully normalized WebSocket URL including scheme.
network (str) –

Network type string (e.g., "clearnet", "tor").
discovered_at (int) –

Unix timestamp when the relay was first discovered.

Relay `dataclass` ¶

Relay(
    url: str, discovered_at: int = (lambda: int(time()))()
)

Immutable representation of a Nostr relay.

Accepts only URLs already in canonical form. If the input may be dirty (from Nostr events, relay lists, external APIs), pass it through sanitize_relay_url first.

The canonical form enforces:

scheme -- wss:// for clearnet, ws:// for overlay networks
no query string or fragment
no garbage path (control characters, whitespace, embedded URI schemes)
default ports omitted (443 for wss, 80 for ws)
lowercase host, collapsed path slashes, no trailing slash

Attributes:

url (str) –

Canonical normalized URL (init field and primary identity).
network (NetworkType) –

Detected NetworkType enum value.
scheme (str) –

URL scheme (ws or wss).
host (str) –

Hostname or IP address (brackets stripped for IPv6).
port (int | None) –

Explicit port number, or None when using the default.
path (str | None) –

URL path component, or None.
discovered_at (int) –

Unix timestamp when the relay was first discovered.

Raises:

ValueError –

If the URL is not in canonical form, malformed, uses an unsupported scheme, resolves to a local/private address, or contains null bytes.

Examples:

relay = Relay("wss://relay.damus.io")
relay.url       # 'wss://relay.damus.io'
relay.network   # NetworkType.CLEARNET
relay.scheme    # 'wss'
relay.to_db_params()
# RelayDbParams(url='wss://relay.damus.io', network='clearnet', ...)

For untrusted input, sanitize first:

from bigbrotr.models.relay import sanitize_relay_url

dirty = "ws://Relay.Example.Com:443/path?key=val#frag"
clean = sanitize_relay_url(dirty)  # 'wss://relay.example.com/path'
relay = Relay(clean)

Note

Computed fields are set via object.__setattr__ in __post_init__ because the dataclass is frozen. This is the standard workaround and is safe because it runs during __init__ before the instance is exposed.

Functions¶

to_db_params ¶

to_db_params() -> RelayDbParams

Return cached positional parameters for the database insert procedure.

The result is computed once during construction and cached for the lifetime of the (frozen) instance, avoiding repeated network name conversions.

Returns:

RelayDbParams –

RelayDbParams with the
RelayDbParams –

normalized URL, network name, and discovery timestamp.

Source code in src/bigbrotr/models/relay.py

def to_db_params(self) -> RelayDbParams:
    """Return cached positional parameters for the database insert procedure.

    The result is computed once during construction and cached for the
    lifetime of the (frozen) instance, avoiding repeated network name
    conversions.

    Returns:
        [RelayDbParams][bigbrotr.models.relay.RelayDbParams] with the
        normalized URL, network name, and discovery timestamp.
    """
    return self._db_params

Functions¶

sanitize_relay_url ¶

sanitize_relay_url(raw: str) -> str

Normalize and canonicalize a raw relay URL from untrusted input.

Applies the full normalization pipeline:

RFC 3986 parsing — scheme, host, port, path decomposition.
Host normalization — percent-decoding, trailing-dot removal, IDN-to-punycode conversion, whitespace/control-char rejection, IP address canonicalization (IPv6 compression, IPv4 validation).
Network detection — overlay TLD matching (.onion, .i2p, .loki), private/reserved IP rejection, hostname validation.
Scheme enforcement — wss:// for clearnet, ws:// for overlays.
Port normalization — range validation (1-65535), default port omission (443 for wss, 80 for ws).
Path normalization — dot-segment resolution (via rfc3986), slash collapsing, trailing-slash removal, segment validation.
Query/fragment stripping — irrelevant for WebSocket relay identity.

Parameters:

raw (str) –

Raw URL string, potentially malformed.

Returns:

str –

Canonical URL string suitable for :class:Relay construction.

Raises:

ValueError –

If the URL is structurally unrecoverable.

Source code in src/bigbrotr/models/relay.py

def sanitize_relay_url(raw: str) -> str:
    """Normalize and canonicalize a raw relay URL from untrusted input.

    Applies the full normalization pipeline:

    1. **RFC 3986 parsing** — scheme, host, port, path decomposition.
    2. **Host normalization** — percent-decoding, trailing-dot removal,
       IDN-to-punycode conversion, whitespace/control-char rejection,
       IP address canonicalization (IPv6 compression, IPv4 validation).
    3. **Network detection** — overlay TLD matching (``.onion``, ``.i2p``,
       ``.loki``), private/reserved IP rejection, hostname validation.
    4. **Scheme enforcement** — ``wss://`` for clearnet, ``ws://`` for overlays.
    5. **Port normalization** — range validation (1-65535), default port
       omission (443 for ``wss``, 80 for ``ws``).
    6. **Path normalization** — dot-segment resolution (via ``rfc3986``),
       slash collapsing, trailing-slash removal, segment validation.
    7. **Query/fragment stripping** — irrelevant for WebSocket relay identity.

    Args:
        raw: Raw URL string, potentially malformed.

    Returns:
        Canonical URL string suitable for :class:`Relay` construction.

    Raises:
        ValueError: If the URL is structurally unrecoverable.
    """
    uri = uri_reference(_preprocess_idn(raw.strip())).normalize()

    validator = (
        Validator()
        .require_presence_of("scheme", "host")
        .allow_schemes("ws", "wss")
        .check_validity_of("scheme", "host", "port", "path")
    )

    try:
        validator.validate(uri)
    except UnpermittedComponentError:
        raise ValueError("Invalid scheme: must be ws or wss") from None
    except ValidationError as e:
        raise ValueError(f"Invalid URL: {e}") from None

    # --- Host normalization ---
    host = unquote(uri.host.strip("[]")) if uri.host else ""
    host = host.rstrip(".")
    if host != host.strip() or any(c in host for c in " \t\n\r\x00\\"):
        raise ValueError(f"Invalid host: '{host[:50]}'")

    # --- Network classification and rejection ---
    network = _detect_network(host)
    if network == NetworkType.LOCAL:
        raise ValueError("Local addresses not allowed")
    if network == NetworkType.UNKNOWN:
        raise ValueError(f"Invalid host: '{host}'")

    host = _normalize_ip(host)
    scheme = "wss" if network == NetworkType.CLEARNET else "ws"

    # --- Port validation ---
    port = int(uri.port) if uri.port else None
    if port is not None and not _PORT_MIN <= port <= _PORT_MAX:
        raise ValueError(f"Port out of range: {port}")

    # --- Path normalization ---
    path = _sanitize_path(uri.path)

    # --- URL reconstruction ---
    formatted_host = f"[{host}]" if ":" in host else host
    default_port = _PORT_WSS if scheme == "wss" else _PORT_WS
    if port and port != default_port:
        url = f"{scheme}://{formatted_host}:{port}{path}"
    else:
        url = f"{scheme}://{formatted_host}{path}"

    if len(url) > _MAX_URL_LENGTH:
        raise ValueError(
            f"URL exceeds maximum length ({len(url)} > {_MAX_URL_LENGTH}): '{url[:80]}...'"
        )

    return url

relay

relay ¶

Classes¶

RelayDbParams ¶

Relay dataclass ¶

Functions¶

to_db_params ¶

Functions¶

sanitize_relay_url ¶

Relay `dataclass` ¶