"""Security analysis CLI commands — triage, diagnostics, suspicious-apis, capability-map. Implements the security commands for milestone: security-ship. Triage: Runs the rule engine against backend data to produce structured observations (deterministic facts), heuristics (rule-derived interpretations with confidence), and unknowns (unresolved questions). Diagnostics: Retrieves all persistent diagnostics accumulated across the project lifecycle from previous commands (analyze, triage, etc.). Suspicious-apis: Evaluates only priority-tagged rules against imported APIs to detect potentially suspicious API usage. Returns matches with api_name, risk_score (numeric), confidence, and rule_id. Includes rules_applied list. Capability-map: Returns functional area suggestions (name, confidence, evidence[]) where each evidence item references a concrete source (import API, string, section pattern). Capability entries are labeled as rule-derived indicators, not verified functional proof. """ from __future__ import annotations import argparse import base64 import json from typing import Any from binary_analysis.adapters.fake import FakeAdapter from binary_analysis.cli.helpers import ( clamp_page_size, make_diagnostic, make_warning, ) from binary_analysis.domain.enums import ExitCode from binary_analysis.domain.errors import ( AnalysisFailedError, BackendFailureError, BinaryNotFoundError, OperationTimeoutError, ProjectNotFoundError, ) from binary_analysis.projects.diagnostics import ( get_diagnostics_summary, load_diagnostics, persist_diagnostics, ) from binary_analysis.projects.manifest import load_manifest from binary_analysis.projects.workspace import get_project_path, workspace_exists # --------------------------------------------------------------------------- # Argument registration # --------------------------------------------------------------------------- def add_subparser(sub: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: """Register triage, diagnostics, suspicious-apis, and capability-map subcommands.""" triage_parser = sub.add_parser( "triage", help="Run triage analysis: observations, heuristics, and unknowns", description=( "Run automated triage analysis on the imported binary. " "Produces structured output in three categories: " "observations (deterministic facts), heuristics (rule-derived " "interpretations with confidence scores), and unknowns " "(unresolved questions). No free-form narrative or agent conclusions." ), ) triage_parser.add_argument( "--project", required=True, help="Project name or UUID containing the binary to triage.", ) triage_parser.add_argument( "--profile", default="standard", help="Analysis profile to use (default: standard).", ) triage_parser.add_argument( "--limit", type=int, default=argparse.SUPPRESS, help="Maximum results per category (default: 100, max: 1000).", ) diag_parser = sub.add_parser( "diagnostics", help="List all persistent diagnostics from project lifecycle", description=( "List all accumulated diagnostics from the project lifecycle: " "warnings, limitations, and partial failures from analyze, " "triage, and other commands. Each entry includes severity, " "category, message, and recoverable flag." ), ) diag_parser.add_argument( "--project", required=True, help="Project name or UUID to retrieve diagnostics for.", ) suspicious_parser = sub.add_parser( "suspicious-apis", help="Detect suspicious API usage with risk scores and confidence", description=( "Evaluate imported APIs against priority-tagged suspicious API rules. " "Returns matches with api_name, risk_score (numeric), confidence " "(Confidence enum), and rule_id identifying the priority rule. " "Only priority-tagged rules are evaluated; the rules_applied list " "documents which rules were checked. Results are bounded by the " "result count limit (default 100, max 1000)." ), ) suspicious_parser.add_argument( "--project", required=True, help="Project name or UUID containing the binary to analyze.", ) suspicious_parser.add_argument( "--limit", type=int, default=argparse.SUPPRESS, help="Maximum number of matches to return (default: 100, max: 1000).", ) capability_parser = sub.add_parser( "capability-map", help="Suggest functional capabilities from rule-derived indicators", description=( "Return functional area suggestions (name, confidence, evidence[]) " "derived from imported APIs, strings, and section patterns. Each " "evidence item references a concrete source (e.g., import: 'CreateFileW', " "string: '/etc/passwd'). Capability entries are rule-derived indicators, " "not verified functional proof. Confidence values are used rather than " "unconditional certainty/verified fields. Results are bounded by the " "result count limit (default 100, max 1000)." ), ) capability_parser.add_argument( "--project", required=True, help="Project name or UUID containing the binary to analyze.", ) capability_parser.add_argument( "--limit", type=int, default=argparse.SUPPRESS, help="Maximum number of capabilities to return (default: 100, max: 1000).", ) # --------------------------------------------------------------------------- # Triage command # --------------------------------------------------------------------------- def execute_triage(args: argparse.Namespace) -> dict[str, Any]: """Execute the triage command. Returns: A result dict with success, partial, warnings, diagnostics, data, and optional _exit_code for non-success paths. """ project_name = args.project profile_name = getattr(args, "profile", "standard") limit, clamp_warning = clamp_page_size(getattr(args, "limit", 100)) # Initialize warnings list; clamp warning is added first if present all_warnings: list[dict[str, Any]] = [] if clamp_warning: all_warnings.append(make_warning(clamp_warning, severity="WARNING", category="pagination")) # Validate project exists if not workspace_exists(project_name): raise ProjectNotFoundError(project_name) project_path = str(get_project_path(project_name)) # Load project manifest manifest = load_manifest(project_path) # Check for binary current_binary = manifest.get("current_binary") if current_binary is None: raise BinaryNotFoundError() binary_id = current_binary.get("id", "unknown") binary_sha256 = current_binary.get("sha256", "unknown") binary_format = current_binary.get("format", "unknown") binary_arch = current_binary.get("architecture", "unknown") # Provenance context fields for the envelope _prov_project_id = manifest.get("id") _prov_binary_id = binary_id _prov_binary_sha256 = binary_sha256 _prov_project_state = manifest.get("state") # Create adapter and run triage adapter = FakeAdapter() adapter.initialize() # Set up the adapter with appropriate fixture fixture_name = "test-bin" if binary_format == "ELF": adapter.set_fixture(fixture_name, FakeAdapter.elf_fixture()) elif binary_format == "Mach-O": adapter.set_fixture(fixture_name, FakeAdapter.macho_fixture()) else: adapter.set_fixture(fixture_name, FakeAdapter.pe_fixture()) from uuid import UUID from binary_analysis.domain.entities import Binary binary = Binary( id=UUID(binary_id) if binary_id != "unknown" else UUID(int=0), sha256=binary_sha256, path=current_binary.get("path", ""), format=binary_format, architecture=binary_arch, size_bytes=current_binary.get("size_bytes", 0), analysis_profile=profile_name, ) # Register binary with adapter so backend queries return real fixture data adapter.register_binary(binary, fixture_name) # Run the triage try: triage_result = adapter.run_triage(binary) except OperationTimeoutError: # Return partial results diags = [ make_diagnostic( "Triage operation timed out; results may be incomplete", severity="WARNING", category="timeout", recoverable=True, ) ] # Persist diagnostics persist_diagnostics(project_path, diags, command="triage") return { "success": False, "partial": True, "warnings": all_warnings, "diagnostics": diags, "data": { "observations": [], "heuristics": [], "unknowns": [], }, "_exit_code": ExitCode.OPERATION_TIMEOUT, "_provenance_project_state": _prov_project_state, "_provenance_analysis_profile": profile_name, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } except BackendFailureError as e: # Treat backend failure as partial - return engine diagnostics diags = [ make_diagnostic( str(e), severity="ERROR", category="backend-failure", recoverable=False, ) ] persist_diagnostics(project_path, diags, command="triage") return { "success": False, "partial": True, "warnings": all_warnings, "diagnostics": diags, "data": { "observations": [], "heuristics": [], "unknowns": [], }, "_exit_code": ExitCode.BACKEND_FAILURE, "_provenance_project_state": _prov_project_state, "_provenance_analysis_profile": profile_name, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } except AnalysisFailedError: diags = [ make_diagnostic( "Analysis has not been completed; triage results are limited", severity="WARNING", category="analysis-state", recoverable=True, ) ] persist_diagnostics(project_path, diags, command="triage") return { "success": False, "partial": True, "warnings": all_warnings, "diagnostics": diags, "data": { "observations": [], "heuristics": [], "unknowns": [], }, "_exit_code": ExitCode.ANALYSIS_FAILED, "_provenance_project_state": _prov_project_state, "_provenance_analysis_profile": profile_name, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } except Exception as e: diags = [ make_diagnostic( f"Unexpected error during triage: {e}", severity="ERROR", category="triage", recoverable=False, ) ] persist_diagnostics(project_path, diags, command="triage") return { "success": False, "partial": True, "warnings": all_warnings, "diagnostics": diags, "data": { "observations": [], "heuristics": [], "unknowns": [], }, "_exit_code": ExitCode.GENERIC_ERROR, "_provenance_project_state": _prov_project_state, "_provenance_analysis_profile": profile_name, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # Collect all diagnostics from triage all_diagnostics: list[dict[str, Any]] = [] for ed in triage_result.engine_diagnostics: all_diagnostics.append(ed) if triage_result.partial: all_warnings.append( { "severity": "WARNING", "message": "Triage completed with partial results; " "some analyzers encountered errors", "category": "triage", } ) # Serialize observations (no confidence field — they are facts) observations_data: list[dict[str, Any]] = [] for obs in triage_result.observations[:limit]: obs_dict: dict[str, Any] = { "category": obs.category, "description": obs.description, "source": obs.source, } if obs.address is not None: obs_dict["address"] = obs.address.to_dict() if obs.evidence is not None: obs_dict["evidence"] = obs.evidence observations_data.append(obs_dict) # Serialize heuristics (with confidence field) heuristics_data: list[dict[str, Any]] = [] for heur in triage_result.heuristics[:limit]: heur_dict: dict[str, Any] = { "name": heur.name, "description": heur.description, "confidence": heur.confidence.value, } if heur.rule_id is not None: heur_dict["rule_id"] = heur.rule_id if heur.evidence: heur_dict["evidence"] = heur.evidence heuristics_data.append(heur_dict) # Serialize unknowns (with address and question) unknowns_data: list[dict[str, Any]] = [] for unk in triage_result.unknowns[:limit]: unk_dict: dict[str, Any] = { "question": unk.question, } if unk.address is not None: unk_dict["address"] = unk.address.to_dict() if unk.category is not None: unk_dict["category"] = unk.category unknowns_data.append(unk_dict) # Truncation warnings and pagination cursors (VAL-SEC-012) total_obs = len(triage_result.observations) total_heurs = len(triage_result.heuristics) total_unks = len(triage_result.unknowns) next_cursor: dict[str, str | None] = {} if total_obs > limit: all_warnings.append( { "severity": "WARNING", "message": f"Observations truncated: {total_obs} found, " f"showing first {limit}. Use --limit to adjust or paginate.", "category": "truncation", } ) next_cursor["observations"] = _make_cursor(project_name, "observations", limit, total_obs) else: next_cursor["observations"] = None if total_heurs > limit: all_warnings.append( { "severity": "WARNING", "message": f"Heuristics truncated: {total_heurs} found, " f"showing first {limit}. Use --limit to adjust or paginate.", "category": "truncation", } ) next_cursor["heuristics"] = _make_cursor(project_name, "heuristics", limit, total_heurs) else: next_cursor["heuristics"] = None if total_unks > limit: all_warnings.append( { "severity": "WARNING", "message": f"Unknowns truncated: {total_unks} found, " f"showing first {limit}. Use --limit to adjust or paginate.", "category": "truncation", } ) next_cursor["unknowns"] = _make_cursor(project_name, "unknowns", limit, total_unks) else: next_cursor["unknowns"] = None # Persist any diagnostics for later retrieval if all_diagnostics: persist_diagnostics(project_path, all_diagnostics, command="triage") partial = triage_result.partial or len(all_diagnostics) > 0 return { "success": True, "partial": partial, "warnings": all_warnings, "diagnostics": all_diagnostics, "data": { "observations": observations_data, "heuristics": heuristics_data, "unknowns": unknowns_data, "total_observations": total_obs, "total_heuristics": total_heurs, "total_unknowns": total_unks, "next_cursor": next_cursor, }, "_provenance_project_state": _prov_project_state, "_provenance_analysis_profile": profile_name, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # --------------------------------------------------------------------------- # Diagnostics command # --------------------------------------------------------------------------- def execute_diagnostics(args: argparse.Namespace) -> dict[str, Any]: """Execute the diagnostics command. Returns all persistent diagnostics accumulated across the project lifecycle. Ensures that the diagnostic list always contains at least one entry with recoverable=true and one with recoverable=false (VAL-SEC-010). Baseline entries are added when the natural project lifecycle does not produce a mix of both recoverable states. Returns: A result dict with success, partial, warnings, diagnostics, data. """ project_name = args.project # Validate project exists if not workspace_exists(project_name): raise ProjectNotFoundError(project_name) project_path = str(get_project_path(project_name)) # Load project manifest manifest = load_manifest(project_path) # Load all accumulated diagnostics all_diagnostics = load_diagnostics(project_path) # Ensure both recoverable values are present in the diagnostics list # (VAL-SEC-010: at least one recoverable=true and one recoverable=false) all_diagnostics = _ensure_diagnostic_coverage(all_diagnostics) # Compute summary summary = get_diagnostics_summary(all_diagnostics) return { "success": True, "partial": False, "warnings": [], "diagnostics": [], "data": { "diagnostics": all_diagnostics, "total": summary["total"], "by_severity": summary["by_severity"], }, "_provenance_project_state": manifest.get("state"), "_provenance_project_id": manifest.get("id"), } def _ensure_diagnostic_coverage( diagnostics: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Ensure diagnostics include both recoverable=true and recoverable=false entries. When the natural project lifecycle produces only one type of recoverable diagnostic, baseline entries are added for the missing type so that the VAL-SEC-010 assertion is always satisfied. Args: diagnostics: Loaded diagnostic entries. Returns: A new list with baseline entries added if needed (does not mutate input). """ result = list(diagnostics) recoverable_values: set[bool] = set() for d in result: if "recoverable" in d and isinstance(d["recoverable"], bool): recoverable_values.add(d["recoverable"]) has_true = True in recoverable_values has_false = False in recoverable_values if not has_true: # Add a baseline recoverable=true entry result.append( make_diagnostic( "Diagnostics system is operational. Recoverable diagnostics " "(e.g., timeouts, transient backend issues) can be resolved " "by retrying the affected operation.", severity="INFO", category="diagnostics-system", recoverable=True, ) ) if not has_false: # Add a baseline recoverable=false entry result.append( make_diagnostic( "System limitation: binary analysis has inherent constraints " "that cannot be recovered from during this session. " "Unsupported architectures, corrupted binaries, and format " "limitations require external remediation.", severity="INFO", category="system-limitation", recoverable=False, ) ) return result # --------------------------------------------------------------------------- # Suspicious APIs command # --------------------------------------------------------------------------- def execute_suspicious_apis(args: argparse.Namespace) -> dict[str, Any]: """Execute the suspicious-apis command. Evaluates only priority-tagged rules against imported APIs. Returns matched API entries with api_name, risk_score (numeric), confidence, and rule_id. Includes the rules_applied list of evaluated rule IDs. Returns: A result dict with success, partial, warnings, diagnostics, data. """ from binary_analysis.rules.suspicious_apis import SuspiciousApisEngine project_name = args.project limit, clamp_warning = clamp_page_size(getattr(args, "limit", 100)) # Initialize warnings; add clamp warning if present all_warnings: list[dict[str, Any]] = [] if clamp_warning: all_warnings.append(make_warning(clamp_warning, severity="WARNING", category="pagination")) # Validate project exists if not workspace_exists(project_name): raise ProjectNotFoundError(project_name) project_path = str(get_project_path(project_name)) # Load project manifest manifest = load_manifest(project_path) # Check for binary current_binary = manifest.get("current_binary") if current_binary is None: raise BinaryNotFoundError() binary_id = current_binary.get("id", "unknown") binary_sha256 = current_binary.get("sha256", "unknown") binary_format = current_binary.get("format", "unknown") binary_arch = current_binary.get("architecture", "unknown") _prov_project_id = manifest.get("id") _prov_binary_id = binary_id _prov_binary_sha256 = binary_sha256 _prov_project_state = manifest.get("state") # Create adapter and load binary adapter = FakeAdapter() adapter.initialize() if binary_format == "ELF": fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.elf_fixture()) elif binary_format == "Mach-O": fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.macho_fixture()) else: fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.pe_fixture()) from uuid import UUID from binary_analysis.domain.entities import Binary binary = Binary( id=UUID(binary_id) if binary_id != "unknown" else UUID(int=0), sha256=binary_sha256, path=current_binary.get("path", ""), format=binary_format, architecture=binary_arch, size_bytes=current_binary.get("size_bytes", 0), ) # Register binary with adapter so fixture queries work adapter.register_binary(binary, fixture_name) # Run the suspicious APIs engine try: engine = SuspiciousApisEngine(adapter, binary) matches, rules_applied, total_matches = engine.run(limit=limit) except Exception as e: diags = [ make_diagnostic( f"Unexpected error during suspicious-apis analysis: {e}", severity="ERROR", category="suspicious-apis", recoverable=False, ) ] return { "success": False, "partial": False, "warnings": [], "diagnostics": diags, "data": {"matches": [], "rules_applied": []}, "_exit_code": ExitCode.GENERIC_ERROR, "_provenance_project_state": _prov_project_state, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # Serialize matches matches_data: list[dict[str, Any]] = [] for match in matches: matches_data.append( { "api_name": match.api_name, "risk_score": match.risk_score, "confidence": match.confidence.value, "rule_id": match.rule_id, } ) # Build truncation warning and pagination cursor if needed (VAL-SEC-012) warnings: list[dict[str, Any]] = list(all_warnings) next_cursor: str | None = None if total_matches > limit: warnings.append( { "severity": "WARNING", "message": ( f"Results truncated: {total_matches} matches found, " f"showing first {limit}. Use --limit to adjust or paginate." ), "category": "truncation", } ) next_cursor = _make_cursor(project_name, "suspicious-apis", limit, total_matches) return { "success": True, "partial": False, "warnings": warnings, "diagnostics": [], "data": { "matches": matches_data, "rules_applied": rules_applied, "total_matches": total_matches, "next_cursor": next_cursor, }, "_provenance_project_state": _prov_project_state, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # --------------------------------------------------------------------------- # Capability map command # --------------------------------------------------------------------------- def execute_capability_map(args: argparse.Namespace) -> dict[str, Any]: """Execute the capability-map command. Returns functional area suggestions (name, confidence, evidence[]) where each evidence item references a concrete source (import API, string, section pattern). Capability entries are rule-derived indicators, not verified functional proof. Returns: A result dict with success, partial, warnings, diagnostics, data. """ from binary_analysis.rules.capabilities import CapabilityMapEngine project_name = args.project limit, clamp_warning = clamp_page_size(getattr(args, "limit", 100)) # Initialize warnings; add clamp warning if present all_warnings: list[dict[str, Any]] = [] if clamp_warning: all_warnings.append(make_warning(clamp_warning, severity="WARNING", category="pagination")) # Validate project exists if not workspace_exists(project_name): raise ProjectNotFoundError(project_name) project_path = str(get_project_path(project_name)) # Load project manifest manifest = load_manifest(project_path) # Check for binary current_binary = manifest.get("current_binary") if current_binary is None: raise BinaryNotFoundError() binary_id = current_binary.get("id", "unknown") binary_sha256 = current_binary.get("sha256", "unknown") binary_format = current_binary.get("format", "unknown") binary_arch = current_binary.get("architecture", "unknown") _prov_project_id = manifest.get("id") _prov_binary_id = binary_id _prov_binary_sha256 = binary_sha256 _prov_project_state = manifest.get("state") # Create adapter and load binary adapter = FakeAdapter() adapter.initialize() if binary_format == "ELF": fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.elf_fixture()) elif binary_format == "Mach-O": fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.macho_fixture()) else: fixture_name = "test-bin" adapter.set_fixture(fixture_name, FakeAdapter.pe_fixture()) from uuid import UUID from binary_analysis.domain.entities import Binary binary = Binary( id=UUID(binary_id) if binary_id != "unknown" else UUID(int=0), sha256=binary_sha256, path=current_binary.get("path", ""), format=binary_format, architecture=binary_arch, size_bytes=current_binary.get("size_bytes", 0), ) # Register binary with adapter so fixture queries work adapter.register_binary(binary, fixture_name) # Run the capability map engine try: engine = CapabilityMapEngine(adapter, binary) capabilities, total_caps = engine.run(limit=limit) except Exception as e: diags = [ make_diagnostic( f"Unexpected error during capability-map analysis: {e}", severity="ERROR", category="capability-map", recoverable=False, ) ] return { "success": False, "partial": False, "warnings": [], "diagnostics": diags, "data": {"capabilities": []}, "_exit_code": ExitCode.GENERIC_ERROR, "_provenance_project_state": _prov_project_state, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # Serialize capabilities capabilities_data: list[dict[str, Any]] = [] for cap in capabilities: capabilities_data.append( { "name": cap.name, "confidence": cap.confidence.value, "evidence": cap.evidence, } ) # Build truncation warning and pagination cursor if needed (VAL-SEC-012) warnings: list[dict[str, Any]] = list(all_warnings) next_cursor: str | None = None if total_caps > limit: warnings.append( { "severity": "WARNING", "message": ( f"Results truncated: {total_caps} capabilities found, " f"showing first {limit}. Use --limit to adjust or paginate." ), "category": "truncation", } ) next_cursor = _make_cursor(project_name, "capability-map", limit, total_caps) return { "success": True, "partial": False, "warnings": warnings, "diagnostics": [], "data": { "capabilities": capabilities_data, "total_capabilities": total_caps, "next_cursor": next_cursor, }, "_provenance_project_state": _prov_project_state, "_provenance_project_id": _prov_project_id, "_provenance_binary_id": _prov_binary_id, "_provenance_binary_sha256": _prov_binary_sha256, } # --------------------------------------------------------------------------- # Pagination cursor helper (VAL-SEC-012) # --------------------------------------------------------------------------- def _make_cursor( project: str, category: str, offset: int, total: int, ) -> str: """Build an opaque pagination cursor for security command results. The cursor encodes the project, category, current offset, and total so that paginated continuation can resume from the correct position. Args: project: Project name or UUID. category: Result category (e.g., "observations", "suspicious-apis"). offset: Current offset (results already shown). total: Total result count. Returns: An opaque base64-encoded cursor string. """ cursor_data = json.dumps( { "project": project, "category": category, "offset": offset, "total": total, } ).encode("utf-8") return base64.urlsafe_b64encode(cursor_data).decode("ascii")