Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-SkillsFramework mappings
MITRE ATT&CK
NIST CSF 2.0
When to Use
- When investigating security incidents that require detecting supply chain attacks in ci cd
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques
Prerequisites
- Familiarity with security operations concepts and tools
- Access to a test or lab environment for safe execution
- Python 3.8+ with required dependencies installed
- Appropriate authorization for any testing activities
Instructions
Scan CI/CD workflow files for supply chain risks by parsing GitHub Actions YAML, checking for unpinned dependencies, script injection vectors, and secrets exposure.
import yaml
from pathlib import Path
for wf in Path(".github/workflows").glob("*.yml"):
with open(wf) as f:
workflow = yaml.safe_load(f)
for job_name, job in workflow.get("jobs", {}).items():
for step in job.get("steps", []):
uses = step.get("uses", "")
if uses and "@" in uses and not uses.split("@")[1].startswith("sha"):
print(f"Unpinned action: {uses} in {wf.name}")Key supply chain risks:
- Unpinned GitHub Actions (using @main instead of SHA)
- Script injection via ${{ github.event }} expressions
- Overly permissive GITHUB_TOKEN permissions
- Third-party actions with write access to repo
- Dependency confusion via public/private package name collision
Examples
# Check for script injection in run steps
for step in job.get("steps", []):
run_cmd = step.get("run", "")
if "${{" in run_cmd and "github.event" in run_cmd:
print(f"Script injection risk: {run_cmd[:80]}")Source materials
References and resources
Everything below is rendered for inspection. Script files are read-only and never run.
References 1
api-reference.md1.6 KB
API Reference: Detecting Supply Chain Attacks in CI/CD
GitHub Actions Workflow Parsing
import yaml
with open(".github/workflows/ci.yml") as f:
wf = yaml.safe_load(f)
# Key fields
wf["permissions"] # Workflow-level permissions
wf["jobs"]["build"]["steps"] # Step list
step["uses"] # Action reference (owner/repo@ref)
step["run"] # Shell script
step["env"] # Environment variablesSupply Chain Risk Patterns
| Risk | Pattern | Severity |
|---|---|---|
| Unpinned action | uses: owner/action@main |
CRITICAL |
| Mutable tag | uses: owner/action@v1 |
MEDIUM |
| Script injection | run: echo ${{ github.event.issue.title }} |
CRITICAL |
| Write permissions | permissions: write-all |
HIGH |
| Curl pipe bash | RUN curl | bash |
HIGH |
| Latest image tag | FROM image:latest |
MEDIUM |
Dependency Confusion Check
import requests
# Check if private package exists on public registry
resp = requests.get(f"https://registry.npmjs.org/{pkg}")
exists = resp.status_code == 200
resp = requests.get(f"https://pypi.org/pypi/{pkg}/json")
exists = resp.status_code == 200Pinning Actions to SHA
# Bad: mutable reference
uses: actions/checkout@main
# Good: pinned to commit SHA
uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608References
- GitHub Actions security hardening: https://docs.github.com/en/actions/security-guides
- StepSecurity: https://github.com/step-security/harden-runner
- Dependency confusion: https://medium.com/@alex.birsan/dependency-confusion-4a5d60fec610
Scripts 1
agent.py7.4 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for detecting supply chain attacks in CI/CD pipelines."""
import re
import json
import argparse
from pathlib import Path
from datetime import datetime
import yaml
import requests
def scan_workflow_file(workflow_path):
"""Scan a single GitHub Actions workflow file for supply chain risks."""
with open(workflow_path) as f:
workflow = yaml.safe_load(f)
if not workflow or not isinstance(workflow, dict):
return []
findings = []
permissions = workflow.get("permissions", {})
if permissions == "write-all" or (isinstance(permissions, dict) and
permissions.get("contents") == "write"):
findings.append({
"file": str(workflow_path),
"issue": "Overly permissive workflow permissions",
"severity": "HIGH",
"detail": f"permissions: {permissions}",
})
for job_name, job in workflow.get("jobs", {}).items():
for i, step in enumerate(job.get("steps", [])):
uses = step.get("uses", "")
if uses:
findings.extend(check_action_pinning(str(workflow_path), job_name, uses))
run_cmd = step.get("run", "")
if run_cmd:
findings.extend(check_script_injection(str(workflow_path), job_name, run_cmd))
env_vars = step.get("env", {})
for key, val in (env_vars or {}).items():
if isinstance(val, str) and "${{" in val and "secrets." in val:
if "run" in step:
findings.append({
"file": str(workflow_path),
"job": job_name,
"issue": "Secret passed to environment variable in run step",
"severity": "MEDIUM",
"detail": f"{key}={val[:60]}",
})
return findings
def check_action_pinning(filepath, job_name, uses):
"""Check if a GitHub Action is pinned to a commit SHA."""
findings = []
if "@" not in uses:
return findings
ref = uses.split("@")[1]
if re.match(r'^[0-9a-f]{40}$', ref):
return findings
if ref in ("main", "master", "latest"):
findings.append({
"file": filepath,
"job": job_name,
"issue": f"Action pinned to mutable branch: {uses}",
"severity": "CRITICAL",
})
elif re.match(r'^v\d+', ref):
findings.append({
"file": filepath,
"job": job_name,
"issue": f"Action pinned to mutable tag: {uses}",
"severity": "MEDIUM",
"recommendation": "Pin to full commit SHA instead of tag",
})
return findings
def check_script_injection(filepath, job_name, run_cmd):
"""Check for script injection via GitHub context expressions."""
findings = []
injection_patterns = [
r"\$\{\{\s*github\.event\.issue\.title",
r"\$\{\{\s*github\.event\.issue\.body",
r"\$\{\{\s*github\.event\.pull_request\.title",
r"\$\{\{\s*github\.event\.pull_request\.body",
r"\$\{\{\s*github\.event\.comment\.body",
r"\$\{\{\s*github\.event\.review\.body",
r"\$\{\{\s*github\.head_ref",
]
for pattern in injection_patterns:
if re.search(pattern, run_cmd):
findings.append({
"file": filepath,
"job": job_name,
"issue": f"Script injection via untrusted input",
"severity": "CRITICAL",
"pattern": pattern,
"detail": run_cmd[:100],
})
return findings
def scan_directory(directory):
"""Scan all workflow files in a directory."""
all_findings = []
workflow_dir = Path(directory) / ".github" / "workflows"
if not workflow_dir.exists():
workflow_dir = Path(directory)
for wf in workflow_dir.glob("*.yml"):
all_findings.extend(scan_workflow_file(str(wf)))
for wf in workflow_dir.glob("*.yaml"):
all_findings.extend(scan_workflow_file(str(wf)))
return all_findings
def check_dependency_confusion(package_name, registry="npm"):
"""Check if a private package name exists on public registries."""
urls = {
"npm": f"https://registry.npmjs.org/{package_name}",
"pypi": f"https://pypi.org/pypi/{package_name}/json",
}
url = urls.get(registry)
if not url:
return {"exists_publicly": False}
try:
resp = requests.get(url, timeout=10)
return {
"package": package_name,
"registry": registry,
"exists_publicly": resp.status_code == 200,
"severity": "HIGH" if resp.status_code == 200 else "INFO",
}
except requests.RequestException:
return {"package": package_name, "exists_publicly": False}
def scan_dockerfile(dockerfile_path):
"""Scan Dockerfile for supply chain risks."""
findings = []
with open(dockerfile_path) as f:
lines = f.readlines()
for i, line in enumerate(lines, 1):
stripped = line.strip()
if stripped.startswith("FROM") and ":latest" in stripped:
findings.append({
"file": str(dockerfile_path),
"line": i,
"issue": "Using :latest tag in FROM",
"severity": "MEDIUM",
"detail": stripped,
})
if "curl" in stripped and "| sh" in stripped or "| bash" in stripped:
findings.append({
"file": str(dockerfile_path),
"line": i,
"issue": "Piping curl output to shell",
"severity": "HIGH",
"detail": stripped[:100],
})
return findings
def main():
parser = argparse.ArgumentParser(description="CI/CD Supply Chain Attack Detection Agent")
parser.add_argument("--directory", default=".", help="Repository root to scan")
parser.add_argument("--check-packages", nargs="*", help="Package names to check for confusion")
parser.add_argument("--output", default="supply_chain_report.json")
parser.add_argument("--action", choices=[
"workflows", "dockerfiles", "packages", "full_scan"
], default="full_scan")
args = parser.parse_args()
report = {"generated_at": datetime.utcnow().isoformat(), "findings": {}}
if args.action in ("workflows", "full_scan"):
wf_findings = scan_directory(args.directory)
report["findings"]["workflow_risks"] = wf_findings
print(f"[+] Workflow risks found: {len(wf_findings)}")
if args.action in ("dockerfiles", "full_scan"):
df_findings = []
for df in Path(args.directory).rglob("Dockerfile*"):
df_findings.extend(scan_dockerfile(str(df)))
report["findings"]["dockerfile_risks"] = df_findings
print(f"[+] Dockerfile risks found: {len(df_findings)}")
if args.action in ("packages", "full_scan") and args.check_packages:
pkg_results = []
for pkg in args.check_packages:
pkg_results.append(check_dependency_confusion(pkg))
report["findings"]["dependency_confusion"] = pkg_results
public = [p for p in pkg_results if p.get("exists_publicly")]
print(f"[+] Dependency confusion risks: {len(public)}")
with open(args.output, "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"[+] Report saved to {args.output}")
if __name__ == "__main__":
main()
Keep exploring