Proxies and proxy maps

proxylib.proxy

Core types: Proxy (one proxy), the ProxyMap protocol (request URL to the proxies to try) and lookup(), which reads any proxy map's answer.

A proxy map is anything indexable by request URL -- a plain dict included. proxymap[url] answers None (DIRECT), one proxy, or an iterable of them to try in order, where a None entry means DIRECT; a proxy is a Proxy, a URL string, a PAC result string, or any object with a url attribute. KeyError means the map has no opinion. :func:lookup turns every one of those answers into a tuple of Proxy | None.

ProxyAnswer = Union[None, ProxyLike, Iterable[Optional[ProxyLike]]] module-attribute

What proxymap[url] may return.

ProxyLike = Union[Proxy, str, Any] module-attribute

A Proxy, a proxy URL or PAC result string, or any object with a url attribute (credentials from the URL, else its username/password).

__all__ = ['Proxy', 'ProxyLike', 'ProxyAnswer', 'ProxyMap', 'lookup', 'alookup', 'SimpleProxyMap', 'ChainProxyMap'] module-attribute

ChainProxyMap(*maps)

Bases: ProxyMap

Asks each map in order; the first with an opinion answers.

A map's KeyError passes to the next one; any answer, DIRECT included, ends the chain. If every map has no opinion, neither does the chain, so a chain can be a link in a larger chain. Links can be any proxy map, dicts included.

Source code in src/proxylib/proxy.py

def __init__(self, *maps: ProxyMap) -> None:
    self.maps = maps

__slots__ = ('maps',) class-attribute instance-attribute

maps = maps instance-attribute

__getitem__(url)

Source code in src/proxylib/proxy.py

def __getitem__(self, url: str) -> "Tuple[Optional[Proxy], ...]":
    for proxymap in self.maps:
        try:
            return lookup(proxymap, url)
        except KeyError:
            continue
    raise KeyError(url)

__repr__()

Source code in src/proxylib/proxy.py

def __repr__(self) -> str:
    return f"ChainProxyMap({', '.join(map(repr, self.maps))})"

close()

Close every link that has close().

Source code in src/proxylib/proxy.py

def close(self) -> None:
    """Close every link that has ``close()``."""
    for proxymap in self.maps:
        _close(proxymap)

Proxy(scheme, host, port=None, username=None, password=None, *, path=None, headers=None)

One proxy: scheme://[username[:password]@]host[:port]. Immutable.

url is the address without credentials (what logs and pathapi-style consumers read); uri() adds the credentials, percent-encoded. The password never appears in repr() or str().

headers are sent to this proxy only (a token in Proxy-Authorization, or a header of another name): in the CONNECT of a tunnel and on a plain-http request to the proxy, never to the server. repr() shows their names, not their values. A SOCKS proxy takes none.

The schemes prefix+http and prefix+https name a URL-prefix gateway: the request is sent directly to <scheme>://host[:port]<path><target URL> (path defaults to /). Its credentials go as a Basic Proxy-Authorization header of that request, with headers.

Source code in src/proxylib/proxy.py

def __init__(
    self,
    scheme: str,
    host: str,
    port: Optional[int] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    *,
    path: Optional[str] = None,
    headers: "Union[Mapping[str, str], Iterable[Tuple[str, str]], None]" = None,
) -> None:
    scheme = (scheme or "").lower()
    scheme = _SCHEME_ALIASES.get(scheme, scheme)
    if not scheme:
        raise ProxyParseError("a proxy needs a scheme")
    if not host:
        raise ProxyParseError(f"a {scheme} proxy needs a host")
    if port is not None and not (isinstance(port, int) and 0 < port < 65536):
        raise ProxyParseError(f"invalid proxy port {port!r}")
    if scheme in _GATEWAY_SCHEMES:
        path = path or "/"
        if not path.startswith("/") or "?" in path or "#" in path:
            raise ProxyParseError(f"a gateway path is an absolute path with no query: {path!r}")
    elif scheme.startswith(_GATEWAY_PREFIX):
        raise ProxyParseError(f"a gateway is prefix+http or prefix+https, not {scheme}")
    elif path not in (None, "", "/"):
        raise ProxyParseError(f"only a prefix+http(s) gateway has a path, not a {scheme} proxy")
    else:
        path = None
    pairs = tuple((headers.items() if isinstance(headers, Mapping) else headers) or ())
    for name, value in pairs:
        if not isinstance(name, str) or not _TOKEN.match(name):
            raise ProxyParseError(f"invalid proxy header name {name!r}")
        if not isinstance(value, str) or any(c in value for c in "\r\n\0"):
            raise ProxyParseError(f"invalid value for proxy header {name!r}")
    if pairs and scheme.startswith("socks"):
        raise ProxyParseError("a SOCKS proxy takes no headers")
    for name, value in (
        ("scheme", scheme),
        ("host", host.lower()),
        ("port", port),
        ("username", username or None),
        ("password", password or None),
        ("path", path),
        ("_headers", pairs),
    ):
        object.__setattr__(self, name, value)

__slots__ = ('scheme', 'host', 'port', 'username', 'password', 'path', '_headers') class-attribute instance-attribute

headers property

The headers sent to this proxy only, as a new dict.

host instance-attribute

is_gateway property

A URL-prefix gateway (prefix+http/prefix+https), not a proxy.

password instance-attribute

path instance-attribute

port instance-attribute

scheme instance-attribute

url property

scheme://host[:port] (plus a gateway's path), no credentials; an IPv6 host is bracketed.

username instance-attribute

__delattr__(name)

Source code in src/proxylib/proxy.py

def __delattr__(self, name: str) -> None:
    raise AttributeError(f"{type(self).__name__} is immutable")

__eq__(other)

Source code in src/proxylib/proxy.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, Proxy):
        return NotImplemented
    return self._key() == other._key()

__hash__()

Source code in src/proxylib/proxy.py

def __hash__(self) -> int:
    return hash(self._key())

__reduce__()

Source code in src/proxylib/proxy.py

def __reduce__(self) -> tuple:
    return (_rebuild, (self.scheme, self.host, self.port, self.username, self.password, self.path, self._headers))

__repr__()

Source code in src/proxylib/proxy.py

def __repr__(self) -> str:
    parts = [f"scheme={self.scheme!r}", f"host={self.host!r}"]
    if self.port is not None:
        parts.append(f"port={self.port!r}")
    if self.username is not None:
        parts.append(f"username={self.username!r}")
    if self.password is not None:
        parts.append("password='***'")
    if self.path is not None:
        parts.append(f"path={self.path!r}")
    if self._headers:
        parts.append("headers={%s}" % ", ".join(f"{name!r}: '***'" for name, _ in self._headers))
    return f"Proxy({', '.join(parts)})"

__setattr__(name, value)

Source code in src/proxylib/proxy.py

def __setattr__(self, name: str, value: Any) -> None:
    raise AttributeError(f"{type(self).__name__} is immutable")

__str__()

Source code in src/proxylib/proxy.py

def __str__(self) -> str:
    return self.url

gateway_url(target)

The URL a gateway is asked for to fetch target: <scheme>://host[:port]<path><target>, the target appended as given.

Source code in src/proxylib/proxy.py

def gateway_url(self, target: str) -> str:
    """The URL a gateway is asked for to fetch ``target``:
    ``<scheme>://host[:port]<path><target>``, the target appended as given."""
    if not self.is_gateway:
        raise ValueError(f"{self.url} is a proxy, not a URL-prefix gateway")
    scheme = self.scheme[len(_GATEWAY_PREFIX):]
    return f"{scheme}://{_net.join_host(self.host, self.port)}{self.path}{target}"

parse(text) classmethod

Parse a proxy URL (http://user:pass@host:3128). A missing scheme means http; an IPv6 host is written in brackets. A path other than / is refused: that is a PAC location, not a proxy -- except on a gateway (prefix+https://gw.example/fetch/).

Source code in src/proxylib/proxy.py

@classmethod
def parse(cls, text: str) -> "Proxy":
    """Parse a proxy URL (``http://user:pass@host:3128``). A missing scheme
    means ``http``; an IPv6 host is written in brackets. A path other than
    ``/`` is refused: that is a PAC location, not a proxy -- except on a
    gateway (``prefix+https://gw.example/fetch/``)."""
    text = text.strip()
    if "://" not in text:
        text = "http://" + text
    try:
        parts = urlsplit(text)
        port = parts.port
    except ValueError as exc:
        raise ProxyParseError(f"invalid proxy URL {text!r}: {exc}") from None
    gateway = parts.scheme.lower().startswith(_GATEWAY_PREFIX)
    if parts.query or parts.fragment or (parts.path not in ("", "/") and not gateway):
        raise ProxyParseError(f"proxy URL {text!r} has a path; a PAC location is not a proxy")
    if not parts.hostname:
        raise ProxyParseError(f"proxy URL {text!r} has no host")
    return cls(
        parts.scheme,
        parts.hostname,
        port,
        unquote(parts.username) if parts.username else None,
        unquote(parts.password) if parts.password else None,
        path=parts.path if gateway else None,
    )

parse_pac_result(text) classmethod

Parse a PAC FindProxyForURL result ("PROXY a:1; SOCKS5 b:2; DIRECT").

Keywords are case-insensitive; DIRECT is None. PREFIX <URL> (proxylib's own) names a URL-prefix gateway, PREFIX https://gw.example/fetch/. An entry with an unknown keyword or no usable host is dropped with a warning, and a result with nothing usable left is (None,), as browsers treat it.

Source code in src/proxylib/proxy.py

@classmethod
def parse_pac_result(cls, text: str) -> "Tuple[Optional[Proxy], ...]":
    """Parse a PAC ``FindProxyForURL`` result (``"PROXY a:1; SOCKS5 b:2; DIRECT"``).

    Keywords are case-insensitive; ``DIRECT`` is ``None``. ``PREFIX <URL>``
    (proxylib's own) names a URL-prefix gateway, ``PREFIX
    https://gw.example/fetch/``. An entry with an unknown keyword or no
    usable host is dropped with a warning, and a result with nothing usable
    left is ``(None,)``, as browsers treat it.
    """
    entries: "list[Optional[Proxy]]" = []
    dropped: "list[str]" = []
    for raw in text.split(";"):
        entry = raw.strip()
        if not entry:
            continue
        keyword, *rest_parts = entry.split(None, 1)
        keyword = keyword.upper()
        rest = rest_parts[0] if rest_parts else ""
        if keyword == "DIRECT":
            entries.append(None)
            continue
        if keyword == "PREFIX":
            target = rest.strip()
            try:
                if not target.lower().startswith(("http://", "https://")):
                    raise ProxyParseError(target)
                entries.append(cls.parse(_GATEWAY_PREFIX + target))
            except ProxyParseError:
                dropped.append(entry)
            continue
        scheme = _PAC_KEYWORDS.get(keyword)
        host_port = _net.split_host_port(rest.strip()) if scheme and rest.strip() else None
        if scheme is None or host_port is None or not host_port[0]:
            dropped.append(entry)
            continue
        try:
            entries.append(cls(scheme, host_port[0], host_port[1]))
        except ProxyParseError:
            dropped.append(entry)
    if dropped:
        warnings.warn(f"ignoring unusable PAC result entries: {dropped!r}", stacklevel=2)
    return tuple(entries) or (None,)

uri()

url with the credentials, percent-encoded, for a client to connect with.

Source code in src/proxylib/proxy.py

def uri(self) -> str:
    """``url`` with the credentials, percent-encoded, for a client to connect with."""
    if self.username is None:
        return self.url
    userinfo = quote(self.username, safe="")
    if self.password is not None:
        userinfo += ":" + quote(self.password, safe="")
    return f"{self.scheme}://{userinfo}@{_net.join_host(self.host, self.port)}{self.path or ''}"

ProxyMap

Bases: Protocol

Anything indexable by request URL: proxymap[url] returns a :data:ProxyAnswer or raises KeyError for "no opinion".

Structural: a dict and any class with __getitem__ qualify without importing proxylib. Read an answer with :func:lookup.

Calling it builds a map from a configured value, with no I/O: ProxyMap(None), a proxy URL, a PAC result string, a Proxy or a list/tuple of them gives a :class:SimpleProxyMap; a PAC location (file: URL, inline script, or an http(s) URL with a path) gives a map that loads the PAC on its first lookup; anything else with __getitem__ is returned as is.

__getitem__(url)

Source code in src/proxylib/proxy.py

def __getitem__(self, url: str, /) -> ProxyAnswer: ...

SimpleProxyMap(proxy=None)

Bases: ProxyMap

A proxy map with one fixed answer for every URL.

Takes anything :func:lookup reads: None (DIRECT), a Proxy, a proxy URL, a PAC result string ("PROXY a:1; DIRECT") or an iterable of those. An empty iterable is DIRECT.

Source code in src/proxylib/proxy.py

def __init__(self, proxy: ProxyAnswer = None) -> None:
    self.proxies: "Tuple[Optional[Proxy], ...]" = _normalise(proxy)

__slots__ = ('proxies',) class-attribute instance-attribute

proxies = _normalise(proxy) instance-attribute

__getitem__(url)

Source code in src/proxylib/proxy.py

def __getitem__(self, url: str) -> "Tuple[Optional[Proxy], ...]":
    return self.proxies

__repr__()

Source code in src/proxylib/proxy.py

def __repr__(self) -> str:
    return f"SimpleProxyMap({self.proxies!r})"

close()

Nothing to free; every built-in map has close().

Source code in src/proxylib/proxy.py

def close(self) -> None:
    """Nothing to free; every built-in map has ``close()``."""

alookup(proxymap, url) async

:func:lookup for asyncio code, run in a worker thread: a PAC script, a download or an OS resolver may block. Same answer and errors.

Source code in src/proxylib/proxy.py

async def alookup(proxymap: ProxyMap, url: str) -> "Tuple[Optional[Proxy], ...]":
    """:func:`lookup` for asyncio code, run in a worker thread: a PAC script,
    a download or an OS resolver may block. Same answer and errors."""
    import asyncio

    return await asyncio.to_thread(lookup, proxymap, url)

lookup(proxymap, url)

The proxies proxymap says to try for url, in order; None means DIRECT. Never empty.

A plain mapping that has no entry for the full URL is also asked with the keys requests uses (scheme://host, scheme, all://host, all). Raises KeyError when the map has no opinion, and TypeError for an answer that names no proxy.

Source code in src/proxylib/proxy.py

def lookup(proxymap: ProxyMap, url: str) -> "Tuple[Optional[Proxy], ...]":
    """The proxies ``proxymap`` says to try for ``url``, in order; ``None``
    means DIRECT. Never empty.

    A plain mapping that has no entry for the full URL is also asked with the
    keys ``requests`` uses (``scheme://host``, ``scheme``, ``all://host``,
    ``all``). Raises ``KeyError`` when the map has no opinion, and
    ``TypeError`` for an answer that names no proxy.
    """
    try:
        answer = proxymap[url]
    except KeyError:
        if not isinstance(proxymap, Mapping):
            raise KeyError(url) from None
        for key in _requests_keys(url):
            if key in proxymap:
                answer = proxymap[key]
                break
        else:
            raise KeyError(url) from None
    return _normalise(answer)