whitehat-security/src/whitehat_security/capacity.py

from __future__ import annotations

from dataclasses import asdict, dataclass

from .model import Outcome, utc_now


@dataclass(frozen=True)
class CapacitySample:
    consumer: str
    latency_ms: float
    error_rate: float
    throughput_per_second: float


@dataclass(frozen=True)
class CapacityResult:
    outcome: Outcome
    governor_bound: bool
    aggressor_peak: float
    aggressor_ceiling: float
    neighbour_degradation: dict[str, dict[str, float]]
    reasons: list[str]


def characterize(*, baseline: list[CapacitySample], loaded: list[CapacitySample],
                 governor_bound: bool, aggressor_peak: float,
                 aggressor_ceiling: float) -> CapacityResult:
    before = {sample.consumer: sample for sample in baseline}
    after = {sample.consumer: sample for sample in loaded}
    reasons: list[str] = []
    if not governor_bound:
        reasons.append("declared governor did not bind")
    if aggressor_peak > aggressor_ceiling:
        reasons.append("probe exceeded its authorized allowance")
    missing = sorted(before.keys() - after.keys())
    if missing:
        reasons.append("loaded samples missing consumers: " + ", ".join(missing))

    degradation: dict[str, dict[str, float]] = {}
    for consumer in sorted(before.keys() & after.keys()):
        old, new = before[consumer], after[consumer]
        degradation[consumer] = {
            "latency_increase_percent": _change(old.latency_ms, new.latency_ms),
            "error_rate_increase_points": new.error_rate - old.error_rate,
            "throughput_decrease_percent": _decrease(old.throughput_per_second,
                                                       new.throughput_per_second),
        }
    outcome: Outcome = "pass"
    if aggressor_peak > aggressor_ceiling:
        outcome = "aborted"
    elif reasons:
        outcome = "finding"
    return CapacityResult(
        outcome=outcome,
        governor_bound=governor_bound,
        aggressor_peak=aggressor_peak,
        aggressor_ceiling=aggressor_ceiling,
        neighbour_degradation=degradation,
        reasons=reasons,
    )


def _change(before: float, after: float) -> float:
    if before == 0:
        return 0.0 if after == 0 else float("inf")
    return round((after - before) / before * 100, 3)


def _decrease(before: float, after: float) -> float:
    if before == 0:
        return 0.0
    return round((before - after) / before * 100, 3)


def _case(name: str, result: CapacityResult) -> dict:
    payload = asdict(result)
    payload["case"] = name
    return payload


def capacity_calibration() -> dict:
    """In-process known-good/known-bad outcomes. Generates no load."""
    started = utc_now()
    baseline = [
        CapacitySample("aggressor", 10, 0, 100),
        CapacitySample("neighbour", 12, 0, 80),
    ]
    loaded = [
        CapacitySample("aggressor", 25, 0.01, 120),
        CapacitySample("neighbour", 18, 0.02, 60),
    ]
    good = characterize(
        baseline=baseline, loaded=loaded, governor_bound=True,
        aggressor_peak=10, aggressor_ceiling=10,
    )
    unbound = characterize(
        baseline=baseline, loaded=loaded, governor_bound=False,
        aggressor_peak=10, aggressor_ceiling=10,
    )
    exceeded = characterize(
        baseline=baseline, loaded=loaded, governor_bound=True,
        aggressor_peak=11, aggressor_ceiling=10,
    )
    missing = characterize(
        baseline=baseline, loaded=[loaded[0]], governor_bound=True,
        aggressor_peak=10, aggressor_ceiling=10,
    )
    ok = (
        good.outcome == "pass"
        and unbound.outcome == "finding"
        and exceeded.outcome == "aborted"
        and missing.outcome == "finding"
    )
    return {
        "schema_version": "whitehat-capacity-calibration/v1",
        "evidence_class": "fixture",
        "run_id": f"capacity-calibration-{started}",
        "started_at": started,
        "ended_at": utc_now(),
        "outcome": "pass" if ok else "finding",
        "known_good": [_case("governor_bound_within_ceiling", good)],
        "known_bad": [
            _case("unbound_governor", unbound),
            _case("exceeded_ceiling", exceeded),
            _case("missing_neighbour", missing),
        ],
        "limitations": [
            "Offline capacity calibration evaluates the harness; it is not target assurance.",
            "No network load, database connection, or live credential was used.",
            "Zero neighbour degradation is not an expected assertion on shared infrastructure.",
        ],
    }