tegwick
7e83a66573
Register in-process E3 and capacity fixtures, keep live database and substrate targets pending, and ask ops-mason for namespace-only provision. No packets, no credentials, no cancelled engagement IDs. Assistant: grok Assistant-Session: 01a02670-3345-76f2-a014-70fde8e2a2bb
107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Callable, Literal
|
|
|
|
from .model import Outcome, utc_now
|
|
|
|
Expectation = Literal["zero_rows", "statement_rejected", "false", "documented_limit"]
|
|
Query = Callable[[str], object]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class E3Probe:
|
|
probe_id: str
|
|
sql_key: str
|
|
expectation: Expectation
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class E3Result:
|
|
probe_id: str
|
|
outcome: Outcome
|
|
reason: str
|
|
|
|
|
|
PROBES = (
|
|
E3Probe("conformance-view-empty", "conformance", "zero_rows"),
|
|
E3Probe("unset-guc-reads-none", "unset_guc", "zero_rows"),
|
|
E3Probe("tenant-a-cannot-read-b", "wrong_tenant_read", "zero_rows"),
|
|
E3Probe("tenant-a-cannot-insert-b", "wrong_tenant_insert", "statement_rejected"),
|
|
E3Probe("runtime-lacks-bypassrls", "runtime_bypassrls", "false"),
|
|
E3Probe("unsafe-definer-inventory-empty", "unsafe_definer", "zero_rows"),
|
|
E3Probe("sql-compromise-reset", "reset_to_b", "documented_limit"),
|
|
)
|
|
|
|
|
|
def evaluate(probe: E3Probe, *, rows: int = 0, rejected: bool = False,
|
|
boolean: bool | None = None) -> E3Result:
|
|
if probe.expectation == "documented_limit":
|
|
return E3Result(probe.probe_id, "inconclusive",
|
|
"E3-B observation records the documented SQL-compromise limit")
|
|
passed = {
|
|
"zero_rows": rows == 0,
|
|
"statement_rejected": rejected,
|
|
"false": boolean is False,
|
|
}[probe.expectation]
|
|
if passed:
|
|
return E3Result(probe.probe_id, "pass", f"expectation met: {probe.expectation}")
|
|
return E3Result(probe.probe_id, "finding", f"expectation failed: {probe.expectation}")
|
|
|
|
|
|
def fixture_results(*, enforce: bool) -> list[E3Result]:
|
|
"""In-process known-good/known-bad outcomes. No database connection."""
|
|
results: list[E3Result] = []
|
|
for probe in PROBES:
|
|
if probe.expectation == "documented_limit":
|
|
results.append(evaluate(probe))
|
|
continue
|
|
if enforce:
|
|
results.append(evaluate(probe, rows=0, rejected=True, boolean=False))
|
|
else:
|
|
results.append(evaluate(probe, rows=1, rejected=False, boolean=True))
|
|
return results
|
|
|
|
|
|
def e3_calibration() -> dict:
|
|
started = utc_now()
|
|
good = fixture_results(enforce=True)
|
|
bad = fixture_results(enforce=False)
|
|
|
|
def expected(result: E3Result, *, enforce: bool) -> bool:
|
|
probe = next(item for item in PROBES if item.probe_id == result.probe_id)
|
|
if probe.expectation == "documented_limit":
|
|
return result.outcome == "inconclusive"
|
|
return result.outcome == ("pass" if enforce else "finding")
|
|
|
|
ok = all(expected(item, enforce=True) for item in good) and all(
|
|
expected(item, enforce=False) for item in bad
|
|
)
|
|
return {
|
|
"schema_version": "whitehat-e3-calibration/v1",
|
|
"evidence_class": "fixture",
|
|
"run_id": f"e3-calibration-{started}",
|
|
"started_at": started,
|
|
"ended_at": utc_now(),
|
|
"outcome": "pass" if ok else "finding",
|
|
"cadence": CADENCE,
|
|
"known_good": [result.__dict__ for result in good],
|
|
"known_bad": [result.__dict__ for result in bad],
|
|
"limitations": [
|
|
"Offline E3 calibration evaluates the harness; it is not target assurance.",
|
|
"No database connection or live credential was used.",
|
|
"sql-compromise-reset is E3's documented limit and stays inconclusive.",
|
|
],
|
|
}
|
|
|
|
|
|
CADENCE = {
|
|
"interval": "24h",
|
|
"maximum_detection_window": "24h plus run and reporting latency",
|
|
"reset_triggers": [
|
|
"schema migration", "role or grant change", "RLS policy change",
|
|
"security-definer function change", "posture mechanism change",
|
|
],
|
|
"triggered_run_deadline": "before deployment promotion",
|
|
}
|
|
|