marius/Open_Ultra_SAST
1
0
Fork 0
mirror of https://github.com/norandom/OpenUltraSAST.git synced 2026-10-16 19:59:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

ci: add quality gates and release workflows

Browse Source
This commit is contained in:
Marius Ciepluch
2026-06-18 10:06:21 +02:00
parent 5519413c82
commit 1e6c1f4436
22 changed files with 1606 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openultrasast/calibration.py
+4 -2
View File
@@ -6,7 +6,7 @@ from enum import StrEnum
from pathlib import Path
from .findings import StaticFinding
from .rank import RankingScore, composite_priority
from .rank import RankingScore
class FalsePositiveReason(StrEnum):
@@ -140,7 +140,9 @@ def calibrate_ranking(ranking: RankingScore, learnings: list[FalsePositiveLearni
retrieval_adjustments: dict[str, str] = {}
for learning in applicable:
calibrated = max(1.0, calibrated - learning.demotion)
prompt_constraints.append(f"Do not repeat unsupported {learning.vulnerability_class} claim for scope {learning.scope} without new evidence.")
prompt_constraints.append(
f"Do not repeat unsupported {learning.vulnerability_class} claim for scope {learning.scope} without new evidence."
)
retrieval_adjustments[learning.vulnerability_class] = learning.scope
return RankingCalibration(
path=ranking.path,
src/openultrasast/config.py
+11 -4
View File
@@ -6,7 +6,6 @@ import tomllib
from dataclasses import asdict, dataclass
from pathlib import Path
DEFAULT_VECTOR_STORE = "json-local"
@@ -112,9 +111,9 @@ def _load_sandbox(value: object) -> SandboxConfig:
return SandboxConfig(
network=bool(data.get("network", False)),
workspace_readonly=bool(data.get("workspace_readonly", True)),
memory_mb=int(data.get("memory_mb", 2048)),
timeout_seconds=int(data.get("timeout_seconds", 300)),
pids_limit=int(data.get("pids_limit", 512)),
memory_mb=_int_value(data.get("memory_mb"), 2048),
timeout_seconds=_int_value(data.get("timeout_seconds"), 300),
pids_limit=_int_value(data.get("pids_limit"), 512),
)
@@ -135,6 +134,14 @@ def _load_evidence(value: object) -> EvidenceConfig:
)
def _int_value(value: object, default: int) -> int:
if isinstance(value, bool):
return default
if isinstance(value, int | str | bytes | bytearray):
return int(value)
return default
def _load_static_analysis(value: object) -> StaticAnalysisConfig:
data = _section(value)
paths = data.get("sarif_paths", [])
src/openultrasast/harness.py
+2 -1
View File
@@ -2,10 +2,11 @@ from __future__ import annotations
import hashlib
import json
from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Callable, Literal, Protocol
from typing import Any, Literal, Protocol
from .config import ResolvedConfig
src/openultrasast/index.py
+7 -3
View File
@@ -9,7 +9,6 @@ from typing import Protocol
from .preprocess import FileTarget, RepoSnapshot
VALID_NAMESPACES = {"repo_code", "docs", "static_findings", "mechanisms", "skills", "traces"}
@@ -122,9 +121,14 @@ def chunk_text_namespace(
start_line = start + 1
end_line = start + len(chunk_lines)
chunk_text = "\n".join(chunk_lines)
chunk_metadata = {"path": path, "namespace": namespace, "start_line": start_line, "end_line": end_line}
chunk_metadata: dict[str, str | int | bool] = {
"path": path,
"namespace": namespace,
"start_line": start_line,
"end_line": end_line,
}
if metadata:
chunk_metadata.update(metadata)
chunk_metadata.update({key: value for key, value in metadata.items() if isinstance(value, str | int | bool)})
chunks.append(
CodeChunk(
chunk_id=_chunk_id(f"{namespace}:{path}", start_line, end_line, chunk_text),
src/openultrasast/mapping.py
+124 -14
View File
@@ -4,7 +4,7 @@ import ast
import json
from dataclasses import asdict, dataclass, replace
from pathlib import Path
from typing import Any, Literal
from typing import Any, Literal, cast
from .preprocess import FileTarget
@@ -233,7 +233,9 @@ def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord
for number, line in enumerate(text.splitlines(), start=1):
stripped = line.strip()
if stripped.startswith("if __name__") and "__main__" in stripped:
records.append(_entry(target, number, number, "__main__", "__main__", "cli", "local-only", "process_invocation", [stripped], []))
records.append(
_entry(target, number, number, "__main__", "__main__", "cli", "local-only", "process_invocation", [stripped], [])
)
if "argparse." in stripped or "click.command" in stripped:
records.append(_entry(target, number, number, None, "cli_parser", "cli", "local-only", "process_invocation", [stripped], []))
return records
@@ -245,9 +247,35 @@ def _js_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
for number, line in enumerate(text.splitlines(), start=1):
stripped = line.strip()
if any(token in stripped for token in route_tokens) and any(prefix in stripped for prefix in ("app", "router", "server")):
records.append(_entry(target, number, number, _js_handler_name(stripped), "http_handler", "route", "public", "http_request", [stripped], _line_conditions(stripped)))
records.append(
_entry(
target,
number,
number,
_js_handler_name(stripped),
"http_handler",
"route",
"public",
"http_request",
[stripped],
_line_conditions(stripped),
)
)
if "process.argv" in stripped:
records.append(_entry(target, number, number, None, "process.argv", "cli", "local-only", "process_invocation", [stripped], _line_conditions(stripped)))
records.append(
_entry(
target,
number,
number,
None,
"process.argv",
"cli",
"local-only",
"process_invocation",
[stripped],
_line_conditions(stripped),
)
)
return records
@@ -256,9 +284,35 @@ def _c_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
for number, line in enumerate(text.splitlines(), start=1):
stripped = line.strip()
if " main(" in f" {stripped}" or stripped.startswith("main("):
records.append(_entry(target, number, number, "main", "main", "cli", "local-only", "process_invocation", [stripped], _line_conditions(stripped)))
records.append(
_entry(
target,
number,
number,
"main",
"main",
"cli",
"local-only",
"process_invocation",
[stripped],
_line_conditions(stripped),
)
)
if "LLVMFuzzerTestOneInput" in stripped:
records.append(_entry(target, number, number, "LLVMFuzzerTestOneInput", "LLVMFuzzerTestOneInput", "fuzz", "public", "fuzzer_input", [stripped], _line_conditions(stripped)))
records.append(
_entry(
target,
number,
number,
"LLVMFuzzerTestOneInput",
"LLVMFuzzerTestOneInput",
"fuzz",
"public",
"fuzzer_input",
[stripped],
_line_conditions(stripped),
)
)
return records
@@ -272,12 +326,25 @@ def _solidity_entry_points(target: FileTarget, text: str) -> list[EntryPointReco
continue
evidence = [token for token in ("onlyOwner", "onlyRole", "requiresAuth") if token in stripped]
conditions = [token for token in ("whenNotPaused", "whenPaused", "featureEnabled") if token in stripped]
access = "role-restricted" if evidence else "public"
access: AccessLevel = "role-restricted" if evidence else "public"
name = stripped.removeprefix("function ").split("(", 1)[0]
if "callback" in name.lower() or name.startswith("onERC"):
access = "contract-only/callback"
evidence.append("callback naming convention")
records.append(_entry(target, number, number, name, name, "smart_contract_state_change", access, "contract_call", evidence or ["external/public function"], conditions))
records.append(
_entry(
target,
number,
number,
name,
name,
"smart_contract_state_change",
access,
"contract_call",
evidence or ["external/public function"],
conditions,
)
)
return records
@@ -285,13 +352,54 @@ def _tag_entry_points(target: FileTarget) -> list[EntryPointRecord]:
records: list[EntryPointRecord] = []
tags = set(target.tags)
if "parser" in tags or "deserialization" in tags:
records.append(_entry(target, None, None, None, "parser_input", "parser", "public", "attacker_controlled_input", ["parser/deserialization file tag"], []))
records.append(
_entry(
target,
None,
None,
None,
"parser_input",
"parser",
"public",
"attacker_controlled_input",
["parser/deserialization file tag"],
[],
)
)
if "network_entry" in tags:
records.append(_entry(target, None, None, None, "network_surface", "route", "public", "network_request", ["network entry file tag"], []))
records.append(
_entry(target, None, None, None, "network_surface", "route", "public", "network_request", ["network entry file tag"], [])
)
if "auth_boundary" in tags:
records.append(_entry(target, None, None, None, "auth_boundary", "privileged_surface", "authenticated", "identity_boundary", ["auth boundary file tag"], []))
records.append(
_entry(
target,
None,
None,
None,
"auth_boundary",
"privileged_surface",
"authenticated",
"identity_boundary",
["auth boundary file tag"],
[],
)
)
if "syscall_entry" in tags or "filesystem_entry" in tags:
records.append(_entry(target, None, None, None, "local_privileged_boundary", "privileged_surface", "local-only", "local_process", ["syscall/filesystem file tag"], []))
records.append(
_entry(
target,
None,
None,
None,
"local_privileged_boundary",
"privileged_surface",
"local-only",
"local_process",
["syscall/filesystem file tag"],
[],
)
)
return records
@@ -363,7 +471,9 @@ def _line_conditions(line: str) -> list[str]:
def _looks_conditional(text: str) -> bool:
lowered = text.lower()
return any(token in lowered for token in ("feature", "flag", "toggle", "enabled", "disabled", "experiment", "beta", "rollout", "paused"))
return any(
token in lowered for token in ("feature", "flag", "toggle", "enabled", "disabled", "experiment", "beta", "rollout", "paused")
)
def _js_handler_name(line: str) -> str | None:
@@ -414,7 +524,7 @@ def _rules_by_id(rules: object) -> dict[str, dict[str, object]]:
for rule in _items(rules):
rule_id = rule.get("id") if isinstance(rule, dict) else None
if isinstance(rule_id, str):
by_id[rule_id] = rule
by_id[rule_id] = cast("dict[str, object]", rule)
return by_id
src/openultrasast/preprocess.py
-1
View File
@@ -6,7 +6,6 @@ import subprocess
from dataclasses import asdict, dataclass, field
from pathlib import Path
IGNORED_DIRS = {".git", ".hg", ".svn", ".openultrasast", "__pycache__", "node_modules", ".venv", "venv"}
LANGUAGE_BY_EXTENSION = {
src/openultrasast/provider/openrouter.py
+2 -2
View File
@@ -18,7 +18,7 @@ class OpenRouterChatClient:
base_url: str = "https://openrouter.ai/api/v1"
@classmethod
def from_env(cls) -> "OpenRouterChatClient":
def from_env(cls) -> OpenRouterChatClient:
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
raise OpenRouterError("OPENROUTER_API_KEY is required for OpenRouter chat calls")
@@ -51,7 +51,7 @@ class OpenRouterEmbeddingClient:
base_url: str = "https://openrouter.ai/api/v1"
@classmethod
def from_env(cls) -> "OpenRouterEmbeddingClient":
def from_env(cls) -> OpenRouterEmbeddingClient:
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
raise OpenRouterError("OPENROUTER_API_KEY is required for OpenRouter embedding calls")
src/openultrasast/rank.py
+2 -4
View File
@@ -2,8 +2,8 @@ from __future__ import annotations
import json
from dataclasses import asdict, dataclass
from typing import Protocol
from pathlib import Path
from typing import Protocol
from .preprocess import FileTarget
@@ -231,8 +231,6 @@ def _reachability_from_hints(hints: list[dict[str, object]]) -> int:
score = max(score, 4)
elif access_level == "role-restricted":
score = max(score, 3)
elif access_level == "review-required":
score = max(score, 2)
elif access_level == "local-only":
elif access_level == "review-required" or access_level == "local-only":
score = max(score, 2)
return score
src/openultrasast/vectorstore.py
-1
View File
@@ -6,7 +6,6 @@ from pathlib import Path
from .index import VectorIndex, index_reuse_key, query_vector_index, read_vector_index, write_vector_index
from .preprocess import RepoSnapshot
SELECTED_LOCAL_VECTOR_STORE = "json-local"
src/openultrasast/verification.py
+3 -1
View File
@@ -49,7 +49,9 @@ def validate_evidence_transition(current: EvidenceLevel | str, target: EvidenceL
def is_report_verified(evidence_level: EvidenceLevel | str, status: VerificationStatus | str) -> bool:
level = EvidenceLevel(evidence_level)
verification_status = VerificationStatus(status)
return verification_status == VerificationStatus.ACCEPTED and _EVIDENCE_ORDER[level] >= _EVIDENCE_ORDER[EvidenceLevel.STATIC_CORROBORATION]
return (
verification_status == VerificationStatus.ACCEPTED and _EVIDENCE_ORDER[level] >= _EVIDENCE_ORDER[EvidenceLevel.STATIC_CORROBORATION]
)
def build_independent_verifier_context(finding: StaticFinding) -> dict[str, object]: