Admit fixture-asm for ASM T-06 and calibrate known-bad replay
Reassess the T-06 blocker against the secrets-engine consume client. Gate House cited 3cd9955; approval_consume.py first appears at 4b4d556. An in-process CAS disables different-digest conflict for known-bad and drives consume_approval through a mock opener. No network or OpenBao. Live asm-t06 stays pending. Assistant: grok Assistant-Session: 01a05e32-c776-72a3-86ec-c490e027aca9
This commit is contained in:
parent
75b79a0167
commit
75deaf073f
13 changed files with 424 additions and 21 deletions
232
src/whitehat_security/asm.py
Normal file
232
src/whitehat_security/asm.py
Normal file
|
|
@ -0,0 +1,232 @@
|
|||
"""In-process ASM fixtures. No network, OpenBao, or live approval-engine."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
from urllib.error import HTTPError
|
||||
from urllib.request import Request
|
||||
|
||||
from .model import Outcome, utc_now
|
||||
from .plane import repo_root
|
||||
|
||||
DIGEST_A = "sha256:" + ("ab" * 32)
|
||||
DIGEST_B = "sha256:" + ("cd" * 32)
|
||||
CANDIDATE_CLAIMED = "3cd9955ac935be398f0fbb6aa7263660a9888721"
|
||||
CONSUME_MODULE_REV = "4b4d556d6264a5da933f7e14aae0fe27ba1f4300"
|
||||
SYNTHETIC_TOKEN = "whitehat-t06-synthetic-not-for-use"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AsmResult:
|
||||
probe_id: str
|
||||
outcome: Outcome
|
||||
reason: str
|
||||
openbao_calls: int
|
||||
|
||||
|
||||
class _Response:
|
||||
def __init__(self, status: int, payload: dict[str, Any]) -> None:
|
||||
self._status = status
|
||||
self._raw = json.dumps(payload).encode("utf-8")
|
||||
|
||||
def getcode(self) -> int:
|
||||
return self._status
|
||||
|
||||
def read(self, _n: int = -1) -> bytes:
|
||||
return self._raw
|
||||
|
||||
def close(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class ConsumeCAS:
|
||||
"""In-process consume store. Known-bad disables different-digest conflict."""
|
||||
|
||||
def __init__(self, *, enforce: bool) -> None:
|
||||
self.enforce = enforce
|
||||
self._digest: dict[str, str] = {}
|
||||
self.openbao_calls = 0
|
||||
|
||||
def opener(self, request: Request, timeout: float) -> _Response:
|
||||
del timeout
|
||||
body = json.loads(request.data.decode("utf-8"))
|
||||
approval_id = request.full_url.rstrip("/").rsplit("/", 2)[1]
|
||||
digest = body["request_digest"]
|
||||
prior = self._digest.get(approval_id)
|
||||
if prior is None:
|
||||
self._digest[approval_id] = digest
|
||||
return _Response(200, _consumed(approval_id, digest, idempotent=False))
|
||||
if prior == digest:
|
||||
return _Response(200, _consumed(approval_id, digest, idempotent=True))
|
||||
if self.enforce:
|
||||
raise HTTPError(
|
||||
request.full_url, 409, "conflict", hdrs=None, fp=BytesIO(b'{"error":"conflict"}')
|
||||
)
|
||||
self._digest[approval_id] = digest
|
||||
return _Response(200, _consumed(approval_id, digest, idempotent=False))
|
||||
|
||||
|
||||
def _consumed(approval_id: str, digest: str, *, idempotent: bool) -> dict[str, Any]:
|
||||
return {
|
||||
"approval_id": approval_id,
|
||||
"status": "consumed",
|
||||
"request_digest": digest,
|
||||
"decision_id": "decision:whitehat-t06",
|
||||
"consumed_at": "2026-09-02T00:00:00Z",
|
||||
"idempotent": idempotent,
|
||||
}
|
||||
|
||||
|
||||
def secrets_engine_src() -> Path | None:
|
||||
env = os.environ.get("WHITEHAT_SECRETS_ENGINE_SRC")
|
||||
if env:
|
||||
path = Path(env)
|
||||
return path if path.exists() else None
|
||||
sibling = repo_root().parent / "secrets-engine" / "src"
|
||||
return sibling if sibling.exists() else None
|
||||
|
||||
|
||||
def _load_consume() -> tuple[Callable, Any, type]:
|
||||
src = secrets_engine_src()
|
||||
if src is None:
|
||||
raise RuntimeError("secrets-engine src is not available")
|
||||
root = str(src)
|
||||
if root not in sys.path:
|
||||
sys.path.insert(0, root)
|
||||
from secrets_engine.approval_consume import ConsumeBinding, consume_approval
|
||||
from secrets_engine.errors import DecisionError
|
||||
return consume_approval, ConsumeBinding, DecisionError
|
||||
|
||||
|
||||
def _token_file() -> Path:
|
||||
handle = tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8")
|
||||
try:
|
||||
handle.write(SYNTHETIC_TOKEN)
|
||||
handle.flush()
|
||||
os.fchmod(handle.fileno(), stat.S_IRUSR | stat.S_IWUSR)
|
||||
finally:
|
||||
handle.close()
|
||||
return Path(handle.name)
|
||||
|
||||
|
||||
def _pep_act(cas: ConsumeCAS, digest: str) -> str:
|
||||
consume_approval, ConsumeBinding, DecisionError = _load_consume()
|
||||
token = _token_file()
|
||||
try:
|
||||
consume_approval(
|
||||
base_url="http://approval.invalid",
|
||||
token_file=token,
|
||||
binding=ConsumeBinding(
|
||||
approval_id="appr_whitehat-t06",
|
||||
request_digest=digest,
|
||||
decision_id="decision:whitehat-t06",
|
||||
),
|
||||
opener=cas.opener,
|
||||
)
|
||||
except DecisionError:
|
||||
return "rejected"
|
||||
finally:
|
||||
token.unlink(missing_ok=True)
|
||||
cas.openbao_calls += 1
|
||||
return "acted"
|
||||
|
||||
|
||||
def t06_results(*, enforce: bool) -> list[AsmResult]:
|
||||
"""Drive secrets-engine consume_approval through an in-process CAS opener."""
|
||||
cas = ConsumeCAS(enforce=enforce)
|
||||
first = _pep_act(cas, DIGEST_A)
|
||||
same = _pep_act(cas, DIGEST_A)
|
||||
before_replay = cas.openbao_calls
|
||||
replay = _pep_act(cas, DIGEST_B)
|
||||
extra = cas.openbao_calls - before_replay
|
||||
first_ok = first == "acted" and same == "acted"
|
||||
if enforce:
|
||||
replay_ok = replay == "rejected" and extra == 0
|
||||
return [
|
||||
AsmResult(
|
||||
"t06-exact-consume",
|
||||
"pass" if first_ok else "finding",
|
||||
"first and same-digest retry consumed",
|
||||
before_replay,
|
||||
),
|
||||
AsmResult(
|
||||
"t06-different-digest-replay",
|
||||
"pass" if replay_ok and first_ok else "finding",
|
||||
"different-digest replay rejected; OpenBao not called"
|
||||
if replay_ok else "enforcing fixture failed T-06 oracle",
|
||||
extra,
|
||||
),
|
||||
]
|
||||
replay_bad = replay == "acted" and extra == 1
|
||||
return [
|
||||
AsmResult(
|
||||
"t06-exact-consume",
|
||||
"pass" if first_ok else "finding",
|
||||
"first and same-digest retry consumed",
|
||||
before_replay,
|
||||
),
|
||||
AsmResult(
|
||||
"t06-different-digest-replay",
|
||||
"finding" if replay_bad else "pass",
|
||||
"known-bad CAS accepted a different-digest replay and the PEP called OpenBao"
|
||||
if replay_bad else "known-bad fixture did not lose the T-06 oracle",
|
||||
extra,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def t06_calibration() -> dict:
|
||||
started = utc_now()
|
||||
try:
|
||||
good = t06_results(enforce=True)
|
||||
bad = t06_results(enforce=False)
|
||||
ok = (
|
||||
all(item.outcome == "pass" for item in good)
|
||||
and any(item.outcome == "finding" and item.probe_id.endswith("replay")
|
||||
for item in bad)
|
||||
)
|
||||
limitations = [
|
||||
"Offline ASM T-06 calibration evaluates the harness against an in-process CAS.",
|
||||
"No network, OpenBao, live approval-engine, or reusable credential was used.",
|
||||
"Gate House cited secrets-engine 3cd9955; approval_consume.py first appears at 4b4d556.",
|
||||
"This is fixture evidence, not live target assurance.",
|
||||
]
|
||||
except Exception as error:
|
||||
return {
|
||||
"schema_version": "whitehat-asm-calibration/v1",
|
||||
"evidence_class": "fixture",
|
||||
"run_id": f"asm-t06-calibration-{started}",
|
||||
"started_at": started,
|
||||
"ended_at": utc_now(),
|
||||
"outcome": "inconclusive",
|
||||
"specification": "asm-assurance-targets.v1",
|
||||
"test_id": "T-06",
|
||||
"limitations": [f"T-06 fixture could not import the candidate: {error}"],
|
||||
}
|
||||
return {
|
||||
"schema_version": "whitehat-asm-calibration/v1",
|
||||
"evidence_class": "fixture",
|
||||
"run_id": f"asm-t06-calibration-{started}",
|
||||
"started_at": started,
|
||||
"ended_at": utc_now(),
|
||||
"outcome": "pass" if ok else "finding",
|
||||
"specification": "asm-assurance-targets.v1",
|
||||
"test_id": "T-06",
|
||||
"target": {
|
||||
"component": "secrets-engine",
|
||||
"claimed_revision": CANDIDATE_CLAIMED,
|
||||
"consume_module_revision": CONSUME_MODULE_REV,
|
||||
"module": "secrets_engine.approval_consume",
|
||||
},
|
||||
"known_good": [item.__dict__ for item in good],
|
||||
"known_bad": [item.__dict__ for item in bad],
|
||||
"limitations": limitations,
|
||||
}
|
||||
|
|
@ -9,6 +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 .e3 import CADENCE, PROBES, e3_calibration
|
||||
from .engagement import AuthorizationError, Engagement
|
||||
from .fixtures import FixtureService, probe_suite
|
||||
|
|
@ -102,6 +103,11 @@ def main(argv: list[str] | None = None) -> None:
|
|||
"capacity-fixture", help="calibrate P1/P2 evaluator offline"
|
||||
)
|
||||
capacity_fix.add_argument("--output")
|
||||
asm_fix = commands.add_parser(
|
||||
"asm-fixtures", help="calibrate ASM probes offline"
|
||||
)
|
||||
asm_fix.add_argument("--test-id", default="T-06")
|
||||
asm_fix.add_argument("--output")
|
||||
message = commands.add_parser("risk-message")
|
||||
message.add_argument("report")
|
||||
conformance = commands.add_parser(
|
||||
|
|
@ -215,6 +221,17 @@ def main(argv: list[str] | None = None) -> None:
|
|||
else:
|
||||
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()
|
||||
rendered = json.dumps(result, indent=2, sort_keys=True) + "\n"
|
||||
if args.output:
|
||||
Path(args.output).write_text(rendered, encoding="utf-8")
|
||||
else:
|
||||
print(rendered, end="")
|
||||
raise SystemExit(0 if result["outcome"] == "pass" else 1)
|
||||
if args.command == "capacity-fixture":
|
||||
result = capacity_calibration()
|
||||
rendered = json.dumps(result, indent=2, sort_keys=True) + "\n"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue