mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5
4ee31142ab
Admission extracted everything before the first parenthesis, so the operation for `const preTax = eval(req.body.preTax)` was `const preTax = eval`. No real declaration can enumerate every assignment a codebase might write, so admission would have rejected genuine findings on operation_semantics_mismatch. The M1b demonstration only passed because its declaration listed the assignment spellings. The symbol is now the trailing qualified callee. Receiver spelling stays part of it, so `$wpdb->prepare` and `prepare` remain different operations and a declaration cannot widen from one to the other. A spelling that is not a qualified name, such as a dynamic `handlers[name](x)`, yields the statement text, matches no declaration and stays ineligible exactly as before. Controls cover the assigned call, the receiver distinction, the dynamic spelling, and that a declaration for a different operation still does not cover it. The recorded-veto report reads the same helper instead of repeating the extraction. 1,403 tests passed, nine skipped. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
266 lines
14 KiB
Python
266 lines
14 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
|
|
|
|
|
|
def test_the_operation_symbol_comes_from_the_call_not_the_statement():
|
|
"""M1b 12.1: a declaration cannot enumerate every assignment a codebase might write."""
|
|
from openultrasast.push.policy import operation_symbol
|
|
|
|
assert operation_symbol("eval(req.body.preTax)") == "eval"
|
|
assert operation_symbol("const preTax = eval(req.body.preTax)") == "eval"
|
|
assert operation_symbol("return os.system(full)") == "os.system"
|
|
# Receiver spelling is part of the symbol: these are different operations.
|
|
assert operation_symbol("$result = $wpdb->query($sql)") == "$wpdb->query"
|
|
assert operation_symbol("$wpdb->prepare($sql)") != operation_symbol("prepare($sql)")
|
|
# A spelling that is not a qualified name matches no declaration and stays ineligible.
|
|
assert operation_symbol("handlers[name](x)") == "handlers[name]"
|
|
|
|
|
|
def test_a_declaration_naming_the_call_now_covers_an_assigned_call():
|
|
candidate, capability, _ = control()
|
|
assigned = replace(
|
|
candidate,
|
|
delta=replace(
|
|
candidate.delta,
|
|
head_operation=replace(candidate.delta.head_operation, operation="const out = exec(command)"),
|
|
site="handler.src:4:handler",
|
|
),
|
|
)
|
|
covered = replace(capability, operation_symbols=("exec",))
|
|
disposition = admit_candidates((assigned,), capabilities=(covered,)).dispositions[0]
|
|
assert "operation_semantics_mismatch" not in disposition.reasons
|
|
# A declaration for a different operation still does not cover it.
|
|
other = replace(capability, operation_symbols=("query",))
|
|
assert "operation_semantics_mismatch" in admit_candidates((assigned,), capabilities=(other,)).dispositions[0].reasons
|