From 50106f4ec4d46713322a9fdbaa6a7cb216a7622c Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Wed, 26 Aug 2026 02:11:25 +0100 Subject: [PATCH 01/12] feat(ci-rules): add AZ-CI-001..004 CI/CD workflow security rules (issue #259 PR 1/3) Implements controls 1-4 from issue #259: - AZ-CI-001: Long-lived Azure credentials instead of workload identity federation - AZ-CI-002: Unnecessarily broad workflow token permissions - AZ-CI-003: Third-party action not pinned to immutable commit SHA - AZ-CI-004: Untrusted PR input reaches privileged workflow context False positive fixes: - FP1 az_ci_004: _checks_out_pr_code uses parsed dict, checks checkout ref: value - FP2 az_ci_001: uses_workload_identity checks job-level permissions blocks - FP3 az_ci_001: credential scan excludes YAML comment lines - FP4 az_ci_002: absent top-level permissions not broad when all jobs declare explicit perms New files: - scanner/github_client.py: GitHub App + PAT auth, Link-header pagination, >1MB guard - scanner/rules/_workflow_common.py: shared YAML helpers - scanner/rules/az_ci_001..004.py: four scanner rule modules - playbooks/cli/fix_az_ci_001..004.sh: remediation playbooks - tests/test_rules_ci_workflow.py: 51 tests including 4 FP regression tests ci.yml: add chromadb CVE-2026-45830 and CVE-2026-45833 to pip-audit ignore list Engine: az_ci_* excluded from AzureClient loader. Compliance: CIS N/A-CI-*, NIST, ISO 27001, SOC2 with trailing newlines. Refs #259 Signed-off-by: Mahfuzur Rahman Emon --- playbooks/cli/fix_az_ci_001.sh | 32 ++ playbooks/cli/fix_az_ci_002.sh | 14 + playbooks/cli/fix_az_ci_003.sh | 20 ++ playbooks/cli/fix_az_ci_004.sh | 20 ++ scanner/engine.py | 6 + scanner/github_client.py | 197 +++++++++++ scanner/rules/_workflow_common.py | 203 +++++++++++ scanner/rules/az_ci_001.py | 79 +++++ scanner/rules/az_ci_002.py | 75 ++++ scanner/rules/az_ci_003.py | 75 ++++ scanner/rules/az_ci_004.py | 106 ++++++ tests/test_rules_ci_workflow.py | 562 ++++++++++++++++++++++++++++++ 12 files changed, 1389 insertions(+) create mode 100755 playbooks/cli/fix_az_ci_001.sh create mode 100755 playbooks/cli/fix_az_ci_002.sh create mode 100755 playbooks/cli/fix_az_ci_003.sh create mode 100755 playbooks/cli/fix_az_ci_004.sh create mode 100644 scanner/github_client.py create mode 100644 scanner/rules/_workflow_common.py create mode 100644 scanner/rules/az_ci_001.py create mode 100644 scanner/rules/az_ci_002.py create mode 100644 scanner/rules/az_ci_003.py create mode 100644 scanner/rules/az_ci_004.py create mode 100644 tests/test_rules_ci_workflow.py diff --git a/playbooks/cli/fix_az_ci_001.sh b/playbooks/cli/fix_az_ci_001.sh new file mode 100755 index 00000000..85f17558 --- /dev/null +++ b/playbooks/cli/fix_az_ci_001.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# Fix AZ-CI-001: Replace long-lived Azure credentials with workload identity federation +# Usage: bash fix_az_ci_001.sh +set -euo pipefail +APP_ID="${1:?Usage: $0 }" +TENANT_ID="${2:?}" +SUBSCRIPTION_ID="${3:?}" +REPO="${4:?}" + +echo "[1/4] Creating federated identity credential for GitHub Actions..." +az ad app federated-credential create \ + --id "$APP_ID" \ + --parameters "{ + \"name\": \"github-oidc\", + \"issuer\": \"https://token.actions.githubusercontent.com\", + \"subject\": \"repo:${REPO}:ref:refs/heads/main\", + \"audiences\": [\"api://AzureADTokenExchange\"] + }" + +echo "[2/4] Update workflow: add id-token: write permission" +echo " Replace azure/login with:" +echo " - uses: azure/login@v2" +echo " with:" +echo " client-id: \${{ secrets.AZURE_CLIENT_ID }}" +echo " tenant-id: \${{ secrets.AZURE_TENANT_ID }}" +echo " subscription-id: \${{ secrets.AZURE_SUBSCRIPTION_ID }}" + +echo "[3/4] Remove AZURE_CLIENT_SECRET from repository secrets" +gh secret delete AZURE_CLIENT_SECRET --repo "$REPO" 2>/dev/null || true +gh secret delete ARM_CLIENT_SECRET --repo "$REPO" 2>/dev/null || true + +echo "[4/4] Done. Validate with: az login --federated-token" diff --git a/playbooks/cli/fix_az_ci_002.sh b/playbooks/cli/fix_az_ci_002.sh new file mode 100755 index 00000000..22c793a4 --- /dev/null +++ b/playbooks/cli/fix_az_ci_002.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Fix AZ-CI-002: Add least-privilege permissions block to workflow +# Usage: bash fix_az_ci_002.sh +set -euo pipefail +WORKFLOW="${1:?Usage: $0 }" + +echo "[INFO] Add the following to the top level of $WORKFLOW:" +echo "" +echo "permissions:" +echo " contents: read" +echo "" +echo "[INFO] Override per-job only where broader access is needed." +echo "[INFO] Never use permissions: write-all in production workflows." +echo "[INFO] See: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs" diff --git a/playbooks/cli/fix_az_ci_003.sh b/playbooks/cli/fix_az_ci_003.sh new file mode 100755 index 00000000..142a5590 --- /dev/null +++ b/playbooks/cli/fix_az_ci_003.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Fix AZ-CI-003: Pin third-party actions to immutable commit SHAs +# Usage: bash fix_az_ci_003.sh +set -euo pipefail +WORKFLOW="${1:?Usage: $0 }" + +echo "[INFO] For each unpinned action in $WORKFLOW, replace the tag with a full SHA:" +echo " Before: uses: actions/checkout@v4" +echo " After: uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2" +echo "" +echo "[INFO] Use pin-github-action to automate:" +echo " pip install pin-github-action" +echo " pin-github-action $WORKFLOW" +echo "" +echo "[INFO] Enable Dependabot to keep pinned SHAs updated:" +echo " Add to .github/dependabot.yml:" +echo " - package-ecosystem: github-actions" +echo " directory: /" +echo " schedule:" +echo " interval: weekly" diff --git a/playbooks/cli/fix_az_ci_004.sh b/playbooks/cli/fix_az_ci_004.sh new file mode 100755 index 00000000..5235ad96 --- /dev/null +++ b/playbooks/cli/fix_az_ci_004.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Fix AZ-CI-004: Prevent untrusted PR code from reaching privileged context +# Usage: bash fix_az_ci_004.sh +set -euo pipefail + +echo "[INFO] For pull_request_target workflows that check out PR code:" +echo "" +echo "[1/3] Never use actions/checkout with PR head ref in pull_request_target:" +echo " Remove: ref: \${{ github.event.pull_request.head.sha }}" +echo "" +echo "[2/3] Split privileged and unprivileged steps:" +echo " - Untrusted build: triggered by pull_request (no secrets)" +echo " - Privileged deploy: triggered by workflow_run on completed build" +echo "" +echo "[3/3] If checkout of PR code is required, remove all secret access:" +echo " permissions:" +echo " contents: read" +echo " env: {} # no secrets" +echo "" +echo "[INFO] See: https://securitylab.github.com/research/github-actions-preventing-pwn-requests/" diff --git a/scanner/engine.py b/scanner/engine.py index 1fef83df..bf7c1714 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -73,6 +73,12 @@ def load_rules(self) -> None: # counts. A misnamed scratch module must never execute against a real # subscription without first passing the rule checks. for rule_path in sorted(RULES_DIR.glob("az_*.py")): + # AZ-CI rules require a GitHubClient, not an AzureClient. + # They are excluded from the Azure scan engine and invoked + # separately by the CI/CD scan entry point. + if rule_path.stem.startswith("az_ci_"): + logger.debug("Skipping CI rule %s (requires GitHubClient)", rule_path.name) + continue try: spec = importlib.util.spec_from_file_location(rule_path.stem, rule_path) module = importlib.util.module_from_spec(spec) # type: ignore[arg-type] diff --git a/scanner/github_client.py b/scanner/github_client.py new file mode 100644 index 00000000..08762eb8 --- /dev/null +++ b/scanner/github_client.py @@ -0,0 +1,197 @@ +"""GitHub REST API client for CI/CD workflow security scan rules. + +Authenticates via GitHub App installation token (preferred) with an explicit +GITHUB_TOKEN or fine-grained PAT as fallback. Returns None on any auth, +permission, or network failure so callers can return UNKNOWN instead of +crashing the scan engine. + +Authentication priority: + 1. GitHub App: GITHUB_APP_ID + GITHUB_APP_PRIVATE_KEY + GITHUB_APP_INSTALLATION_ID + 2. GITHUB_TOKEN (Actions token or fine-grained PAT) +""" + +import base64 +import logging +import os +import time +from typing import Any, Dict, List, Optional + +import requests + +logger = logging.getLogger(__name__) + +_GITHUB_API = "https://api.github.com" +_UNSET = object() + + +def _get_app_token() -> Optional[str]: + """Generate a GitHub App installation access token.""" + app_id = os.environ.get("GITHUB_APP_ID", "") + private_key = os.environ.get("GITHUB_APP_PRIVATE_KEY", "") + installation_id = os.environ.get("GITHUB_APP_INSTALLATION_ID", "") + if not (app_id and private_key and installation_id): + return None + try: + import jwt as pyjwt + + now = int(time.time()) + payload = {"iat": now - 60, "exp": now + 540, "iss": app_id} + app_token = pyjwt.encode(payload, private_key, algorithm="RS256") + resp = requests.post( + f"{_GITHUB_API}/app/installations/{installation_id}/access_tokens", + headers={ + "Authorization": f"Bearer {app_token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + }, + timeout=15, + ) + if resp.status_code == 201: + return resp.json().get("token") + logger.error( + "GitHub App token request failed (HTTP status indicates rejection)" + ) # nosemgrep: python-logger-credential-disclosure + except Exception: + logger.error( + "GitHub App token generation failed (check GITHUB_APP_* env vars)" + ) # nosemgrep: python-logger-credential-disclosure + return None + + +def _get_token() -> Optional[str]: + """Return an authentication token, preferring App auth over PAT.""" + token = _get_app_token() + if token: + return token + token = os.environ.get("GITHUB_TOKEN", "") + return token if token else None + + +class GitHubClient: + """Wraps the GitHub REST API for CI/CD workflow security scan rules. + + Instantiate once per scan and share across all AZ-CI rule modules. + Every method logs on failure and returns None so a missing token or + inaccessible repository never crashes the scan engine. Callers must + treat None as UNKNOWN, not as a compliant result. + """ + + def __init__(self, owner: str, repo: str, token: Optional[str] = None) -> None: + """ + Args: + owner: GitHub organisation or user name. + repo: Repository name. + token: Auth token. Defaults to auto-detection via _get_token(). + """ + self.owner = owner + self.repo = repo + self._token = token or _get_token() + self._workflows_cache: Any = _UNSET + self._workflow_files_cache: Dict[str, Any] = {} + + def _headers(self) -> Dict[str, str]: + headers = { + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + return headers + + def _get(self, path: str, **kwargs: Any) -> Optional[requests.Response]: + """GET a GitHub API path; returns None on any error.""" + url = f"{_GITHUB_API}{path}" + try: + resp = requests.get(url, headers=self._headers(), timeout=15, **kwargs) + if resp.status_code == 403: + logger.warning("GitHub API permission denied for %s", url) + return None + if resp.status_code == 404: + logger.warning("GitHub resource not found: %s", url) + return None + resp.raise_for_status() + return resp + except Exception as exc: + logger.error("GitHub API request failed for %s: %s", url, exc) + return None + + def get_workflows(self) -> Optional[List[Dict[str, Any]]]: + """Return all workflow definitions for the repository. + + Follows the Link header to paginate beyond the 100-item page size, + so repositories with >100 workflows are fully enumerated. + Returns a list (including empty) on success, None on failure. + """ + if self._workflows_cache is not _UNSET: + return self._workflows_cache + workflows: List[Dict[str, Any]] = [] + url: Optional[str] = f"{_GITHUB_API}/repos/{self.owner}/{self.repo}/actions/workflows" + params: Dict[str, Any] = {"per_page": 100} + while url: + resp = self._get(url.replace(_GITHUB_API, ""), params=params) + if resp is None: + self._workflows_cache = None + return None + try: + data = resp.json() + workflows.extend(data.get("workflows", [])) + # Follow Link: ; rel="next" header for next page + link_header = resp.headers.get("Link", "") + next_url = None + for part in link_header.split(","): + part = part.strip() + if 'rel="next"' in part: + next_url = part.split(";")[0].strip().strip("<>") + break + url = next_url + params = {} # URL already contains query params for subsequent pages + except Exception as exc: + logger.error("Failed to parse workflows response: %s", exc) + self._workflows_cache = None + return None + self._workflows_cache = workflows + return self._workflows_cache + + def get_workflow_content(self, path: str) -> Optional[str]: + """Return the decoded YAML content of a workflow file. + + Args: + path: Workflow file path e.g. '.github/workflows/ci.yml' + Returns: + Decoded YAML string, or None on failure/missing permissions. + """ + if path in self._workflow_files_cache: + return self._workflow_files_cache[path] + resp = self._get(f"/repos/{self.owner}/{self.repo}/contents/{path}") + if resp is None: + self._workflow_files_cache[path] = None + return None + try: + data = resp.json() + content_b64 = data.get("content", "") + if not content_b64 and data.get("download_url"): + # File >1 MB: contents API returns empty content with a download_url + logger.warning( + "Workflow file %s exceeds 1 MB — contents API returned empty; treating as UNKNOWN", + path, + ) + self._workflow_files_cache[path] = None + return None + decoded = base64.b64decode(content_b64).decode("utf-8") + self._workflow_files_cache[path] = decoded + return decoded + except Exception as exc: + logger.error("Failed to decode workflow file %s: %s", path, exc) + self._workflow_files_cache[path] = None + return None + + def get_repo_info(self) -> Optional[Dict[str, Any]]: + """Return repository metadata (visibility, default branch, etc.).""" + resp = self._get(f"/repos/{self.owner}/{self.repo}") + if resp is None: + return None + try: + return resp.json() + except Exception as exc: + logger.error("Failed to parse repo info: %s", exc) + return None diff --git a/scanner/rules/_workflow_common.py b/scanner/rules/_workflow_common.py new file mode 100644 index 00000000..34e469ec --- /dev/null +++ b/scanner/rules/_workflow_common.py @@ -0,0 +1,203 @@ +"""Shared helpers for CI/CD workflow security scan rules (AZ-CI-*).""" + +import logging +import re +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +_SHA_PIN_RE = re.compile(r"^[^@]+@[0-9a-f]{40}$") + +_BROAD_PERMISSIONS = { + "contents": "write", + "actions": "write", + "packages": "write", + "deployments": "write", + "id-token": "write", + "pull-requests": "write", + "issues": "write", + "security-events": "write", +} + +_DANGEROUS_TRIGGERS = {"pull_request_target", "workflow_run"} + +_LONGTERM_CRED_PATTERNS = [ + re.compile(r"AZURE_CLIENT_SECRET", re.IGNORECASE), + re.compile(r"AZURE_PASSWORD", re.IGNORECASE), + re.compile(r"ARM_CLIENT_SECRET", re.IGNORECASE), + re.compile(r"ARM_ACCESS_KEY", re.IGNORECASE), + re.compile(r"AZURE_STORAGE_KEY", re.IGNORECASE), +] + + +def parse_workflow(yaml_content: str) -> Optional[Dict[str, Any]]: + """Parse YAML workflow content; returns None on malformed input.""" + try: + import yaml + + return yaml.safe_load(yaml_content) or {} + except Exception as exc: + logger.warning("Failed to parse workflow YAML: %s", exc) + return None + + +def is_action_pinned(uses: str) -> bool: + """Return True if a uses reference is pinned to a full commit SHA.""" + if not uses or uses.startswith("./") or uses.startswith("docker://"): + return True + return bool(_SHA_PIN_RE.match(uses.strip())) + + +def collect_uses(workflow: Dict[str, Any]) -> List[Tuple[str, str]]: + """Return all (job_id, uses) pairs from a parsed workflow dict. + + Collects both step-level uses (regular actions) and job-level uses + (reusable workflow calls). Handles null/empty job definitions safely. + """ + results = [] + jobs = workflow.get("jobs") or {} + for job_id, job in jobs.items(): + if not isinstance(job, dict): + continue + # Job-level uses: reusable workflow call + job_uses = job.get("uses") + if job_uses: + results.append((job_id, job_uses)) + # Step-level uses: regular action references + steps = job.get("steps") or [] + for step in steps: + if not isinstance(step, dict): + continue + uses = step.get("uses") + if uses: + results.append((job_id, uses)) + return results + + +def has_long_lived_credentials(yaml_content: str) -> List[str]: + """Return list of matched long-lived credential env var names found in YAML. + + Excludes YAML comment lines (lines whose first non-whitespace char is #) + to avoid false positives from changelog-style migration comments. + """ + # Strip comment lines before scanning + non_comment_lines = [line for line in yaml_content.splitlines() if not line.lstrip().startswith("#")] + non_comment_content = "\n".join(non_comment_lines) + found = [] + for pattern in _LONGTERM_CRED_PATTERNS: + matches = pattern.findall(non_comment_content) + found.extend(matches) + return found + + +def uses_workload_identity(yaml_content: str, parsed: Optional[Dict[str, Any]] = None) -> bool: + """Return True if the workflow uses OIDC workload identity federation. + + Checks both top-level and job-level permissions blocks for id-token: write, + since per-job least-privilege is a recommended secure pattern. Also verifies + azure/login is present and no long-lived credential is used outside comments. + """ + has_oidc_permission = False + if parsed is not None: + # Check top-level permissions + top_perms = parsed.get("permissions") or {} + if isinstance(top_perms, dict): + has_oidc_permission = top_perms.get("id-token", "").lower() == "write" + # Check job-level permissions if top-level not set + if not has_oidc_permission: + jobs = parsed.get("jobs") or {} + for job in jobs.values(): + if not isinstance(job, dict): + continue + job_perms = job.get("permissions") or {} + if isinstance(job_perms, dict): + if job_perms.get("id-token", "").lower() == "write": + has_oidc_permission = True + break + else: + has_oidc_permission = "id-token: write" in yaml_content + + has_azure_login = "azure/login" in yaml_content + # Check ALL long-lived credential patterns, excluding comment lines + has_any_long_lived_cred = bool(has_long_lived_credentials(yaml_content)) + return has_oidc_permission and has_azure_login and not has_any_long_lived_cred + + +def get_top_level_permissions(workflow: Dict[str, Any]) -> Optional[Dict[str, str]]: + """Return top-level permissions block, or None if not defined.""" + perms = workflow.get("permissions") + if perms is None: + return None + if isinstance(perms, str): + return {"all": perms} + return perms if isinstance(perms, dict) else None + + +def is_permissions_broad( + permissions: Optional[Dict[str, str]], + parsed: Optional[Dict[str, Any]] = None, +) -> bool: + """Return True if permissions grant write-all or multiple broad scopes. + + An absent top-level permissions block is only treated as broad when jobs + also lack explicit per-job permissions declarations. Per-job least privilege + (no top-level block, every job declares its own) is a recommended secure + pattern and must not be flagged. + """ + if permissions is None: + if parsed is not None: + jobs = parsed.get("jobs") or {} + if jobs and all(isinstance(job, dict) and job.get("permissions") is not None for job in jobs.values()): + return False + return True + if permissions.get("all") in ("write-all", "write"): + return True + broad_count = sum(1 for k, v in permissions.items() if v == "write" and k in _BROAD_PERMISSIONS) + return broad_count >= 3 + + +def get_dangerous_triggers(workflow: Dict[str, Any]) -> List[str]: + """Return list of dangerous triggers present in the workflow.""" + on = workflow.get("on") or workflow.get(True) or {} + if isinstance(on, str): + on = {on: {}} + if isinstance(on, list): + on = {t: {} for t in on} + return [t for t in _DANGEROUS_TRIGGERS if t in on] + + +def build_ci_finding( + rule_id: str, + rule_name: str, + severity: str, + category: str, + frameworks: Dict[str, str], + description: str, + remediation: str, + playbook: str, + owner: str, + repo: str, + workflow_path: str, + metadata: Dict[str, Any], +) -> Dict[str, Any]: + """Build a standardised CI/CD finding dict.""" + resource_id = f"github/{owner}/{repo}/workflows/{workflow_path}" + return { + "rule_id": rule_id, + "rule_name": rule_name, + "severity": severity, + "category": category, + "resource_id": resource_id, + "resource_name": workflow_path, + "resource_type": "GitHub/WorkflowFile", + "description": description, + "remediation": remediation, + "playbook": playbook, + "frameworks": frameworks, + "metadata": { + "owner": owner, + "repo": repo, + "workflow_path": workflow_path, + **metadata, + }, + } diff --git a/scanner/rules/az_ci_001.py b/scanner/rules/az_ci_001.py new file mode 100644 index 00000000..f465d713 --- /dev/null +++ b/scanner/rules/az_ci_001.py @@ -0,0 +1,79 @@ +"""AZ-CI-001: CI/CD workflow uses long-lived Azure credentials instead of workload identity federation.""" + +import logging +from typing import Any, Dict, List + +from scanner.rules._workflow_common import ( + build_ci_finding, + has_long_lived_credentials, + parse_workflow, + uses_workload_identity, +) + +RULE_ID = "AZ-CI-001" +RULE_NAME = "CI/CD Workflow Uses Long-Lived Azure Credentials" +SEVERITY = "HIGH" +CATEGORY = "CI/CD Security" +FRAMEWORKS = { + "CIS": "N/A-CI-001", + "NIST": "PR.AC-1", + "ISO27001": "A.9.2.4", + "SOC2": "CC6.1", +} +DESCRIPTION = ( + "The workflow authenticates to Azure using a long-lived client secret or storage key " + "stored as a repository secret. Long-lived credentials have no automatic expiry, cannot " + "be scoped to a single workflow run, and grant persistent access if leaked. Workload " + "identity federation (OIDC) issues short-lived tokens bound to the specific job context " + "with no secret to store or rotate." +) +REMEDIATION = ( + "Replace the long-lived credential with workload identity federation: " + "1. Register a federated identity credential on the Azure AD application or managed identity. " + "2. Add permissions: id-token: write to the workflow. " + "3. Use azure/login with client-id, tenant-id, subscription-id (no client-secret). " + "4. Remove AZURE_CLIENT_SECRET and ARM_CLIENT_SECRET from repository secrets." +) +PLAYBOOK = "playbooks/cli/fix_az_ci_001.sh" +logger = logging.getLogger(__name__) + + +def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: + """Detect workflows that use long-lived Azure credentials.""" + findings: List[Dict[str, Any]] = [] + workflows = github_client.get_workflows() + if workflows is None: + logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) + return findings + for wf in workflows: + path = wf.get("path", "") + if not path: + continue + content = github_client.get_workflow_content(path) + if content is None: + logger.warning("%s: could not read %s", RULE_ID, path) + continue + parsed = parse_workflow(content) + if parsed is None: + continue + creds = has_long_lived_credentials(content) + # Only suppress if OIDC is present AND the specific credential is absent. + # A workflow using OIDC but also referencing ARM_ACCESS_KEY still fails. + if creds and not uses_workload_identity(content, parsed): + findings.append( + build_ci_finding( + rule_id=RULE_ID, + rule_name=RULE_NAME, + severity=SEVERITY, + category=CATEGORY, + frameworks=FRAMEWORKS, + description=DESCRIPTION, + remediation=REMEDIATION, + playbook=PLAYBOOK, + owner=owner, + repo=repo, + workflow_path=path, + metadata={"long_lived_credentials_found": creds}, + ) + ) + return findings diff --git a/scanner/rules/az_ci_002.py b/scanner/rules/az_ci_002.py new file mode 100644 index 00000000..0daf00c0 --- /dev/null +++ b/scanner/rules/az_ci_002.py @@ -0,0 +1,75 @@ +"""AZ-CI-002: CI/CD workflow has unnecessarily broad token permissions.""" + +import logging +from typing import Any, Dict, List + +from scanner.rules._workflow_common import ( + build_ci_finding, + get_top_level_permissions, + is_permissions_broad, + parse_workflow, +) + +RULE_ID = "AZ-CI-002" +RULE_NAME = "CI/CD Workflow Has Unnecessarily Broad Token Permissions" +SEVERITY = "HIGH" +CATEGORY = "CI/CD Security" +FRAMEWORKS = { + "CIS": "N/A-CI-002", + "NIST": "PR.AC-4", + "ISO27001": "A.9.2.3", + "SOC2": "CC6.3", +} +DESCRIPTION = ( + "The workflow does not declare a top-level permissions block or declares write access " + "to multiple sensitive scopes. Without an explicit permissions block GitHub grants " + "read/write access to all scopes by default. Over-broad GITHUB_TOKEN permissions allow " + "a compromised step to push code, modify releases, or exfiltrate secrets." +) +REMEDIATION = ( + "Add a top-level permissions block that grants only the minimum scopes required: " + "permissions:\n contents: read\n" + "Override per-job only where broader access is needed. " + "Never use permissions: write-all in production workflows." +) +PLAYBOOK = "playbooks/cli/fix_az_ci_002.sh" +logger = logging.getLogger(__name__) + + +def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: + """Detect workflows with broad or absent token permissions.""" + findings: List[Dict[str, Any]] = [] + workflows = github_client.get_workflows() + if workflows is None: + logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) + return findings + for wf in workflows: + path = wf.get("path", "") + if not path: + continue + content = github_client.get_workflow_content(path) + if content is None: + logger.warning("%s: could not read %s", RULE_ID, path) + continue + parsed = parse_workflow(content) + if parsed is None: + continue + perms = get_top_level_permissions(parsed) + if is_permissions_broad(perms, parsed): + findings.append( + build_ci_finding( + rule_id=RULE_ID, + rule_name=RULE_NAME, + severity=SEVERITY, + category=CATEGORY, + frameworks=FRAMEWORKS, + description=DESCRIPTION, + remediation=REMEDIATION, + playbook=PLAYBOOK, + owner=owner, + repo=repo, + workflow_path=path, + metadata={"permissions_declared": perms}, + ) + ) + return findings diff --git a/scanner/rules/az_ci_003.py b/scanner/rules/az_ci_003.py new file mode 100644 index 00000000..75d0c6ca --- /dev/null +++ b/scanner/rules/az_ci_003.py @@ -0,0 +1,75 @@ +"""AZ-CI-003: Third-party workflow action not pinned to an immutable commit SHA.""" + +import logging +from typing import Any, Dict, List + +from scanner.rules._workflow_common import ( + build_ci_finding, + collect_uses, + is_action_pinned, + parse_workflow, +) + +RULE_ID = "AZ-CI-003" +RULE_NAME = "Third-Party Workflow Action Not Pinned to Immutable Commit SHA" +SEVERITY = "HIGH" +CATEGORY = "CI/CD Security" +FRAMEWORKS = { + "CIS": "N/A-CI-003", + "NIST": "PR.IP-1", + "ISO27001": "A.12.1.2", + "SOC2": "CC7.1", +} +DESCRIPTION = ( + "One or more workflow steps reference a third-party action using a mutable tag or branch " + "instead of a full 40-character commit SHA. A tag can be silently moved to point at " + "malicious code, so any future run of the workflow would execute the attacker-controlled " + "version without any change to the workflow file itself." +) +REMEDIATION = ( + "Pin every third-party action to a full commit SHA: " + "uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2\n" + "Use a tool such as pin-github-action or Dependabot to keep pinned SHAs up to date. " + "Local actions (./path) and Docker actions are exempt." +) +PLAYBOOK = "playbooks/cli/fix_az_ci_003.sh" +logger = logging.getLogger(__name__) + + +def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: + """Detect workflow steps that reference unpinned third-party actions.""" + findings: List[Dict[str, Any]] = [] + workflows = github_client.get_workflows() + if workflows is None: + logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) + return findings + for wf in workflows: + path = wf.get("path", "") + if not path: + continue + content = github_client.get_workflow_content(path) + if content is None: + logger.warning("%s: could not read %s", RULE_ID, path) + continue + parsed = parse_workflow(content) + if parsed is None: + continue + unpinned = [uses for _, uses in collect_uses(parsed) if not is_action_pinned(uses)] + if unpinned: + findings.append( + build_ci_finding( + rule_id=RULE_ID, + rule_name=RULE_NAME, + severity=SEVERITY, + category=CATEGORY, + frameworks=FRAMEWORKS, + description=DESCRIPTION, + remediation=REMEDIATION, + playbook=PLAYBOOK, + owner=owner, + repo=repo, + workflow_path=path, + metadata={"unpinned_actions": unpinned}, + ) + ) + return findings diff --git a/scanner/rules/az_ci_004.py b/scanner/rules/az_ci_004.py new file mode 100644 index 00000000..2a7120cc --- /dev/null +++ b/scanner/rules/az_ci_004.py @@ -0,0 +1,106 @@ +"""AZ-CI-004: Untrusted pull-request input reaches a privileged workflow context.""" + +import logging +from typing import Any, Dict, List + +from scanner.rules._workflow_common import ( + build_ci_finding, + get_dangerous_triggers, + parse_workflow, +) + +RULE_ID = "AZ-CI-004" +RULE_NAME = "Untrusted Pull-Request Input Reaches Privileged Workflow Context" +SEVERITY = "HIGH" +CATEGORY = "CI/CD Security" +FRAMEWORKS = { + "CIS": "N/A-CI-004", + "NIST": "PR.AC-4", + "ISO27001": "A.14.2.5", + "SOC2": "CC8.1", +} +DESCRIPTION = ( + "The workflow is triggered by pull_request_target or workflow_run, which run in the " + "context of the base repository and have access to repository secrets and write " + "permissions. If the workflow checks out or executes code from the pull request branch " + "without explicit protection, an attacker can submit a malicious PR to exfiltrate " + "secrets or push code — a pwn-request attack." +) +REMEDIATION = ( + "Never check out pull-request code in a pull_request_target workflow that also has " + "secret access or write permissions. If both are needed, separate the privileged steps " + "into a workflow triggered by workflow_run on a completed, already-vetted run. " + "See: https://securitylab.github.com/research/github-actions-preventing-pwn-requests/" +) +PLAYBOOK = "playbooks/cli/fix_az_ci_004.sh" +logger = logging.getLogger(__name__) + +_PR_HEAD_REF_PATTERNS = ( + "github.event.pull_request.head.sha", + "github.event.pull_request.head.ref", + "github.head_ref", + "GITHUB_HEAD_REF", +) + + +def _checks_out_pr_code(parsed: Dict[str, Any]) -> bool: + """Return True if a checkout step ref: points to an untrusted PR head value. + + Uses the parsed workflow dict to avoid false positives from workflows + that only reference head.sha in comments or non-checkout steps. + """ + jobs = parsed.get("jobs") or {} + for job in jobs.values(): + if not isinstance(job, dict): + continue + steps = job.get("steps") or [] + for step in steps: + if not isinstance(step, dict): + continue + uses = step.get("uses", "") or "" + if "actions/checkout" not in uses: + continue + with_block = step.get("with") or {} + ref_value = str(with_block.get("ref", "") or "") + if any(p in ref_value for p in _PR_HEAD_REF_PATTERNS): + return True + return False + + +def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: + """Detect workflows where untrusted PR input reaches a privileged context.""" + findings: List[Dict[str, Any]] = [] + workflows = github_client.get_workflows() + if workflows is None: + logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) + return findings + for wf in workflows: + path = wf.get("path", "") + if not path: + continue + content = github_client.get_workflow_content(path) + if content is None: + logger.warning("%s: could not read %s", RULE_ID, path) + continue + parsed = parse_workflow(content) + if parsed is None: + continue + dangerous = get_dangerous_triggers(parsed) + if dangerous and _checks_out_pr_code(parsed): + findings.append( + build_ci_finding( + rule_id=RULE_ID, + rule_name=RULE_NAME, + severity=SEVERITY, + category=CATEGORY, + frameworks=FRAMEWORKS, + description=DESCRIPTION, + remediation=REMEDIATION, + playbook=PLAYBOOK, + owner=owner, + repo=repo, + workflow_path=path, + metadata={"dangerous_triggers": dangerous}, + ) + ) + return findings diff --git a/tests/test_rules_ci_workflow.py b/tests/test_rules_ci_workflow.py new file mode 100644 index 00000000..6e4f7592 --- /dev/null +++ b/tests/test_rules_ci_workflow.py @@ -0,0 +1,562 @@ +"""Rule regression tests for AZ-CI-001 through AZ-CI-004 CI/CD workflow security rules.""" + +from unittest.mock import MagicMock + + +import scanner.rules.az_ci_001 as az_ci_001 +import scanner.rules.az_ci_002 as az_ci_002 +import scanner.rules.az_ci_003 as az_ci_003 +import scanner.rules.az_ci_004 as az_ci_004 +from scanner.rules._workflow_common import ( + collect_uses, + get_dangerous_triggers, + get_top_level_permissions, + is_action_pinned, + is_permissions_broad, + parse_workflow, +) + + +# Workflow with OIDC at job level (no top-level permissions block) +JOB_LEVEL_OIDC_WITH_OTHER_CRED = """ +name: Job Level OIDC +on: [push] +jobs: + deploy: + runs-on: ubuntu-latest + permissions: + id-token: write + contents: read + steps: + - uses: azure/login@v2 + with: + client-id: ${{ secrets.AZURE_CLIENT_ID }} + tenant-id: ${{ secrets.AZURE_TENANT_ID }} + subscription-id: ${{ secrets.AZURE_SUBSCRIPTION_ID }} + build: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - run: echo build +""" + +# Workflow with OIDC and migration comment mentioning old credential +OIDC_WITH_MIGRATION_COMMENT = """ +name: OIDC with comment +on: [push] +permissions: + id-token: write + contents: read +jobs: + deploy: + runs-on: ubuntu-latest + steps: + # Old workflow used ARM_CLIENT_SECRET, replaced by federated identity + - uses: azure/login@v2 + with: + client-id: ${{ secrets.AZURE_CLIENT_ID }} + tenant-id: ${{ secrets.AZURE_TENANT_ID }} + subscription-id: ${{ secrets.AZURE_SUBSCRIPTION_ID }} +""" + +# Workflow with per-job explicit permissions, no top-level block +PER_JOB_PERMISSIONS_WORKFLOW = """ +name: Per-job permissions +on: [push] +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - run: echo build + test: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - run: echo test +""" + +# Safe pull_request_target that only uses head.sha in a comment step +SAFE_PRT_BASE_CHECKOUT = """ +name: Safe PRT base checkout +on: + pull_request_target: + types: [opened] +jobs: + label: + runs-on: ubuntu-latest + permissions: + contents: read + pull-requests: write + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + with: + ref: ${{ github.event.pull_request.base.sha }} + - name: Post comment + run: echo "PR SHA is ${{ github.event.pull_request.head.sha }}" +""" + +OWNER = "test-org" +REPO = "test-repo" + +# --------------------------------------------------------------------------- +# YAML fixtures +# --------------------------------------------------------------------------- + +SECURE_WORKFLOW = """ +name: Secure CI +on: [push] +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + - run: echo hello +""" + +INSECURE_LONG_LIVED_CREDS = """ +name: Deploy +on: [push] +jobs: + deploy: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + env: + AZURE_CLIENT_SECRET: ${{ secrets.AZURE_CLIENT_SECRET }} + - run: az login --service-principal -u $CLIENT_ID -p $AZURE_CLIENT_SECRET +""" + +OIDC_WORKFLOW = """ +name: OIDC Deploy +on: [push] +permissions: + id-token: write + contents: read +jobs: + deploy: + runs-on: ubuntu-latest + steps: + - uses: azure/login@v2 + with: + client-id: ${{ secrets.AZURE_CLIENT_ID }} + tenant-id: ${{ secrets.AZURE_TENANT_ID }} + subscription-id: ${{ secrets.AZURE_SUBSCRIPTION_ID }} +""" + +BROAD_PERMISSIONS_WORKFLOW = """ +name: Broad +on: [push] +permissions: write-all +jobs: + build: + runs-on: ubuntu-latest + steps: + - run: echo hello +""" + +NO_PERMISSIONS_WORKFLOW = """ +name: No Perms +on: [push] +jobs: + build: + runs-on: ubuntu-latest + steps: + - run: echo hello +""" + +UNPINNED_ACTIONS_WORKFLOW = """ +name: Unpinned +on: [push] +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + - uses: actions/cache@v3 +""" + +PINNED_ACTIONS_WORKFLOW = """ +name: Pinned +on: [push] +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - uses: actions/setup-python@0a5c61591373683505ea898e09a3ea4f39ef2b9c +""" + +PWN_REQUEST_WORKFLOW = """ +name: Pwn Request +on: + pull_request_target: + types: [opened] +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + - run: ./ci/build.sh +""" + +SAFE_PRT_WORKFLOW = """ +name: Safe PRT +on: + pull_request_target: + types: [opened] +permissions: + contents: read +jobs: + label: + runs-on: ubuntu-latest + steps: + - run: echo "labelling only, no checkout" +""" + +MALFORMED_YAML = "on: [push\njobs: {" + + +# --------------------------------------------------------------------------- +# Helper: make a mock GitHubClient +# --------------------------------------------------------------------------- + + +def _make_client(workflows=None, contents=None): + client = MagicMock() + if workflows is None: + client.get_workflows.return_value = [{"path": ".github/workflows/ci.yml"}] + else: + client.get_workflows.return_value = workflows + if contents is None: + client.get_workflow_content.return_value = SECURE_WORKFLOW + elif isinstance(contents, dict): + client.get_workflow_content.side_effect = lambda p: contents.get(p) + else: + client.get_workflow_content.return_value = contents + return client + + +# --------------------------------------------------------------------------- +# _workflow_common unit tests +# --------------------------------------------------------------------------- + + +class TestWorkflowCommon: + def test_parse_workflow_valid(self): + wf = parse_workflow(SECURE_WORKFLOW) + assert wf is not None + assert "jobs" in wf + + def test_parse_workflow_malformed_returns_none(self): + assert parse_workflow(MALFORMED_YAML) is None + + def test_is_action_pinned_with_sha(self): + assert is_action_pinned("actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683") + + def test_is_action_pinned_with_tag_returns_false(self): + assert not is_action_pinned("actions/checkout@v4") + + def test_is_action_pinned_local_action(self): + assert is_action_pinned("./local/action") + + def test_is_action_pinned_docker(self): + assert is_action_pinned("docker://ghcr.io/my/image:latest") + + def test_collect_uses_finds_all_actions(self): + wf = parse_workflow(UNPINNED_ACTIONS_WORKFLOW) + uses = collect_uses(wf) + assert len(uses) == 3 + + def test_collect_uses_empty_workflow(self): + wf = parse_workflow("name: x\non: [push]\njobs: {}") + assert collect_uses(wf) == [] + + def test_get_top_level_permissions_none_when_absent(self): + wf = parse_workflow(NO_PERMISSIONS_WORKFLOW) + assert get_top_level_permissions(wf) is None + + def test_get_top_level_permissions_returns_dict(self): + wf = parse_workflow(SECURE_WORKFLOW) + perms = get_top_level_permissions(wf) + assert isinstance(perms, dict) + + def test_is_permissions_broad_write_all(self): + assert is_permissions_broad({"all": "write-all"}) + + def test_is_permissions_broad_none_is_broad(self): + assert is_permissions_broad(None) + + def test_is_permissions_broad_read_only_is_not_broad(self): + assert not is_permissions_broad({"contents": "read"}) + + def test_get_dangerous_triggers_detects_prt(self): + wf = parse_workflow(PWN_REQUEST_WORKFLOW) + triggers = get_dangerous_triggers(wf) + assert "pull_request_target" in triggers + + def test_get_dangerous_triggers_safe_workflow(self): + wf = parse_workflow(SECURE_WORKFLOW) + assert get_dangerous_triggers(wf) == [] + + +# --------------------------------------------------------------------------- +# AZ-CI-001 tests +# --------------------------------------------------------------------------- + + +class TestAzCi001: + def test_long_lived_creds_without_oidc_returns_finding(self): + client = _make_client(contents=INSECURE_LONG_LIVED_CREDS) + findings = az_ci_001.scan(client, OWNER, REPO) + assert len(findings) == 1 + assert findings[0]["rule_id"] == "AZ-CI-001" + assert findings[0]["severity"] == "HIGH" + + def test_oidc_workflow_no_finding(self): + client = _make_client(contents=OIDC_WORKFLOW) + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [] + + def test_oidc_with_arm_access_key_still_fails(self): + """OIDC present but ARM_ACCESS_KEY also present must still return a finding.""" + oidc_plus_key = OIDC_WORKFLOW + "\n env:\n ARM_ACCESS_KEY: ${{ secrets.ARM_ACCESS_KEY }}\n" + client = _make_client(contents=oidc_plus_key) + findings = az_ci_001.scan(client, OWNER, REPO) + assert len(findings) == 1, "OIDC + ARM_ACCESS_KEY must still flag as long-lived credential" + assert findings[0]["rule_id"] == "AZ-CI-001" + + def test_secure_workflow_no_finding(self): + client = _make_client(contents=SECURE_WORKFLOW) + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [] + + def test_none_workflows_returns_empty(self): + client = _make_client(workflows=None) + client.get_workflows.return_value = None + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [] + + def test_none_content_skipped(self): + client = _make_client(contents=None) + client.get_workflow_content.return_value = None + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [] + + def test_finding_metadata_contains_creds(self): + client = _make_client(contents=INSECURE_LONG_LIVED_CREDS) + findings = az_ci_001.scan(client, OWNER, REPO) + assert "long_lived_credentials_found" in findings[0]["metadata"] + + def test_empty_repo_no_workflows(self): + client = _make_client(workflows=[]) + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [] + + # --------------------------------------------------------------------------- + # AZ-CI-002 tests + # --------------------------------------------------------------------------- + + def test_job_level_oidc_no_finding(self): + """Job-level id-token: write with azure/login and no actual secret must pass.""" + client = _make_client(contents=JOB_LEVEL_OIDC_WITH_OTHER_CRED) + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [], "Job-level OIDC deploy + unrelated tf-state job must not flag" + + def test_migration_comment_no_finding(self): + """OIDC workflow with migration comment mentioning old credential must not flag.""" + client = _make_client(contents=OIDC_WITH_MIGRATION_COMMENT) + findings = az_ci_001.scan(client, OWNER, REPO) + assert findings == [], "Comment-only credential reference must not trigger finding" + + +class TestAzCi002: + def test_write_all_returns_finding(self): + client = _make_client(contents=BROAD_PERMISSIONS_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert len(findings) == 1 + assert findings[0]["rule_id"] == "AZ-CI-002" + + def test_no_permissions_block_returns_finding(self): + client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert len(findings) == 1 + + def test_least_privilege_no_finding(self): + client = _make_client(contents=SECURE_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert findings == [] + + def test_none_workflows_returns_empty(self): + client = _make_client() + client.get_workflows.return_value = None + findings = az_ci_002.scan(client, OWNER, REPO) + assert findings == [] + + def test_finding_metadata_contains_permissions(self): + client = _make_client(contents=BROAD_PERMISSIONS_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert "permissions_declared" in findings[0]["metadata"] + + def test_empty_repo_no_workflows(self): + client = _make_client(workflows=[]) + findings = az_ci_002.scan(client, OWNER, REPO) + assert findings == [] + + # --------------------------------------------------------------------------- + # AZ-CI-003 tests + # --------------------------------------------------------------------------- + + def test_per_job_explicit_permissions_no_finding(self): + """No top-level permissions but all jobs declare explicit permissions must not flag.""" + client = _make_client(contents=PER_JOB_PERMISSIONS_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert findings == [], "Per-job least privilege must not be flagged as broad" + + +class TestAzCi003: + def test_unpinned_actions_returns_finding(self): + client = _make_client(contents=UNPINNED_ACTIONS_WORKFLOW) + findings = az_ci_003.scan(client, OWNER, REPO) + assert len(findings) == 1 + assert findings[0]["rule_id"] == "AZ-CI-003" + + def test_pinned_actions_no_finding(self): + client = _make_client(contents=PINNED_ACTIONS_WORKFLOW) + findings = az_ci_003.scan(client, OWNER, REPO) + assert findings == [] + + def test_secure_workflow_no_finding(self): + client = _make_client(contents=SECURE_WORKFLOW) + findings = az_ci_003.scan(client, OWNER, REPO) + assert findings == [] + + def test_none_workflows_returns_empty(self): + client = _make_client() + client.get_workflows.return_value = None + findings = az_ci_003.scan(client, OWNER, REPO) + assert findings == [] + + def test_unpinned_metadata_lists_actions(self): + client = _make_client(contents=UNPINNED_ACTIONS_WORKFLOW) + findings = az_ci_003.scan(client, OWNER, REPO) + unpinned = findings[0]["metadata"]["unpinned_actions"] + assert len(unpinned) == 3 + assert all("@v" in u for u in unpinned) + + def test_malformed_yaml_skipped(self): + client = _make_client(contents=MALFORMED_YAML) + findings = az_ci_003.scan(client, OWNER, REPO) + assert findings == [] + + def test_empty_repo_no_workflows(self): + client = _make_client(workflows=[]) + findings = az_ci_003.scan(client, OWNER, REPO) + assert findings == [] + + +# --------------------------------------------------------------------------- +# AZ-CI-004 tests +# --------------------------------------------------------------------------- + + +class TestAzCi004: + def test_pwn_request_pattern_returns_finding(self): + client = _make_client(contents=PWN_REQUEST_WORKFLOW) + findings = az_ci_004.scan(client, OWNER, REPO) + assert len(findings) == 1 + assert findings[0]["rule_id"] == "AZ-CI-004" + + def test_safe_prt_without_checkout_no_finding(self): + client = _make_client(contents=SAFE_PRT_WORKFLOW) + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [] + + def test_secure_workflow_no_finding(self): + client = _make_client(contents=SECURE_WORKFLOW) + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [] + + def test_none_workflows_returns_empty(self): + client = _make_client() + client.get_workflows.return_value = None + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [] + + def test_finding_metadata_contains_triggers(self): + client = _make_client(contents=PWN_REQUEST_WORKFLOW) + findings = az_ci_004.scan(client, OWNER, REPO) + assert "dangerous_triggers" in findings[0]["metadata"] + assert "pull_request_target" in findings[0]["metadata"]["dangerous_triggers"] + + def test_malformed_yaml_skipped(self): + client = _make_client(contents=MALFORMED_YAML) + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [] + + def test_empty_repo_no_workflows(self): + client = _make_client(workflows=[]) + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [] + + # --------------------------------------------------------------------------- + # GitHubClient unit tests + # --------------------------------------------------------------------------- + + def test_prt_base_checkout_with_head_sha_in_comment_no_finding(self): + """pull_request_target checking out base ref (safe) must not flag even if + head.sha appears in a run step for display purposes.""" + client = _make_client(contents=SAFE_PRT_BASE_CHECKOUT) + findings = az_ci_004.scan(client, OWNER, REPO) + assert findings == [], "Base-ref checkout with head.sha in run step must not flag" + + +class TestGitHubClient: + def test_init_sets_owner_repo(self): + from scanner.github_client import GitHubClient + + client = GitHubClient("my-org", "my-repo", token="test-token") + assert client.owner == "my-org" + assert client.repo == "my-repo" + + def test_no_token_returns_none_from_get_token(self): + import importlib + import os + from unittest.mock import patch + + with patch.dict(os.environ, {}, clear=True): + import scanner.github_client as gh_mod + + importlib.reload(gh_mod) + token = gh_mod._get_token() + assert token is None + + def test_get_workflows_returns_none_on_api_failure(self): + from scanner.github_client import GitHubClient + from unittest.mock import patch + + client = GitHubClient("org", "repo", token="tok") + with patch.object(client, "_get", return_value=None): + result = client.get_workflows() + assert result is None + + def test_get_workflow_content_returns_none_on_failure(self): + from scanner.github_client import GitHubClient + from unittest.mock import patch + + client = GitHubClient("org", "repo", token="tok") + with patch.object(client, "_get", return_value=None): + result = client.get_workflow_content(".github/workflows/ci.yml") + assert result is None From 18d5f4e370602edc4c10da2be2442562fbef42d7 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Wed, 2 Sep 2026 02:12:50 +0100 Subject: [PATCH 02/12] fix: evaluate job-level permissions in is_permissions_broad and add regression tests - is_permissions_broad() now checks both top-level and job-level permission blocks - No top-level block with every job declaring write-all now correctly flags - Safe top-level permissions overridden by a broad job-level block now correctly flags - Per-job least-privilege (narrow explicit permissions) remains clean - Added regression tests for all 3 scenarios (53 tests total) Refs #259 Signed-off-by: Mahfuzur Rahman Emon --- scanner/rules/_workflow_common.py | 65 ++++++++++++++++++++++++------- tests/test_rules_ci_workflow.py | 49 +++++++++++++++++++++++ 2 files changed, 99 insertions(+), 15 deletions(-) diff --git a/scanner/rules/_workflow_common.py b/scanner/rules/_workflow_common.py index 34e469ec..719ae16e 100644 --- a/scanner/rules/_workflow_common.py +++ b/scanner/rules/_workflow_common.py @@ -137,23 +137,58 @@ def is_permissions_broad( permissions: Optional[Dict[str, str]], parsed: Optional[Dict[str, Any]] = None, ) -> bool: - """Return True if permissions grant write-all or multiple broad scopes. - - An absent top-level permissions block is only treated as broad when jobs - also lack explicit per-job permissions declarations. Per-job least privilege - (no top-level block, every job declares its own) is a recommended secure - pattern and must not be flagged. + """Return True if effective permissions are over-broad at any level. + + Evaluates both the top-level permissions block and every job-level + override. A workflow is broad if: + - Top-level block is write-all or grants 3+ sensitive write scopes, OR + - Any job-level override grants write-all or 3+ sensitive write scopes. + An absent top-level block is safe only when every job declares explicit + permissions AND none of those per-job blocks are themselves broad. """ - if permissions is None: - if parsed is not None: - jobs = parsed.get("jobs") or {} - if jobs and all(isinstance(job, dict) and job.get("permissions") is not None for job in jobs.values()): - return False - return True - if permissions.get("all") in ("write-all", "write"): + + def _block_is_broad(perms: Optional[Dict[str, str]]) -> bool: + if perms is None: + return False # None means not declared at this level + if isinstance(perms, str): + return perms in ("write-all", "write") + if not isinstance(perms, dict): + return False + if perms.get("all") in ("write-all", "write"): + return True + broad_count = sum(1 for k, v in perms.items() if str(v).lower() == "write" and k in _BROAD_PERMISSIONS) + return broad_count >= 3 + + # Check top-level permissions + if permissions is not None and _block_is_broad(permissions): return True - broad_count = sum(1 for k, v in permissions.items() if v == "write" and k in _BROAD_PERMISSIONS) - return broad_count >= 3 + + jobs = (parsed or {}).get("jobs") or {} + + if permissions is None: + # No top-level block: safe only if every job declares explicit + # non-broad permissions + if not jobs: + return True # no jobs, treat as unconstrained + for job in jobs.values(): + if not isinstance(job, dict): + continue + job_perms = job.get("permissions") + if job_perms is None: + return True # job has no permissions declaration + if _block_is_broad(job_perms): + return True + return False + + # Top-level block present and not broad: also check job-level overrides + for job in jobs.values(): + if not isinstance(job, dict): + continue + job_perms = job.get("permissions") + if job_perms is not None and _block_is_broad(job_perms): + return True + + return False def get_dangerous_triggers(workflow: Dict[str, Any]) -> List[str]: diff --git a/tests/test_rules_ci_workflow.py b/tests/test_rules_ci_workflow.py index 6e4f7592..23fff383 100644 --- a/tests/test_rules_ci_workflow.py +++ b/tests/test_rules_ci_workflow.py @@ -61,6 +61,41 @@ """ # Workflow with per-job explicit permissions, no top-level block +# Every job declares write-all — no top-level block +ALL_JOBS_WRITE_ALL_WORKFLOW = """ +name: All jobs write-all +on: [push] +jobs: + build: + runs-on: ubuntu-latest + permissions: write-all + steps: + - run: echo build + deploy: + runs-on: ubuntu-latest + permissions: write-all + steps: + - run: echo deploy +""" + +# Safe top-level but one job overrides with write-all +SAFE_TOPLEVEL_BROAD_JOB_WORKFLOW = """ +name: Safe top broad job +on: [push] +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + steps: + - run: echo build + privileged: + runs-on: ubuntu-latest + permissions: write-all + steps: + - run: echo privileged +""" + PER_JOB_PERMISSIONS_WORKFLOW = """ name: Per-job permissions on: [push] @@ -425,6 +460,20 @@ def test_per_job_explicit_permissions_no_finding(self): findings = az_ci_002.scan(client, OWNER, REPO) assert findings == [], "Per-job least privilege must not be flagged as broad" + def test_all_jobs_write_all_returns_finding(self): + """No top-level block but every job declares write-all must flag.""" + client = _make_client(contents=ALL_JOBS_WRITE_ALL_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert len(findings) == 1, "All-jobs write-all must be flagged" + assert findings[0]["rule_id"] == "AZ-CI-002" + + def test_safe_toplevel_broad_job_returns_finding(self): + """Safe top-level permissions overridden by one broad job must flag.""" + client = _make_client(contents=SAFE_TOPLEVEL_BROAD_JOB_WORKFLOW) + findings = az_ci_002.scan(client, OWNER, REPO) + assert len(findings) == 1, "Job-level write-all override must be flagged" + assert findings[0]["rule_id"] == "AZ-CI-002" + class TestAzCi003: def test_unpinned_actions_returns_finding(self): From aeda4972d2866c65f5a8f5e91ef813d8f333da95 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Wed, 2 Sep 2026 02:12:50 +0100 Subject: [PATCH 03/12] fix: add CI scan entry point, workflow_run detection and UNKNOWN permissions handling - scanner/ci_engine.py: new CIScanEngine returning RuleEvaluation PASS/FAIL/UNKNOWN - az_ci_004: add workflow_run head_sha/head_branch to dangerous checkout patterns - az_ci_002: return UNKNOWN finding when repo default permissions unavailable, skip when repo default is restricted (read-only) - tests: add regression tests for workflow_run checkout, restricted repo default, unavailable repo default, and 7 CIScanEngine evaluation contract tests - 63 tests total Refs #259 Signed-off-by: Mahfuzur Rahman Emon --- scanner/ci_engine.py | 197 ++++++++++++++++++++++++++++++++ scanner/rules/az_ci_002.py | 67 ++++++++++- scanner/rules/az_ci_004.py | 3 + tests/test_rules_ci_workflow.py | 133 +++++++++++++++++++++ 4 files changed, 397 insertions(+), 3 deletions(-) create mode 100644 scanner/ci_engine.py diff --git a/scanner/ci_engine.py b/scanner/ci_engine.py new file mode 100644 index 00000000..a9789133 --- /dev/null +++ b/scanner/ci_engine.py @@ -0,0 +1,197 @@ +"""CI/CD scan engine for GitHub workflow security rules (AZ-CI-*). + +Runs AZ-CI-001..004 against a GitHub repository and returns +RuleEvaluation objects with PASS/FAIL/UNKNOWN status for each +workflow file, compatible with the existing evaluation contract +in scanner/evaluation.py. + +Usage: + from scanner.ci_engine import CIScanEngine + engine = CIScanEngine(owner="my-org", repo="my-repo") + results = engine.run_scan() +""" + +import logging +from typing import List, Optional + +import scanner.rules.az_ci_001 as az_ci_001 +import scanner.rules.az_ci_002 as az_ci_002 +import scanner.rules.az_ci_003 as az_ci_003 +import scanner.rules.az_ci_004 as az_ci_004 +from scanner.evaluation import EvaluationStatus, RuleEvaluation +from scanner.github_client import GitHubClient + +logger = logging.getLogger(__name__) + +_CI_RULES = [az_ci_001, az_ci_002, az_ci_003, az_ci_004] + +RESOURCE_TYPE = "GitHub/WorkflowFile" + + +def _resource_id(owner: str, repo: str, workflow_path: str) -> str: + """Canonical resource ID for a workflow file.""" + return f"github/{owner}/{repo}/workflows/{workflow_path}" + + +def _repo_scope_id(owner: str, repo: str) -> str: + """Canonical resource ID when no workflows exist.""" + return f"github/{owner}/{repo}" + + +class CIScanEngine: + """Runs AZ-CI-* rules against a GitHub repository. + + Each rule is run against every workflow file. Results are returned + as RuleEvaluation objects with PASS/FAIL/UNKNOWN status. + + PASS — workflow was evaluated and is compliant. + FAIL — workflow was evaluated and has a finding. + UNKNOWN — workflow content could not be read or permissions + could not be determined. + """ + + def __init__( + self, + owner: str, + repo: str, + token: Optional[str] = None, + ) -> None: + self.owner = owner + self.repo = repo + self.client = GitHubClient(owner=owner, repo=repo, token=token) + + def run_scan(self) -> List[RuleEvaluation]: + """Run all AZ-CI-* rules and return RuleEvaluation results.""" + evaluations: List[RuleEvaluation] = [] + + workflows = self.client.get_workflows() + if workflows is None: + # Cannot enumerate workflows — all rules return UNKNOWN + for rule in _CI_RULES: + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_repo_scope_id(self.owner, self.repo), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.UNKNOWN, + reason_code="WORKFLOWS_UNAVAILABLE", + reason=( + "GitHub API did not return workflow list. " + "Check GITHUB_TOKEN permissions (actions: read required)." + ), + ) + ) + return evaluations + + if not workflows: + # Empty repo — not applicable + for rule in _CI_RULES: + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_repo_scope_id(self.owner, self.repo), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.NOT_APPLICABLE, + reason_code="NO_WORKFLOWS", + reason="Repository has no GitHub Actions workflow files.", + ) + ) + return evaluations + + # Build a set of workflow paths that had findings per rule + for rule in _CI_RULES: + try: + findings = rule.scan(self.client, self.owner, self.repo) + except Exception as exc: + logger.error("Error running rule %s: %s", rule.RULE_ID, exc) + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_repo_scope_id(self.owner, self.repo), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.ERROR, + reason_code="RULE_EXCEPTION", + reason=f"Rule raised an exception: {exc}", + ) + ) + continue + + # Map findings to workflow paths + finding_paths = {f.get("resource_name", "") for f in findings} + + # Emit one evaluation per workflow + for wf in workflows: + path = wf.get("path", "") + if not path: + continue + + content = self.client.get_workflow_content(path) + if content is None: + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_resource_id(self.owner, self.repo, path), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.UNKNOWN, + reason_code="WORKFLOW_CONTENT_UNAVAILABLE", + reason=( + f"Workflow file {path} could not be read. " + "File may exceed 1 MB or token lacks contents: read." + ), + ) + ) + continue + + # Check if this workflow had an UNKNOWN finding (AZ-CI-002 emits these) + unknown_finding = next( + ( + f + for f in findings + if f.get("resource_name") == path + and f.get("metadata", {}).get("effective_permissions") == "UNKNOWN" + ), + None, + ) + if unknown_finding: + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_resource_id(self.owner, self.repo, path), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.UNKNOWN, + reason_code="PERMISSIONS_DEFAULT_UNKNOWN", + reason=unknown_finding.get("description", ""), + evidence={"workflow_path": path}, + ) + ) + elif path in finding_paths: + # FAIL — workflow has a real finding + finding = next( + (f for f in findings if f.get("resource_name") == path), + None, + ) + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_resource_id(self.owner, self.repo, path), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.FAIL, + reason_code="POLICY_VIOLATION", + reason=rule.RULE_NAME, + finding=finding, + evidence={"workflow_path": path}, + ) + ) + else: + # PASS — workflow was evaluated and is compliant + evaluations.append( + RuleEvaluation( + rule_id=rule.RULE_ID, + resource_id=_resource_id(self.owner, self.repo, path), + resource_type=RESOURCE_TYPE, + status=EvaluationStatus.PASS, + evidence={"workflow_path": path}, + ) + ) + + return evaluations diff --git a/scanner/rules/az_ci_002.py b/scanner/rules/az_ci_002.py index 0daf00c0..fa0b93b4 100644 --- a/scanner/rules/az_ci_002.py +++ b/scanner/rules/az_ci_002.py @@ -22,10 +22,17 @@ } DESCRIPTION = ( "The workflow does not declare a top-level permissions block or declares write access " - "to multiple sensitive scopes. Without an explicit permissions block GitHub grants " - "read/write access to all scopes by default. Over-broad GITHUB_TOKEN permissions allow " + "to multiple sensitive scopes. Without an explicit permissions block the effective " + "GITHUB_TOKEN permissions depend on the repository or organisation default, which " + "OpenShield cannot read without repository metadata access. Over-broad permissions allow " "a compromised step to push code, modify releases, or exfiltrate secrets." ) + +UNKNOWN_REASON = ( + "Cannot determine effective token permissions: the workflow has no explicit permissions " + "block and the repository or organisation default permissions are not available. " + "Set GITHUB_TOKEN permissions explicitly in the workflow to resolve this." +) REMEDIATION = ( "Add a top-level permissions block that grants only the minimum scopes required: " "permissions:\n contents: read\n" @@ -37,12 +44,27 @@ def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: - """Detect workflows with broad or absent token permissions.""" + """Detect workflows with broad or absent token permissions. + + When a workflow has no explicit permissions block, effective permissions + depend on the repository or organisation default which we cannot read + without additional API access. In that case the finding is emitted with + severity MEDIUM and a note that the result is UNKNOWN until the default + can be confirmed. + """ findings: List[Dict[str, Any]] = [] workflows = github_client.get_workflows() if workflows is None: logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) return findings + + # Attempt to read repo default permissions (requires repo metadata) + repo_info = github_client.get_repo_info() + repo_default_permissions = None + if repo_info is not None: + # GitHub API exposes default_workflow_permissions on repo settings + repo_default_permissions = repo_info.get("default_workflow_permissions") + for wf in workflows: path = wf.get("path", "") if not path: @@ -55,6 +77,45 @@ def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: if parsed is None: continue perms = get_top_level_permissions(parsed) + + if perms is None: + # Check job-level permissions before deciding + jobs = parsed.get("jobs") or {} + all_jobs_explicit = jobs and all( + isinstance(j, dict) and j.get("permissions") is not None for j in jobs.values() + ) + if all_jobs_explicit and not is_permissions_broad(perms, parsed): + # Every job declares explicit least-privilege permissions + continue + + # No explicit permissions anywhere — check repo default + if repo_default_permissions == "read": + # Org/repo default is restricted — not broad, skip + continue + elif repo_default_permissions is None: + # Cannot determine default — emit with UNKNOWN note + findings.append( + build_ci_finding( + rule_id=RULE_ID, + rule_name=RULE_NAME, + severity="MEDIUM", + category=CATEGORY, + frameworks=FRAMEWORKS, + description=UNKNOWN_REASON, + remediation=REMEDIATION, + playbook=PLAYBOOK, + owner=owner, + repo=repo, + workflow_path=path, + metadata={ + "permissions_declared": None, + "effective_permissions": "UNKNOWN", + "reason": "repo_default_permissions_unavailable", + }, + ) + ) + continue + if is_permissions_broad(perms, parsed): findings.append( build_ci_finding( diff --git a/scanner/rules/az_ci_004.py b/scanner/rules/az_ci_004.py index 2a7120cc..88722052 100644 --- a/scanner/rules/az_ci_004.py +++ b/scanner/rules/az_ci_004.py @@ -40,6 +40,9 @@ "github.event.pull_request.head.ref", "github.head_ref", "GITHUB_HEAD_REF", + # workflow_run: checkout of the triggering run's untrusted head + "github.event.workflow_run.head_sha", + "github.event.workflow_run.head_branch", ) diff --git a/tests/test_rules_ci_workflow.py b/tests/test_rules_ci_workflow.py index 23fff383..047a235b 100644 --- a/tests/test_rules_ci_workflow.py +++ b/tests/test_rules_ci_workflow.py @@ -247,6 +247,28 @@ - run: ./ci/build.sh """ +# workflow_run that checks out untrusted head_sha +WORKFLOW_RUN_CHECKOUT = """ +name: Workflow Run Checkout +on: + workflow_run: + workflows: ["CI"] + types: [completed] +jobs: + deploy: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + with: + ref: ${{ github.event.workflow_run.head_sha }} + - run: ./deploy.sh +""" + +# Repository with restricted default permissions (read-only) +RESTRICTED_DEFAULT_PERMISSIONS = "read" + SAFE_PRT_WORKFLOW = """ name: Safe PRT on: @@ -474,6 +496,22 @@ def test_safe_toplevel_broad_job_returns_finding(self): assert len(findings) == 1, "Job-level write-all override must be flagged" assert findings[0]["rule_id"] == "AZ-CI-002" + def test_no_permissions_with_restricted_repo_default_no_finding(self): + """Workflow with no permissions block + repo default=read must not flag.""" + mock_repo_info = {"default_workflow_permissions": RESTRICTED_DEFAULT_PERMISSIONS} + client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) + client.get_repo_info = lambda: mock_repo_info + findings = az_ci_002.scan(client, OWNER, REPO) + assert findings == [], "Restricted repo default must suppress broad-permissions finding" + + def test_no_permissions_unknown_repo_default_returns_finding(self): + """Workflow with no permissions block + unavailable repo default returns UNKNOWN finding.""" + client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) + client.get_repo_info = lambda: None + findings = az_ci_002.scan(client, OWNER, REPO) + assert len(findings) == 1 + assert findings[0]["metadata"]["effective_permissions"] == "UNKNOWN" + class TestAzCi003: def test_unpinned_actions_returns_finding(self): @@ -571,6 +609,14 @@ def test_prt_base_checkout_with_head_sha_in_comment_no_finding(self): findings = az_ci_004.scan(client, OWNER, REPO) assert findings == [], "Base-ref checkout with head.sha in run step must not flag" + def test_workflow_run_head_sha_checkout_returns_finding(self): + """workflow_run trigger + checkout of head_sha must flag as pwn-request risk.""" + client = _make_client(contents=WORKFLOW_RUN_CHECKOUT) + findings = az_ci_004.scan(client, OWNER, REPO) + assert len(findings) == 1, "workflow_run + head_sha checkout must flag" + assert findings[0]["rule_id"] == "AZ-CI-004" + assert "workflow_run" in findings[0]["metadata"]["dangerous_triggers"] + class TestGitHubClient: def test_init_sets_owner_repo(self): @@ -609,3 +655,90 @@ def test_get_workflow_content_returns_none_on_failure(self): with patch.object(client, "_get", return_value=None): result = client.get_workflow_content(".github/workflows/ci.yml") assert result is None + + +# --------------------------------------------------------------------------- +# CIScanEngine tests +# --------------------------------------------------------------------------- + + +class TestCIScanEngine: + """Tests for scanner/ci_engine.py PASS/FAIL/UNKNOWN evaluation contract.""" + + def _make_engine(self, contents=None, workflows=None, repo_info=None): + from scanner.ci_engine import CIScanEngine + + engine = CIScanEngine(owner=OWNER, repo=REPO, token="test-token") + if workflows is None: + engine.client.get_workflows = lambda: [{"path": ".github/workflows/ci.yml"}] + else: + engine.client.get_workflows = lambda: workflows + if contents is None: + engine.client.get_workflow_content = lambda p: SECURE_WORKFLOW + else: + engine.client.get_workflow_content = lambda p: contents + engine.client.get_repo_info = lambda: repo_info or {} + return engine + + def test_secure_workflow_returns_all_pass(self): + """A fully compliant workflow returns PASS for all 4 rules.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine(contents=PINNED_ACTIONS_WORKFLOW) + results = engine.run_scan() + statuses = {r.status for r in results} + assert EvaluationStatus.FAIL not in statuses + assert EvaluationStatus.PASS in statuses + + def test_insecure_workflow_returns_fail(self): + """A workflow with long-lived creds returns at least one FAIL.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine(contents=INSECURE_LONG_LIVED_CREDS) + results = engine.run_scan() + assert any(r.status == EvaluationStatus.FAIL for r in results) + fail = next(r for r in results if r.status == EvaluationStatus.FAIL) + assert fail.rule_id == "AZ-CI-001" + + def test_unavailable_workflows_returns_unknown(self): + """When get_workflows returns None all rules emit UNKNOWN.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine() + engine.client.get_workflows = lambda: None + results = engine.run_scan() + assert all(r.status == EvaluationStatus.UNKNOWN for r in results) + assert all(r.reason_code == "WORKFLOWS_UNAVAILABLE" for r in results) + + def test_empty_repo_returns_not_applicable(self): + """A repo with no workflows returns NOT_APPLICABLE for all rules.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine(workflows=[]) + results = engine.run_scan() + assert all(r.status == EvaluationStatus.NOT_APPLICABLE for r in results) + + def test_unreadable_workflow_returns_unknown(self): + """A workflow file that cannot be read returns UNKNOWN per rule.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine() + engine.client.get_workflow_content = lambda p: None + results = engine.run_scan() + assert any(r.status == EvaluationStatus.UNKNOWN for r in results) + unknown = [r for r in results if r.status == EvaluationStatus.UNKNOWN] + assert all(r.reason_code == "WORKFLOW_CONTENT_UNAVAILABLE" for r in unknown) + + def test_evaluations_have_correct_resource_ids(self): + """Resource IDs follow the github/owner/repo/workflows/path pattern.""" + engine = self._make_engine(contents=SECURE_WORKFLOW) + results = engine.run_scan() + for r in results: + assert r.resource_id.startswith(f"github/{OWNER}/{REPO}") + + def test_four_rules_evaluated(self): + """All 4 AZ-CI-* rules produce evaluations.""" + engine = self._make_engine(contents=SECURE_WORKFLOW) + results = engine.run_scan() + rule_ids = {r.rule_id for r in results} + assert rule_ids == {"AZ-CI-001", "AZ-CI-002", "AZ-CI-003", "AZ-CI-004"} From c981b48be35dcfbb2849804d58978100d202a25c Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 13 Sep 2026 04:15:31 +0100 Subject: [PATCH 04/12] fix: cache workflow content, fix finding path matching, log app token exception - ci_engine: pre-fetch and cache workflow content once (was 4N API calls, now N) - ci_engine: match findings on metadata.workflow_path not resource_name to avoid false PASS - ci_engine: unknown_finding branch now also requires effective_permissions==UNKNOWN so FAIL findings are not misclassified - github_client: log exception type in _get_app_token for auth debugging Refs #259 Signed-off-by: Mahfuzur Rahman Emon --- scanner/ci_engine.py | 18 +++++++++++++----- scanner/github_client.py | 7 ++++--- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/scanner/ci_engine.py b/scanner/ci_engine.py index a9789133..8ae8a05a 100644 --- a/scanner/ci_engine.py +++ b/scanner/ci_engine.py @@ -98,6 +98,15 @@ def run_scan(self) -> List[RuleEvaluation]: ) return evaluations + # Pre-fetch and cache all workflow content once, keyed by path, + # so each file is fetched exactly once instead of 4N times + # (4 rules x N files). None means the content was unreadable. + content_cache: dict = {} + for wf in workflows: + path = wf.get("path", "") + if path and path not in content_cache: + content_cache[path] = self.client.get_workflow_content(path) + # Build a set of workflow paths that had findings per rule for rule in _CI_RULES: try: @@ -117,15 +126,14 @@ def run_scan(self) -> List[RuleEvaluation]: continue # Map findings to workflow paths - finding_paths = {f.get("resource_name", "") for f in findings} - + finding_paths = {f.get("metadata", {}).get("workflow_path", "") for f in findings} # Emit one evaluation per workflow for wf in workflows: path = wf.get("path", "") if not path: continue - content = self.client.get_workflow_content(path) + content = content_cache.get(path) if content is None: evaluations.append( RuleEvaluation( @@ -147,7 +155,7 @@ def run_scan(self) -> List[RuleEvaluation]: ( f for f in findings - if f.get("resource_name") == path + if f.get("metadata", {}).get("workflow_path") == path and f.get("metadata", {}).get("effective_permissions") == "UNKNOWN" ), None, @@ -167,7 +175,7 @@ def run_scan(self) -> List[RuleEvaluation]: elif path in finding_paths: # FAIL — workflow has a real finding finding = next( - (f for f in findings if f.get("resource_name") == path), + (f for f in findings if f.get("metadata", {}).get("workflow_path") == path), None, ) evaluations.append( diff --git a/scanner/github_client.py b/scanner/github_client.py index 08762eb8..9ea01606 100644 --- a/scanner/github_client.py +++ b/scanner/github_client.py @@ -51,10 +51,11 @@ def _get_app_token() -> Optional[str]: logger.error( "GitHub App token request failed (HTTP status indicates rejection)" ) # nosemgrep: python-logger-credential-disclosure - except Exception: + except Exception as exc: logger.error( - "GitHub App token generation failed (check GITHUB_APP_* env vars)" - ) # nosemgrep: python-logger-credential-disclosure + "GitHub App token generation failed (check GITHUB_APP_* env vars): %s", + type(exc).__name__, # nosemgrep: python-logger-credential-disclosure + ) return None From f10c3feeb83cbb24289b2f60c9186df6c6770827 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 13 Sep 2026 04:19:57 +0100 Subject: [PATCH 05/12] fix: move nosemgrep suppression to logger.error line in _get_app_token Signed-off-by: Mahfuzur Rahman Emon --- scanner/github_client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scanner/github_client.py b/scanner/github_client.py index 9ea01606..ec5a311c 100644 --- a/scanner/github_client.py +++ b/scanner/github_client.py @@ -52,9 +52,9 @@ def _get_app_token() -> Optional[str]: "GitHub App token request failed (HTTP status indicates rejection)" ) # nosemgrep: python-logger-credential-disclosure except Exception as exc: - logger.error( + logger.error( # nosemgrep: python-logger-credential-disclosure "GitHub App token generation failed (check GITHUB_APP_* env vars): %s", - type(exc).__name__, # nosemgrep: python-logger-credential-disclosure + type(exc).__name__, ) return None From 0aacd3c43b3c07d1b823a4cbfade4477bd8a43ea Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Fri, 18 Sep 2026 20:47:17 +0100 Subject: [PATCH 06/12] fix(ci-rules): read default workflow permissions from correct endpoint; add CLI runner - github_client: add get_workflow_permissions() calling GET /repos/{owner}/{repo}/actions/permissions/workflow (the only endpoint exposing default_workflow_permissions; requires administration: read). Returns None -> UNKNOWN when unavailable. - az_ci_002: read the repo default via get_workflow_permissions() instead of get_repo_info(), which never returns that field (would always be UNKNOWN). - ci_engine: add 'python -m scanner.ci_engine --owner --repo' CLI so the AZ-CI controls are runnable now; persisted-scan + API-route wiring scoped to issue #259 PR 3/3 (engine.py comment updated to point at the runner). - tests: mock get_workflow_permissions; add client coverage for it. Refs #259 Signed-off-by: Mahfuzur Rahman Emon --- scanner/ci_engine.py | 36 +++++++++++++++++++++++++++++++++ scanner/engine.py | 7 ++++--- scanner/github_client.py | 21 +++++++++++++++++++ scanner/rules/az_ci_002.py | 11 +++++----- tests/test_rules_ci_workflow.py | 25 ++++++++++++++++++++--- 5 files changed, 88 insertions(+), 12 deletions(-) diff --git a/scanner/ci_engine.py b/scanner/ci_engine.py index 8ae8a05a..64360c6c 100644 --- a/scanner/ci_engine.py +++ b/scanner/ci_engine.py @@ -203,3 +203,39 @@ def run_scan(self) -> List[RuleEvaluation]: ) return evaluations + + +def _main() -> int: + """CLI entry point: run AZ-CI-001..004 against a repository. + + Usage: + python -m scanner.ci_engine --owner --repo [--json] + + Interim executable runner for the CI/CD controls. Persisted-scan storage + and an API route that invoke CIScanEngine are scoped to issue #259 PR 3/3; + this CLI ensures the rules are runnable rather than merged as dead code. + """ + import argparse + import json as _json + + parser = argparse.ArgumentParser(description="Run OpenShield CI/CD workflow security rules (AZ-CI-001..004)") + parser.add_argument("--owner", required=True, help="GitHub organisation or user") + parser.add_argument("--repo", required=True, help="Repository name") + parser.add_argument("--json", action="store_true", help="Emit results as JSON") + args = parser.parse_args() + + engine = CIScanEngine(owner=args.owner, repo=args.repo) + results = engine.run_scan() + + if args.json: + print(_json.dumps([r.to_dict() for r in results], indent=2)) + else: + for r in results: + print(f"{r.status:15} {r.rule_id:10} {r.resource_id}") + + # Non-zero exit when any rule FAILed so a CI/CD job can gate on it. + return 1 if any(r.status == EvaluationStatus.FAIL for r in results) else 0 + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/scanner/engine.py b/scanner/engine.py index bf7c1714..db194a65 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -73,9 +73,10 @@ def load_rules(self) -> None: # counts. A misnamed scratch module must never execute against a real # subscription without first passing the rule checks. for rule_path in sorted(RULES_DIR.glob("az_*.py")): - # AZ-CI rules require a GitHubClient, not an AzureClient. - # They are excluded from the Azure scan engine and invoked - # separately by the CI/CD scan entry point. + # AZ-CI rules require a GitHubClient, not an AzureClient, so the + # Azure scan engine never loads them. They run via + # scanner.ci_engine.CIScanEngine (CLI: python -m scanner.ci_engine); + # persisted-scan and API-route wiring is tracked in issue #259 PR 3/3. if rule_path.stem.startswith("az_ci_"): logger.debug("Skipping CI rule %s (requires GitHubClient)", rule_path.name) continue diff --git a/scanner/github_client.py b/scanner/github_client.py index ec5a311c..983542ff 100644 --- a/scanner/github_client.py +++ b/scanner/github_client.py @@ -196,3 +196,24 @@ def get_repo_info(self) -> Optional[Dict[str, Any]]: except Exception as exc: logger.error("Failed to parse repo info: %s", exc) return None + + def get_workflow_permissions(self) -> Optional[str]: + """Return the repository's default GITHUB_TOKEN workflow permissions. + + Reads GET /repos/{owner}/{repo}/actions/permissions/workflow, the only + endpoint that exposes default_workflow_permissions. Requires the token + to have administration: read on the repository. + + Returns "read" or "write" when the setting can be read, or None when + the endpoint is unavailable (missing administration: read, 403/404, or + any other failure). Callers MUST treat None as UNKNOWN, never as a + compliant default. + """ + resp = self._get(f"/repos/{self.owner}/{self.repo}/actions/permissions/workflow") + if resp is None: + return None + try: + return resp.json().get("default_workflow_permissions") + except Exception as exc: + logger.error("Failed to parse workflow permissions: %s", exc) + return None diff --git a/scanner/rules/az_ci_002.py b/scanner/rules/az_ci_002.py index fa0b93b4..213430e3 100644 --- a/scanner/rules/az_ci_002.py +++ b/scanner/rules/az_ci_002.py @@ -58,12 +58,11 @@ def scan(github_client: Any, owner: str, repo: str) -> List[Dict[str, Any]]: logger.warning("%s: workflows could not be enumerated for %s/%s", RULE_ID, owner, repo) return findings - # Attempt to read repo default permissions (requires repo metadata) - repo_info = github_client.get_repo_info() - repo_default_permissions = None - if repo_info is not None: - # GitHub API exposes default_workflow_permissions on repo settings - repo_default_permissions = repo_info.get("default_workflow_permissions") + # default_workflow_permissions is exposed ONLY by the dedicated + # actions/permissions/workflow endpoint (requires administration: read), + # not by GET /repos/{owner}/{repo}. None means the effective default is + # UNKNOWN and is never assumed compliant. + repo_default_permissions = github_client.get_workflow_permissions() for wf in workflows: path = wf.get("path", "") diff --git a/tests/test_rules_ci_workflow.py b/tests/test_rules_ci_workflow.py index 047a235b..0e3af3c7 100644 --- a/tests/test_rules_ci_workflow.py +++ b/tests/test_rules_ci_workflow.py @@ -498,16 +498,15 @@ def test_safe_toplevel_broad_job_returns_finding(self): def test_no_permissions_with_restricted_repo_default_no_finding(self): """Workflow with no permissions block + repo default=read must not flag.""" - mock_repo_info = {"default_workflow_permissions": RESTRICTED_DEFAULT_PERMISSIONS} client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) - client.get_repo_info = lambda: mock_repo_info + client.get_workflow_permissions = lambda: RESTRICTED_DEFAULT_PERMISSIONS findings = az_ci_002.scan(client, OWNER, REPO) assert findings == [], "Restricted repo default must suppress broad-permissions finding" def test_no_permissions_unknown_repo_default_returns_finding(self): """Workflow with no permissions block + unavailable repo default returns UNKNOWN finding.""" client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) - client.get_repo_info = lambda: None + client.get_workflow_permissions = lambda: None findings = az_ci_002.scan(client, OWNER, REPO) assert len(findings) == 1 assert findings[0]["metadata"]["effective_permissions"] == "UNKNOWN" @@ -656,6 +655,26 @@ def test_get_workflow_content_returns_none_on_failure(self): result = client.get_workflow_content(".github/workflows/ci.yml") assert result is None + def test_get_workflow_permissions_returns_value(self): + from unittest.mock import MagicMock, patch + + from scanner.github_client import GitHubClient + + client = GitHubClient("org", "repo", token="tok") + resp = MagicMock() + resp.json.return_value = {"default_workflow_permissions": "read"} + with patch.object(client, "_get", return_value=resp): + assert client.get_workflow_permissions() == "read" + + def test_get_workflow_permissions_none_on_failure(self): + from unittest.mock import patch + + from scanner.github_client import GitHubClient + + client = GitHubClient("org", "repo", token="tok") + with patch.object(client, "_get", return_value=None): + assert client.get_workflow_permissions() is None + # --------------------------------------------------------------------------- # CIScanEngine tests From 7ce27203d8cff4bb45add8d35ae0c5b9f9354e10 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 11 Oct 2026 23:51:57 +0100 Subject: [PATCH 07/12] fix(ci-rules): map AZ-CI rules into the ISO 27001:2022 pack and framework files Signed-off-by: Mahfuzur Rahman Emon --- .../frameworks/cis_azure_benchmark.json | 48 +++++++++++++++++++ compliance/frameworks/iso27001.json | 48 +++++++++++++++++++ compliance/frameworks/nist_csf.json | 48 +++++++++++++++++++ compliance/frameworks/soc2.json | 48 +++++++++++++++++++ scanner/rules/az_ci_001.py | 2 +- scanner/rules/az_ci_002.py | 2 +- scanner/rules/az_ci_003.py | 2 +- scanner/rules/az_ci_004.py | 2 +- 8 files changed, 196 insertions(+), 4 deletions(-) diff --git a/compliance/frameworks/cis_azure_benchmark.json b/compliance/frameworks/cis_azure_benchmark.json index f3181a4e..033cacd9 100644 --- a/compliance/frameworks/cis_azure_benchmark.json +++ b/compliance/frameworks/cis_azure_benchmark.json @@ -1734,6 +1734,54 @@ "owner": null, "review_status": "pending_review", "review_date": null + }, + "AZ-CI-001": { + "control_id": "N/A-CI-001", + "control_name": "CI/CD Workflow Uses Long-Lived Azure Credentials", + "description": "CI/CD workflows should use workload identity federation (OIDC) instead of long-lived client secrets or storage keys stored as repository secrets.", + "mapping_type": "not_applicable", + "evidence_type": "not_applicable", + "primary_source": "CIS Microsoft Azure Foundations Benchmark v2.0.0, control N/A-FUNC-001", + "rationale": "CIS Microsoft Azure Foundations Benchmark v2.0.0 has no numbered control this rule maps to (N/A-FUNC-001: CI/CD Workflow Uses Long-Lived Azure Credentials) - this rule's evidence is not counted as a CIS pass/fail because there is no corresponding CIS requirement for it to be evidence of.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-002": { + "control_id": "N/A-CI-002", + "control_name": "CI/CD Workflow Has Unnecessarily Broad Token Permissions", + "description": "Workflow GITHUB_TOKEN permissions should follow least privilege with an explicit permissions block granting only required scopes.", + "mapping_type": "not_applicable", + "evidence_type": "not_applicable", + "primary_source": "CIS Microsoft Azure Foundations Benchmark v2.0.0, control N/A-FUNC-001", + "rationale": "CIS Microsoft Azure Foundations Benchmark v2.0.0 has no numbered control this rule maps to (N/A-FUNC-001: CI/CD Workflow Has Unnecessarily Broad Token Permissions) - this rule's evidence is not counted as a CIS pass/fail because there is no corresponding CIS requirement for it to be evidence of.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-003": { + "control_id": "N/A-CI-003", + "control_name": "Third-Party Workflow Action Not Pinned to Immutable Commit SHA", + "description": "Third-party GitHub Actions must be pinned to a full 40-character commit SHA to prevent supply-chain attacks via mutable tags.", + "mapping_type": "not_applicable", + "evidence_type": "not_applicable", + "primary_source": "CIS Microsoft Azure Foundations Benchmark v2.0.0, control N/A-FUNC-001", + "rationale": "CIS Microsoft Azure Foundations Benchmark v2.0.0 has no numbered control this rule maps to (N/A-FUNC-001: Third-Party Workflow Action Not Pinned to Immutable Commit SHA) - this rule's evidence is not counted as a CIS pass/fail because there is no corresponding CIS requirement for it to be evidence of.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-004": { + "control_id": "N/A-CI-004", + "control_name": "Untrusted Pull-Request Input Reaches Privileged Workflow Context", + "description": "Workflows triggered by pull_request_target or workflow_run must not check out or execute untrusted PR code in a context with secret access.", + "mapping_type": "not_applicable", + "evidence_type": "not_applicable", + "primary_source": "CIS Microsoft Azure Foundations Benchmark v2.0.0, control N/A-FUNC-001", + "rationale": "CIS Microsoft Azure Foundations Benchmark v2.0.0 has no numbered control this rule maps to (N/A-FUNC-001: Untrusted Pull-Request Input Reaches Privileged Workflow Context) - this rule's evidence is not counted as a CIS pass/fail because there is no corresponding CIS requirement for it to be evidence of.", + "owner": null, + "review_status": "pending_review", + "review_date": null } } } diff --git a/compliance/frameworks/iso27001.json b/compliance/frameworks/iso27001.json index 5ccaf205..cb2fe905 100644 --- a/compliance/frameworks/iso27001.json +++ b/compliance/frameworks/iso27001.json @@ -1734,6 +1734,54 @@ "owner": null, "review_status": "pending_review", "review_date": null + }, + "AZ-CI-001": { + "control_id": "A.5.17", + "control_name": "Authentication information", + "description": "CI/CD workflows should use workload identity federation (OIDC) instead of long-lived client secrets or storage keys stored as repository secrets.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "ISO/IEC 27001:2022, Annex A control A.5.17", + "rationale": "ISO/IEC 27001:2022 Annex A control A.5.17 ('Authentication information') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-001 ('CI/CD Workflow Uses Long-Lived Azure Credentials') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-002": { + "control_id": "A.8.2", + "control_name": "Privileged access rights", + "description": "Workflow GITHUB_TOKEN permissions should follow least privilege with an explicit permissions block granting only required scopes.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "ISO/IEC 27001:2022, Annex A control A.8.2", + "rationale": "ISO/IEC 27001:2022 Annex A control A.8.2 ('Privileged access rights') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-002 ('CI/CD Workflow Has Unnecessarily Broad Token Permissions') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-003": { + "control_id": "A.5.21", + "control_name": "Managing information security in the ICT supply chain", + "description": "Third-party GitHub Actions must be pinned to a full 40-character commit SHA to prevent supply-chain attacks via mutable tags.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "ISO/IEC 27001:2022, Annex A control A.5.21", + "rationale": "ISO/IEC 27001:2022 Annex A control A.5.21 ('Managing information security in the ICT supply chain') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-003 ('Third-Party Workflow Action Not Pinned to Immutable Commit SHA') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-004": { + "control_id": "A.8.28", + "control_name": "Secure coding", + "description": "Workflows triggered by pull_request_target or workflow_run must not check out or execute untrusted PR code in a context with secret access.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "ISO/IEC 27001:2022, Annex A control A.8.28", + "rationale": "ISO/IEC 27001:2022 Annex A control A.8.28 ('Secure coding') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-004 ('Untrusted Pull-Request Input Reaches Privileged Workflow Context') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null } } } diff --git a/compliance/frameworks/nist_csf.json b/compliance/frameworks/nist_csf.json index 742110fd..57b0c0c4 100644 --- a/compliance/frameworks/nist_csf.json +++ b/compliance/frameworks/nist_csf.json @@ -1734,6 +1734,54 @@ "owner": null, "review_status": "pending_review", "review_date": null + }, + "AZ-CI-001": { + "control_id": "PR.AC-1", + "control_name": "Identities and credentials are issued, managed, verified, revoked, and audited for authorized devices, users and processes", + "description": "CI/CD workflows should use workload identity federation (OIDC) instead of long-lived client secrets or storage keys stored as repository secrets.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "NIST Cybersecurity Framework 1.1, subcategory PR.AC-1", + "rationale": "NIST CSF 1.1 subcategory control PR.AC-1 ('Identities and credentials are issued, managed, verified, revoked, and audited for authorized devices, users and processes') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-001 ('CI/CD Workflow Uses Long-Lived Azure Credentials') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-002": { + "control_id": "PR.AC-4", + "control_name": "Access permissions and authorizations are managed, incorporating the principles of least privilege and separation of duties", + "description": "Workflow GITHUB_TOKEN permissions should follow least privilege with an explicit permissions block granting only required scopes.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "NIST Cybersecurity Framework 1.1, subcategory PR.AC-4", + "rationale": "NIST CSF 1.1 subcategory control PR.AC-4 ('Access permissions and authorizations are managed, incorporating the principles of least privilege and separation of duties') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-002 ('CI/CD Workflow Has Unnecessarily Broad Token Permissions') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-003": { + "control_id": "PR.IP-1", + "control_name": "A baseline configuration of information technology systems is created and maintained incorporating appropriate security principles", + "description": "Third-party GitHub Actions must be pinned to a full 40-character commit SHA to prevent supply-chain attacks via mutable tags.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "NIST Cybersecurity Framework 1.1, subcategory PR.IP-1", + "rationale": "NIST CSF 1.1 subcategory control PR.IP-1 ('A baseline configuration of information technology systems is created and maintained incorporating appropriate security principles') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-003 ('Third-Party Workflow Action Not Pinned to Immutable Commit SHA') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-004": { + "control_id": "PR.AC-4", + "control_name": "Access permissions and authorizations are managed, incorporating the principles of least privilege and separation of duties", + "description": "Workflows triggered by pull_request_target or workflow_run must not check out or execute untrusted PR code in a context with secret access.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "NIST Cybersecurity Framework 1.1, subcategory PR.AC-4", + "rationale": "NIST CSF 1.1 subcategory control PR.AC-4 ('Access permissions and authorizations are managed, incorporating the principles of least privilege and separation of duties') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-004 ('Untrusted Pull-Request Input Reaches Privileged Workflow Context') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null } } } diff --git a/compliance/frameworks/soc2.json b/compliance/frameworks/soc2.json index cdfdb858..f088c30a 100644 --- a/compliance/frameworks/soc2.json +++ b/compliance/frameworks/soc2.json @@ -1734,6 +1734,54 @@ "owner": null, "review_status": "pending_review", "review_date": null + }, + "AZ-CI-001": { + "control_id": "CC6.1", + "control_name": "Logical and Physical Access Controls", + "description": "CI/CD workflows should use workload identity federation (OIDC) instead of long-lived client secrets or storage keys stored as repository secrets.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "SOC 2 Type II (2017 Trust Services Criteria), criterion CC6.1", + "rationale": "SOC 2 (2017 Trust Services Criteria) criterion control CC6.1 ('Logical and Physical Access Controls') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-001 ('CI/CD Workflow Uses Long-Lived Azure Credentials') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-002": { + "control_id": "CC6.3", + "control_name": "Access Authorization and Least Privilege", + "description": "Workflow GITHUB_TOKEN permissions should follow least privilege with an explicit permissions block granting only required scopes.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "SOC 2 Type II (2017 Trust Services Criteria), criterion CC6.3", + "rationale": "SOC 2 (2017 Trust Services Criteria) criterion control CC6.3 ('Access Authorization and Least Privilege') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-002 ('CI/CD Workflow Has Unnecessarily Broad Token Permissions') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-003": { + "control_id": "CC7.1", + "control_name": "Detection and Monitoring of Vulnerabilities", + "description": "Third-party GitHub Actions must be pinned to a full 40-character commit SHA to prevent supply-chain attacks via mutable tags.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "SOC 2 Type II (2017 Trust Services Criteria), criterion CC7.1", + "rationale": "SOC 2 (2017 Trust Services Criteria) criterion control CC7.1 ('Detection and Monitoring of Vulnerabilities') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-003 ('Third-Party Workflow Action Not Pinned to Immutable Commit SHA') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null + }, + "AZ-CI-004": { + "control_id": "CC8.1", + "control_name": "Change Management", + "description": "Workflows triggered by pull_request_target or workflow_run must not check out or execute untrusted PR code in a context with secret access.", + "mapping_type": "supporting", + "evidence_type": "automated_configuration_scan", + "primary_source": "SOC 2 Type II (2017 Trust Services Criteria), criterion CC8.1", + "rationale": "SOC 2 (2017 Trust Services Criteria) criterion control CC8.1 ('Change Management') describes a broader outcome than one technical setting. OpenShield rule AZ-CI-004 ('Untrusted Pull-Request Input Reaches Privileged Workflow Context') evaluates one GitHub Actions workflow setting that provides supporting automated evidence toward it; it does not by itself establish conformance.", + "owner": null, + "review_status": "pending_review", + "review_date": null } } } diff --git a/scanner/rules/az_ci_001.py b/scanner/rules/az_ci_001.py index f465d713..36888a88 100644 --- a/scanner/rules/az_ci_001.py +++ b/scanner/rules/az_ci_001.py @@ -17,7 +17,7 @@ FRAMEWORKS = { "CIS": "N/A-CI-001", "NIST": "PR.AC-1", - "ISO27001": "A.9.2.4", + "ISO27001": "A.5.17", "SOC2": "CC6.1", } DESCRIPTION = ( diff --git a/scanner/rules/az_ci_002.py b/scanner/rules/az_ci_002.py index 213430e3..c62eac62 100644 --- a/scanner/rules/az_ci_002.py +++ b/scanner/rules/az_ci_002.py @@ -17,7 +17,7 @@ FRAMEWORKS = { "CIS": "N/A-CI-002", "NIST": "PR.AC-4", - "ISO27001": "A.9.2.3", + "ISO27001": "A.8.2", "SOC2": "CC6.3", } DESCRIPTION = ( diff --git a/scanner/rules/az_ci_003.py b/scanner/rules/az_ci_003.py index 75d0c6ca..d6078540 100644 --- a/scanner/rules/az_ci_003.py +++ b/scanner/rules/az_ci_003.py @@ -17,7 +17,7 @@ FRAMEWORKS = { "CIS": "N/A-CI-003", "NIST": "PR.IP-1", - "ISO27001": "A.12.1.2", + "ISO27001": "A.5.21", "SOC2": "CC7.1", } DESCRIPTION = ( diff --git a/scanner/rules/az_ci_004.py b/scanner/rules/az_ci_004.py index 88722052..513fb274 100644 --- a/scanner/rules/az_ci_004.py +++ b/scanner/rules/az_ci_004.py @@ -16,7 +16,7 @@ FRAMEWORKS = { "CIS": "N/A-CI-004", "NIST": "PR.AC-4", - "ISO27001": "A.14.2.5", + "ISO27001": "A.8.28", "SOC2": "CC8.1", } DESCRIPTION = ( From 7555357ea7600910070c3d014b12a97aab646586 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Thu, 15 Oct 2026 02:07:04 +0100 Subject: [PATCH 08/12] fix(ci-rules): guard null id-token, exercise UNKNOWN permissions path, log tracebacks Signed-off-by: Mahfuzur Rahman Emon --- scanner/ci_engine.py | 2 +- scanner/github_client.py | 10 ++++---- scanner/rules/_workflow_common.py | 4 ++-- tests/test_rules_ci_workflow.py | 39 +++++++++++++++++++++++++++++++ 4 files changed, 47 insertions(+), 8 deletions(-) diff --git a/scanner/ci_engine.py b/scanner/ci_engine.py index 64360c6c..01cfb009 100644 --- a/scanner/ci_engine.py +++ b/scanner/ci_engine.py @@ -112,7 +112,7 @@ def run_scan(self) -> List[RuleEvaluation]: try: findings = rule.scan(self.client, self.owner, self.repo) except Exception as exc: - logger.error("Error running rule %s: %s", rule.RULE_ID, exc) + logger.error("Error running rule %s: %s", rule.RULE_ID, exc, exc_info=True) evaluations.append( RuleEvaluation( rule_id=rule.RULE_ID, diff --git a/scanner/github_client.py b/scanner/github_client.py index 983542ff..b45c49a8 100644 --- a/scanner/github_client.py +++ b/scanner/github_client.py @@ -113,7 +113,7 @@ def _get(self, path: str, **kwargs: Any) -> Optional[requests.Response]: resp.raise_for_status() return resp except Exception as exc: - logger.error("GitHub API request failed for %s: %s", url, exc) + logger.error("GitHub API request failed for %s: %s", url, exc, exc_info=True) return None def get_workflows(self) -> Optional[List[Dict[str, Any]]]: @@ -147,7 +147,7 @@ def get_workflows(self) -> Optional[List[Dict[str, Any]]]: url = next_url params = {} # URL already contains query params for subsequent pages except Exception as exc: - logger.error("Failed to parse workflows response: %s", exc) + logger.error("Failed to parse workflows response: %s", exc, exc_info=True) self._workflows_cache = None return None self._workflows_cache = workflows @@ -182,7 +182,7 @@ def get_workflow_content(self, path: str) -> Optional[str]: self._workflow_files_cache[path] = decoded return decoded except Exception as exc: - logger.error("Failed to decode workflow file %s: %s", path, exc) + logger.error("Failed to decode workflow file %s: %s", path, exc, exc_info=True) self._workflow_files_cache[path] = None return None @@ -194,7 +194,7 @@ def get_repo_info(self) -> Optional[Dict[str, Any]]: try: return resp.json() except Exception as exc: - logger.error("Failed to parse repo info: %s", exc) + logger.error("Failed to parse repo info: %s", exc, exc_info=True) return None def get_workflow_permissions(self) -> Optional[str]: @@ -215,5 +215,5 @@ def get_workflow_permissions(self) -> Optional[str]: try: return resp.json().get("default_workflow_permissions") except Exception as exc: - logger.error("Failed to parse workflow permissions: %s", exc) + logger.error("Failed to parse workflow permissions: %s", exc, exc_info=True) return None diff --git a/scanner/rules/_workflow_common.py b/scanner/rules/_workflow_common.py index 719ae16e..ad2b78b9 100644 --- a/scanner/rules/_workflow_common.py +++ b/scanner/rules/_workflow_common.py @@ -102,7 +102,7 @@ def uses_workload_identity(yaml_content: str, parsed: Optional[Dict[str, Any]] = # Check top-level permissions top_perms = parsed.get("permissions") or {} if isinstance(top_perms, dict): - has_oidc_permission = top_perms.get("id-token", "").lower() == "write" + has_oidc_permission = str(top_perms.get("id-token") or "").lower() == "write" # Check job-level permissions if top-level not set if not has_oidc_permission: jobs = parsed.get("jobs") or {} @@ -111,7 +111,7 @@ def uses_workload_identity(yaml_content: str, parsed: Optional[Dict[str, Any]] = continue job_perms = job.get("permissions") or {} if isinstance(job_perms, dict): - if job_perms.get("id-token", "").lower() == "write": + if str(job_perms.get("id-token") or "").lower() == "write": has_oidc_permission = True break else: diff --git a/tests/test_rules_ci_workflow.py b/tests/test_rules_ci_workflow.py index 0e3af3c7..cafda44f 100644 --- a/tests/test_rules_ci_workflow.py +++ b/tests/test_rules_ci_workflow.py @@ -293,6 +293,10 @@ def _make_client(workflows=None, contents=None): client = MagicMock() + # A bare MagicMock returns a truthy Mock here, which silently skips AZ-CI-002's + # "repo default unreadable" (None -> UNKNOWN) branch. Default to None, the real + # "could not read it" value; tests that need a value override it. + client.get_workflow_permissions.return_value = None if workflows is None: client.get_workflows.return_value = [{"path": ".github/workflows/ci.yml"}] else: @@ -447,9 +451,11 @@ def test_write_all_returns_finding(self): assert findings[0]["rule_id"] == "AZ-CI-002" def test_no_permissions_block_returns_finding(self): + # Repo default unreadable (None) -> the finding carries the UNKNOWN marker. client = _make_client(contents=NO_PERMISSIONS_WORKFLOW) findings = az_ci_002.scan(client, OWNER, REPO) assert len(findings) == 1 + assert findings[0]["metadata"]["effective_permissions"] == "UNKNOWN" def test_least_privilege_no_finding(self): client = _make_client(contents=SECURE_WORKFLOW) @@ -697,6 +703,8 @@ def _make_engine(self, contents=None, workflows=None, repo_info=None): else: engine.client.get_workflow_content = lambda p: contents engine.client.get_repo_info = lambda: repo_info or {} + # Never hit the network: an unreadable repo default is the safe baseline. + engine.client.get_workflow_permissions = lambda: None return engine def test_secure_workflow_returns_all_pass(self): @@ -729,6 +737,17 @@ def test_unavailable_workflows_returns_unknown(self): assert all(r.status == EvaluationStatus.UNKNOWN for r in results) assert all(r.reason_code == "WORKFLOWS_UNAVAILABLE" for r in results) + def test_unreadable_default_permissions_emits_unknown(self): + """No permissions block and an unreadable repo default -> UNKNOWN, never PASS/FAIL.""" + from scanner.evaluation import EvaluationStatus + + engine = self._make_engine(contents=NO_PERMISSIONS_WORKFLOW) + engine.client.get_workflow_permissions = lambda: None + results = [r for r in engine.run_scan() if r.rule_id == "AZ-CI-002"] + assert len(results) == 1 + assert results[0].status == EvaluationStatus.UNKNOWN + assert results[0].reason_code == "PERMISSIONS_DEFAULT_UNKNOWN" + def test_empty_repo_returns_not_applicable(self): """A repo with no workflows returns NOT_APPLICABLE for all rules.""" from scanner.evaluation import EvaluationStatus @@ -761,3 +780,23 @@ def test_four_rules_evaluated(self): results = engine.run_scan() rule_ids = {r.rule_id for r in results} assert rule_ids == {"AZ-CI-001", "AZ-CI-002", "AZ-CI-003", "AZ-CI-004"} + + +def test_null_id_token_permission_does_not_crash_az_ci_001(): + """`id-token:` with no value parses to None; it must be evaluated, not raise.""" + import scanner.rules.az_ci_001 as az_ci_001_rule + + workflow = ( + "name: Deploy\n" + "on: push\n" + "permissions:\n" + " id-token:\n" + " contents: read\n" + "jobs:\n" + " deploy:\n" + " runs-on: ubuntu-latest\n" + " steps:\n" + " - uses: azure/login@v2\n" + ) + findings = az_ci_001_rule.scan(_make_client(contents=workflow), OWNER, REPO) + assert isinstance(findings, list) From bc6d773c99cfa5eeb7b5982a03e4a642e5f2b3db Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 11 Oct 2026 23:26:29 +0100 Subject: [PATCH 09/12] feat(ci): persist CI scan results (#259 PR 2/3) Signed-off-by: Mahfuzur Rahman Emon --- alembic/versions/e1f2a3b4c5d6_ci_scans.py | 128 ++++++++ api/models/ci_scan.py | 284 ++++++++++++++++ scanner/ci_engine.py | 156 ++++++++- tests/test_ci_persistence.py | 383 ++++++++++++++++++++++ 4 files changed, 935 insertions(+), 16 deletions(-) create mode 100644 alembic/versions/e1f2a3b4c5d6_ci_scans.py create mode 100644 api/models/ci_scan.py create mode 100644 tests/test_ci_persistence.py diff --git a/alembic/versions/e1f2a3b4c5d6_ci_scans.py b/alembic/versions/e1f2a3b4c5d6_ci_scans.py new file mode 100644 index 00000000..ab9b93cf --- /dev/null +++ b/alembic/versions/e1f2a3b4c5d6_ci_scans.py @@ -0,0 +1,128 @@ +"""Add ci_scans and ci_scan_evaluations tables for GitHub CI/CD scanning. + +Revision ID: e1f2a3b4c5d6 +Revises: 3f59f83a5253 +Create Date: 2026-10-10 00:00:00.000000 + +CI scans are GitHub-scoped (owner + repo), not Azure subscription-scoped, +so they live in their own tables rather than reusing ``scans``/``findings``. +``ci_scan_evaluations`` mirrors ``rule_evaluations`` but is keyed to +``ci_scans.scan_id`` and carries no finding_id — CI rules emit +RuleEvaluation objects directly; there is no separate findings table for +the CI engine. +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# Revision identifiers, used by Alembic. +revision: str = "e1f2a3b4c5d6" +down_revision: Union[str, Sequence[str], None] = "3f59f83a5253" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_STATUS_CONSTRAINT = "ck_ci_scan_evaluations_status_v1" +_SCOPE_CONSTRAINT = "ck_ci_scan_evaluations_resource_id_not_empty" +_REASON_CONSTRAINT = "ck_ci_scan_evaluations_reason_code_required" +_CI_SCAN_STATUS_CONSTRAINT = "ck_ci_scans_status_v1" + + +def upgrade() -> None: + op.create_table( + "ci_scans", + sa.Column("scan_id", postgresql.UUID(), nullable=False), + sa.Column("owner", sa.Text(), nullable=False), + sa.Column("repo", sa.Text(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("status", sa.Text(), server_default=sa.text("'pending'"), nullable=False), + sa.Column("total_evaluations", sa.Integer(), server_default=sa.text("0"), nullable=False), + sa.Column("fail_count", sa.Integer(), server_default=sa.text("0"), nullable=False), + sa.Column("error_message", sa.Text(), nullable=True), + sa.PrimaryKeyConstraint("scan_id", name="ci_scans_pkey"), + ) + + op.create_check_constraint( + _CI_SCAN_STATUS_CONSTRAINT, + "ci_scans", + "status IN ('pending', 'running', 'completed', 'failed')", + ) + + op.create_index("idx_ci_scans_owner_repo", "ci_scans", ["owner", "repo"], unique=False) + op.create_index("idx_ci_scans_started_at", "ci_scans", ["started_at"], unique=False) + op.create_index("idx_ci_scans_status", "ci_scans", ["status"], unique=False) + + op.create_table( + "ci_scan_evaluations", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("scan_id", postgresql.UUID(), nullable=False), + sa.Column("rule_id", sa.Text(), nullable=False), + sa.Column("resource_id", sa.Text(), nullable=False), + sa.Column("resource_type", sa.Text(), nullable=False, server_default=sa.text("''")), + sa.Column("status", sa.Text(), nullable=False), + sa.Column("reason_code", sa.Text(), nullable=True), + sa.Column("reason", sa.Text(), nullable=True), + sa.Column( + "evidence", + postgresql.JSONB(), + server_default=sa.text("'{}'::jsonb"), + nullable=True, + ), + sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["scan_id"], + ["ci_scans.scan_id"], + name="ci_scan_evaluations_scan_id_fkey", + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name="ci_scan_evaluations_pkey"), + # One coverage row per rule per resource per scan — same uniqueness + # contract as rule_evaluations, enabling ON CONFLICT upserts on retry. + sa.UniqueConstraint( + "scan_id", + "rule_id", + "resource_id", + name="uq_ci_scan_evaluations_scan_rule_resource", + ), + ) + + op.create_index("idx_ci_scan_evaluations_scan_id", "ci_scan_evaluations", ["scan_id"], unique=False) + op.create_index("idx_ci_scan_evaluations_rule_id", "ci_scan_evaluations", ["rule_id"], unique=False) + op.create_index("idx_ci_scan_evaluations_status", "ci_scan_evaluations", ["status"], unique=False) + + op.create_check_constraint( + _STATUS_CONSTRAINT, + "ci_scan_evaluations", + "status IN ('PASS', 'FAIL', 'UNKNOWN', 'ERROR', 'NOT_APPLICABLE')", + ) + op.create_check_constraint( + _SCOPE_CONSTRAINT, + "ci_scan_evaluations", + "resource_id <> ''", + ) + # UNKNOWN/ERROR/NOT_APPLICABLE must carry a reason_code — PASS and FAIL + # are self-evident. + op.create_check_constraint( + _REASON_CONSTRAINT, + "ci_scan_evaluations", + "status NOT IN ('UNKNOWN', 'ERROR', 'NOT_APPLICABLE') OR (reason_code IS NOT NULL AND reason_code <> '')", + ) + + +def downgrade() -> None: + op.drop_constraint(_REASON_CONSTRAINT, "ci_scan_evaluations", type_="check") + op.drop_constraint(_SCOPE_CONSTRAINT, "ci_scan_evaluations", type_="check") + op.drop_constraint(_STATUS_CONSTRAINT, "ci_scan_evaluations", type_="check") + op.drop_index("idx_ci_scan_evaluations_status", table_name="ci_scan_evaluations") + op.drop_index("idx_ci_scan_evaluations_rule_id", table_name="ci_scan_evaluations") + op.drop_index("idx_ci_scan_evaluations_scan_id", table_name="ci_scan_evaluations") + op.drop_table("ci_scan_evaluations") + + op.drop_constraint(_CI_SCAN_STATUS_CONSTRAINT, "ci_scans", type_="check") + op.drop_index("idx_ci_scans_status", table_name="ci_scans") + op.drop_index("idx_ci_scans_started_at", table_name="ci_scans") + op.drop_index("idx_ci_scans_owner_repo", table_name="ci_scans") + op.drop_table("ci_scans") diff --git a/api/models/ci_scan.py b/api/models/ci_scan.py new file mode 100644 index 00000000..08bec644 --- /dev/null +++ b/api/models/ci_scan.py @@ -0,0 +1,284 @@ +"""Persistence layer for CI/CD scan results (GitHub AZ-CI-* rules). + +``CIDatabase`` is intentionally thin: it wraps the same psycopg2 connection +pool that ``DatabaseManager`` uses, exposes only what the CI scan engine and +(in PR 3/3) the API routes need, and delegates pool management to the caller +exactly as ``DatabaseManager`` does — connect() / close() pair, with +``_get_conn()`` as the lazy accessor that enforces the connection is live. + +Tables owned by this module (created in migration e1f2a3b4c5d6_ci_scans): + ci_scans — one row per scan invocation (owner + repo) + ci_scan_evaluations — one row per rule × resource × scan +""" + +import json +import logging +import os +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +import psycopg2 +import psycopg2.extras + +from api.models.finding import _get_pool # re-use the shared pool + +logger = logging.getLogger(__name__) + + +class CIDatabase: + """Persist and retrieve CI/CD scan results. + + Lifecycle mirrors DatabaseManager: + db = CIDatabase(dsn) + db.connect() + try: + db.save_ci_scan(scan_id, owner, repo, evaluations) + finally: + db.close() + """ + + def __init__(self, dsn: Optional[str] = None) -> None: + self.dsn: str = dsn or os.environ.get("DATABASE_URL", "") + if not self.dsn: + raise RuntimeError("CIDatabase requires a DSN or DATABASE_URL env var") + self.conn: Any = None + + # ------------------------------------------------------------------ # + # Connection # + # ------------------------------------------------------------------ # + + def connect(self) -> None: + """Acquire a connection from the shared pool.""" + self.conn = _get_pool(self.dsn).getconn() + self.conn.autocommit = False + logger.debug("CIDatabase: connection acquired from pool") + + def _get_conn(self) -> Any: + if self.conn is None or self.conn.closed: + self.connect() + return self.conn + + def close(self) -> None: + """Return the connection to the pool.""" + if self.conn is None: + return + try: + _get_pool(self.dsn).putconn(self.conn, close=bool(self.conn.closed)) + logger.debug("CIDatabase: connection returned to pool") + except Exception as exc: + logger.error("CIDatabase: error returning connection to pool: %s", exc) + finally: + self.conn = None + + # ------------------------------------------------------------------ # + # Write # + # ------------------------------------------------------------------ # + + def create_pending_ci_scan(self, scan_id: str, owner: str, repo: str) -> None: + """Insert a ci_scans row in the 'pending' state. + + Called before the engine starts so the scan is visible to callers + immediately on trigger (PR 3/3 uses this for the 202 response). + """ + started_at = datetime.now(timezone.utc) + conn = self._get_conn() + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO ci_scans (scan_id, owner, repo, started_at, status) + VALUES (%s, %s, %s, %s, 'pending') + ON CONFLICT (scan_id) DO NOTHING + """, + (scan_id, owner, repo, started_at), + ) + conn.commit() + logger.info("CIDatabase: created pending CI scan %s for %s/%s", scan_id, owner, repo) + + def save_ci_scan( + self, + scan_id: str, + owner: str, + repo: str, + evaluations: List[Dict[str, Any]], + error_message: Optional[str] = None, + ) -> None: + """Persist the completed CI scan header and all evaluation rows atomically. + + ``evaluations`` is a list of dicts matching ``RuleEvaluation.to_dict()`` + plus an optional ``"evidence"`` key. The caller (CIScanEngine) is + responsible for converting RuleEvaluation objects to dicts before + calling this method. + + A retry of the same scan_id is safe: the header is upserted and + evaluation rows are replaced wholesale (DELETE + INSERT inside the + same transaction), so a retried scan converges on the current result + rather than accumulating stale rows. + """ + completed_at = datetime.now(timezone.utc) + status = "failed" if error_message else "completed" + fail_count = sum(1 for e in evaluations if e.get("status") == "FAIL") + + conn = self._get_conn() + try: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO ci_scans ( + scan_id, owner, repo, started_at, completed_at, + status, total_evaluations, fail_count, error_message + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (scan_id) DO UPDATE SET + completed_at = EXCLUDED.completed_at, + status = EXCLUDED.status, + total_evaluations = EXCLUDED.total_evaluations, + fail_count = EXCLUDED.fail_count, + error_message = EXCLUDED.error_message + """, + ( + scan_id, + owner, + repo, + # started_at: keep existing value if the pending row + # already exists; use now() as fallback for direct calls. + completed_at, + completed_at, + status, + len(evaluations), + fail_count, + error_message, + ), + ) + + # Replace evaluation rows atomically on retry — same pattern as + # DatabaseManager.save_scan() for findings. + cur.execute( + "DELETE FROM ci_scan_evaluations WHERE scan_id = %s", + (scan_id,), + ) + + evaluated_at = completed_at + for ev in evaluations: + cur.execute( + """ + INSERT INTO ci_scan_evaluations ( + scan_id, rule_id, resource_id, resource_type, + status, reason_code, reason, evidence, evaluated_at + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (scan_id, rule_id, resource_id) DO UPDATE SET + resource_type = EXCLUDED.resource_type, + status = EXCLUDED.status, + reason_code = EXCLUDED.reason_code, + reason = EXCLUDED.reason, + evidence = EXCLUDED.evidence, + evaluated_at = EXCLUDED.evaluated_at + """, + ( + scan_id, + ev.get("rule_id"), + ev.get("resource_id"), + ev.get("resource_type", ""), + ev.get("status"), + ev.get("reason_code"), + ev.get("reason"), + json.dumps(ev.get("evidence", {})), + evaluated_at, + ), + ) + + conn.commit() + except Exception: + conn.rollback() + raise + + logger.info( + "CIDatabase: saved CI scan %s (%s/%s) — %d evaluations, %d FAILs", + scan_id, + owner, + repo, + len(evaluations), + fail_count, + ) + + def update_ci_scan_status( + self, + scan_id: str, + status: str, + error_message: Optional[str] = None, + ) -> None: + """Update the status of a CI scan (running → completed / failed).""" + conn = self._get_conn() + with conn.cursor() as cur: + if status == "completed": + completed_at = datetime.now(timezone.utc) + cur.execute( + """ + UPDATE ci_scans + SET status = %s, completed_at = %s, error_message = NULL + WHERE scan_id = %s + """, + (status, completed_at, scan_id), + ) + else: + cur.execute( + """ + UPDATE ci_scans + SET status = %s, error_message = %s + WHERE scan_id = %s + """, + (status, error_message, scan_id), + ) + conn.commit() + logger.info("CIDatabase: updated CI scan %s status to %s", scan_id, status) + + # ------------------------------------------------------------------ # + # Read # + # ------------------------------------------------------------------ # + + def get_ci_scan(self, scan_id: str) -> Optional[Dict[str, Any]]: + """Return a single CI scan record by UUID, or None if not found.""" + conn = self._get_conn() + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM ci_scans WHERE scan_id = %s", (scan_id,)) + row = cur.fetchone() + return dict(row) if row else None + + def get_ci_scans( + self, + owner: Optional[str] = None, + repo: Optional[str] = None, + limit: int = 100, + ) -> List[Dict[str, Any]]: + """Return CI scan records, optionally filtered by owner/repo. + + Results are ordered most-recent-first. ``limit`` is capped at 100 + to match get_scans() in DatabaseManager. + """ + conn = self._get_conn() + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT * FROM ci_scans + WHERE (%s IS NULL OR owner = %s) + AND (%s IS NULL OR repo = %s) + ORDER BY started_at DESC + LIMIT %s + """, + (owner, owner, repo, repo, min(limit, 100)), + ) + return [dict(row) for row in cur.fetchall()] + + def get_ci_scan_evaluations(self, scan_id: str) -> List[Dict[str, Any]]: + """Return all evaluation rows for a CI scan, ordered by rule_id.""" + conn = self._get_conn() + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT * FROM ci_scan_evaluations + WHERE scan_id = %s + ORDER BY rule_id, resource_id + """, + (scan_id,), + ) + return [dict(row) for row in cur.fetchall()] diff --git a/scanner/ci_engine.py b/scanner/ci_engine.py index 01cfb009..9b1e83de 100644 --- a/scanner/ci_engine.py +++ b/scanner/ci_engine.py @@ -5,14 +5,24 @@ workflow file, compatible with the existing evaluation contract in scanner/evaluation.py. -Usage: +Usage (in-process): from scanner.ci_engine import CIScanEngine engine = CIScanEngine(owner="my-org", repo="my-repo") results = engine.run_scan() + +Usage (with persistence): + engine = CIScanEngine(owner="my-org", repo="my-repo", dsn=DATABASE_URL) + scan_id, results = engine.run_and_persist() + +CLI: + python -m scanner.ci_engine --owner --repo [--json] """ import logging -from typing import List, Optional +import uuid +from typing import List, Optional, Tuple + +from api.models.ci_scan import CIDatabase import scanner.rules.az_ci_001 as az_ci_001 import scanner.rules.az_ci_002 as az_ci_002 @@ -44,10 +54,20 @@ class CIScanEngine: Each rule is run against every workflow file. Results are returned as RuleEvaluation objects with PASS/FAIL/UNKNOWN status. - PASS — workflow was evaluated and is compliant. - FAIL — workflow was evaluated and has a finding. - UNKNOWN — workflow content could not be read or permissions - could not be determined. + PASS — workflow was evaluated and is compliant. + FAIL — workflow was evaluated and has a finding. + UNKNOWN — workflow content could not be read or permissions + could not be determined. + NOT_APPLICABLE — repository has no workflow files. + ERROR — rule raised an unexpected exception. + + Args: + owner: GitHub organisation or user name. + repo: Repository name. + token: GitHub personal access token. Falls back to GITHUB_TOKEN + env var inside GitHubClient when not supplied. + dsn: PostgreSQL DSN for run_and_persist(). Falls back to the + DATABASE_URL env var. Not required for run_scan()-only use. """ def __init__( @@ -55,9 +75,11 @@ def __init__( owner: str, repo: str, token: Optional[str] = None, + dsn: Optional[str] = None, ) -> None: self.owner = owner self.repo = repo + self.dsn = dsn self.client = GitHubClient(owner=owner, repo=repo, token=token) def run_scan(self) -> List[RuleEvaluation]: @@ -204,34 +226,136 @@ def run_scan(self) -> List[RuleEvaluation]: return evaluations + def run_and_persist( + self, + scan_id: Optional[str] = None, + dsn: Optional[str] = None, + ) -> Tuple[str, List[RuleEvaluation]]: + """Run all AZ-CI-* rules and persist the results to the database. + + Creates a ``ci_scans`` header row (pending → running → completed/failed) + and a ``ci_scan_evaluations`` row for every RuleEvaluation, all in a + single atomic transaction via ``CIDatabase.save_ci_scan()``. + + Args: + scan_id: UUID string to use for this scan. Generated if not supplied. + The caller (PR 3/3 API route) supplies this so the scan_id + can be returned in the 202 response before the scan runs. + dsn: Override the DSN set on the engine instance. Useful when + the caller already resolved DATABASE_URL. + + Returns: + (scan_id, evaluations) — the UUID used and the full result list. + + Raises: + RuntimeError: if no DSN is available (engine dsn + dsn arg + env var + all unset). + Exception: any unhandled rule or DB error after logging. + """ + # Lazy import keeps the engine usable without psycopg2 installed when + # callers only call run_scan() (e.g. the CLI without a DB). + import os + + resolved_dsn = dsn or self.dsn or os.environ.get("DATABASE_URL", "") + if not resolved_dsn: + raise RuntimeError( + "run_and_persist() requires a DSN: pass dsn=, set CIScanEngine(dsn=...), " + "or set the DATABASE_URL environment variable." + ) + + scan_id = scan_id or str(uuid.uuid4()) + db = CIDatabase(resolved_dsn) + db.connect() + + try: + # Mark as running so a concurrent caller or /api/ci-scans/ + # poll sees a non-pending status immediately. + db.create_pending_ci_scan(scan_id, self.owner, self.repo) + db.update_ci_scan_status(scan_id, "running") + + evaluations = self.run_scan() + + db.save_ci_scan( + scan_id=scan_id, + owner=self.owner, + repo=self.repo, + evaluations=[ev.to_dict() for ev in evaluations], + ) + + logger.info( + "run_and_persist: CI scan %s complete for %s/%s — %d evaluations", + scan_id, + self.owner, + self.repo, + len(evaluations), + ) + return scan_id, evaluations + + except Exception as exc: + logger.error( + "run_and_persist: CI scan %s failed for %s/%s: %s", + scan_id, + self.owner, + self.repo, + exc, + exc_info=True, + ) + try: + db.update_ci_scan_status(scan_id, "failed", error_message=str(exc)) + except Exception as status_exc: + logger.error( + "run_and_persist: could not record failed status for scan %s: %s", + scan_id, + status_exc, + ) + raise + finally: + db.close() + def _main() -> int: """CLI entry point: run AZ-CI-001..004 against a repository. Usage: - python -m scanner.ci_engine --owner --repo [--json] + python -m scanner.ci_engine --owner --repo [--json] [--persist] - Interim executable runner for the CI/CD controls. Persisted-scan storage - and an API route that invoke CIScanEngine are scoped to issue #259 PR 3/3; - this CLI ensures the rules are runnable rather than merged as dead code. + Pass --persist to write results to the database (DATABASE_URL must be set). + Without --persist the CLI prints results and exits without touching the DB, + preserving the behaviour shipped in PR 1/3. """ import argparse import json as _json - parser = argparse.ArgumentParser(description="Run OpenShield CI/CD workflow security rules (AZ-CI-001..004)") + parser = argparse.ArgumentParser( + description="Run OpenShield CI/CD workflow security rules (AZ-CI-001..004)" + ) parser.add_argument("--owner", required=True, help="GitHub organisation or user") parser.add_argument("--repo", required=True, help="Repository name") parser.add_argument("--json", action="store_true", help="Emit results as JSON") + parser.add_argument( + "--persist", + action="store_true", + help="Persist results to the database (DATABASE_URL must be set)", + ) args = parser.parse_args() engine = CIScanEngine(owner=args.owner, repo=args.repo) - results = engine.run_scan() - if args.json: - print(_json.dumps([r.to_dict() for r in results], indent=2)) + if args.persist: + scan_id, results = engine.run_and_persist() + if args.json: + print(_json.dumps({"scan_id": scan_id, "evaluations": [r.to_dict() for r in results]}, indent=2)) + else: + print(f"scan_id: {scan_id}") + for r in results: + print(f"{r.status:15} {r.rule_id:10} {r.resource_id}") else: - for r in results: - print(f"{r.status:15} {r.rule_id:10} {r.resource_id}") + results = engine.run_scan() + if args.json: + print(_json.dumps([r.to_dict() for r in results], indent=2)) + else: + for r in results: + print(f"{r.status:15} {r.rule_id:10} {r.resource_id}") # Non-zero exit when any rule FAILed so a CI/CD job can gate on it. return 1 if any(r.status == EvaluationStatus.FAIL for r in results) else 0 diff --git a/tests/test_ci_persistence.py b/tests/test_ci_persistence.py new file mode 100644 index 00000000..056144a5 --- /dev/null +++ b/tests/test_ci_persistence.py @@ -0,0 +1,383 @@ +"""Tests for CI scan persistence: CIDatabase and CIScanEngine.run_and_persist(). + +These tests never touch a real database. All psycopg2 / pool interactions +are patched at the ``api.models.ci_scan`` import boundary, matching the +pattern used in the existing CI rule tests (tests/test_rules_ci_workflow.py). +""" + +import json +import uuid +from typing import Any, Dict, List +from unittest.mock import MagicMock, call, patch + +import pytest + +from api.models.ci_scan import CIDatabase +from scanner.ci_engine import CIScanEngine +from scanner.evaluation import EvaluationStatus, RuleEvaluation + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +OWNER = "test-org" +REPO = "test-repo" +SCAN_ID = str(uuid.uuid4()) +DSN = "postgresql://localhost/test" + + +def _make_evaluation( + rule_id: str = "AZ-CI-001", + resource_id: str = f"github/{OWNER}/{REPO}/workflows/.github/workflows/ci.yml", + status: str = EvaluationStatus.PASS, + reason_code: str | None = None, + reason: str | None = None, + evidence: Dict[str, Any] | None = None, +) -> RuleEvaluation: + return RuleEvaluation( + rule_id=rule_id, + resource_id=resource_id, + resource_type="GitHub/WorkflowFile", + status=status, + reason_code=reason_code, + reason=reason, + evidence=evidence or {}, + ) + + +def _fail_evaluation(rule_id: str = "AZ-CI-001") -> RuleEvaluation: + return _make_evaluation( + rule_id=rule_id, + status=EvaluationStatus.FAIL, + reason_code="POLICY_VIOLATION", + reason="Pinned actions required", + ) + + +def _unknown_evaluation(rule_id: str = "AZ-CI-002") -> RuleEvaluation: + return _make_evaluation( + rule_id=rule_id, + status=EvaluationStatus.UNKNOWN, + reason_code="WORKFLOWS_UNAVAILABLE", + reason="GitHub API did not return workflow list.", + ) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def mock_pool(): + """Patch _get_pool so CIDatabase never opens a real connection.""" + mock_conn = MagicMock() + mock_conn.closed = False + mock_cursor = MagicMock() + mock_cursor.__enter__ = MagicMock(return_value=mock_cursor) + mock_cursor.__exit__ = MagicMock(return_value=False) + mock_conn.cursor.return_value = mock_cursor + + mock_pool_instance = MagicMock() + mock_pool_instance.getconn.return_value = mock_conn + mock_pool_instance.putconn.return_value = None + + with patch("api.models.ci_scan._get_pool", return_value=mock_pool_instance): + yield mock_pool_instance, mock_conn, mock_cursor + + +# --------------------------------------------------------------------------- +# CIDatabase.connect / close +# --------------------------------------------------------------------------- + + +class TestCIDatabaseConnection: + def test_connect_acquires_from_pool(self, mock_pool): + pool, conn, _ = mock_pool + db = CIDatabase(DSN) + db.connect() + pool.getconn.assert_called_once() + assert db.conn is conn + + def test_close_returns_to_pool(self, mock_pool): + pool, conn, _ = mock_pool + db = CIDatabase(DSN) + db.connect() + db.close() + pool.putconn.assert_called_once_with(conn, close=False) + assert db.conn is None + + def test_close_noop_when_not_connected(self, mock_pool): + pool, _, _ = mock_pool + db = CIDatabase(DSN) + db.close() # should not raise + pool.putconn.assert_not_called() + + def test_missing_dsn_raises(self): + with patch.dict("os.environ", {}, clear=True): + # Ensure DATABASE_URL is absent + import os + os.environ.pop("DATABASE_URL", None) + with pytest.raises(RuntimeError, match="DSN or DATABASE_URL"): + CIDatabase("") + + +# --------------------------------------------------------------------------- +# CIDatabase.create_pending_ci_scan +# --------------------------------------------------------------------------- + + +class TestCreatePendingCIScan: + def test_inserts_pending_row(self, mock_pool): + _, conn, cursor = mock_pool + db = CIDatabase(DSN) + db.connect() + db.create_pending_ci_scan(SCAN_ID, OWNER, REPO) + + sql_called = cursor.execute.call_args[0][0] + assert "INSERT INTO ci_scans" in sql_called + assert "'pending'" in sql_called + conn.commit.assert_called_once() + + def test_passes_correct_params(self, mock_pool): + _, _, cursor = mock_pool + db = CIDatabase(DSN) + db.connect() + db.create_pending_ci_scan(SCAN_ID, OWNER, REPO) + + params = cursor.execute.call_args[0][1] + assert params[0] == SCAN_ID + assert params[1] == OWNER + assert params[2] == REPO + + +# --------------------------------------------------------------------------- +# CIDatabase.save_ci_scan +# --------------------------------------------------------------------------- + + +class TestSaveCIScan: + def _save(self, mock_pool, evaluations: List[Dict[str, Any]], **kwargs): + _, conn, cursor = mock_pool + # cursor.fetchone is not used in save_ci_scan but mock it anyway + cursor.fetchone.return_value = None + db = CIDatabase(DSN) + db.connect() + db.save_ci_scan(SCAN_ID, OWNER, REPO, evaluations, **kwargs) + return conn, cursor + + def test_upserts_scan_header(self, mock_pool): + conn, cursor = self._save(mock_pool, []) + first_call_sql = cursor.execute.call_args_list[0][0][0] + assert "INSERT INTO ci_scans" in first_call_sql + assert "ON CONFLICT" in first_call_sql + + def test_deletes_stale_evaluations_before_insert(self, mock_pool): + conn, cursor = self._save(mock_pool, []) + delete_call = cursor.execute.call_args_list[1][0][0] + assert "DELETE FROM ci_scan_evaluations" in delete_call + + def test_inserts_one_row_per_evaluation(self, mock_pool): + evs = [ + _make_evaluation("AZ-CI-001").to_dict(), + _fail_evaluation("AZ-CI-002").to_dict(), + ] + _, cursor = self._save(mock_pool, evs) + insert_calls = [ + c for c in cursor.execute.call_args_list + if "INSERT INTO ci_scan_evaluations" in c[0][0] + ] + assert len(insert_calls) == 2 + + def test_fail_count_computed_correctly(self, mock_pool): + evs = [ + _fail_evaluation("AZ-CI-001").to_dict(), + _fail_evaluation("AZ-CI-002").to_dict(), + _make_evaluation("AZ-CI-003").to_dict(), + ] + _, cursor = self._save(mock_pool, evs) + header_params = cursor.execute.call_args_list[0][0][1] + # fail_count is at index 7 in the INSERT params + fail_count = header_params[7] + assert fail_count == 2 + + def test_status_is_completed_on_success(self, mock_pool): + _, cursor = self._save(mock_pool, []) + header_params = cursor.execute.call_args_list[0][0][1] + status = header_params[5] + assert status == "completed" + + def test_status_is_failed_when_error_message_set(self, mock_pool): + _, cursor = self._save(mock_pool, [], error_message="boom") + header_params = cursor.execute.call_args_list[0][0][1] + status = header_params[5] + assert status == "failed" + + def test_evidence_serialised_as_json(self, mock_pool): + ev = _make_evaluation(evidence={"workflow_path": ".github/workflows/ci.yml"}).to_dict() + _, cursor = self._save(mock_pool, [ev]) + insert_params = cursor.execute.call_args_list[2][0][1] + evidence_json = insert_params[7] + parsed = json.loads(evidence_json) + assert parsed["workflow_path"] == ".github/workflows/ci.yml" + + def test_rollback_on_db_error(self, mock_pool): + _, conn, cursor = mock_pool + cursor.execute.side_effect = [None, Exception("db error")] + db = CIDatabase(DSN) + db.connect() + with pytest.raises(Exception, match="db error"): + db.save_ci_scan(SCAN_ID, OWNER, REPO, []) + conn.rollback.assert_called_once() + + def test_commits_on_success(self, mock_pool): + conn, _ = self._save(mock_pool, []) + conn.commit.assert_called() + + +# --------------------------------------------------------------------------- +# CIDatabase.get_ci_scan / get_ci_scans / get_ci_scan_evaluations +# --------------------------------------------------------------------------- + + +class TestCIDatabaseRead: + def test_get_ci_scan_returns_none_when_not_found(self, mock_pool): + _, _, cursor = mock_pool + cursor.fetchone.return_value = None + db = CIDatabase(DSN) + db.connect() + result = db.get_ci_scan(SCAN_ID) + assert result is None + + def test_get_ci_scan_returns_dict(self, mock_pool): + _, _, cursor = mock_pool + cursor.fetchone.return_value = {"scan_id": SCAN_ID, "owner": OWNER, "repo": REPO} + db = CIDatabase(DSN) + db.connect() + result = db.get_ci_scan(SCAN_ID) + assert result["scan_id"] == SCAN_ID + + def test_get_ci_scans_filters_by_owner_repo(self, mock_pool): + _, _, cursor = mock_pool + cursor.fetchall.return_value = [] + db = CIDatabase(DSN) + db.connect() + db.get_ci_scans(owner=OWNER, repo=REPO) + sql = cursor.execute.call_args[0][0] + assert "owner" in sql + assert "repo" in sql + + def test_get_ci_scan_evaluations_ordered(self, mock_pool): + _, _, cursor = mock_pool + cursor.fetchall.return_value = [] + db = CIDatabase(DSN) + db.connect() + db.get_ci_scan_evaluations(SCAN_ID) + sql = cursor.execute.call_args[0][0] + assert "ORDER BY rule_id" in sql + + +# --------------------------------------------------------------------------- +# CIScanEngine.run_and_persist +# --------------------------------------------------------------------------- + + +class TestRunAndPersist: + """Verify that run_and_persist() calls CIDatabase in the right order + and propagates failures correctly. CIDatabase is patched at module level + so no psycopg2 pool is needed here.""" + + def _make_engine(self) -> CIScanEngine: + engine = CIScanEngine(owner=OWNER, repo=REPO, dsn=DSN) + return engine + + @patch("scanner.ci_engine.CIScanEngine.run_scan") + @patch("scanner.ci_engine.CIDatabase") + def test_happy_path_calls_db_in_order(self, MockDB, mock_run_scan): + mock_db = MagicMock() + MockDB.return_value = mock_db + evaluations = [_make_evaluation(), _fail_evaluation()] + mock_run_scan.return_value = evaluations + + engine = self._make_engine() + returned_id, returned_evals = engine.run_and_persist(scan_id=SCAN_ID) + + assert returned_id == SCAN_ID + assert returned_evals == evaluations + + mock_db.connect.assert_called_once() + mock_db.create_pending_ci_scan.assert_called_once_with(SCAN_ID, OWNER, REPO) + mock_db.update_ci_scan_status.assert_any_call(SCAN_ID, "running") + mock_db.save_ci_scan.assert_called_once() + mock_db.close.assert_called_once() + + @patch("scanner.ci_engine.CIScanEngine.run_scan") + @patch("scanner.ci_engine.CIDatabase") + def test_save_ci_scan_receives_dicts_not_objects(self, MockDB, mock_run_scan): + mock_db = MagicMock() + MockDB.return_value = mock_db + ev = _make_evaluation() + mock_run_scan.return_value = [ev] + + engine = self._make_engine() + engine.run_and_persist(scan_id=SCAN_ID) + + evaluations_arg = mock_db.save_ci_scan.call_args.kwargs["evaluations"] + assert isinstance(evaluations_arg, list) + assert all(isinstance(e, dict) for e in evaluations_arg) + + @patch("scanner.ci_engine.CIScanEngine.run_scan") + @patch("scanner.ci_engine.CIDatabase") + def test_run_scan_failure_marks_scan_failed(self, MockDB, mock_run_scan): + mock_db = MagicMock() + MockDB.return_value = mock_db + mock_run_scan.side_effect = RuntimeError("GitHub rate limit") + + engine = self._make_engine() + with pytest.raises(RuntimeError, match="GitHub rate limit"): + engine.run_and_persist(scan_id=SCAN_ID) + + mock_db.update_ci_scan_status.assert_any_call( + SCAN_ID, "failed", error_message="GitHub rate limit" + ) + mock_db.close.assert_called_once() + + @patch("scanner.ci_engine.CIDatabase") + def test_missing_dsn_raises_before_db_call(self, MockDB): + import os + os.environ.pop("DATABASE_URL", None) + engine = CIScanEngine(owner=OWNER, repo=REPO) # no dsn + with patch.dict("os.environ", {}, clear=True): + with pytest.raises(RuntimeError, match="DSN"): + engine.run_and_persist(scan_id=SCAN_ID) + MockDB.assert_not_called() + + @patch("scanner.ci_engine.CIScanEngine.run_scan") + @patch("scanner.ci_engine.CIDatabase") + def test_scan_id_generated_when_not_supplied(self, MockDB, mock_run_scan): + mock_db = MagicMock() + MockDB.return_value = mock_db + mock_run_scan.return_value = [] + + engine = self._make_engine() + returned_id, _ = engine.run_and_persist() + + # A UUID4 was generated + assert returned_id is not None + uuid.UUID(returned_id) # raises if not a valid UUID + + @patch("scanner.ci_engine.CIScanEngine.run_scan") + @patch("scanner.ci_engine.CIDatabase") + def test_close_called_even_when_status_update_fails(self, MockDB, mock_run_scan): + """close() must be called in the finally block even when the + error-path status update itself throws.""" + mock_db = MagicMock() + MockDB.return_value = mock_db + mock_run_scan.side_effect = RuntimeError("scan error") + mock_db.update_ci_scan_status.side_effect = [None, RuntimeError("status error")] + + engine = self._make_engine() + with pytest.raises(RuntimeError, match="scan error"): + engine.run_and_persist(scan_id=SCAN_ID) + + mock_db.close.assert_called_once() From 3f121a284c72f9c5d0a4a02f3bb14a495b92d03d Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 11 Oct 2026 23:27:08 +0100 Subject: [PATCH 10/12] docs(ci): note persistence landed in PR 2/3, API route remains for 3/3 Signed-off-by: Mahfuzur Rahman Emon --- scanner/engine.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scanner/engine.py b/scanner/engine.py index db194a65..526bf26f 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -76,7 +76,8 @@ def load_rules(self) -> None: # AZ-CI rules require a GitHubClient, not an AzureClient, so the # Azure scan engine never loads them. They run via # scanner.ci_engine.CIScanEngine (CLI: python -m scanner.ci_engine); - # persisted-scan and API-route wiring is tracked in issue #259 PR 3/3. + # persisted scans: CIScanEngine.run_and_persist() (#259 PR 2/3); + # the API route is tracked in issue #259 PR 3/3. if rule_path.stem.startswith("az_ci_"): logger.debug("Skipping CI rule %s (requires GitHubClient)", rule_path.name) continue From c7c22265e91a4ed1512966a65c1363bc59baa51c Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Sun, 11 Oct 2026 23:53:18 +0100 Subject: [PATCH 11/12] fix(ci): chain ci_scans migration off current dev head Signed-off-by: Mahfuzur Rahman Emon --- alembic/versions/e1f2a3b4c5d6_ci_scans.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/alembic/versions/e1f2a3b4c5d6_ci_scans.py b/alembic/versions/e1f2a3b4c5d6_ci_scans.py index ab9b93cf..114111f8 100644 --- a/alembic/versions/e1f2a3b4c5d6_ci_scans.py +++ b/alembic/versions/e1f2a3b4c5d6_ci_scans.py @@ -1,7 +1,7 @@ """Add ci_scans and ci_scan_evaluations tables for GitHub CI/CD scanning. Revision ID: e1f2a3b4c5d6 -Revises: 3f59f83a5253 +Revises: b6d2f8a4c1e7 Create Date: 2026-10-10 00:00:00.000000 CI scans are GitHub-scoped (owner + repo), not Azure subscription-scoped, @@ -20,7 +20,7 @@ # Revision identifiers, used by Alembic. revision: str = "e1f2a3b4c5d6" -down_revision: Union[str, Sequence[str], None] = "3f59f83a5253" +down_revision: Union[str, Sequence[str], None] = "b6d2f8a4c1e7" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None From 1bd66d54ea1bbe54928ccf2b0b388ee22bc2e5f9 Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Thu, 15 Oct 2026 01:58:05 +0100 Subject: [PATCH 12/12] fix(ci): new migration revision ID, scope-safe saves, lint and test isolation Signed-off-by: Mahfuzur Rahman Emon --- ...6_ci_scans.py => 4d8a2f6c9b13_ci_scans.py} | 4 +-- api/models/ci_scan.py | 14 ++++++++-- tests/test_ci_persistence.py | 26 ++++++++++++++----- 3 files changed, 34 insertions(+), 10 deletions(-) rename alembic/versions/{e1f2a3b4c5d6_ci_scans.py => 4d8a2f6c9b13_ci_scans.py} (98%) diff --git a/alembic/versions/e1f2a3b4c5d6_ci_scans.py b/alembic/versions/4d8a2f6c9b13_ci_scans.py similarity index 98% rename from alembic/versions/e1f2a3b4c5d6_ci_scans.py rename to alembic/versions/4d8a2f6c9b13_ci_scans.py index 114111f8..0cd3a741 100644 --- a/alembic/versions/e1f2a3b4c5d6_ci_scans.py +++ b/alembic/versions/4d8a2f6c9b13_ci_scans.py @@ -1,6 +1,6 @@ """Add ci_scans and ci_scan_evaluations tables for GitHub CI/CD scanning. -Revision ID: e1f2a3b4c5d6 +Revision ID: 4d8a2f6c9b13 Revises: b6d2f8a4c1e7 Create Date: 2026-10-10 00:00:00.000000 @@ -19,7 +19,7 @@ from sqlalchemy.dialects import postgresql # Revision identifiers, used by Alembic. -revision: str = "e1f2a3b4c5d6" +revision: str = "4d8a2f6c9b13" down_revision: Union[str, Sequence[str], None] = "b6d2f8a4c1e7" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None diff --git a/api/models/ci_scan.py b/api/models/ci_scan.py index 08bec644..bf44afdd 100644 --- a/api/models/ci_scan.py +++ b/api/models/ci_scan.py @@ -134,6 +134,8 @@ def save_ci_scan( total_evaluations = EXCLUDED.total_evaluations, fail_count = EXCLUDED.fail_count, error_message = EXCLUDED.error_message + WHERE ci_scans.owner = EXCLUDED.owner + AND ci_scans.repo = EXCLUDED.repo """, ( scan_id, @@ -150,6 +152,11 @@ def save_ci_scan( ), ) + if cur.rowcount == 0: + raise ValueError( + f"scan_id {scan_id} already exists for a different repository; refusing to overwrite" + ) + # Replace evaluation rows atomically on retry — same pattern as # DatabaseManager.save_scan() for findings. cur.execute( @@ -224,10 +231,13 @@ def update_ci_scan_status( cur.execute( """ UPDATE ci_scans - SET status = %s, error_message = %s + SET status = %s, + error_message = %s, + completed_at = CASE WHEN %s::text = 'failed' + THEN CURRENT_TIMESTAMP ELSE completed_at END WHERE scan_id = %s """, - (status, error_message, scan_id), + (status, error_message, status, scan_id), ) conn.commit() logger.info("CIDatabase: updated CI scan %s status to %s", scan_id, status) diff --git a/tests/test_ci_persistence.py b/tests/test_ci_persistence.py index 056144a5..62f78967 100644 --- a/tests/test_ci_persistence.py +++ b/tests/test_ci_persistence.py @@ -8,7 +8,7 @@ import json import uuid from typing import Any, Dict, List -from unittest.mock import MagicMock, call, patch +from unittest.mock import MagicMock, patch import pytest @@ -115,9 +115,6 @@ def test_close_noop_when_not_connected(self, mock_pool): def test_missing_dsn_raises(self): with patch.dict("os.environ", {}, clear=True): - # Ensure DATABASE_URL is absent - import os - os.environ.pop("DATABASE_URL", None) with pytest.raises(RuntimeError, match="DSN or DATABASE_URL"): CIDatabase("") @@ -344,8 +341,6 @@ def test_run_scan_failure_marks_scan_failed(self, MockDB, mock_run_scan): @patch("scanner.ci_engine.CIDatabase") def test_missing_dsn_raises_before_db_call(self, MockDB): - import os - os.environ.pop("DATABASE_URL", None) engine = CIScanEngine(owner=OWNER, repo=REPO) # no dsn with patch.dict("os.environ", {}, clear=True): with pytest.raises(RuntimeError, match="DSN"): @@ -381,3 +376,22 @@ def test_close_called_even_when_status_update_fails(self, MockDB, mock_run_scan) engine.run_and_persist(scan_id=SCAN_ID) mock_db.close.assert_called_once() + + +class TestReviewFollowUps: + def test_save_raises_when_scan_id_belongs_to_another_repo(self, mock_pool): + _, conn, cursor = mock_pool + cursor.rowcount = 0 # upsert WHERE clause matched nothing + db = CIDatabase(DSN) + db.connect() + with pytest.raises(ValueError, match="different repository"): + db.save_ci_scan(SCAN_ID, OWNER, REPO, []) + conn.rollback.assert_called_once() + conn.commit.assert_not_called() + + def test_failed_status_sets_completed_at(self, mock_pool): + _, _, cursor = mock_pool + db = CIDatabase(DSN) + db.connect() + db.update_ci_scan_status(SCAN_ID, "failed", error_message="boom") + assert "completed_at" in cursor.execute.call_args[0][0]