From 50106f4ec4d46713322a9fdbaa6a7cb216a7622c Mon Sep 17 00:00:00 2001 From: Mahfuzur Rahman Emon Date: Wed, 26 Aug 2026 02:11:25 +0100 Subject: [PATCH 1/8] 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 2/8] 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 3/8] 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 4/8] 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 5/8] 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 6/8] 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 7/8] 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 8/8] 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)