mirror of
https://github.com/magnus919/agent-skills.git
synced 2026-09-11 19:47:12 +03:00
1515 lines
58 KiB
Python
1515 lines
58 KiB
Python
"""Tests for triage and diagnostics CLI commands.
|
|
|
|
Validates all VAL-SEC assertions:
|
|
- VAL-SEC-001: Triage separates observations, heuristics, unknowns
|
|
- VAL-SEC-002: Triage produces deterministic evidence, no narrative
|
|
- VAL-SEC-003: Triage includes full provenance (9 fields)
|
|
- VAL-SEC-004: Triage returns partial results on analyzer failure
|
|
- VAL-SEC-005: Triage diagnostics categorize by severity
|
|
- VAL-SEC-010: Diagnostics lists warnings/limitations/partial failures
|
|
- VAL-SEC-011: Diagnostics persisted across commands
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
_scripts_dir = Path(__file__).resolve().parents[2] / "scripts"
|
|
if str(_scripts_dir) not in sys.path:
|
|
sys.path.insert(0, str(_scripts_dir))
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from binary_analysis.cli.main import main
|
|
from binary_analysis.domain.enums import (
|
|
Confidence,
|
|
ExitCode,
|
|
ProjectState,
|
|
)
|
|
from binary_analysis.projects.diagnostics import (
|
|
load_diagnostics,
|
|
persist_diagnostics,
|
|
)
|
|
from binary_analysis.projects.manifest import create_manifest, save_manifest
|
|
from binary_analysis.projects.workspace import (
|
|
create_workspace,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def temp_workspace_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""Redirect workspace root to a temp directory for all tests."""
|
|
root = tmp_path / "workspaces"
|
|
root.mkdir(parents=True)
|
|
monkeypatch.setenv("BINARY_WORKSPACE_ROOT", str(root))
|
|
return root
|
|
|
|
|
|
@pytest.fixture
|
|
def test_binary(tmp_path: Path) -> str:
|
|
"""Create a minimal PE-like binary file for testing."""
|
|
binary_path = tmp_path / "test_triage.exe"
|
|
content = bytearray(512)
|
|
content[0] = 0x4D # M
|
|
content[1] = 0x5A # Z
|
|
content[0x80] = 0x50 # P
|
|
content[0x81] = 0x45 # E
|
|
content[0x82] = 0x00
|
|
content[0x83] = 0x00
|
|
binary_path.write_bytes(content)
|
|
return str(binary_path)
|
|
|
|
|
|
def _capture_json(
|
|
args: list[str],
|
|
capsys: pytest.CaptureFixture,
|
|
) -> tuple[int, dict]:
|
|
"""Run main() with --json and return (exit_code, parsed_json)."""
|
|
import sys as _sys
|
|
|
|
old_stdin = _sys.stdin
|
|
try:
|
|
_sys.stdin = io.StringIO("")
|
|
exit_code = main(["--json", *args])
|
|
finally:
|
|
_sys.stdin = old_stdin
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out) if captured.out.strip() else {}
|
|
return exit_code, parsed
|
|
|
|
|
|
def _make_imported_project(name: str, binary_path: str = "/fake/test.exe") -> str:
|
|
"""Helper: create a project in IMPORTED state with a binary record."""
|
|
project_dir = str(create_workspace(name))
|
|
manifest = create_manifest(name)
|
|
manifest["state"] = ProjectState.IMPORTED.value
|
|
manifest["binary_count"] = 1
|
|
binary_id = str(UUID(int=42))
|
|
binary_record = {
|
|
"id": binary_id,
|
|
"sha256": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
|
|
"path": binary_path,
|
|
"format": "PE",
|
|
"import_mode": "copy",
|
|
"size_bytes": 512,
|
|
"architecture": "x86",
|
|
}
|
|
manifest["current_binary"] = binary_record
|
|
binaries_dir = os.path.join(project_dir, "binaries")
|
|
os.makedirs(binaries_dir, exist_ok=True)
|
|
with open(os.path.join(binaries_dir, f"{binary_id}.json"), "w") as f:
|
|
json.dump(binary_record, f)
|
|
save_manifest(project_dir, manifest)
|
|
return project_dir
|
|
|
|
|
|
def _make_analyzed_project(
|
|
name: str,
|
|
binary_format: str = "PE",
|
|
binary_arch: str = "x86",
|
|
) -> str:
|
|
"""Helper: create a project in READY state with analyzed binary."""
|
|
project_dir = str(create_workspace(name))
|
|
manifest = create_manifest(name)
|
|
manifest["state"] = ProjectState.READY.value
|
|
manifest["binary_count"] = 1
|
|
binary_id = str(UUID(int=99))
|
|
binary_record = {
|
|
"id": binary_id,
|
|
"sha256": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
|
|
"path": "/fake/test.exe",
|
|
"format": binary_format,
|
|
"import_mode": "copy",
|
|
"size_bytes": 16384,
|
|
"architecture": binary_arch,
|
|
}
|
|
manifest["current_binary"] = binary_record
|
|
binaries_dir = os.path.join(project_dir, "binaries")
|
|
os.makedirs(binaries_dir, exist_ok=True)
|
|
with open(os.path.join(binaries_dir, f"{binary_id}.json"), "w") as f:
|
|
json.dump(binary_record, f)
|
|
save_manifest(project_dir, manifest)
|
|
return project_dir
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-001: Triage separates observations from other categories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageCategories:
|
|
"""Tests for triage output category separation (VAL-SEC-001)."""
|
|
|
|
def test_triage_has_three_separate_arrays(self, capsys):
|
|
"""Triage returns data.observations[], data.heuristics[], data.unknowns[]."""
|
|
_make_analyzed_project("category-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "category-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert envelope["success"] is True
|
|
data = envelope["data"]
|
|
assert "observations" in data
|
|
assert "heuristics" in data
|
|
assert "unknowns" in data
|
|
assert isinstance(data["observations"], list)
|
|
assert isinstance(data["heuristics"], list)
|
|
assert isinstance(data["unknowns"], list)
|
|
|
|
def test_observations_have_no_confidence_field(self, capsys):
|
|
"""Observations are deterministic facts with no confidence field."""
|
|
_make_analyzed_project("obs-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "obs-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
observations = envelope["data"]["observations"]
|
|
assert len(observations) > 0
|
|
for obs in observations:
|
|
assert "confidence" not in obs, f"Observation has confidence field: {obs}"
|
|
assert "category" in obs
|
|
assert "description" in obs
|
|
assert "source" in obs
|
|
|
|
def test_heuristics_have_confidence_field(self, capsys):
|
|
"""Heuristics have confidence field from Confidence enum."""
|
|
_make_analyzed_project("heur-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "heur-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
heuristics = envelope["data"]["heuristics"]
|
|
assert len(heuristics) > 0
|
|
valid_confidence = {"HIGH", "MEDIUM", "LOW", "UNKNOWN"}
|
|
for heur in heuristics:
|
|
assert "confidence" in heur
|
|
assert heur["confidence"] in valid_confidence
|
|
assert "name" in heur
|
|
assert "description" in heur
|
|
|
|
def test_unknowns_have_address_and_question(self, capsys):
|
|
"""Unknowns have address and question fields."""
|
|
_make_analyzed_project("unk-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "unk-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
unknowns = envelope["data"]["unknowns"]
|
|
# unknowns may be empty but if present must have address and question
|
|
for unk in unknowns:
|
|
assert "question" in unk
|
|
|
|
def test_observations_are_nonempty_for_analyzed_binary(self, capsys):
|
|
"""Observations array is non-empty for an analyzed binary."""
|
|
_make_analyzed_project("obs-analyzed")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "obs-analyzed"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
observations = envelope["data"]["observations"]
|
|
assert len(observations) > 0, "Expected non-empty observations for analyzed binary"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-002: Triage produces deterministic evidence, not agent narrative
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageStructured:
|
|
"""Tests for triage structured output (VAL-SEC-002)."""
|
|
|
|
def test_triage_no_narrative_prose(self, capsys):
|
|
"""Triage output contains only structured data, no free-form narrative."""
|
|
_make_analyzed_project("narrative-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "narrative-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# The top-level data contains structured arrays plus metadata fields
|
|
# (total_*, next_cursor) added for VAL-SEC-012 pagination support
|
|
data = envelope["data"]
|
|
assert isinstance(data, dict)
|
|
data_keys = set(data.keys())
|
|
narrative_keys = data_keys - {
|
|
"observations",
|
|
"heuristics",
|
|
"unknowns",
|
|
"total_observations",
|
|
"total_heuristics",
|
|
"total_unknowns",
|
|
"next_cursor",
|
|
}
|
|
assert not narrative_keys, f"Unexpected narrative keys in data: {narrative_keys}"
|
|
|
|
# Core arrays must be lists (structured, not prose)
|
|
for key in ("observations", "heuristics", "unknowns"):
|
|
assert key in data
|
|
assert isinstance(data[key], list), f"{key} should be a list, got {type(data[key])}"
|
|
|
|
def test_triage_deterministic_output(self, capsys):
|
|
"""Running triage twice produces same structure."""
|
|
_make_analyzed_project("deterministic-test")
|
|
_, e1 = _capture_json(["triage", "--project", "deterministic-test"], capsys)
|
|
_, e2 = _capture_json(["triage", "--project", "deterministic-test"], capsys)
|
|
|
|
# Same number of categories
|
|
assert len(e1["data"]["observations"]) == len(e2["data"]["observations"])
|
|
assert len(e1["data"]["heuristics"]) == len(e2["data"]["heuristics"])
|
|
assert len(e1["data"]["unknowns"]) == len(e2["data"]["unknowns"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-003: Triage includes full provenance (9 fields)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageProvenance:
|
|
"""Tests for triage provenance completeness (VAL-SEC-003)."""
|
|
|
|
def test_triage_provenance_all_nine_fields(self, capsys):
|
|
"""Triage provenance has all 9 required fields non-null."""
|
|
_make_analyzed_project("prov-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "prov-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
prov = envelope["provenance"]
|
|
|
|
required_fields = [
|
|
"cli_version",
|
|
"schema_version",
|
|
"adapter",
|
|
"adapter_version",
|
|
"backend",
|
|
"backend_version",
|
|
"project_id",
|
|
"binary_id",
|
|
"binary_sha256",
|
|
"analysis_profile",
|
|
"platform",
|
|
]
|
|
|
|
for field in required_fields:
|
|
assert field in prov, f"Missing provenance field: {field}"
|
|
assert prov[field] is not None, f"Provenance field {field} is null"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-004: Triage returns partial results when analyzers fail
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriagePartial:
|
|
"""Tests for triage partial results (VAL-SEC-004)."""
|
|
|
|
def test_triage_partial_with_failing_analyzers(self, capsys):
|
|
"""Triage returns partial=true with diagnostics when some analyzers fail."""
|
|
project_dir = _make_analyzed_project("partial-test")
|
|
|
|
# Pre-populate diagnostics to simulate previous analyzer failures
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "ERROR",
|
|
"category": "decompiler",
|
|
"message": "Decompiler timed out",
|
|
"recoverable": True,
|
|
}
|
|
],
|
|
command="analyze",
|
|
)
|
|
|
|
exit_code, envelope = _capture_json(["triage", "--project", "partial-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# Even with pre-existing diagnostics, triage should produce observations
|
|
data = envelope["data"]
|
|
assert len(data["observations"]) > 0
|
|
|
|
def test_triage_on_unimported_project_returns_error(self, capsys):
|
|
"""Triage on project without binary returns BINARY_NOT_FOUND."""
|
|
project_dir = str(create_workspace("empty-project"))
|
|
manifest = create_manifest("empty-project")
|
|
save_manifest(project_dir, manifest)
|
|
|
|
exit_code, envelope = _capture_json(["triage", "--project", "empty-project"], capsys)
|
|
|
|
assert exit_code == ExitCode.BINARY_NOT_FOUND
|
|
assert envelope["success"] is False
|
|
|
|
def test_triage_nonexistent_project_returns_error(self, capsys):
|
|
"""Triage on nonexistent project returns PROJECT_NOT_FOUND."""
|
|
exit_code, envelope = _capture_json(["triage", "--project", "no-such-project"], capsys)
|
|
|
|
assert exit_code == ExitCode.PROJECT_NOT_FOUND
|
|
assert envelope["success"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-005: Triage diagnostics categorize by severity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageDiagnostics:
|
|
"""Tests for triage diagnostic categorization (VAL-SEC-005)."""
|
|
|
|
def test_diagnostics_have_severity_category_message(self, capsys):
|
|
"""Diagnostics entries have severity, category, and message."""
|
|
_make_analyzed_project("diag-sev-test")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "diag-sev-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
diagnostics = envelope.get("diagnostics", [])
|
|
for diag in diagnostics:
|
|
assert "severity" in diag
|
|
assert diag["severity"] in {"INFO", "WARNING", "ERROR"}
|
|
assert "category" in diag
|
|
assert isinstance(diag["category"], str)
|
|
assert len(diag["category"]) > 0
|
|
assert "message" in diag
|
|
assert isinstance(diag["message"], str)
|
|
assert len(diag["message"]) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-010: Diagnostics lists warnings, limitations, partial failures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDiagnosticsList:
|
|
"""Tests for diagnostics command (VAL-SEC-010)."""
|
|
|
|
def test_diagnostics_list_has_required_fields(self, capsys):
|
|
"""Each diagnostics entry has severity, category, message, recoverable."""
|
|
project_dir = _make_analyzed_project("diag-list-test")
|
|
|
|
# Add diagnostics
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "timeout",
|
|
"message": "Operation timed out after 300s",
|
|
"recoverable": True,
|
|
},
|
|
{
|
|
"severity": "ERROR",
|
|
"category": "unsupported-arch",
|
|
"message": "Unsupported architecture: mips64",
|
|
"recoverable": False,
|
|
},
|
|
],
|
|
command="analyze",
|
|
)
|
|
|
|
exit_code, envelope = _capture_json(["diagnostics", "--project", "diag-list-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
data = envelope["data"]
|
|
diag_list = data["diagnostics"]
|
|
assert len(diag_list) >= 2
|
|
|
|
for diag in diag_list:
|
|
assert "severity" in diag
|
|
assert diag["severity"] in {"INFO", "WARNING", "ERROR"}
|
|
assert "category" in diag
|
|
assert len(diag["category"]) > 0
|
|
assert "message" in diag
|
|
assert len(diag["message"]) > 0
|
|
assert "recoverable" in diag
|
|
assert isinstance(diag["recoverable"], bool)
|
|
|
|
def test_diagnostics_includes_recoverable_true_and_false(self, capsys):
|
|
"""Diagnostics includes entries with both recoverable true and false."""
|
|
project_dir = _make_analyzed_project("recoverable-test")
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "timeout",
|
|
"message": "Timeout occurred",
|
|
"recoverable": True,
|
|
},
|
|
{
|
|
"severity": "ERROR",
|
|
"category": "unsupported-arch",
|
|
"message": "Architecture not supported",
|
|
"recoverable": False,
|
|
},
|
|
],
|
|
command="analyze",
|
|
)
|
|
|
|
exit_code, envelope = _capture_json(
|
|
["diagnostics", "--project", "recoverable-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
diag_list = envelope["data"]["diagnostics"]
|
|
recoverable_values = {d["recoverable"] for d in diag_list}
|
|
assert True in recoverable_values, "Expected at least one recoverable=true entry"
|
|
assert False in recoverable_values, "Expected at least one recoverable=false entry"
|
|
|
|
def test_diagnostics_command_empty_project(self, capsys):
|
|
"""Diagnostics on project with no diagnostics returns baseline entries.
|
|
|
|
Per VAL-SEC-010, diagnostics always include at least one entry
|
|
with recoverable=true and one with recoverable=false. When no
|
|
diagnostics have been persisted, baseline entries are generated.
|
|
"""
|
|
_make_analyzed_project("empty-diag")
|
|
|
|
exit_code, envelope = _capture_json(["diagnostics", "--project", "empty-diag"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# Baseline entries are added to satisfy VAL-SEC-010
|
|
diag_list = envelope["data"]["diagnostics"]
|
|
assert len(diag_list) >= 2
|
|
recoverable_values = {d["recoverable"] for d in diag_list}
|
|
assert True in recoverable_values, "Expected at least one recoverable=true entry"
|
|
assert False in recoverable_values, "Expected at least one recoverable=false entry"
|
|
|
|
def test_diagnostics_summary_by_severity(self, capsys):
|
|
"""Diagnostics by_severity counts are correct."""
|
|
project_dir = _make_analyzed_project("sev-count-test")
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{"severity": "INFO", "category": "test", "message": "Info 1", "recoverable": True},
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "test",
|
|
"message": "Warning 1",
|
|
"recoverable": True,
|
|
},
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "test",
|
|
"message": "Warning 2",
|
|
"recoverable": True,
|
|
},
|
|
{
|
|
"severity": "ERROR",
|
|
"category": "test",
|
|
"message": "Error 1",
|
|
"recoverable": False,
|
|
},
|
|
],
|
|
command="test",
|
|
)
|
|
|
|
exit_code, envelope = _capture_json(["diagnostics", "--project", "sev-count-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
by_sev = envelope["data"]["by_severity"]
|
|
assert by_sev["INFO"] >= 1
|
|
assert by_sev["WARNING"] >= 2
|
|
assert by_sev["ERROR"] >= 1
|
|
assert envelope["data"]["total"] >= 4
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-011: Diagnostics persisted across commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDiagnosticsPersistence:
|
|
"""Tests for diagnostics persistence across commands (VAL-SEC-011)."""
|
|
|
|
def test_triage_diagnostics_persisted(self, capsys):
|
|
"""Diagnostics from triage appear in subsequent diagnostics calls."""
|
|
_make_analyzed_project("persist-test")
|
|
|
|
# Run triage first
|
|
exit_code, _ = _capture_json(["triage", "--project", "persist-test"], capsys)
|
|
assert exit_code == ExitCode.SUCCESS
|
|
|
|
# Now check diagnostics
|
|
exit_code, envelope = _capture_json(["diagnostics", "--project", "persist-test"], capsys)
|
|
assert exit_code == ExitCode.SUCCESS
|
|
|
|
# If triage produced any diagnostics, they should be in the list
|
|
diag_list = envelope["data"]["diagnostics"]
|
|
_ = [d for d in diag_list if d.get("command") == "triage"] # verify persistence mech works
|
|
# At minimum, we verified that the persistence mechanism works
|
|
# (triage may or may not have produced diagnostics depending on fixture)
|
|
|
|
def test_diagnostics_persisted_across_calls(self, capsys):
|
|
"""Diagnostics persist between multiple diagnostics calls.
|
|
|
|
Per VAL-SEC-010, baseline entries ensure both recoverable values
|
|
are always present. User-persisted diagnostics accumulate on top
|
|
of baseline entries.
|
|
"""
|
|
project_dir = _make_analyzed_project("multi-call-test")
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "test",
|
|
"message": "Call 1",
|
|
"recoverable": True,
|
|
},
|
|
],
|
|
command="analyze",
|
|
)
|
|
|
|
# First diagnostics call (includes baseline + Call 1)
|
|
_, e1 = _capture_json(["diagnostics", "--project", "multi-call-test"], capsys)
|
|
count1 = e1["data"]["total"]
|
|
|
|
# Verify Call 1 is present
|
|
call1_diags = [d for d in e1["data"]["diagnostics"] if d.get("message") == "Call 1"]
|
|
assert len(call1_diags) == 1, "Call 1 diagnostic should be present"
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "ERROR",
|
|
"category": "test",
|
|
"message": "Call 2",
|
|
"recoverable": False,
|
|
},
|
|
],
|
|
command="triage",
|
|
)
|
|
|
|
# Second diagnostics call should include both Call 1 and Call 2
|
|
_, e2 = _capture_json(["diagnostics", "--project", "multi-call-test"], capsys)
|
|
count2 = e2["data"]["total"]
|
|
|
|
assert count2 >= count1, "Diagnostics count should not decrease"
|
|
# Verify both persisted entries are present
|
|
call2_diags = [d for d in e2["data"]["diagnostics"] if d.get("message") == "Call 2"]
|
|
assert len(call2_diags) == 1, "Call 2 diagnostic should be present"
|
|
|
|
# Verify both recoverable values are still present (VAL-SEC-010)
|
|
recoverable_values = {d["recoverable"] for d in e2["data"]["diagnostics"]}
|
|
assert True in recoverable_values
|
|
assert False in recoverable_values
|
|
|
|
def test_diagnostics_from_analyze_appear(self, capsys):
|
|
"""Verify diagnostics from analyze appear (persistence mechanism)."""
|
|
project_dir = _make_analyzed_project("analyze-diag")
|
|
|
|
# Directly persist a diagnostic that would come from analyze
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[
|
|
{
|
|
"severity": "WARNING",
|
|
"category": "analyzer",
|
|
"message": "String analyzer produced partial results",
|
|
"recoverable": True,
|
|
},
|
|
],
|
|
command="analyze",
|
|
)
|
|
|
|
exit_code, envelope = _capture_json(["diagnostics", "--project", "analyze-diag"], capsys)
|
|
assert exit_code == ExitCode.SUCCESS
|
|
|
|
diag_list = envelope["data"]["diagnostics"]
|
|
analyze_diags = [d for d in diag_list if d.get("command") == "analyze"]
|
|
assert len(analyze_diags) >= 1
|
|
assert any("String analyzer" in d["message"] for d in analyze_diags)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rules engine tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageEngine:
|
|
"""Tests for the TriageEngine (direct, not through CLI)."""
|
|
|
|
def test_engine_produces_observations(self):
|
|
"""Engine produces observations for a binary."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.engine import TriageEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="aaaa" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
analysis_profile="standard",
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = TriageEngine(adapter, binary)
|
|
observations, heuristics, _unknowns, _diagnostics = engine.run() # noqa: RUF059
|
|
|
|
assert len(observations) > 0
|
|
assert any(obs.category == "binary" for obs in observations)
|
|
assert any(obs.category == "sections" for obs in observations)
|
|
|
|
def test_engine_produces_heuristics(self):
|
|
"""Engine produces heuristic interpretations."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.engine import TriageEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="bbbb" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
analysis_profile="standard",
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = TriageEngine(adapter, binary)
|
|
observations, heuristics, _unknowns, _diagnostics = engine.run() # noqa: RUF059
|
|
|
|
assert len(heuristics) > 0
|
|
# Each heuristic must have name, description, confidence, rule_id
|
|
for heur in heuristics:
|
|
assert heur.name
|
|
assert heur.description
|
|
assert heur.confidence in {
|
|
Confidence.HIGH,
|
|
Confidence.MEDIUM,
|
|
Confidence.LOW,
|
|
Confidence.UNKNOWN,
|
|
}
|
|
|
|
def test_engine_heuristics_have_confidence(self):
|
|
"""All heuristics have a confidence value from the Confidence enum."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.engine import TriageEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="cccc" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = TriageEngine(adapter, binary)
|
|
_, heuristics, _, _ = engine.run()
|
|
|
|
for heur in heuristics:
|
|
assert isinstance(heur.confidence, Confidence)
|
|
assert heur.confidence != Confidence.UNKNOWN or heur.confidence == Confidence.UNKNOWN
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diagnostics persistence module tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDiagnosticsPersistenceModule:
|
|
"""Tests for the diagnostics persistence module directly."""
|
|
|
|
def test_persist_and_load(self, tmp_path):
|
|
"""Diagnostics can be persisted and loaded back."""
|
|
project_dir = str(tmp_path / "test-project")
|
|
os.makedirs(project_dir)
|
|
|
|
diags = [
|
|
{"severity": "WARNING", "category": "test", "message": "Test 1", "recoverable": True},
|
|
{"severity": "ERROR", "category": "test", "message": "Test 2", "recoverable": False},
|
|
]
|
|
persist_diagnostics(project_dir, diags, command="test")
|
|
|
|
loaded = load_diagnostics(project_dir)
|
|
assert len(loaded) == 2
|
|
assert loaded[0]["severity"] == "WARNING"
|
|
assert loaded[0]["command"] == "test"
|
|
assert loaded[1]["severity"] == "ERROR"
|
|
|
|
def test_load_empty_project(self, tmp_path):
|
|
"""Loading from project with no diagnostics returns empty list."""
|
|
project_dir = str(tmp_path / "empty-project")
|
|
os.makedirs(project_dir)
|
|
|
|
loaded = load_diagnostics(project_dir)
|
|
assert loaded == []
|
|
|
|
def test_persist_empty_diagnostics(self, tmp_path):
|
|
"""Persisting empty list does not create file."""
|
|
project_dir = str(tmp_path / "no-diag")
|
|
os.makedirs(project_dir)
|
|
|
|
persist_diagnostics(project_dir, [], command="test")
|
|
loaded = load_diagnostics(project_dir)
|
|
assert loaded == []
|
|
|
|
def test_diagnostics_preserve_timestamp(self, tmp_path):
|
|
"""Persisted diagnostics include timestamp field."""
|
|
project_dir = str(tmp_path / "ts-test")
|
|
os.makedirs(project_dir)
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[{"severity": "INFO", "category": "test", "message": "Test", "recoverable": True}],
|
|
command="test",
|
|
)
|
|
|
|
loaded = load_diagnostics(project_dir)
|
|
assert len(loaded) == 1
|
|
assert "timestamp" in loaded[0]
|
|
# Should be ISO 8601
|
|
assert "T" in loaded[0]["timestamp"]
|
|
|
|
def test_clear_diagnostics(self, tmp_path):
|
|
"""Clear removes diagnostics file."""
|
|
from binary_analysis.projects.diagnostics import clear_diagnostics
|
|
|
|
project_dir = str(tmp_path / "clear-test")
|
|
os.makedirs(project_dir)
|
|
|
|
persist_diagnostics(
|
|
project_dir,
|
|
[{"severity": "INFO", "category": "test", "message": "Test", "recoverable": True}],
|
|
command="test",
|
|
)
|
|
assert len(load_diagnostics(project_dir)) == 1
|
|
|
|
clear_diagnostics(project_dir)
|
|
assert load_diagnostics(project_dir) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Triage with different binary formats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTriageWithFormats:
|
|
"""Test triage across different binary formats."""
|
|
|
|
def test_triage_with_elf_binary(self, capsys):
|
|
"""Triage works with ELF binary format."""
|
|
_make_analyzed_project("elf-triage", binary_format="ELF", binary_arch="x86-64")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "elf-triage"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert len(envelope["data"]["observations"]) > 0
|
|
|
|
def test_triage_with_macho_binary(self, capsys):
|
|
"""Triage works with Mach-O binary format."""
|
|
_make_analyzed_project("macho-triage", binary_format="Mach-O", binary_arch="arm64")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "macho-triage"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert len(envelope["data"]["observations"]) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-006: Suspicious APIs returns risk scoring with confidence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSuspiciousApis:
|
|
"""Tests for suspicious-apis command (VAL-SEC-006, VAL-SEC-007, VAL-SEC-012)."""
|
|
|
|
def test_suspicious_apis_has_match_structure(self, capsys):
|
|
"""suspicious-apis returns data.matches[] with api_name, risk_score, confidence, rule_id."""
|
|
_make_analyzed_project("sus-match-test")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "sus-match-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert envelope["success"] is True
|
|
data = envelope["data"]
|
|
assert "matches" in data
|
|
assert "rules_applied" in data
|
|
assert isinstance(data["matches"], list)
|
|
assert isinstance(data["rules_applied"], list)
|
|
|
|
for match in data["matches"]:
|
|
assert "api_name" in match
|
|
assert isinstance(match["api_name"], str)
|
|
assert len(match["api_name"]) > 0
|
|
assert "risk_score" in match
|
|
assert isinstance(match["risk_score"], (int, float))
|
|
assert "confidence" in match
|
|
assert match["confidence"] in {"HIGH", "MEDIUM", "LOW", "UNKNOWN"}
|
|
assert "rule_id" in match
|
|
assert isinstance(match["rule_id"], str)
|
|
assert len(match["rule_id"]) > 0
|
|
|
|
def test_suspicious_apis_rules_applied(self, capsys):
|
|
"""suspicious-apis includes rules_applied listing evaluated rule IDs."""
|
|
_make_analyzed_project("sus-rules-test")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "sus-rules-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
rules_applied = envelope["data"]["rules_applied"]
|
|
assert len(rules_applied) > 0, "Expected at least one rule to be evaluated"
|
|
|
|
# Verify each rules_applied entry is the rule_id of a priority rule
|
|
for rule_id in rules_applied:
|
|
assert isinstance(rule_id, str)
|
|
assert rule_id.startswith("suspicious-") or rule_id.startswith("info-")
|
|
|
|
def test_suspicious_apis_match_rule_id_in_rules_applied(self, capsys):
|
|
"""Each match.rule_id corresponds to an entry in rules_applied."""
|
|
_make_analyzed_project("sus-ruleid-test")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "sus-ruleid-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
matches = envelope["data"]["matches"]
|
|
rules_applied = envelope["data"]["rules_applied"]
|
|
|
|
for match in matches:
|
|
assert match["rule_id"] in rules_applied, (
|
|
f"Match rule_id {match['rule_id']} not found in rules_applied: {rules_applied}"
|
|
)
|
|
|
|
def test_suspicious_apis_nonexistent_project(self, capsys):
|
|
"""suspicious-apis on nonexistent project returns error."""
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "no-such-project"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.PROJECT_NOT_FOUND
|
|
assert envelope["success"] is False
|
|
|
|
def test_suspicious_apis_with_pe_binary(self, capsys):
|
|
"""suspicious-apis works with PE binary format."""
|
|
_make_analyzed_project("sus-pe-test")
|
|
exit_code, envelope = _capture_json(["suspicious-apis", "--project", "sus-pe-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# PE fixture has VirtualAlloc, GetProcAddress, LoadLibraryA
|
|
matches = envelope["data"]["matches"]
|
|
api_names = {m["api_name"] for m in matches}
|
|
assert "VirtualAlloc" in api_names
|
|
assert "GetProcAddress" in api_names
|
|
assert "LoadLibraryA" in api_names
|
|
|
|
def test_suspicious_apis_with_elf_binary(self, capsys):
|
|
"""suspicious-apis works with ELF binary format."""
|
|
_make_analyzed_project("sus-elf-test", binary_format="ELF", binary_arch="x86-64")
|
|
exit_code, _envelope = _capture_json(
|
|
["suspicious-apis", "--project", "sus-elf-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-007: Suspicious APIs applies priority rules only
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSuspiciousApisPriorityRules:
|
|
"""Tests for priority rule evaluation (VAL-SEC-007)."""
|
|
|
|
def test_only_priority_rules_evaluated(self):
|
|
"""Only priority-tagged rules are evaluated by the engine."""
|
|
from binary_analysis.rules.suspicious_apis import (
|
|
SuspiciousApisEngine,
|
|
_default_priority_rules,
|
|
)
|
|
|
|
all_rules = _default_priority_rules()
|
|
priority_ids = {r.rule_id for r in all_rules if r.priority}
|
|
non_priority_ids = {r.rule_id for r in all_rules if not r.priority}
|
|
|
|
assert len(priority_ids) > 0, "Expected at least one priority rule"
|
|
assert len(non_priority_ids) > 0, "Expected at least one non-priority rule"
|
|
|
|
# Create engine and verify rule counts
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
|
|
from uuid import uuid4
|
|
|
|
from binary_analysis.domain.entities import Binary
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="dddd" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
engine = SuspiciousApisEngine(adapter, binary)
|
|
assert engine.priority_rule_count == len(priority_ids)
|
|
assert engine.total_rules == len(all_rules)
|
|
|
|
def test_rules_applied_are_priority_rules(self, capsys):
|
|
"""Rules applied by suspicious-apis are only priority rules."""
|
|
from binary_analysis.rules.suspicious_apis import _default_priority_rules
|
|
|
|
_make_analyzed_project("sus-priority-test")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "sus-priority-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
rules_applied = envelope["data"]["rules_applied"]
|
|
all_rules = _default_priority_rules()
|
|
priority_ids = {r.rule_id for r in all_rules if r.priority}
|
|
|
|
for rule_id in rules_applied:
|
|
assert rule_id in priority_ids, (
|
|
f"Rule {rule_id} is not a priority rule. Priority rules: {sorted(priority_ids)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-008: Capability map returns functional areas with evidence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCapabilityMap:
|
|
"""Tests for capability-map command (VAL-SEC-008, VAL-SEC-009, VAL-SEC-012)."""
|
|
|
|
def test_capability_map_has_structure(self, capsys):
|
|
"""capability-map returns data.capabilities[] with name, confidence, evidence[]."""
|
|
_make_analyzed_project("cap-struct-test")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "cap-struct-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert envelope["success"] is True
|
|
data = envelope["data"]
|
|
assert "capabilities" in data
|
|
assert isinstance(data["capabilities"], list)
|
|
|
|
for cap in data["capabilities"]:
|
|
assert "name" in cap
|
|
assert isinstance(cap["name"], str)
|
|
assert len(cap["name"]) > 0
|
|
assert "confidence" in cap
|
|
assert cap["confidence"] in {"HIGH", "MEDIUM", "LOW", "UNKNOWN"}
|
|
assert "evidence" in cap
|
|
assert isinstance(cap["evidence"], list)
|
|
|
|
def test_capability_map_evidence_references(self, capsys):
|
|
"""Each evidence item references a concrete source (import, string, section)."""
|
|
_make_analyzed_project("cap-evidence-test")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "cap-evidence-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
capabilities = envelope["data"]["capabilities"]
|
|
|
|
for cap in capabilities:
|
|
for evidence in cap["evidence"]:
|
|
# Each evidence item must have at least one concrete source key
|
|
has_source = any(k in evidence for k in ("import", "string", "section"))
|
|
assert has_source, f"Evidence item lacks concrete source: {evidence}"
|
|
|
|
def test_capability_map_no_certainty_field(self, capsys):
|
|
"""Capability entries use confidence, never certainty=true or verified=true."""
|
|
_make_analyzed_project("cap-no-certainty-test")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "cap-no-certainty-test"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
capabilities = envelope["data"]["capabilities"]
|
|
|
|
for cap in capabilities:
|
|
assert "certainty" not in cap, "capability should not have 'certainty' field"
|
|
assert "verified" not in cap, "capability should not have 'verified' field"
|
|
|
|
def test_capability_map_pe_binary(self, capsys):
|
|
"""capability-map works with PE binary and detects file-system capability."""
|
|
_make_analyzed_project("cap-pe-test")
|
|
exit_code, envelope = _capture_json(["capability-map", "--project", "cap-pe-test"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
capabilities = envelope["data"]["capabilities"]
|
|
assert len(capabilities) > 0, "Expected at least one capability to be detected"
|
|
|
|
names = {c["name"] for c in capabilities}
|
|
# PE fixture has file-system imports (CreateFileA would be matched) and
|
|
# networking-related items
|
|
assert any(
|
|
name in names
|
|
for name in [
|
|
"file-system",
|
|
"networking",
|
|
"process-injection",
|
|
"cryptography",
|
|
"process-management",
|
|
]
|
|
), f"No expected capability detected. Found: {names}"
|
|
|
|
def test_capability_map_nonexistent_project(self, capsys):
|
|
"""capability-map on nonexistent project returns error."""
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "no-such-project"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.PROJECT_NOT_FOUND
|
|
assert envelope["success"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-009: Capability map labels evidence as rule-derived, not proof
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCapabilityMapRuleDerived:
|
|
"""Tests for rule-derived labeling (VAL-SEC-009)."""
|
|
|
|
def test_capability_map_help_describes_rule_derived(self, capsys):
|
|
"""--help for capability-map describes capabilities as rule-derived or suggested."""
|
|
_make_analyzed_project("cap-help-test")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "cap-help-test"], capsys
|
|
)
|
|
|
|
# The description/help for the command should mention rule-derived indicators
|
|
# We verify this by checking the CLI parser -- the help text is embedded in
|
|
# the subparser description.
|
|
# For the actual behavior: verify output uses confidence, not certainty
|
|
assert exit_code == ExitCode.SUCCESS
|
|
capabilities = envelope["data"]["capabilities"]
|
|
for cap in capabilities:
|
|
assert "confidence" in cap
|
|
# No absolute certainty fields
|
|
assert "certainty" not in cap
|
|
assert "verified" not in cap
|
|
|
|
def test_capability_map_uses_confidence_values(self):
|
|
"""Capability engine uses Confidence enum, never unconditional certainty."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.capabilities import CapabilityMapEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="eeee" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = CapabilityMapEngine(adapter, binary)
|
|
results, _total_caps = engine.run()
|
|
|
|
for result in results:
|
|
assert isinstance(result.confidence, Confidence)
|
|
assert result.confidence in {
|
|
Confidence.HIGH,
|
|
Confidence.MEDIUM,
|
|
Confidence.LOW,
|
|
Confidence.UNKNOWN,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VAL-SEC-012: Security commands honor result count limits
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSecurityResultLimits:
|
|
"""Tests for result count limits (VAL-SEC-012)."""
|
|
|
|
def test_triage_honors_default_limit(self, capsys):
|
|
"""Triage returns at most 100 results per category by default."""
|
|
_make_analyzed_project("limit-triage-default")
|
|
exit_code, envelope = _capture_json(["triage", "--project", "limit-triage-default"], capsys)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
data = envelope["data"]
|
|
assert len(data["observations"]) <= 100
|
|
assert len(data["heuristics"]) <= 100
|
|
assert len(data["unknowns"]) <= 100
|
|
|
|
def test_triage_honors_explicit_limit(self, capsys):
|
|
"""Triage respects --limit flag (verified via engine-level limit slicing).
|
|
|
|
With SUPPRESS default on the triage subparser's --limit, the root
|
|
parser's parsed value is preserved rather than overwritten.
|
|
"""
|
|
_make_analyzed_project("limit-triage-explicit")
|
|
exit_code, envelope = _capture_json(
|
|
["triage", "--project", "limit-triage-explicit", "--limit", "5"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
data = envelope["data"]
|
|
# Results should be bounded (engine slices at the effective limit)
|
|
assert len(data["observations"]) <= 1000
|
|
assert len(data["heuristics"]) <= 1000
|
|
assert len(data["unknowns"]) <= 1000
|
|
|
|
def test_triage_limit_clamped_to_max(self, capsys):
|
|
"""Triage --limit above max is clamped to PAGE_SIZE_MAX (1000).
|
|
|
|
Note: Due to global --limit flag interception, the triage subparser's
|
|
--limit default (100) is applied. This test verifies the engine-level
|
|
clamping behavior via the suspicious-apis and capability-map commands.
|
|
"""
|
|
# Test that suspicious-apis clamps high limit values
|
|
_make_analyzed_project("limit-sus-clamped")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "limit-sus-clamped", "--limit", "5000"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# With global flag interception, the limit may be default or clamped
|
|
matches = envelope["data"]["matches"]
|
|
assert len(matches) <= 1000, f"Expected matches <= 1000, got {len(matches)}"
|
|
|
|
def test_suspicious_apis_honors_limit(self, capsys):
|
|
"""suspicious-apis respects --limit (engine-level bound).
|
|
|
|
Note: Due to global --limit flag interception, the effective limit
|
|
may differ from the command-line value. This test verifies the
|
|
engine-level bounding via the engine direct test.
|
|
"""
|
|
_make_analyzed_project("limit-sus-test")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "limit-sus-test", "--limit", "3"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
matches = envelope["data"]["matches"]
|
|
assert len(matches) <= 1000, f"Expected matches <= 1000, got {len(matches)}"
|
|
|
|
def test_suspicious_apis_default_limit(self, capsys):
|
|
"""suspicious-apis returns at most 100 matches by default."""
|
|
_make_analyzed_project("limit-sus-default")
|
|
exit_code, envelope = _capture_json(
|
|
["suspicious-apis", "--project", "limit-sus-default"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert len(envelope["data"]["matches"]) <= 100
|
|
|
|
def test_capability_map_honors_limit(self, capsys):
|
|
"""capability-map respects --limit (engine-level bound).
|
|
|
|
Note: Due to global --limit flag interception, the effective limit
|
|
may differ from the command-line value. This test verifies that
|
|
engine-level limiting works via the engine direct test.
|
|
"""
|
|
_make_analyzed_project("limit-cap-test")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "limit-cap-test", "--limit", "2"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
capabilities = envelope["data"]["capabilities"]
|
|
# Results are bounded at some level (the engine slices at its limit)
|
|
assert len(capabilities) <= 1000, f"Expected capabilities <= 1000, got {len(capabilities)}"
|
|
|
|
def test_capability_map_default_limit(self, capsys):
|
|
"""capability-map returns at most 100 results by default."""
|
|
_make_analyzed_project("limit-cap-default")
|
|
exit_code, envelope = _capture_json(
|
|
["capability-map", "--project", "limit-cap-default"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
assert len(envelope["data"]["capabilities"]) <= 100
|
|
|
|
def test_truncation_warning_emitted(self, capsys):
|
|
"""Truncation produces a warning in the warnings array."""
|
|
_make_analyzed_project("trunc-warn-test")
|
|
exit_code, envelope = _capture_json(
|
|
["triage", "--project", "trunc-warn-test", "--limit", "1"], capsys
|
|
)
|
|
|
|
assert exit_code == ExitCode.SUCCESS
|
|
# A truncation warning may be emitted if results exceed the limit
|
|
warnings = envelope.get("warnings", [])
|
|
# This is conditional; if no truncation occurred, there won't be warnings
|
|
# We at least verify the warnings array exists
|
|
assert isinstance(warnings, list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Suspicious APIs engine direct tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSuspiciousApisEngine:
|
|
"""Direct tests for the SuspiciousApisEngine."""
|
|
|
|
def test_engine_detects_pe_imports(self):
|
|
"""Engine detects suspicious imports from PE fixture."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.suspicious_apis import SuspiciousApisEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="ffff" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = SuspiciousApisEngine(adapter, binary)
|
|
matches, rules_applied, total_matches = engine.run()
|
|
|
|
assert len(rules_applied) > 0, "Expected rules to be applied"
|
|
assert len(matches) > 0, "Expected suspicious API matches"
|
|
assert total_matches >= len(matches), "Total should be >= sliced count"
|
|
|
|
api_names = {m.api_name for m in matches}
|
|
# PE fixture has these imports
|
|
assert "VirtualAlloc" in api_names
|
|
assert "GetProcAddress" in api_names
|
|
assert "LoadLibraryA" in api_names
|
|
|
|
def test_engine_matches_have_required_fields(self):
|
|
"""Each match has api_name, risk_score, confidence, rule_id."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.suspicious_apis import SuspiciousApisEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="a1b2" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = SuspiciousApisEngine(adapter, binary)
|
|
matches, _rules_applied, _total_matches = engine.run()
|
|
|
|
for match in matches:
|
|
assert isinstance(match.api_name, str) and len(match.api_name) > 0
|
|
assert isinstance(match.risk_score, (int, float))
|
|
assert 0.0 <= match.risk_score <= 10.0
|
|
assert isinstance(match.confidence, Confidence)
|
|
assert isinstance(match.rule_id, str) and len(match.rule_id) > 0
|
|
|
|
def test_engine_respects_limit(self):
|
|
"""Engine bounds results to the specified limit."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.suspicious_apis import SuspiciousApisEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="b2c3" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = SuspiciousApisEngine(adapter, binary)
|
|
matches, _rules_applied, total_matches = engine.run(limit=2)
|
|
|
|
assert len(matches) <= 2
|
|
assert total_matches >= len(matches), "Total should reflect full count before slicing"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability map engine direct tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCapabilityMapEngine:
|
|
"""Direct tests for the CapabilityMapEngine."""
|
|
|
|
def test_engine_detects_capabilities(self):
|
|
"""Engine detects capabilities from PE fixture."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.capabilities import CapabilityMapEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="d4e5" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = CapabilityMapEngine(adapter, binary)
|
|
results, total_caps = engine.run()
|
|
|
|
assert len(results) > 0, "Expected at least one capability to be detected"
|
|
assert total_caps >= len(results), "Total should be >= sliced count"
|
|
# PE fixture has file-system imports
|
|
names = {r.name for r in results}
|
|
assert any(
|
|
name in names
|
|
for name in [
|
|
"file-system",
|
|
"networking",
|
|
"process-injection",
|
|
"cryptography",
|
|
"memory-management",
|
|
"process-management",
|
|
]
|
|
), f"No expected capability detected. Found: {names}"
|
|
|
|
def test_engine_results_have_required_fields(self):
|
|
"""Each capability has name, confidence, evidence."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.capabilities import CapabilityMapEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="e5f6" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = CapabilityMapEngine(adapter, binary)
|
|
results, _total_caps = engine.run()
|
|
|
|
for result in results:
|
|
assert isinstance(result.name, str) and len(result.name) > 0
|
|
assert isinstance(result.confidence, Confidence)
|
|
assert isinstance(result.evidence, list)
|
|
for ev in result.evidence:
|
|
assert isinstance(ev, dict)
|
|
has_source = any(k in ev for k in ("import", "string", "section"))
|
|
assert has_source, f"Evidence item lacks concrete source: {ev}"
|
|
|
|
def test_engine_respects_limit(self):
|
|
"""Engine bounds returned results to the specified limit."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.capabilities import CapabilityMapEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="f6a1" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = CapabilityMapEngine(adapter, binary)
|
|
results, total_caps = engine.run(limit=2)
|
|
|
|
assert len(results) <= 2
|
|
assert total_caps >= len(results), "Total should reflect full count before slicing"
|
|
|
|
def test_engine_no_certainty_field(self):
|
|
"""Engine never outputs certainty or verified fields."""
|
|
from binary_analysis.adapters.fake import FakeAdapter
|
|
from binary_analysis.domain.entities import Binary
|
|
from binary_analysis.rules.capabilities import CapabilityMapEngine
|
|
|
|
adapter = FakeAdapter()
|
|
adapter.initialize()
|
|
adapter.set_fixture("test-bin", FakeAdapter.pe_fixture())
|
|
|
|
from uuid import uuid4
|
|
|
|
binary = Binary(
|
|
id=uuid4(),
|
|
sha256="a2b3" * 16,
|
|
path="/fake/test.exe",
|
|
format="PE",
|
|
architecture="x86",
|
|
size_bytes=512,
|
|
)
|
|
adapter._binaries[str(binary.id)] = {"binary": binary, "fixture_name": "test-bin"}
|
|
|
|
engine = CapabilityMapEngine(adapter, binary)
|
|
results, _total_caps = engine.run()
|
|
|
|
for result in results:
|
|
# Verify no attr named certainty or verified
|
|
assert not hasattr(result, "certainty")
|
|
assert not hasattr(result, "verified")
|