English fallback

This translation is missing; the canonical English reference follows.

Internal bounded original-content response worker

The installed worker sends one exact anonymous GET to an explicitly selected canonical literal-loopback URL. One five-second monotonic deadline covers connection, send, reads and actual EOF. It admits complete bounded Content-Length responses with UTF-8 HTML or plain-text bodies before emitting original byte descriptors. Closed refusals contain only a fixed safe category.

This is an Internal worker protocol, not a supported extension API. Direct invocation grants no acquisition or retention permission. The governed workflow owns authorization, cancellation, original evidence custody and persistence; the worker performs no security assessment. See docs/OWNED_CONTENT_RESPONSE_CAPTURE_CONTRACT.md.

fuzzy1337.owned_content_response_worker

One exact anonymous loopback GET, complete content admission and no rejected output.

Remaining(deadline: float) -> float

Consume the original connect/send/read/EOF clock without renewal on progress.

Parameters:

Name	Type	Description	Default
deadline	float	One absolute five-second monotonic boundary.	required

Returns:

Type	Description
float	Positive remaining allowance; expiry raises TimeoutError without response values.

Source code in fuzzy1337/owned_content_response_worker.py

def Remaining(deadline: float) -> float:
    """Consume the original connect/send/read/EOF clock without renewal on progress.

    Args:
        deadline: One absolute five-second monotonic boundary.

    Returns:
        Positive remaining allowance; expiry raises TimeoutError without response values.
    """

    remaining = deadline - time.monotonic()

    if remaining <= 0:
        raise TimeoutError

    return remaining

Probe(url: str) -> dict[str, object]

Admit one actual complete original response before producing any byte descriptors.

Parameters:

Name	Type	Description	Default
url	str	Exact canonical anonymous loopback URL, without a path/body catalog.	required

Returns:

Type	Description
dict[str, object]	Captured record after complete framing, actual timely EOF and UTF-8 admission.

Source code in fuzzy1337/owned_content_response_worker.py

def Probe(url: str) -> dict[str, object]:
    """Admit one actual complete original response before producing any byte descriptors.

    Args:
        url: Exact canonical anonymous loopback URL, without a path/body catalog.

    Returns:
        Captured record after complete framing, actual timely EOF and UTF-8 admission.
    """

    address, port, _, _ = Destination(url)
    request = RequestBytes(url)
    deadline = time.monotonic() + 5.0
    family = socket.AF_INET6 if address == "::1" else socket.AF_INET

    with socket.socket(family, socket.SOCK_STREAM) as connection:
        connection.settimeout(Remaining(deadline))
        connection.connect((address, port))
        Remaining(deadline)
        peer = connection.getpeername()

        if peer[0] != address or peer[1] != port:
            raise ContentResponsePolicyError("protocol")

        connection.settimeout(Remaining(deadline))
        connection.sendall(request)
        Remaining(deadline)
        buffer = bytearray()
        boundary = -1

        while boundary < 0:
            if len(buffer) >= MAX_HEADER_BYTES:
                raise ContentResponsePolicyError("response-policy")

            connection.settimeout(Remaining(deadline))
            chunk = connection.recv(min(4096, MAX_HEADER_BYTES - len(buffer)))
            Remaining(deadline)

            if not chunk:
                raise ContentResponsePolicyError("protocol")

            buffer.extend(chunk)
            boundary = buffer.find(b"\r\n\r\n")

        headers = bytes(buffer[:boundary + 4])
        length, media = AdmitHeaders(headers)
        body = bytearray(buffer[boundary + 4:])

        if len(body) > length:
            raise ContentResponsePolicyError("protocol")

        while len(body) < length:
            connection.settimeout(Remaining(deadline))
            chunk = connection.recv(min(4096, length - len(body)))
            Remaining(deadline)

            if not chunk:
                raise ContentResponsePolicyError("protocol")

            body.extend(chunk)

        connection.settimeout(Remaining(deadline))
        extra = connection.recv(1)
        Remaining(deadline)

        if extra:
            raise ContentResponsePolicyError("protocol")

    original_body = bytes(body)
    AdmitBody(original_body, length)
    response = headers + original_body

    return {"schema_version": 1, "profile": OWNED_CONTENT_RESPONSE_PROFILE, "outcome": "captured",
        "url": url, "address": address, "port": port, "media_type": media,
        "request": BytePart(request), "response_headers": BytePart(headers),
        "response_body": BytePart(original_body),
        "response": {"size_bytes": len(response), "sha256": hashlib.sha256(response).hexdigest()}}

Main(argv: Sequence[str] | None = None) -> int

Emit only an admitted canonical record or safe refusal, without stderr diagnostics.

Parameters:

Name	Type	Description	Default
argv	Sequence[str] | None	Exactly one canonical URL; no flags or timeout configuration.	None

Returns:

Type	Description
int	Zero for captured original bytes, two for a fixed four-field refusal.

Source code in fuzzy1337/owned_content_response_worker.py

def Main(argv: Sequence[str] | None = None) -> int:
    """Emit only an admitted canonical record or safe refusal, without stderr diagnostics.

    Args:
        argv: Exactly one canonical URL; no flags or timeout configuration.

    Returns:
        Zero for captured original bytes, two for a fixed four-field refusal.
    """

    arguments = sys.argv[1:] if argv is None else argv
    error: str | None = None

    try:
        if len(arguments) != 1:
            raise ValueError

        record = Probe(arguments[0])

    except ContentResponsePolicyError as refusal:
        error = refusal.error

    except TimeoutError:
        error = "timeout"

    except OSError:
        error = "connection"

    except (ValueError, TypeError):
        error = "input"

    if error is not None:
        record = {"schema_version": 1, "profile": OWNED_CONTENT_RESPONSE_PROFILE,
                  "outcome": "refused", "error": error}

    output = SerializeContract(record).encode("ascii")

    if len(output) > MAX_RECORD_BYTES:
        raise ValueError("content response record exceeds its fixed bound")

    sys.stdout.buffer.write(output)
    sys.stdout.buffer.flush()

    return 0 if error is None else 2