flayer.diagnostics.benchmark

Module source

flayer.diagnostics.benchmark

Bounded HTTP measurements against an explicitly authorized endpoint.

MAX_TIMEOUT_SECONDS module-attribute

MAX_TIMEOUT_SECONDS = 30.0

MAX_REQUEST_COUNT module-attribute

MAX_REQUEST_COUNT = 10

MAX_BYTE_LIMIT module-attribute

MAX_BYTE_LIMIT = 1024 * 1024

BenchmarkLimits dataclass

BenchmarkLimits(timeout_seconds: float = 3.0, count: int = 3, byte_limit: int = 64 * 1024)

Hard per-request deadline, request count, and response payload limits.

timeout_seconds class-attribute instance-attribute

timeout_seconds: float = 3.0

count class-attribute instance-attribute

count: int = 3

byte_limit class-attribute instance-attribute

byte_limit: int = 64 * 1024

EndpointMeasurement dataclass

EndpointMeasurement(status: DiagnosticStatus, duration_seconds: float, received_bytes: int = 0, http_status: int | None = None)

Safe metrics exclude response content, credentials, and target identifiers.

status instance-attribute

status: DiagnosticStatus

duration_seconds instance-attribute

duration_seconds: float

received_bytes class-attribute instance-attribute

received_bytes: int = 0

http_status class-attribute instance-attribute

http_status: int | None = None

EndpointTransport

Bases: Protocol

A measurement implementation enforces supplied limits without emitting raw data.

__call__

__call__(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement

Return one bounded HTTP measurement without exposing response payloads.

Source code in installed/flayer/diagnostics/benchmark.py

def __call__(
    self, endpoint: str, timeout_seconds: float, byte_limit: int
) -> EndpointMeasurement:
    """Return one bounded HTTP measurement without exposing response payloads."""

    ...

ValidateEndpoint

ValidateEndpoint(endpoint: str) -> SplitResult

Allow explicit HTTP(S) URLs without credentials, query strings, or fragments.

Source code in installed/flayer/diagnostics/benchmark.py

def ValidateEndpoint(endpoint: str) -> SplitResult:
    """Allow explicit HTTP(S) URLs without credentials, query strings, or fragments."""

    try:
        parsed = urlsplit(endpoint)
        port = parsed.port

        if (
            parsed.scheme not in ("http", "https")
            or not parsed.hostname
            or parsed.username is not None
            or parsed.password is not None
            or parsed.query
            or parsed.fragment
            or any(character.isspace() or ord(character) < 32 for character in endpoint)
            or (port is not None and port < 1)
        ):
            raise ValueError("Invalid endpoint")

    except ValueError:
        raise ValueError("Endpoint must be an HTTP(S) URL without credentials, query or fragment") \
            from None

    return parsed

HttpMeasurement

HttpMeasurement(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement

Read a limited payload with TLS verification and no redirects or proxy discovery.

Source code in installed/flayer/diagnostics/benchmark.py

def HttpMeasurement(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement:
    """Read a limited payload with TLS verification and no redirects or proxy discovery."""

    parsed = ValidateEndpoint(endpoint)
    connection_type = (
        http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection
    )
    connection = connection_type(str(parsed.hostname), port=parsed.port, timeout=timeout_seconds)
    started = time.monotonic()

    try:
        connection.request(
            "GET", parsed.path or "/",
            headers={"Range": f"bytes=0-{byte_limit - 1}", "Connection": "close"},
        )
        response = connection.getresponse()
        received_bytes = len(response.read(byte_limit))
        status = DiagnosticStatus.OK if 200 <= response.status < 300 else DiagnosticStatus.FAILED

        return EndpointMeasurement(status, time.monotonic() - started, received_bytes, response.status)

    finally:
        connection.close()

EndpointWorker

EndpointWorker(sender: Connection, endpoint: str, timeout_seconds: float, byte_limit: int) -> None

Isolate network work so DNS, TLS, connect, and response reads share a hard deadline.

Source code in installed/flayer/diagnostics/benchmark.py

def EndpointWorker(
    sender: Connection, endpoint: str, timeout_seconds: float, byte_limit: int
) -> None:
    """Isolate network work so DNS, TLS, connect, and response reads share a hard deadline."""

    try:
        measurement = HttpMeasurement(endpoint, timeout_seconds, byte_limit)

    except Exception:
        measurement = EndpointMeasurement(DiagnosticStatus.FAILED, 0.0)

    try:
        sender.send(measurement)

    finally:
        sender.close()

MeasureEndpoint

MeasureEndpoint(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement

Terminate isolated network work at the deadline, including a stalled DNS resolver.

Source code in installed/flayer/diagnostics/benchmark.py

def MeasureEndpoint(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement:
    """Terminate isolated network work at the deadline, including a stalled DNS resolver."""

    ValidateEndpoint(endpoint)
    BenchmarkLimits(timeout_seconds=timeout_seconds, count=1, byte_limit=byte_limit)
    context = multiprocessing.get_context("spawn")
    receiver, sender = context.Pipe(duplex=False)
    process = context.Process(
        target=EndpointWorker, args=(sender, endpoint, timeout_seconds, byte_limit), daemon=True
    )
    started = time.monotonic()
    launched = False

    try:
        process.start()
        launched = True
        sender.close()
        remaining = max(0.0, timeout_seconds - (time.monotonic() - started))

        if receiver.poll(remaining):
            measurement: EndpointMeasurement = receiver.recv()

            return measurement

        return EndpointMeasurement(DiagnosticStatus.FAILED, timeout_seconds)

    except (OSError, EOFError, RuntimeError):
        return EndpointMeasurement(DiagnosticStatus.FAILED, time.monotonic() - started)

    finally:
        sender.close()
        receiver.close()

        if launched:
            if process.is_alive():
                process.terminate()

            process.join(timeout=1.0)

            if process.is_alive():
                process.kill()
                process.join(timeout=1.0)

        process.close()

SafeMeasurement

SafeMeasurement(endpoint: str, limits: BenchmarkLimits, transport: EndpointTransport) -> EndpointMeasurement

Reject invalid adapter metrics and suppress all unstructured exception output.

Source code in installed/flayer/diagnostics/benchmark.py

def SafeMeasurement(
    endpoint: str, limits: BenchmarkLimits, transport: EndpointTransport
) -> EndpointMeasurement:
    """Reject invalid adapter metrics and suppress all unstructured exception output."""

    try:
        measurement = transport(endpoint, limits.timeout_seconds, limits.byte_limit)

        if (
            not isinstance(measurement, EndpointMeasurement)
            or not isinstance(measurement.status, DiagnosticStatus)
            or isinstance(measurement.duration_seconds, bool)
            or not math.isfinite(measurement.duration_seconds)
            or not 0 <= measurement.duration_seconds <= limits.timeout_seconds
            or isinstance(measurement.received_bytes, bool)
            or not isinstance(measurement.received_bytes, int)
            or not 0 <= measurement.received_bytes <= limits.byte_limit
            or (
                measurement.http_status is not None
                and (
                    isinstance(measurement.http_status, bool)
                    or not isinstance(measurement.http_status, int)
                    or not 100 <= measurement.http_status <= 599
                )
            )
            or (
                measurement.status is DiagnosticStatus.OK
                and (measurement.http_status is None or not 200 <= measurement.http_status < 300)
            )
        ):
            raise ValueError("Adapter returned invalid metrics")

        return measurement

    except Exception:
        return EndpointMeasurement(DiagnosticStatus.FAILED, 0.0)

ProbeEndpoint

ProbeEndpoint(endpoint: str, timeout_seconds: float = 3.0, transport: EndpointTransport = MeasureEndpoint) -> CheckResult

Check explicit HTTP reachability without pretending it validates guest hardening.

Source code in installed/flayer/diagnostics/benchmark.py

def ProbeEndpoint(
    endpoint: str, timeout_seconds: float = 3.0,
    transport: EndpointTransport = MeasureEndpoint,
) -> CheckResult:
    """Check explicit HTTP reachability without pretending it validates guest hardening."""

    ValidateEndpoint(endpoint)
    limits = BenchmarkLimits(timeout_seconds=timeout_seconds, count=1, byte_limit=1)
    measurement = SafeMeasurement(endpoint, limits, transport)

    return CheckResult(
        "endpoint", measurement.status,
        "Endpoint answered successfully" if measurement.status is DiagnosticStatus.OK
        else "Endpoint probe did not succeed",
        {"duration_seconds": measurement.duration_seconds, "http_status": measurement.http_status},
    )

RunBenchmark

RunBenchmark(endpoint: str, limits: BenchmarkLimits | None = None, transport: EndpointTransport = MeasureEndpoint) -> tuple[CheckResult, ...]

Measure a fixed count of bounded downloads without inferring upload or tunnel speed.

Source code in installed/flayer/diagnostics/benchmark.py

def RunBenchmark(
    endpoint: str, limits: BenchmarkLimits | None = None,
    transport: EndpointTransport = MeasureEndpoint,
) -> tuple[CheckResult, ...]:
    """Measure a fixed count of bounded downloads without inferring upload or tunnel speed."""

    ValidateEndpoint(endpoint)
    resolved_limits = limits or BenchmarkLimits()
    checks: list[CheckResult] = []

    for index in range(resolved_limits.count):
        measurement = SafeMeasurement(endpoint, resolved_limits, transport)
        status = measurement.status

        if status is DiagnosticStatus.OK and measurement.received_bytes == 0:
            status = DiagnosticStatus.FAILED

        throughput = (
            measurement.received_bytes * 8 / measurement.duration_seconds / 1_000_000
            if measurement.duration_seconds > 0 else None
        )
        checks.append(CheckResult(
            f"http-download-{index + 1}", status,
            "Bounded HTTP download measured" if status is DiagnosticStatus.OK
            else "Bounded HTTP download did not succeed",
            {
                "received_bytes": measurement.received_bytes,
                "duration_seconds": measurement.duration_seconds,
                "throughput_mbps": throughput,
                "http_status": measurement.http_status,
                "byte_limit": resolved_limits.byte_limit,
            },
        ))

    return tuple(checks)