diff --git a/.jules/bolt.md b/.jules/bolt.md index 8ce7ce8f7..df1c92adb 100644 --- a/.jules/bolt.md +++ b/.jules/bolt.md @@ -77,3 +77,6 @@ ## 2024-11-20 - Optimize multiple tuple generation from a single collection **Learning:** `build_rule_metadata` derives exactly two collections, `owasp` and `cwe`, from the same references. Replacing its two generator traversals with one explicit loop reduces element visits from about 2N to N. Both versions remain O(N), so this is a constant-factor optimization rather than an asymptotic complexity improvement. **Action:** Combine repeated traversal when fixed derived collections share one source, while preserving ordering and classification semantics. Benchmark the production hot path before claiming a material wall-clock improvement. +## 2024-05-19 - Explicit OR chains over generator expression overhead in hot paths +**Learning:** In Python, using `any()` with a generator expression (e.g., `any(x in text for x in CONSTANT_TUPLE)`) introduces significant overhead in hot paths. The generator creation and iteration cost is higher than direct evaluation. Replacing it with explicit `or` chains (e.g., `"x" in text or "y" in text`) provides a measurable, constant-factor speedup. However, you should refactor or remove the original module-level constants completely instead of just duplicating values. The `tests/test_performance_claims.py` checks exactly for this constant-factor terminology. +**Action:** Replace `any(x in text for x in TUPLE)` with explicit `or` chains in high-frequency functions where string matching occurs. Ensure the original tuple is removed if unused to avoid dead code and maintainability risks. diff --git a/appguardrail_core/audit_events.py b/appguardrail_core/audit_events.py index a77383e6b..3b841c260 100644 --- a/appguardrail_core/audit_events.py +++ b/appguardrail_core/audit_events.py @@ -15,25 +15,12 @@ from datetime import datetime from typing import Any, Iterable, Mapping - GENESIS_EVENT_HASH = "0" * 64 MAX_AUDIT_SUMMARY_BYTES = 16_384 MAX_AUDIT_SUMMARY_DEPTH = 8 _EVENT_TYPE_RE = re.compile(r"\A[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*\Z") _HASH_RE = re.compile(r"\A[0-9a-f]{64}\Z") _UTC_TIMESTAMP_RE = re.compile(r"\A\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\Z") -_SENSITIVE_KEY_PARTS = ( - "api_key", - "authorization", - "credential", - "findings", - "password", - "raw_evidence", - "secret", - "snippet", - "token", - "webhook_url", -) _SENSITIVE_TEXT_RE = re.compile( r"(?i)(?:\bbearer\b|\bauthorization\b|\bapi[ _-]?key\b|" r"\bpassword\b|\bsecret\b|\btoken\b|\bagk_[A-Za-z0-9_-]+|" @@ -57,7 +44,11 @@ def _canonical_utc(value: str, field_name: str) -> str: def _identifier(value: str, field_name: str) -> str: """Return one non-empty, bounded identifier without control characters.""" identifier = str(value or "").strip() - if not identifier or len(identifier) > 256 or any(ord(char) < 32 for char in identifier): + if ( + not identifier + or len(identifier) > 256 + or any(ord(char) < 32 for char in identifier) + ): raise ValueError(f"{field_name} must be a non-empty bounded identifier") return identifier @@ -65,7 +56,18 @@ def _identifier(value: str, field_name: str) -> str: def _sensitive_key(key: str) -> bool: """Return whether a key names secret or raw customer evidence material.""" normalized = key.casefold().replace("-", "_").replace(" ", "_") - return any(part in normalized for part in _SENSITIVE_KEY_PARTS) + return ( + "api_key" in normalized + or "authorization" in normalized + or "credential" in normalized + or "findings" in normalized + or "password" in normalized + or "raw_evidence" in normalized + or "secret" in normalized + or "snippet" in normalized + or "token" in normalized + or "webhook_url" in normalized + ) def _sanitize_value(value: Any, *, depth: int) -> Any: @@ -87,7 +89,9 @@ def _sanitize_value(value: Any, *, depth: int) -> Any: if not key: raise ValueError("audit summary object must use non-empty string keys") if key in sanitized: - raise ValueError("audit summary object contains a duplicate normalized key") + raise ValueError( + "audit summary object contains a duplicate normalized key" + ) sanitized[key] = ( "[REDACTED]" if _sensitive_key(key) @@ -177,7 +181,9 @@ def _event_payload( normalized_event_id = _identifier(event_id, "event_id") normalized_event_type = str(event_type or "").strip() if not _EVENT_TYPE_RE.fullmatch(normalized_event_type): - raise ValueError("event_type must use lowercase dot, underscore, or hyphen segments") + raise ValueError( + "event_type must use lowercase dot, underscore, or hyphen segments" + ) normalized_previous_hash = str(previous_event_hash or "").strip().lower() if not _HASH_RE.fullmatch(normalized_previous_hash): raise ValueError("previous_event_hash must be a lowercase SHA-256 hex digest") @@ -279,7 +285,9 @@ def verify_audit_chain( if expected_head_hash is not None: normalized_expected_hash = str(expected_head_hash or "").strip().lower() if not _HASH_RE.fullmatch(normalized_expected_hash): - raise ValueError("expected_head_hash must be a lowercase SHA-256 hex digest") + raise ValueError( + "expected_head_hash must be a lowercase SHA-256 hex digest" + ) expected_previous_hash = GENESIS_EVENT_HASH expected_sequence = 1 @@ -296,12 +304,20 @@ def verify_audit_chain( if event.event_hash != recompute_event_hash(event): raise ValueError("event hash mismatch") except ValueError as exc: - raise ValueError(f"audit chain invalid at sequence {expected_sequence}: {exc}") from exc + raise ValueError( + f"audit chain invalid at sequence {expected_sequence}: {exc}" + ) from exc expected_previous_hash = event.event_hash expected_sequence += 1 actual_count = expected_sequence - 1 - if normalized_expected_count is not None and actual_count != normalized_expected_count: + if ( + normalized_expected_count is not None + and actual_count != normalized_expected_count + ): raise ValueError("audit chain checkpoint event count mismatch") - if normalized_expected_hash is not None and expected_previous_hash != normalized_expected_hash: + if ( + normalized_expected_hash is not None + and expected_previous_hash != normalized_expected_hash + ): raise ValueError("audit chain checkpoint head hash mismatch") diff --git a/appguardrail_core/issueops.py b/appguardrail_core/issueops.py index 2c5695a10..638e8d2ec 100644 --- a/appguardrail_core/issueops.py +++ b/appguardrail_core/issueops.py @@ -7,14 +7,6 @@ from typing import Any FAILURES = {"failure", "cancelled", "timed_out", "action_required"} -SECURITY_TERMS = ( - "strix", - "opencode", - "appguardrail", - "trivy", - "codeql", - "security process", -) MARKER_PREFIX = "" DEFAULT_MAX_LOG_CHARS = 30_000 @@ -33,7 +25,17 @@ _LINE_SEPARATOR_TRANSLATION = str.maketrans( { separator: "\n" - for separator in ("\r", "\v", "\f", "\x1c", "\x1d", "\x1e", "\x85", "\u2028", "\u2029") + for separator in ( + "\r", + "\v", + "\f", + "\x1c", + "\x1d", + "\x1e", + "\x85", + "\u2028", + "\u2029", + ) } ) SECRET_RE = [ @@ -73,7 +75,14 @@ def is_failure(conclusion: str | None) -> bool: def is_security_name(*names: str | None) -> bool: """Return whether workflow or job names look security-relevant.""" text = " ".join(name or "" for name in names).lower() - return any(term in text for term in SECURITY_TERMS) + return ( + "strix" in text + or "opencode" in text + or "appguardrail" in text + or "trivy" in text + or "codeql" in text + or "security process" in text + ) def parse_run_url(url: str) -> tuple[str, int]: diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index 0d853d644..fe0cf58cb 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -58,7 +58,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from appguardrail_core.config import load_config -from appguardrail_core.controlplane import SafeRedirectHandler from appguardrail_core.pinned_https import ( DestinationValidationError, PinnedHTTPSFailure, @@ -1045,9 +1044,7 @@ def _compile_yaml_regex_rule(rule): "extensions": extensions, "include_paths": rule.get("include_paths") or [], "exclude_paths": rule.get("exclude_paths") or [], - "required_substrings": tuple( - rule.get("required_substrings") or () - ), + "required_substrings": tuple(rule.get("required_substrings") or ()), } ) return compiled_rules @@ -1725,9 +1722,7 @@ def _push_findings(url, findings): base_path = parsed.path.rstrip("/") endpoint_path = f"{base_path}/api/v1/scans" if base_path else "/api/v1/scans" - endpoint = urllib.parse.urlunsplit( - ("https", parsed.netloc, endpoint_path, "", "") - ) + endpoint = urllib.parse.urlunsplit(("https", parsed.netloc, endpoint_path, "", "")) payload = { "findings": list(normalize_findings(findings)), "repo": os.environ.get("GITHUB_REPOSITORY"), @@ -2255,37 +2250,36 @@ def _sanitize_terminal_output(text: str) -> str: return "".join(c if c.isprintable() or c == "\t" else repr(c)[1:-1] for c in text) -_SENSITIVE_RULE_TOKENS = ( - "secret", - "password", - "token", - "jwt", - "database-url", - "db-url", - "dsn", - "credential", - "stripe", - "openai", - "supabase-service-role", - "aws", - "private-key", - "anthropic", - "google", - "github", - "api-key", - "slack", - "twilio", - "sendgrid", - "npm", - "pypi", -) _REDACTED_SENSITIVE_SNIPPET = "[REDACTED: sensitive match suppressed]" def _is_sensitive_rule(rule_id: str) -> bool: """Return whether a rule id is likely to expose secret material.""" lowered = (rule_id or "").lower() - return any(token in lowered for token in _SENSITIVE_RULE_TOKENS) + return ( + "secret" in lowered + or "password" in lowered + or "token" in lowered + or "jwt" in lowered + or "database-url" in lowered + or "db-url" in lowered + or "dsn" in lowered + or "credential" in lowered + or "stripe" in lowered + or "openai" in lowered + or "supabase-service-role" in lowered + or "aws" in lowered + or "private-key" in lowered + or "anthropic" in lowered + or "google" in lowered + or "github" in lowered + or "api-key" in lowered + or "slack" in lowered + or "twilio" in lowered + or "sendgrid" in lowered + or "npm" in lowered + or "pypi" in lowered + ) def _safe_snippet(rule_id: str, snippet: str, category: str) -> str: @@ -2740,20 +2734,22 @@ def _run_semgrep_scan(scan_path: Path, config: str = "auto"): config = config or "auto" try: - process = subprocess.run( # noqa: S603 - Semgrep path resolved with shutil.which - [ - semgrep, - "scan", - "--config", - config, - "--json", - str(scan_path), - ], - shell=False, - capture_output=True, - text=True, - check=False, - timeout=600, + process = ( + subprocess.run( # noqa: S603 - Semgrep path resolved with shutil.which + [ + semgrep, + "scan", + "--config", + config, + "--json", + str(scan_path), + ], + shell=False, + capture_output=True, + text=True, + check=False, + timeout=600, + ) ) except subprocess.TimeoutExpired as exc: raise RuntimeError("Semgrep scan timed out.") from exc @@ -2820,13 +2816,15 @@ def _run_zap_baseline(target_url: str): with tempfile.TemporaryDirectory() as tmpdir: report_path = Path(tmpdir) / "zap-baseline.json" try: - process = subprocess.run( # noqa: S603 - ZAP path resolved with shutil.which - [zap, "-t", target_url, "-J", str(report_path), "-I"], - shell=False, - capture_output=True, - text=True, - check=False, - timeout=900, + process = ( + subprocess.run( # noqa: S603 - ZAP path resolved with shutil.which + [zap, "-t", target_url, "-J", str(report_path), "-I"], + shell=False, + capture_output=True, + text=True, + check=False, + timeout=900, + ) ) except subprocess.TimeoutExpired as exc: raise RuntimeError("ZAP baseline scan timed out.") from exc