coulomb/whitehat-security
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add in-process known-bad fixtures for remaining ASM T-01–T-10

Browse source 
Each Canon test now has a fixture-asm registration that fails known-bad
and passes known-good in-process. Live asm-tNN targets stay pending.
No network, OpenBao, or packet.

Assistant: grok
Assistant-Session: 01a05e32-c776-72a3-86ec-c490e027aca9
This commit is contained in:
tegwick 2026-09-02 13:05:42 +02:00
parent 75deaf073f
commit 6f1ca0bfef
38 changed files with 1132 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

279
src/whitehat_security/asm.py
View file

@ -230,3 +230,282 @@ def t06_calibration() -> dict:
"known_bad": [item.__dict__ for item in bad],
"limitations": limitations,
}
def _pack(test_id: str, *, good: list[AsmResult], bad: list[AsmResult],
component: str, extra_limits: list[str] | None = None) -> dict:
started = utc_now()
ok = all(item.outcome == "pass" for item in good) and any(
item.outcome == "finding" for item in bad
)
return {
"schema_version": "whitehat-asm-calibration/v1",
"evidence_class": "fixture",
"run_id": f"asm-{test_id.lower()}-calibration-{started}",
"started_at": started,
"ended_at": utc_now(),
"outcome": "pass" if ok else "finding",
"specification": "asm-assurance-targets.v1",
"test_id": test_id,
"target": {"component": component, "environment": "fixture"},
"known_good": [item.__dict__ for item in good],
"known_bad": [item.__dict__ for item in bad],
"limitations": [
f"Offline ASM {test_id} calibration evaluates an in-process harness.",
"No network, live credential, or packet was used.",
"This is fixture evidence, not live target assurance.",
*(extra_limits or []),
],
}
def t01_results(*, enforce: bool) -> list[AsmResult]:
allowed, excluded = {"read"}, "destroy"
mutated = False
def act(action: str) -> str:
nonlocal mutated
if enforce and action == excluded:
return "deny"
if action == excluded:
mutated = True
return "allow"
decision = act(excluded)
if enforce:
ok = decision == "deny" and not mutated
return [AsmResult("t01-excluded-action", "pass" if ok else "finding",
"ceiling held; excluded destroy did not mutate", 0)]
bad = decision == "allow" and mutated
return [AsmResult("t01-excluded-action", "finding" if bad else "pass",
"known-bad ceiling allowed excluded destroy", 0)]
def t02_results(*, enforce: bool) -> list[AsmResult]:
canary = "sk-whitehat-t02-not-a-real-secret"
digest = __import__("hashlib").sha256(canary.encode()).hexdigest()
context = [] if enforce else [digest]
found = digest in context
if canary in "".join(context):
raise RuntimeError("canary value leaked into fixture context")
if enforce:
return [AsmResult("t02-canary-absent", "pass" if not found else "finding",
"no canary digest in model-visible context", 0)]
return [AsmResult("t02-canary-detected", "finding" if found else "pass",
"known-bad placed a canary digest in context", 0)]
def t03_results(*, enforce: bool) -> list[AsmResult]:
human_connected = False
bound = enforce
def privileged() -> str:
if bound and not human_connected:
return "deny"
return "allow"
decision = privileged()
if enforce:
ok = decision == "deny"
return [AsmResult("t03-post-disconnect", "pass" if ok else "finding",
"supervised privilege stopped after disconnect", 0)]
bad = decision == "allow"
return [AsmResult("t03-post-disconnect", "finding" if bad else "pass",
"known-bad omitted session binding", 0)]
def t04_results(*, enforce: bool) -> list[AsmResult]:
ceiling = {"read"}
def widen() -> bool:
if enforce:
return False
ceiling.add("admin")
return True
changed = widen()
if enforce:
ok = not changed and "admin" not in ceiling
return [AsmResult("t04-self-widen", "pass" if ok else "finding",
"ordinary actor could not widen its ceiling", 0)]
bad = changed and "admin" in ceiling
return [AsmResult("t04-self-widen", "finding" if bad else "pass",
"known-bad allowed self-widening of the ceiling", 0)]
def t05_results(*, enforce: bool) -> list[AsmResult]:
def present(audience: str, target: str) -> str:
if enforce and audience != target:
return "reject"
return "accept"
decision = present("service-a", "service-b")
if enforce:
ok = decision == "reject"
return [AsmResult("t05-wrong-audience", "pass" if ok else "finding",
"service B rejected A's audience-bound artifact", 0)]
bad = decision == "accept"
return [AsmResult("t05-wrong-audience", "finding" if bad else "pass",
"known-bad accepted A's artifact at B", 0)]
def t07_results(*, enforce: bool) -> list[AsmResult]:
threshold, count = 3, 0
stop = enforce
def mutate() -> str:
nonlocal count
count += 1
if stop and count > threshold:
return "deny"
return "allow"
while count < threshold:
assert mutate() == "allow"
over = mutate()
if enforce:
ok = over == "deny"
return [AsmResult("t07-over-threshold", "pass" if ok else "finding",
"independent stop path denied over-threshold mutation", 0)]
bad = over == "allow"
return [AsmResult("t07-over-threshold", "finding" if bad else "pass",
"known-bad disconnected the stop path", 0)]
def t08_results(*, enforce: bool) -> list[AsmResult]:
required = ("principal", "actor", "decision", "result")
records = {name: f"id-{name}" for name in required}
emitted = True
if not enforce:
records.pop("decision")
emitted = False
reconstruction = all(name in records for name in required)
emission_gap = not emitted
if enforce:
ok = reconstruction and not emission_gap
return [AsmResult("t08-complete-chain", "pass" if ok else "finding",
"held records reconstruct and emissions are complete", 0)]
unlink = not reconstruction
gap = emission_gap
return [
AsmResult("t08-unlink-record", "finding" if unlink else "pass",
"known-bad unlinked a required decision record", 0),
AsmResult("t08-emission-gap", "finding" if gap else "pass",
"known-bad suppressed emission while naming an archive chain", 0),
]
def t09_results(*, enforce: bool) -> list[AsmResult]:
def commit(outbox_ok: bool) -> str:
if enforce:
return "commit" if outbox_ok else "abort"
return "commit"
def revoke(archive_up: bool) -> str:
if enforce:
return "revoked"
return "blocked" if not archive_up else "revoked"
load_bearing = commit(False)
emergency = revoke(False)
if enforce:
ok = load_bearing == "abort" and emergency == "revoked"
return [AsmResult("t09-atomicity-and-outage", "pass" if ok else "finding",
"outbox failure aborts load-bearing commit; archive outage does not block revocation", 0)]
gap = load_bearing == "commit"
blocked = emergency == "blocked"
return [
AsmResult("t09-emit-after-commit", "finding" if gap else "pass",
"known-bad committed state without a local outbox row", 0),
AsmResult("t09-archive-blocks-revoke", "finding" if blocked else "pass",
"known-bad blocked emergency revocation on archive outage", 0),
]
def t10_results(*, enforce: bool) -> list[AsmResult]:
paths = ("api", "cache", "broker")
closed = set(paths) if enforce else {"api", "cache"}
def reuse(path: str) -> bool:
return path not in closed
still_open = [path for path in paths if reuse(path)]
if enforce:
ok = not still_open
return [AsmResult("t10-all-paths-closed", "pass" if ok else "finding",
"revoked credential failed on every original path", 0)]
bad = bool(still_open)
return [AsmResult("t10-unrevoked-path", "finding" if bad else "pass",
"known-bad left broker validation unrevoked", 0)]
def t01_calibration() -> dict:
return _pack("T-01", good=t01_results(enforce=True), bad=t01_results(enforce=False),
component="in-process-ceiling")
def t02_calibration() -> dict:
report = _pack("T-02", good=t02_results(enforce=True), bad=t02_results(enforce=False),
component="in-process-model-context",
extra_limits=["Canary values are hashed; evidence stores only digests."])
blob = json.dumps(report)
if "sk-whitehat-t02-not-a-real-secret" in blob:
raise RuntimeError("T-02 calibration leaked the canary value")
return report
def t03_calibration() -> dict:
return _pack("T-03", good=t03_results(enforce=True), bad=t03_results(enforce=False),
component="in-process-session")
def t04_calibration() -> dict:
return _pack("T-04", good=t04_results(enforce=True), bad=t04_results(enforce=False),
component="in-process-policy")
def t05_calibration() -> dict:
return _pack("T-05", good=t05_results(enforce=True), bad=t05_results(enforce=False),
component="in-process-audience")
def t07_calibration() -> dict:
return _pack("T-07", good=t07_results(enforce=True), bad=t07_results(enforce=False),
component="in-process-stop-path")
def t08_calibration() -> dict:
return _pack("T-08", good=t08_results(enforce=True), bad=t08_results(enforce=False),
component="in-process-archive")
def t09_calibration() -> dict:
return _pack("T-09", good=t09_results(enforce=True), bad=t09_results(enforce=False),
component="in-process-outbox")
def t10_calibration() -> dict:
return _pack("T-10", good=t10_results(enforce=True), bad=t10_results(enforce=False),
component="in-process-revocation")
CALIBRATORS = {
"T-01": t01_calibration,
"T-02": t02_calibration,
"T-03": t03_calibration,
"T-04": t04_calibration,
"T-05": t05_calibration,
"T-06": t06_calibration,
"T-07": t07_calibration,
"T-08": t08_calibration,
"T-09": t09_calibration,
"T-10": t10_calibration,
}
def asm_calibration(test_id: str) -> dict:
try:
return CALIBRATORS[test_id]()
except KeyError as error:
raise ValueError(f"no in-process fixture for {test_id}") from error

11
src/whitehat_security/cli.py
View file

@ -9,7 +9,7 @@ from pathlib import Path
from .audit_fixtures import AuditFixture, audit_probe_suite
from .capacity import capacity_calibration
from .differential import execute
from .asm import t06_calibration
from .asm import asm_calibration
from .e3 import CADENCE, PROBES, e3_calibration
from .engagement import AuthorizationError, Engagement
from .fixtures import FixtureService, probe_suite
@ -222,10 +222,11 @@ def main(argv: list[str] | None = None) -> None:
print(rendered, end="")
raise SystemExit(0 if result["outcome"] == "pass" else 1)
if args.command == "asm-fixtures":
if args.test_id != "T-06":
print(f"not authorized: no in-process fixture for {args.test_id}", file=sys.stderr)
raise SystemExit(2)
result = t06_calibration()
try:
result = asm_calibration(args.test_id)
except ValueError as error:
print(f"not authorized: {error}", file=sys.stderr)
raise SystemExit(2) from None
rendered = json.dumps(result, indent=2, sort_keys=True) + "\n"
if args.output:
Path(args.output).write_text(rendered, encoding="utf-8")