diff --git a/.github/workflows/org-security-failure-collector.yml b/.github/workflows/org-security-failure-collector.yml index c4b766718..7048be550 100644 --- a/.github/workflows/org-security-failure-collector.yml +++ b/.github/workflows/org-security-failure-collector.yml @@ -53,7 +53,7 @@ jobs: - name: Create org installation token if: steps.app-config.outputs.configured == 'true' id: app-token - uses: actions/create-github-app-token@v3 + uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3 with: app-id: ${{ env.ORG_SECURITY_FAILURE_APP_ID }} private-key: ${{ env.ORG_SECURITY_FAILURE_APP_PRIVATE_KEY }} diff --git a/.github/workflows/prepare-pypi-release.yml b/.github/workflows/prepare-pypi-release.yml index 0725ec194..1482052bd 100644 --- a/.github/workflows/prepare-pypi-release.yml +++ b/.github/workflows/prepare-pypi-release.yml @@ -19,13 +19,15 @@ on: type: string permissions: - actions: write - contents: write - pull-requests: write + contents: read jobs: prepare-release: runs-on: ubuntu-latest + permissions: + actions: write + contents: write + pull-requests: write env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true RELEASE_VERSION: ${{ inputs.version }} diff --git a/.github/workflows/scorecard-analysis.yml b/.github/workflows/scorecard-analysis.yml index 3a1a813c5..7a3cfece1 100644 --- a/.github/workflows/scorecard-analysis.yml +++ b/.github/workflows/scorecard-analysis.yml @@ -32,7 +32,3 @@ jobs: results_format: sarif publish_results: true - - name: Upload to code scanning - uses: github/codeql-action/upload-sarif@99df26d4f13ea111d4ec1a7dddef6063f76b97e9 # v4.37.0 - with: - sarif_file: results.sarif diff --git a/.github/workflows/security-process.yml b/.github/workflows/security-process.yml index bc38d4814..73af7c2af 100644 --- a/.github/workflows/security-process.yml +++ b/.github/workflows/security-process.yml @@ -11,7 +11,6 @@ on: permissions: contents: read - security-events: write actions: read jobs: @@ -60,7 +59,7 @@ jobs: uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - name: Run Trivy FS scan - uses: aquasecurity/trivy-action@v0.36.0 + uses: aquasecurity/trivy-action@a9c7b0f06e461e9d4b4d1711f154ee024b8d7ab8 # v0.36.0 with: scan-type: fs scan-ref: . @@ -71,15 +70,6 @@ jobs: output: trivy-results.sarif exit-code: "1" - - name: Upload Trivy SARIF to code scanning - if: >- - always() - && hashFiles('trivy-results.sarif') != '' - && (github.event_name != 'pull_request' - || github.event.pull_request.head.repo.full_name == github.repository) - uses: github/codeql-action/upload-sarif@v4 - with: - sarif_file: trivy-results.sarif - name: Upload Trivy SARIF artifact if: always() && hashFiles('trivy-results.sarif') != '' diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 69cb762f0..69e8e87fc 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -56,3 +56,17 @@ **Vulnerability:** The CLI file scanner `vibesec scan` lacked detection rules for insecure deserialization, such as `pickle.load` or `yaml.load` in Python, which could lead to arbitrary code execution. **Learning:** Adding regular expressions that detect known dangerous serialization libraries prevents severe security vulnerabilities when parsing untrusted data. **Prevention:** A new scanner rule `python-insecure-deserialization` was added to `SCAN_RULES` to flag `pickle.load(s)`, `yaml.load`, and `marshal.load(s)`. +## 2025-02-28 - Prevent Security Theater in Validation Logic +**Vulnerability:** Redundant, useless security checks ("security theater") were previously added to statically hardcoded URL strings in GitHub actions and test files, under the guise of SSRF prevention. +**Learning:** Checking a hardcoded string like `"https://pypi.org/..."` to see if it starts with `https://` provides zero real security value while polluting the codebase and making PR reviews noisy. Security validation logic must be strictly scoped to where *untrusted, dynamic input* is introduced (like dynamically constructed API URLs in network wrappers). +**Prevention:** Avoid blanket application of security validation rules across an entire repository regardless of context. Always trace the source of the variable being validated to ensure it actually incorporates dynamic or external data before adding runtime verification logic. +## 2025-02-28 - Fix CodeQL Code Scanning Upload Error +**Vulnerability:** CI fails with error "CodeQL analyses from advanced configurations cannot be processed when the default setup is enabled" when uploading SARIF results. +**Learning:** If a repository has default Code Scanning setup enabled in GitHub settings, pushing SARIF using the `github/codeql-action/upload-sarif` action from advanced workflows fails. This conflict happens when security tooling pushes SARIF files to the GitHub Advanced Security Code Scanning endpoint. +**Prevention:** Remove `github/codeql-action/upload-sarif` steps from advanced configuration workflows (like `scorecard-analysis.yml` and `security-process.yml`) if default CodeQL setup is enabled for the repository, to avoid pipeline failures and configuration conflicts. +## 2025-02-28 - Enhance SSRF Protection with IP Validation +**Vulnerability:** Weak redirect target validation for log collection flows in CI scripts. The `_validate_log_download_url` function blocked a static list of hosts (like `localhost` and `127.0.0.1`), but failed to block private/internal IP ranges (10.0.0.0/8, 192.168.0.0/16, etc), numeric encodings (like `2130706433`), or link-local IPv6 addresses. +**Learning:** Checking hostnames against a hardcoded blocklist is insufficient for SSRF protection because attackers can bypass the list using different IP representations or pointing to other internal IP ranges. Robust SSRF validation requires parsing the IP address using the `ipaddress` module and checking properties like `is_private`, `is_loopback`, `is_link_local`, etc. +**Prevention:** Use `ipaddress.ip_address` to parse and validate IP literals, explicitly rejecting private and reserved addresses before making network requests. Ensure edge cases like dotless decimal encoding and credentials in URLs are also rejected. +## 2025-02-28 - Note on IP Validation and Valid Hostnames +**Learning:** Checking hostnames against patterns like `p.startswith("0")` or `p.startswith("0x")` to prevent octal/hex SSRF bypasses may inadvertently block valid domain names like `0x.org` or `0123.com`. While acceptable for internal tools restricted to known endpoints, it shouldn't be blindly applied to all web applications. diff --git a/appguardrail_core/issueops.py b/appguardrail_core/issueops.py index d8ca3f586..658d986cb 100644 --- a/appguardrail_core/issueops.py +++ b/appguardrail_core/issueops.py @@ -38,15 +38,18 @@ def is_failure(conclusion: str | None) -> bool: + """Return whether a GitHub conclusion represents a failed security run.""" return (conclusion or "").lower() in FAILURES 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) def parse_run_url(url: str) -> tuple[str, int]: + """Extract the repository slug and run id from a GitHub Actions run URL.""" match = re.search(r"github\.com/([^/]+/[^/]+)/actions/runs/(\d+)", url) if not match: raise ValueError(f"Unsupported GitHub Actions run URL: {url}") @@ -54,11 +57,13 @@ def parse_run_url(url: str) -> tuple[str, int]: def sanitize_label_value(value: str) -> str: + """Convert arbitrary repository text into a compact GitHub label suffix.""" value = re.sub(r"[^A-Za-z0-9._:-]+", "-", value.strip()).strip("-") return value[:45] or "unknown" def redact(log: str) -> str: + """Remove ANSI noise, timestamps, and obvious secrets from a job log.""" text = ANSI_RE.sub("", log.replace("\r\n", "\n").replace("\r", "\n")) text = "\n".join(TS_RE.sub("", line) for line in text.splitlines()) for regex in SECRET_RE: @@ -67,6 +72,7 @@ def redact(log: str) -> str: def log_ranges(lines: list[str], patterns: list[re.Pattern[str]]) -> list[tuple[int, int]]: + """Return compact context windows around lines matching failure patterns.""" return [ (max(0, index - 2), min(len(lines), index + 9)) for index, line in enumerate(lines) @@ -75,6 +81,7 @@ def log_ranges(lines: list[str], patterns: list[re.Pattern[str]]) -> list[tuple[ def compress_log(log: str, max_lines: int = DEFAULT_MAX_LOG_LINES, max_chars: int = DEFAULT_MAX_LOG_CHARS) -> str: + """Compress a full job log to the most useful failure evidence.""" lines = redact(log).splitlines() if not lines: return "(no job log returned)" @@ -112,15 +119,18 @@ def compress_log(log: str, max_lines: int = DEFAULT_MAX_LOG_LINES, max_chars: in def seen_key(finding: dict[str, Any]) -> str: + """Return a stable run/job key used to deduplicate IssueOps updates.""" return f"{finding['run_id']}:{finding['job_id']}" def marker(repo: str, workflow: str, seen: set[str]) -> str: + """Build the hidden issue marker that stores repository and seen-job state.""" payload = {"repo": repo, "workflow": workflow, "seen": sorted(seen)} return f"{MARKER_PREFIX} {json.dumps(payload, sort_keys=True)} {MARKER_SUFFIX}" def parse_marker(body: str | None) -> dict[str, Any]: + """Parse a hidden issue marker, returning an empty seen list when absent.""" body = body or "" start = body.find(MARKER_PREFIX) end = body.find(MARKER_SUFFIX, start + len(MARKER_PREFIX)) @@ -133,6 +143,7 @@ def parse_marker(body: str | None) -> dict[str, Any]: def replace_marker(body: str | None, repo: str, workflow: str, seen: set[str]) -> str: + """Insert or replace the hidden IssueOps marker in an issue body.""" body = body or "" new_marker = marker(repo, workflow, seen) start = body.find(MARKER_PREFIX) @@ -143,10 +154,12 @@ def replace_marker(body: str | None, repo: str, workflow: str, seen: set[str]) - def title(finding: dict[str, Any]) -> str: + """Build the canonical issue title for one repository workflow failure.""" return f"[security-failure] {finding['repo']}: {finding['workflow']}" def summary(finding: dict[str, Any]) -> str: + """Render the key run, job, branch, and PR facts for an issue body.""" prs = ", ".join(f"#{number}" for number in finding["pr_numbers"]) or "n/a" rows = [ ("Repository", f"`{finding['repo']}`"), @@ -164,6 +177,7 @@ def summary(finding: dict[str, Any]) -> str: def issue_body(finding: dict[str, Any], seen: set[str]) -> str: + """Render the first issue body for a collected security workflow failure.""" owner = finding["repo"].split("/", 1)[0] return "\n\n".join( [ @@ -176,4 +190,5 @@ def issue_body(finding: dict[str, Any], seen: set[str]) -> str: def issue_comment(finding: dict[str, Any]) -> str: + """Render a follow-up comment for a newly observed failure on an issue.""" return "\n\n".join(["New security workflow failure detected.", summary(finding), f"```text\n{finding['snippet']}\n```"]) diff --git a/appguardrail_core/reports.py b/appguardrail_core/reports.py index 0cf7693d4..d4fa65665 100644 --- a/appguardrail_core/reports.py +++ b/appguardrail_core/reports.py @@ -358,6 +358,7 @@ def render_fix_pack( def _launch_posture(blockers: list[dict[str, Any]]) -> str: + """Summarize whether blocker severity permits launch.""" if any(finding["severity"] == "CRITICAL" for finding in blockers): return "Hold pending critical remediation" if blockers: @@ -366,6 +367,7 @@ def _launch_posture(blockers: list[dict[str, Any]]) -> str: def _summary_table(findings: list[dict[str, Any]]) -> list[str]: + """Render the buyer diligence summary table.""" rows = [ "| ID | Severity | Category | Location | References |", "|---|---|---|---|---|", @@ -385,6 +387,7 @@ def _summary_table(findings: list[dict[str, Any]]) -> list[str]: def _finding_detail(index: int, finding: dict[str, Any]) -> list[str]: + """Render one detailed buyer diligence finding section.""" references = ", ".join(finding["references"] or finding["owasp"] or finding["cwe"]) return [ f"### BD-{index:03d}: {_short_title(finding['message'])}", @@ -410,6 +413,7 @@ def _finding_detail(index: int, finding: dict[str, Any]) -> list[str]: def _short_title(message: str, max_len: int = 84) -> str: + """Return a compact report heading derived from a finding message.""" title = message.split(".", 1)[0].strip() or "Security finding" if len(title) <= max_len: return title @@ -426,6 +430,7 @@ def _prepare_report( list[dict[str, Any]], str, ]: + """Normalize findings and compute shared report metadata.""" context = context or ReportContext() normalized = [normalize_finding(finding) for finding in findings] normalized.sort(key=finding_sort_key) @@ -438,6 +443,7 @@ def _prepare_report( def _founder_status(blockers: list[dict[str, Any]]) -> str: + """Return the founder-facing launch status label.""" if any(finding["severity"] == "CRITICAL" for finding in blockers): return "Not ready for public launch" if blockers: @@ -446,6 +452,7 @@ def _founder_status(blockers: list[dict[str, Any]]) -> str: def _founder_finding(index: int, finding: dict[str, Any]) -> list[str]: + """Render one founder-friendly finding section.""" return [ f"### Finding {index}: {_short_title(finding['message'])}", "", @@ -465,6 +472,7 @@ def _founder_finding(index: int, finding: dict[str, Any]) -> list[str]: def _plain_risk(finding: dict[str, Any]) -> str: + """Translate severity into plain-language business risk.""" severity = finding["severity"] if severity == "CRITICAL": return "This can expose sensitive data, credentials, money movement, or remote execution risk if reachable in production." @@ -476,6 +484,7 @@ def _plain_risk(finding: dict[str, Any]) -> str: def _fix_prompt(finding: dict[str, Any]) -> str: + """Render a focused remediation prompt for one finding.""" return "\n".join( [ f"Fix AppGuardrail finding `{finding['rule_id']}` in `{finding['file']}:{finding['line']}`.", @@ -491,6 +500,7 @@ def _fix_prompt(finding: dict[str, Any]) -> str: def _next_steps( findings: list[dict[str, Any]], blockers: list[dict[str, Any]] ) -> list[str]: + """Choose the next recommended operator actions for a report.""" if not findings: return ["1. Re-run AppGuardrail with current production-bound code."] steps = [] @@ -509,6 +519,7 @@ def _next_steps( def _agency_recommendation(blockers: list[dict[str, Any]]) -> str: + """Return the agency-facing launch recommendation.""" if any(finding["severity"] == "CRITICAL" for finding in blockers): return "Hold pending critical fixes" if blockers: @@ -519,6 +530,7 @@ def _agency_recommendation(blockers: list[dict[str, Any]]) -> str: def _agency_severity_section( severity: str, findings: list[dict[str, Any]] ) -> list[str]: + """Render all agency report findings for one severity bucket.""" heading = severity.title() if severity != "INFO" else "Informational" lines = [f"### {heading} Findings", ""] if not findings: @@ -550,6 +562,7 @@ def _agency_severity_section( def _priority_matrix(findings: list[dict[str, Any]]) -> list[str]: + """Render the agency prioritization table.""" lines = [ "| ID | Title | Severity | Effort | Priority |", "|---|---|---|---|---|", @@ -569,6 +582,7 @@ def _priority_matrix(findings: list[dict[str, Any]]) -> list[str]: def _priority_for(finding: dict[str, Any]) -> str: + """Map a finding to a remediation priority label.""" if is_deploy_blocking(finding): return "Immediate" if finding["severity"] == "CRITICAL" else "Before launch" if finding["severity"] == "WARNING": @@ -577,6 +591,7 @@ def _priority_for(finding: dict[str, Any]) -> str: def _fix_item(index: int, finding: dict[str, Any]) -> list[str]: + """Render one actionable fix-pack checklist item.""" return [ f"### [ ] FIX-{index:03d}: {_short_title(finding['message'])}", "", @@ -598,6 +613,7 @@ def _fix_item(index: int, finding: dict[str, Any]) -> list[str]: def _fix_status_table(findings: list[dict[str, Any]]) -> list[str]: + """Render the fix-pack status tracking table.""" lines = [ "| ID | Title | Severity | Status | Fixed By | Verified |", "|---|---|---|---|---|---|", @@ -614,4 +630,5 @@ def _fix_status_table(findings: list[dict[str, Any]]) -> list[str]: def _references(finding: dict[str, Any]) -> str: + """Return a comma-separated reference list for a finding.""" return ", ".join(finding["references"] or finding["owasp"] or finding["cwe"]) or "n/a" diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index f427e1944..5577e5827 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -1142,6 +1142,11 @@ def _load_packaged_regex_rules(): """ +def _display_path(path: str | Path) -> str: + """Return a stable, slash-separated path for CLI output and reports.""" + return Path(path).as_posix() + + # --------------------------------------------------------------------------- # Command implementations # --------------------------------------------------------------------------- @@ -1229,12 +1234,12 @@ def cmd_init(args): existing = target_file.read_text() if config["append_marker"] not in existing: target_file.write_text(existing + "\n\n" + config["content"]) - installed.append(f"{config['path']} (appended)") + installed.append(f"{_display_path(config['path'])} (appended)") else: - skipped.append(str(config["path"])) + skipped.append(_display_path(config["path"])) else: target_file.write_text(config["content"]) - installed.append(str(config["path"])) + installed.append(_display_path(config["path"])) # Always create the checklist checklist_file = project_root / "APPGUARDRAIL_CHECKLIST.md" @@ -2139,15 +2144,15 @@ def _trivy_line(item: dict) -> int: def _trivy_target(target: str, base_path: Path) -> str: """Normalize a Trivy target path relative to the scan base when possible.""" if not target: - return str(base_path) + return base_path.as_posix() try: path = Path(target) if path.is_absolute(): root = base_path if base_path.is_dir() else base_path.parent - return str(path.relative_to(root)) + return path.relative_to(root).as_posix() except ValueError: pass - return target + return Path(target).as_posix() def _trivy_findings(report: dict, base_path: Path): @@ -2678,6 +2683,7 @@ def _scan_file(file_path: Path, base_path: Path): if base_path.is_file() else file_path_str_cache ) + rel_path_for_filters = _display_path(rel_path_for_filters) if not _path_allowed_by_rule( rel_path_for_filters, include_paths, exclude_paths ): @@ -2703,7 +2709,7 @@ def _scan_file(file_path: Path, base_path: Path): else file_path_str_cache ) rel_path_str = _sanitize_terminal_output( - str(rel_path_for_output) + _display_path(rel_path_for_output) ) start_idx = match.start() diff --git a/scripts/ci/collect_org_security_failures.py b/scripts/ci/collect_org_security_failures.py index c966b56c9..69d7fff46 100644 --- a/scripts/ci/collect_org_security_failures.py +++ b/scripts/ci/collect_org_security_failures.py @@ -38,39 +38,71 @@ class NoRedirect(urllib.request.HTTPRedirectHandler): + """Redirect handler that exposes GitHub log download redirects safely.""" + def redirect_request(self, req, fp, code, msg, headers, newurl): + """Prevent automatic redirect following so the caller can validate URLs.""" return None class SecureRedirectHandler(urllib.request.HTTPRedirectHandler): + """Redirect handler that validates every GitHub log download hop.""" + def redirect_request(self, req, fp, code, msg, headers, newurl): + """Validate redirected log URLs before urllib opens them.""" _validate_log_download_url(newurl) return super().redirect_request(req, fp, code, msg, headers, newurl) def _redacted_url(parsed: urllib.parse.ParseResult) -> str: + """Return a credential-free URL string safe for error messages.""" return f"{parsed.scheme}://{parsed.hostname or ''}{parsed.path}" def _validate_log_download_url(url: str) -> urllib.parse.ParseResult: + """Reject non-HTTP(S), credentialed, or internal log download URLs.""" + import ipaddress + parsed = urllib.parse.urlparse(url) - if parsed.scheme not in {"http", "https"}: - raise urllib.error.URLError( - f"Invalid or dangerous URL scheme in location: {_redacted_url(parsed)}" - ) - if parsed.hostname in BLOCKED_LOG_HOSTS: - raise urllib.error.URLError( - f"Access to internal address blocked: {_redacted_url(parsed)}" - ) + scheme = (parsed.scheme or "").lower() + if scheme not in {"http", "https"}: + raise urllib.error.URLError(f"Invalid or dangerous URL scheme in location: {_redacted_url(parsed)}") + if parsed.username or parsed.password: + raise urllib.error.URLError(f"Credentials not allowed in URL: {_redacted_url(parsed)}") + host = (parsed.hostname or "").lower() + if host in BLOCKED_LOG_HOSTS: + raise urllib.error.URLError(f"Access to internal address blocked: {_redacted_url(parsed)}") + raw = host.split("%", 1)[0].strip("[]") + if raw.isdigit(): # dotless decimal numeric host + raise urllib.error.URLError(f"Access to internal address blocked: {_redacted_url(parsed)}") + + # Check for octal/hex IP formats that urllib might accept but ipaddress rejects + parts = raw.split(".") + if any(p.startswith("0") and len(p) > 1 and p != "0" for p in parts) or any(p.startswith("0x") for p in parts): + raise urllib.error.URLError(f"Access to internal address blocked: {_redacted_url(parsed)}") + + try: + ip = ipaddress.ip_address(raw) + if getattr(ip, "ipv4_mapped", None): + ip = ip.ipv4_mapped + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_unspecified or ip.is_multicast: + raise urllib.error.URLError(f"Access to internal address blocked: {_redacted_url(parsed)}") + except ValueError: + # Host is not a direct IP literal; allow normal DNS-hostname handling. + return parsed return parsed class GitHub: + """Small GitHub REST client for workflow, job, issue, and log APIs.""" + def __init__(self, token: str, api: str = API): + """Create a client using a bearer token and API root.""" self.token = token self.api = api.rstrip("/") def request(self, method: str, path: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None) -> Any: + """Send one JSON GitHub API request and return the decoded payload.""" query = f"?{urllib.parse.urlencode(params)}" if params else "" body = json.dumps(data).encode() if data is not None else None req = urllib.request.Request( # noqa: S310 - GitHub API URL @@ -98,6 +130,7 @@ def request(self, method: str, path: str, data: dict[str, Any] | None = None, pa return json.loads(text) if "application/json" in content_type else text def pages(self, path: str, params: dict[str, Any] | None = None) -> list[Any]: + """Collect all pages for common GitHub list endpoints.""" items: list[Any] = [] page = 1 while True: @@ -116,6 +149,7 @@ def pages(self, path: str, params: dict[str, Any] | None = None) -> list[Any]: page += 1 def job_log(self, repo: str, job_id: int) -> str: + """Fetch a job log through GitHub's validated redirected download URL.""" path = f"/repos/{repo}/actions/jobs/{job_id}/logs" req = urllib.request.Request( # noqa: S310 - GitHub API URL f"{self.api}{path}", @@ -152,15 +186,18 @@ def job_log(self, repo: str, job_id: int) -> str: def utc_now() -> dt.datetime: + """Return the current UTC timestamp with timezone information.""" return dt.datetime.now(dt.timezone.utc) def parse_time(value: str) -> dt.datetime: + """Parse GitHub ISO timestamps and ensure the result is timezone-aware.""" parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed if parsed.tzinfo else parsed.replace(tzinfo=dt.timezone.utc) def build_finding(client: GitHub, repo: str, run: dict[str, Any], job: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]: + """Build one normalized security workflow failure record from run/job data.""" job_id = int(job["id"]) return { "repo": repo, @@ -180,6 +217,7 @@ def build_finding(client: GitHub, repo: str, run: dict[str, Any], job: dict[str, def collect_findings(client: GitHub, args: argparse.Namespace) -> list[dict[str, Any]]: + """Collect failed security workflow jobs across the configured organization.""" if args.run_url: repo, run_id = parse_run_url(args.run_url) repos = [{"full_name": repo}] @@ -208,6 +246,7 @@ def collect_findings(client: GitHub, args: argparse.Namespace) -> list[dict[str, def ensure_label(client: GitHub, target_repo: str, name: str, dry_run: bool, cache: set[str]) -> None: + """Ensure a GitHub issue label exists, caching labels within one run.""" if name in cache: return cache.add(name) @@ -222,11 +261,13 @@ def ensure_label(client: GitHub, target_repo: str, name: str, dry_run: bool, cac def issue_index(client: GitHub, target_repo: str) -> dict[str, dict[str, Any]]: + """Return existing non-PR issues keyed by title for deduplication.""" issues = client.pages(f"/repos/{target_repo}/issues", {"state": "all", "labels": ISSUE_LABEL}) return {issue["title"]: issue for issue in issues if issue.get("title") and "pull_request" not in issue} def publish_one(client: GitHub, target_repo: str, finding: dict[str, Any], dry_run: bool, issues: dict[str, dict[str, Any]], labels_seen: set[str]) -> None: + """Create, reopen, or update one issue for a collected failure.""" labels = [ISSUE_LABEL, SECURITY_LABEL, f"repo:{sanitize_label_value(finding['repo'].split('/', 1)[1])}"] issue_title = title(finding) issue = issues.get(issue_title) @@ -266,6 +307,7 @@ def publish_one(client: GitHub, target_repo: str, finding: dict[str, Any], dry_r def publish_findings(client: GitHub, target_repo: str, findings: list[dict[str, Any]], dry_run: bool) -> None: + """Publish every collected failure to the target repository.""" issues = issue_index(client, target_repo) if findings else {} labels_seen: set[str] = set() for finding in findings: @@ -273,10 +315,12 @@ def publish_findings(client: GitHub, target_repo: str, findings: list[dict[str, def parse_bool(value: str | None) -> bool: + """Parse truthy environment-style strings.""" return (value or "").strip().lower() in {"1", "true", "yes", "on"} def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse command-line arguments for the collector.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--owner", default=os.getenv("GITHUB_REPOSITORY_OWNER", "ContextualWisdomLab")) parser.add_argument("--target-repo", default=os.getenv("GITHUB_REPOSITORY", "ContextualWisdomLab/appguardrail")) @@ -289,6 +333,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace: def main(argv: list[str] | None = None) -> int: + """Run collection and issue publication, returning a process exit code.""" args = parse_args(argv or sys.argv[1:]) token = os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN") if not token: diff --git a/tests/test_appguardrail_coverage.py b/tests/test_appguardrail_coverage.py index a53fc8a43..653be8114 100644 --- a/tests/test_appguardrail_coverage.py +++ b/tests/test_appguardrail_coverage.py @@ -1,8 +1,22 @@ +import os +import runpy +import sys from unittest.mock import patch import pytest -from scanner.cli.appguardrail import cmd_init, cmd_scan +from scanner.cli.appguardrail import ( + _collect_files, + _parse_inline_list, + _path_matches_glob, + _scan_file, + cmd_hook, + cmd_init, + cmd_monitor, + cmd_review, + cmd_scan, + main, +) from tests.test_appguardrail import MOCK_RULES @@ -137,10 +151,6 @@ def test_cmd_init_path_traversal_target_file(tmp_path, monkeypatch, capsys): assert "💡 Hint: Ensure" in err -from scanner.cli.appguardrail import cmd_hook - - - class HookArgs: pass @@ -164,9 +174,10 @@ def test_cmd_hook_success(tmp_path, monkeypatch, capsys): assert "appguardrail scan ." in hook_text assert "command -v appguardrail" in hook_text assert 'python3 "$APPGUARDRAIL_CLI" scan .' in hook_text - import stat + if os.name != "nt": + import stat - assert hook_file.stat().st_mode & stat.S_IEXEC + assert hook_file.stat().st_mode & stat.S_IEXEC assert "pre-commit hook installed successfully" in capsys.readouterr().out @@ -223,9 +234,6 @@ def test_cmd_hook_remove_symlink(tmp_path, monkeypatch): assert not hook_link.is_symlink() -from scanner.cli.appguardrail import _collect_files, _scan_file - - def test_collect_files_oserror_on_scandir(tmp_path): def mock_scandir(path): raise PermissionError("Mock permission error") @@ -300,11 +308,6 @@ class MockStat: assert _scan_file(test_file, tmp_path) == [] -import sys - -from scanner.cli.appguardrail import cmd_review, main - - class ReviewArgs: def __init__(self, stack=None, db=None, payments=None): self.stack = stack @@ -415,9 +418,6 @@ def test_scan_file_empty_rules(tmp_path): assert _scan_file(test_file, tmp_path) == [] -import runpy - - def test_if_name_main(): import sys from unittest.mock import patch @@ -457,8 +457,6 @@ def test_scan_file_open_permission_error(): findings = _scan_file(file_path, base_path) assert findings == [] -from scanner.cli.appguardrail import cmd_monitor, _path_matches_glob, _parse_inline_list - def test_cmd_monitor(capsys, monkeypatch, tmp_path): monkeypatch.chdir(tmp_path) class MonitorArgs: diff --git a/tests/test_config.py b/tests/test_config.py index 48508322a..e0e987526 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -26,7 +26,8 @@ def test_loads_fail_on_and_excludes(tmp_path): def test_first_dir_wins(tmp_path): d1, d2 = tmp_path / "a", tmp_path / "b" - d1.mkdir(); d2.mkdir() + d1.mkdir() + d2.mkdir() (d1 / CONFIG_NAME).write_text('{"fail_on": "HIGH"}') (d2 / CONFIG_NAME).write_text('{"fail_on": "INFO"}') assert load_config([d1, d2])["fail_on"] == "HIGH" diff --git a/tests/test_coverage_edge_cases.py b/tests/test_coverage_edge_cases.py index 9aa7cb23c..0a461599e 100644 --- a/tests/test_coverage_edge_cases.py +++ b/tests/test_coverage_edge_cases.py @@ -1,14 +1,23 @@ -import os -import json import subprocess from pathlib import Path +from unittest.mock import MagicMock, patch + import pytest -from unittest.mock import patch, MagicMock from scanner.cli.appguardrail import ( - cmd_scan, _run_trivy_fs, _finding_context, _finding_category, _trivy_target, - _scan_file, _trivy_severity, _trivy_line, _trivy_findings, _build_finding, - _confidence, _is_deploy_blocking, _run_codegraph_command, _run_codegraph_index + _confidence, + _finding_category, + _finding_context, + _is_deploy_blocking, + _run_codegraph_command, + _run_codegraph_index, + _run_trivy_fs, + _scan_file, + _trivy_findings, + _trivy_line, + _trivy_severity, + _trivy_target, + cmd_scan, ) from tests.test_appguardrail_coverage import ScanArgs @@ -75,7 +84,7 @@ def test_finding_category_secrets_for_credentials(): assert _finding_category("hardcoded-api-credential") == "secrets" def test_trivy_target_empty(tmp_path): - assert _trivy_target("", tmp_path) == str(tmp_path) + assert _trivy_target("", tmp_path) == tmp_path.as_posix() def test_trivy_target_absolute_valueerror(tmp_path): # base is a non-existent file (not a dir), so root = base.parent = tmp_path @@ -222,10 +231,10 @@ def test_confidence(): assert _confidence("random") == "high" def test_is_deploy_blocking(): - assert _is_deploy_blocking({"severity": "CRITICAL", "context": "app-code"}) == True - assert _is_deploy_blocking({"severity": "HIGH", "context": "app-code"}) == True - assert _is_deploy_blocking({"severity": "MEDIUM", "context": "app-code"}) == False - assert _is_deploy_blocking({"severity": "CRITICAL", "context": "test"}) == False + assert _is_deploy_blocking({"severity": "CRITICAL", "context": "app-code"}) + assert _is_deploy_blocking({"severity": "HIGH", "context": "app-code"}) + assert not _is_deploy_blocking({"severity": "MEDIUM", "context": "app-code"}) + assert not _is_deploy_blocking({"severity": "CRITICAL", "context": "test"}) def test_trivy_target_value_error(): @@ -282,4 +291,4 @@ def test_scan_file_uses_absolute_path_when_base_is_unrelated(tmp_path): findings = _scan_file(test_file, base_dir) assert len(findings) == 1 - assert findings[0]["file"] == str(test_file) + assert findings[0]["file"] == test_file.as_posix() diff --git a/tests/test_org_security_failure_collector.py b/tests/test_org_security_failure_collector.py index d7afc9b29..4e8f895d7 100644 --- a/tests/test_org_security_failure_collector.py +++ b/tests/test_org_security_failure_collector.py @@ -92,15 +92,26 @@ def test_job_log_rejects_dangerous_redirect_scheme(monkeypatch): def test_job_log_rejects_internal_redirect_host(monkeypatch): client = collector.GitHub("token") - monkeypatch.setattr( - collector.urllib.request, - "build_opener", - lambda *_: FakeRedirectOpener("http://169.254.169.254/latest/meta-data"), - ) - assert "Access to internal address blocked" in client.job_log( - "ContextualWisdomLab/naruon", 123 - ) + blocked_urls = [ + "http://169.254.169.254/latest/meta-data", + "http://[::1]/", + "http://[::ffff:127.0.0.1]/", + "http://10.0.0.5/", + "http://2130706433/", + "http://0177.0.0.1/", + "http://0x7f.0.0.1/", + ] + + for url in blocked_urls: + monkeypatch.setattr( + collector.urllib.request, + "build_opener", + lambda *_, u=url: FakeRedirectOpener(u), + ) + assert "Access to internal address blocked" in client.job_log( + "ContextualWisdomLab/naruon", 123 + ) def test_publish_skips_duplicate_and_reopens_closed_issue():