mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 19:59:26 +00:00
Marius CiepluchandClaude Opus 5
86980d8aa0
Rerunning the three NodeGoat revisions after tasks 10.1-10.3 shows the recorded view now classifies the case: the vulnerable revision reaches new with operation_absent_from_comparable_base, and the unchanged revision reaches unchanged with its base location resolved to the same lines. Both previously stopped at operation_correspondence_unresolved and witness_identity_unresolved. The rerun also caught a defect in the repair itself: the new novelty reason was missing from the admission allowlist, so admission rejected the candidate as change_unsupported. Fixed with a control. The production path is still blocked by the last two vetoes, and the measurement shows neither can be scoped or relaxed as the plan described. A taint answer reports traced flows, so an absent row means either that the operation does not exist in the base or that no traced flow reached it. dynamic_external_or_depth_context_unresolved exists to stop the second being read as the first, and removing it would turn a wrong absence into a wrong new finding. It is also computed over every in-scope method, so on any real Express application it is always true and can never be discharged, which is why the tool reports nothing on real projects. The resolution is new evidence rather than a relaxed rule: export the operations the query actually enumerated, separately from the flows it traced, so absence is established from the inventory while a traced-flow claim still requires complete reachability. That is now task 10.4, ahead of the two vetoes it unblocks, and it adds no detector, alias resolver or second source IR. 1,348 tests passed, nine skipped. No capability enabled; rollout stays NO-GO. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
234 lines
12 KiB
Python
234 lines
12 KiB
Python
"""Synthetic admission controls, not evidence of real capability eligibility."""
|
|
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from openultrasast.config import PushConfig
|
|
from openultrasast.model.contracts import QuestionIdentity, RankedQuestion, ScopeDecision
|
|
from openultrasast.push.contracts import ComparisonAnalysis, PushComparison
|
|
from openultrasast.push.policy import (
|
|
AdmissionCandidate,
|
|
AdmissionResult,
|
|
CandidateDelta,
|
|
CapabilityAdmission,
|
|
CapabilityKey,
|
|
EvidenceOperation,
|
|
admission_push_result,
|
|
admit_candidates,
|
|
)
|
|
|
|
|
|
def control(language="javascript", framework="express"):
|
|
q = QuestionIdentity("repository", language, "handler.src", "handler", "injection")
|
|
op = EvidenceOperation(q, "taint", "handler.src", 4, "exec(command)", "request.input", False, "{}")
|
|
delta = CandidateDelta(
|
|
"injection",
|
|
"handler.src:4:handler",
|
|
"model_entailed",
|
|
"request.input -> exec(command)",
|
|
"defect",
|
|
"new",
|
|
"source_connection_absent_from_comparable_base",
|
|
op,
|
|
(),
|
|
("head:handler.src:2-2",),
|
|
"base",
|
|
"head",
|
|
"synthetic-semantics-v1",
|
|
)
|
|
key = CapabilityKey(language, "server", framework, "injection", "taint", "shell-command", "synthetic-semantics-v1")
|
|
comparison = PushComparison("head", "base", "remote", ("refs/heads/main",))
|
|
candidate = AdmissionCandidate(delta, key, comparison)
|
|
capability = CapabilityAdmission(
|
|
key,
|
|
"synthetic-only-evaluation",
|
|
"synthetic://reviewed-controls",
|
|
"PASS",
|
|
"Untrusted input reaching {operation} permits command execution in {context}.",
|
|
"At {operation}, use an argument-vector process API without a shell for {context}.",
|
|
enabled=True,
|
|
positive_controls=2,
|
|
reviewed_alerts=2,
|
|
calibration_artifact="synthetic://scorecard",
|
|
untouched_workload="synthetic-control-population",
|
|
operation_symbols=("exec",),
|
|
)
|
|
scope = ScopeDecision("v1", "static", True, (RankedQuestion(q, 1.0, ()),), (), ())
|
|
analysis = ComparisonAnalysis(comparison, (scope,), (q,), "complete_within_scope")
|
|
return candidate, capability, analysis
|
|
|
|
|
|
@pytest.mark.parametrize("language,framework", [("php", "wordpress"), ("javascript", "express")])
|
|
def test_equivalent_origins_admit_same_claim_and_only_block_when_enabled(language, framework):
|
|
candidate, capability, analysis = control(language, framework)
|
|
admission = admit_candidates((candidate,), capabilities=(capability,))
|
|
assert len(admission.defects) == 1
|
|
defect = admission.defects[0]
|
|
assert defect.locations == (("handler.src", 4),)
|
|
assert defect.witnesses == (candidate.delta.witness,)
|
|
assert "exec(command)" in defect.repairs[0] and "shell-command" in defect.consequences[0]
|
|
for mode, disposition in (("advisory", "allow"), ("blocking", "block")):
|
|
result = admission_push_result(admission, (analysis,), config=PushConfig(mode=mode))
|
|
assert result.finding_status == "actionable"
|
|
assert result.push_disposition == disposition
|
|
assert result.coverage_status == "complete_within_scope"
|
|
assert AdmissionResult.from_payload(admission.to_payload()) == admission
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation,reason",
|
|
[
|
|
(lambda c: replace(c, delta=replace(c.delta, rung="suspicion")), "insufficient_rung"),
|
|
(lambda c: replace(c, delta=replace(c.delta, rung="model_corroborated")), "insufficient_rung"),
|
|
(lambda c: replace(c, delta=replace(c.delta, rung="execution_confirmed")), "insufficient_rung"),
|
|
(lambda c: replace(c, delta=replace(c.delta, novelty="unknown")), "comparison_unknown"),
|
|
(lambda c: replace(c, delta=replace(c.delta, novelty="unchanged")), "unchanged"),
|
|
(lambda c: replace(c, delta=replace(c.delta, witness=None)), "witness_unresolved"),
|
|
(lambda c: replace(c, delta=replace(c.delta, site="elsewhere:1:f")), "location_unresolved"),
|
|
(lambda c: replace(c, delta=replace(c.delta, head_operation=None)), "witness_unresolved"),
|
|
(lambda c: replace(c, delta=replace(c.delta, change_evidence=())), "change_unsupported"),
|
|
(lambda c: replace(c, dependency_gaps=("vendor-summary-missing",)), "dependency_unresolved"),
|
|
(lambda c: replace(c, sanitizer_contexts=("html-output",)), "sanitizer_semantics_mismatch"),
|
|
(lambda c: replace(c, delta=replace(c.delta, head_operation=replace(c.delta.head_operation, discharged=True))), "claim_discharged"),
|
|
(lambda c: replace(c, capability=replace(c.capability, language="python")), "capability_unavailable"),
|
|
],
|
|
)
|
|
def test_rejected_claims_remain_diagnostics(mutation, reason):
|
|
candidate, capability, _ = control()
|
|
result = admit_candidates((mutation(candidate),), capabilities=(capability,))
|
|
assert not result.defects
|
|
assert reason in result.dispositions[0].reasons
|
|
assert not result.dispositions[0].admitted
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation,reason",
|
|
[
|
|
(lambda c: replace(c, enabled=False), "capability_disabled"),
|
|
(lambda c: replace(c, verdict="experimental"), "capability_unevaluated"),
|
|
(lambda c: replace(c, verdict="NO-GO"), "capability_unevaluated"),
|
|
(lambda c: replace(c, key=replace(c.key, semantics="stale")), "capability_unavailable"),
|
|
(lambda c: replace(c, key=replace(c.key, runtime="browser")), "capability_unavailable"),
|
|
(lambda c: replace(c, key=replace(c.key, framework="other")), "capability_unavailable"),
|
|
(lambda c: replace(c, repair_template="sanitize input"), "repair_unsupported"),
|
|
(lambda c: replace(c, consequence_template="potential security issue"), "consequence_unsupported"),
|
|
],
|
|
)
|
|
def test_evaluated_semantics_and_grounded_repairs_required(mutation, reason):
|
|
candidate, capability, _ = control()
|
|
result = admit_candidates((candidate,), capabilities=(mutation(capability),))
|
|
assert not result.defects
|
|
assert reason in result.dispositions[0].reasons
|
|
|
|
|
|
def test_no_default_eligibility_and_unsupported_origin_is_a_coverage_gap():
|
|
candidate, _, analysis = control()
|
|
result = admit_candidates((candidate,))
|
|
assert not result.defects and "capability_unavailable" in result.coverage_reasons
|
|
push = admission_push_result(result, (analysis,), config=PushConfig(mode="blocking"))
|
|
assert push.finding_status == "none" and push.coverage_status == "incomplete" and push.push_disposition == "allow"
|
|
strict = admission_push_result(result, (analysis,), config=PushConfig(mode="blocking", incomplete_coverage_policy="block"))
|
|
assert strict.push_disposition == "block" and strict.finding_status == "none"
|
|
advisory = admission_push_result(result, (analysis,), config=PushConfig(incomplete_coverage_policy="block"))
|
|
assert advisory.push_disposition == "allow"
|
|
|
|
|
|
def test_dedup_preserves_witnesses_and_comparisons_and_does_not_dismiss_printed_before():
|
|
candidate, capability, _ = control()
|
|
second = replace(
|
|
candidate,
|
|
delta=replace(candidate.delta, witness="request.other -> exec(command)", base_revision="other-base"),
|
|
comparison=PushComparison("head", "other-base", "remote", ("refs/heads/release",)),
|
|
)
|
|
result = admit_candidates((candidate, second, candidate), capabilities=(capability,))
|
|
assert len(result.defects) == 1 and len(result.dispositions) == 3
|
|
defect = result.defects[0]
|
|
assert len(defect.witnesses) == 2 and len(defect.comparisons) == 2
|
|
assert defect.refs == ("refs/heads/main", "refs/heads/release")
|
|
assert admit_candidates((candidate, second, candidate), capabilities=(capability,)) == result
|
|
|
|
|
|
def test_unknown_novelty_is_coverage_but_unchanged_debt_is_not():
|
|
candidate, capability, analysis = control()
|
|
for novelty, coverage in (("unknown", "incomplete"), ("unchanged", "complete_within_scope")):
|
|
admission = admit_candidates((replace(candidate, delta=replace(candidate.delta, novelty=novelty)),), capabilities=(capability,))
|
|
assert admission_push_result(admission, (analysis,)).coverage_status == coverage
|
|
|
|
|
|
def test_ambiguous_capability_declarations_fail_closed():
|
|
candidate, capability, _ = control()
|
|
result = admit_candidates((candidate,), capabilities=(capability, replace(capability, evaluation_id="other")))
|
|
assert not result.defects and "capability_ambiguous" in result.coverage_reasons
|
|
|
|
|
|
def test_comparison_ownership_cannot_be_rebound_to_another_base_or_semantics():
|
|
candidate, capability, _ = control()
|
|
for altered in (
|
|
replace(candidate, comparison=replace(candidate.comparison, base_oid="different-base")),
|
|
replace(candidate, comparison=replace(candidate.comparison, head_oid="different-head")),
|
|
replace(candidate, delta=replace(candidate.delta, semantics="other-semantics")),
|
|
replace(candidate, delta=replace(candidate.delta, base_revision=None)),
|
|
):
|
|
result = admit_candidates((altered,), capabilities=(capability,))
|
|
assert not result.defects and "comparison_provenance_mismatch" in result.coverage_reasons
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value", [("positive_controls", 0), ("reviewed_alerts", 0), ("calibration_artifact", None), ("untouched_workload", None)]
|
|
)
|
|
def test_pass_label_without_evaluation_population_is_not_eligibility(field, value):
|
|
candidate, capability, _ = control()
|
|
result = admit_candidates((candidate,), capabilities=(replace(capability, **{field: value}),))
|
|
assert not result.defects and "capability_unevaluated" in result.coverage_reasons
|
|
|
|
|
|
def test_unchanged_on_one_base_does_not_suppress_new_on_another():
|
|
candidate, capability, analysis = control()
|
|
unchanged = replace(
|
|
candidate,
|
|
delta=replace(candidate.delta, base_revision="other-base", novelty="unchanged"),
|
|
comparison=PushComparison("head", "other-base", "remote", ("refs/heads/debt",)),
|
|
)
|
|
result = admit_candidates((unchanged, candidate), capabilities=(capability,))
|
|
assert len(result.defects) == 1 and result.defects[0].refs == candidate.comparison.refs
|
|
assert result.defects[0].comparisons == (candidate.comparison,)
|
|
with pytest.raises(ValueError, match="exact comparison"):
|
|
admission_push_result(result, (replace(analysis, comparison=unchanged.comparison),))
|
|
|
|
|
|
def test_sql_operation_cannot_acquire_shell_consequence_from_same_family():
|
|
candidate, capability, _ = control()
|
|
operation = replace(candidate.delta.head_operation, operation="query(sql)")
|
|
wrong_context = replace(candidate, delta=replace(candidate.delta, head_operation=operation, witness="req.query -> query(sql)"))
|
|
result = admit_candidates((wrong_context,), capabilities=(capability,))
|
|
assert not result.defects
|
|
assert "operation_semantics_mismatch" in result.dispositions[0].reasons
|
|
assert result.dispositions[0].candidate.delta.witness == "req.query -> query(sql)"
|
|
|
|
|
|
@pytest.mark.parametrize("symbols", [(), ("execSync",), ("query",)])
|
|
def test_missing_or_different_evaluated_operation_binding_is_diagnostic(symbols):
|
|
candidate, capability, _ = control()
|
|
result = admit_candidates((candidate,), capabilities=(replace(capability, operation_symbols=symbols),))
|
|
assert not result.defects and "operation_semantics_mismatch" in result.coverage_reasons
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["query('exec(command)')", "custom.exec(command)", "execSync(command)"])
|
|
def test_evaluated_operation_matching_preserves_receiver_and_symbol_boundaries(operation):
|
|
candidate, capability, _ = control()
|
|
candidate = replace(
|
|
candidate, delta=replace(candidate.delta, head_operation=replace(candidate.delta.head_operation, operation=operation))
|
|
)
|
|
assert not admit_candidates((candidate,), capabilities=(capability,)).defects
|
|
|
|
|
|
def test_an_operation_the_change_introduced_is_a_supported_change_basis():
|
|
"""M1b 10.1: the new novelty reason must reach admission, not be rejected as unsupported."""
|
|
candidate, capability, _ = control()
|
|
introduced = replace(candidate, delta=replace(candidate.delta, reason="operation_absent_from_comparable_base"))
|
|
assert admit_candidates((introduced,), capabilities=(capability,)).defects
|
|
invented = replace(candidate, delta=replace(candidate.delta, reason="operation_moved_within_change"))
|
|
disposition = admit_candidates((invented,), capabilities=(capability,)).dispositions[0]
|
|
assert not disposition.admitted and "change_unsupported" in disposition.reasons
|