Skip to content

Fetch

Fetching a URL an agent names, without letting it reach the host's own network.

judge_citation reads any http or https URL the agent gives it. Fetched naively, such a URL could reach the machine the tools run on or its network: a local model server, a router, a cloud metadata service. fetch therefore:

  • accepts http and https only, with no user name or password in the URL;
  • resolves the host and refuses it when any address it resolves to is not a public unicast address: loopback, private, shared, link-local (the cloud metadata address included), unique-local, multicast, unspecified, reserved and documentation ranges, in IPv4 and IPv6, and an IPv6 address that carries an IPv4 one (mapped, NAT64, 6to4, Teredo) is judged by the IPv4 address it carries;
  • connects to the address it checked, never to the name again, keeping the name for the Host header and for TLS (the server name sent and the certificate verified against it), and checks the address the connection reached before sending the request, so a name that resolves to a public address when checked and a private one when connected (DNS rebinding) cannot reach the private one;
  • follows redirects itself, applying all of the above to every hop, up to MAX_HOPS;
  • gives up when a connection or a read waits TIMEOUT seconds, and when the whole fetch, name resolution, redirects and body included, takes MAX_SECONDS, however slowly a server drips its bytes; refuses a body larger than MAX_BYTES, and one that ends before the length its headers declare.

A local file is never named by a URL: file: and every other scheme are refused.

Connection

Bases: Protocol

An HTTP connection opened to one address.

Source code in src/credencegraph_extract/fetch.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Connection(Protocol):
    """An HTTP connection opened to one address."""

    def peer(self) -> str:
        """Return the address the connection reached."""

    def get(self, target: str, headers: Mapping[str, str]) -> tuple[int, Mapping[str, str], Callable[[int], bytes]]:
        """Send a GET request; return the status, the headers by lower-case name, and a reader of the body.

        The reader returns at most the bytes asked for, as soon as some have arrived, and ``b""`` at the end.
        """

    def close(self) -> None:
        """Close the connection."""

close()

Close the connection.

Source code in src/credencegraph_extract/fetch.py
106
107
def close(self) -> None:
    """Close the connection."""

get(target, headers)

Send a GET request; return the status, the headers by lower-case name, and a reader of the body.

The reader returns at most the bytes asked for, as soon as some have arrived, and b"" at the end.

Source code in src/credencegraph_extract/fetch.py
100
101
102
103
104
def get(self, target: str, headers: Mapping[str, str]) -> tuple[int, Mapping[str, str], Callable[[int], bytes]]:
    """Send a GET request; return the status, the headers by lower-case name, and a reader of the body.

    The reader returns at most the bytes asked for, as soon as some have arrived, and ``b""`` at the end.
    """

peer()

Return the address the connection reached.

Source code in src/credencegraph_extract/fetch.py
97
98
def peer(self) -> str:
    """Return the address the connection reached."""

FetchError

Bases: Exception

A URL is refused or cannot be fetched.

Attributes:

Name Type Description
code

refused-url for a URL the policy refuses, fetch-failed for one that could not be read.

Source code in src/credencegraph_extract/fetch.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class FetchError(Exception):
    """A URL is refused or cannot be fetched.

    Attributes:
        code: ``refused-url`` for a URL the policy refuses, ``fetch-failed`` for one that could not be read.
    """

    def __init__(self, code: str, message: str) -> None:
        """Make the error.

        Args:
            code: ``refused-url`` or ``fetch-failed``.
            message: What was refused or failed, in words.
        """
        super().__init__(message)
        self.code = code

__init__(code, message)

Make the error.

Parameters:

Name Type Description Default
code str

refused-url or fetch-failed.

required
message str

What was refused or failed, in words.

required
Source code in src/credencegraph_extract/fetch.py
64
65
66
67
68
69
70
71
72
def __init__(self, code: str, message: str) -> None:
    """Make the error.

    Args:
        code: ``refused-url`` or ``fetch-failed``.
        message: What was refused or failed, in words.
    """
    super().__init__(message)
    self.code = code

Fetched dataclass

What a fetch read.

Attributes:

Name Type Description
url str

The URL asked for.

final_url str

The URL the body came from, after redirects.

hops tuple[str, ...]

Every URL requested, in order, the first and the final included.

content_type str

The media type of the body, without parameters, in lower case.

data bytes

The body.

Source code in src/credencegraph_extract/fetch.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@dataclass(frozen=True)
class Fetched:
    """What a fetch read.

    Attributes:
        url: The URL asked for.
        final_url: The URL the body came from, after redirects.
        hops: Every URL requested, in order, the first and the final included.
        content_type: The media type of the body, without parameters, in lower case.
        data: The body.
    """

    url: str
    final_url: str
    hops: tuple[str, ...]
    content_type: str
    data: bytes

describe(fetched)

What a run records of a fetch: the URLs and the content type, never the body.

Parameters:

Name Type Description Default
fetched Fetched

What a fetch read.

required

Returns:

Type Description
dict[str, Any]

url, final_url, hops and content_type.

Source code in src/credencegraph_extract/fetch.py
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def describe(fetched: Fetched) -> dict[str, Any]:
    """What a run records of a fetch: the URLs and the content type, never the body.

    Args:
        fetched: What a fetch read.

    Returns:
        ``url``, ``final_url``, ``hops`` and ``content_type``.
    """
    return {
        "url": fetched.url,
        "final_url": fetched.final_url,
        "hops": list(fetched.hops),
        "content_type": fetched.content_type,
    }

fetch(url, *, resolve=_resolve, connect=_connect, max_hops=MAX_HOPS, timeout=TIMEOUT, max_seconds=MAX_SECONDS, max_bytes=MAX_BYTES)

Fetch a URL under the policy of this module, following redirects.

Parameters:

Name Type Description Default
url str

An http or https URL.

required
resolve Resolve

Resolves a host name; the system resolver by default.

_resolve
connect Connect

Opens a connection to an address; http.client by default.

_connect
max_hops int

The most requests made, the first included.

MAX_HOPS
timeout float

Seconds a connection or a read may wait.

TIMEOUT
max_seconds float

Seconds the whole fetch may take, redirects and bodies included.

MAX_SECONDS
max_bytes int

The largest body read.

MAX_BYTES

Returns:

Type Description
Fetched

What was read.

Raises:

Type Description
FetchError

refused-url when the URL, a redirect or an address is refused; fetch-failed when the body cannot be read, is too large or too slow, or the final response is not 200.

Source code in src/credencegraph_extract/fetch.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def fetch(  # noqa: PLR0913 - the URL, the policy's limits, and the two seams for tests
    url: str,
    *,
    resolve: Resolve = _resolve,
    connect: Connect = _connect,
    max_hops: int = MAX_HOPS,
    timeout: float = TIMEOUT,
    max_seconds: float = MAX_SECONDS,
    max_bytes: int = MAX_BYTES,
) -> Fetched:
    """Fetch a URL under the policy of this module, following redirects.

    Args:
        url: An ``http`` or ``https`` URL.
        resolve: Resolves a host name; the system resolver by default.
        connect: Opens a connection to an address; http.client by default.
        max_hops: The most requests made, the first included.
        timeout: Seconds a connection or a read may wait.
        max_seconds: Seconds the whole fetch may take, redirects and bodies included.
        max_bytes: The largest body read.

    Returns:
        What was read.

    Raises:
        FetchError: ``refused-url`` when the URL, a redirect or an address is refused; ``fetch-failed``
            when the body cannot be read, is too large or too slow, or the final response is not 200.
    """
    deadline = time.monotonic() + max_seconds
    hops: list[str] = []
    current = url
    while True:
        hops.append(current)
        status, headers, data = _hop(current, resolve, connect, timeout=timeout, max_bytes=max_bytes, deadline=deadline)
        if status in REDIRECTS:
            location = headers.get("location")
            if not location:
                raise FetchError("fetch-failed", f"{current!r} redirects without a location")
            if len(hops) >= max_hops:
                raise FetchError("fetch-failed", f"{url!r} redirects more than {max_hops - 1} times")
            current = urljoin(current, location)
            continue
        if status != http.client.OK:
            raise FetchError("fetch-failed", f"{current!r} answered {status}")
        content_type = headers.get("content-type", "").split(";", 1)[0].strip().lower()
        return Fetched(url, current, tuple(hops), content_type, data)

forbidden(address)

Say why an address may not be fetched from, or None for a public unicast address.

Parameters:

Name Type Description Default
address str

An IPv4 or IPv6 address.

required

Returns:

Type Description
str | None

The reason, or None.

Source code in src/credencegraph_extract/fetch.py
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def forbidden(address: str) -> str | None:
    """Say why an address may not be fetched from, or ``None`` for a public unicast address.

    Args:
        address: An IPv4 or IPv6 address.

    Returns:
        The reason, or ``None``.
    """
    try:
        ip = _ip(address)
    except ValueError:
        return f"{address!r} is not an IP address"
    if isinstance(ip, ipaddress.IPv6Address):
        carried = ip.ipv4_mapped or ip.sixtofour or (ip.teredo[1] if ip.teredo else None)
        if carried is None and ip in _NAT64:
            carried = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
        if carried is not None:
            reason = forbidden(str(carried))
            return f"{ip} carries {carried}, which {reason.split(' ', 1)[1]}" if reason else None
    if ip.is_multicast:
        return f"{ip} is a multicast address"
    if not ip.is_global:
        return f"{ip} is not a public address (loopback, private, link-local or reserved)"
    return None