DevSecOps Explained — Shift-Left Security for CI/CD Pipelines

# sast_demo.py — Simulating a SAST scan on Python code
import ast
import re

class SimpleSAST:
    """A minimal SAST scanner that checks for common vulnerabilities."""

    PATTERNS = {
        "sql_injection": [
            r"execute\(f['\"]",
            r"cursor\.execute\(.*\+",
            r"\.execute\(\s*['\"]\s*SELECT.*\{",
        ],
        "hardcoded_password": [
            r"password\s*=\s*['\"][^'\"]+['\"]",
            r"passwd\s*=\s*['\"][^'\"]+['\"]",
            r"secret_key\s*=\s*['\"][^'\"]+['\"]",
        ],
        "eval_usage": [
            r"\beval\s*\(",
            r"\bexec\s*\(",
        ],
        "insecure_hash": [
            r"hashlib\.md5\b",
            r"hashlib\.sha1\b",
        ],
    }

    def scan_file(self, filepath: str) -> list[dict]:
        """Scan a Python file for security issues."""
        findings = []
        try:
            with open(filepath, 'r') as f:
                content = f.read()
                lines = content.split('\n')
        except FileNotFoundError:
            return [{"file": filepath, "error": "File not found"}]

        for vuln_type, patterns in self.PATTERNS.items():
            for pattern in patterns:
                for i, line in enumerate(lines, 1):
                    if re.search(pattern, line, re.IGNORECASE):
                        findings.append({
                            "file": filepath,
                            "line": i,
                            "type": vuln_type,
                            "severity": "HIGH" if vuln_type == "sql_injection" else "MEDIUM",
                            "snippet": line.strip()
                        })

        return findings

# Example scan
scanner = SimpleSAST()
results = scanner.scan_file("sample_app.py")

print("=== SAST Scan Results ===")
for r in results:
    print(f"[{r['severity']}] {r['type']} at {r['file']}:{r['line']}")
    print(f"  Code: {r['snippet']}")

if not results:
    print("No vulnerabilities found.")

# dast_demo.py — Simulating a DAST scan on a web endpoint
import requests
import json

class SimpleDAST:
    """A minimal DAST scanner that tests common web vulnerabilities."""

    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/')

    def test_sql_injection(self, endpoint: str, param: str) -> dict:
        """Test for SQL injection in a parameter."""
        payloads = ["'", "\"", "' OR '1'='1", "'; DROP TABLE users--"]
        results = []

        for payload in payloads:
            try:
                url = f"{self.base_url}{endpoint}?{param}={payload}"
                response = requests.get(url, timeout=5)

                # Check for error messages indicating SQL injection
                if any(indicator in response.text.lower() for indicator in
                       ["sql", "mysql", "syntax error", "unclosed quotation"]):
                    results.append({
                        "payload": payload,
                        "status": response.status_code,
                        "indicator": "SQL error in response",
                        "finding": "POSSIBLE SQL INJECTION"
                    })

            except requests.RequestException as e:
                results.append({
                    "payload": payload,
                    "error": str(e)
                })

        return {
            "endpoint": endpoint,
            "parameter": param,
            "results": results
        }

    def test_xss(self, endpoint: str, param: str) -> dict:
        """Test for XSS in a parameter."""
        payload = "<script>alert(1)</script>"
        results = []

        try:
            url = f"{self.base_url}{endpoint}?{param}={payload}"
            response = requests.get(url, timeout=5)

            # Check if payload is reflected unescaped
            if payload in response.text:
                results.append({
                    "payload": payload,
                    "finding": "REFLECTED XSS — payload reflected without escaping"
                })
        except requests.RequestException as e:
            results.append({"error": str(e)})

        return {"endpoint": endpoint, "results": results}

# Example usage (replace URL with your target)
# scanner = SimpleDAST("https://example.com")
# results = scanner.test_sql_injection("/search", "q")
# results = scanner.test_xss("/search", "q")

# Using OWASP Dependency-Check (SCA tool)
# Scans dependencies for known CVEs

docker run --rm \
  -v $(pwd):/src \
  owasp/dependency-check \
  --scan /src \
  --format HTML \
  --out /src/reports

# Check for known vulnerabilities in npm packages
npm audit

# Check for known vulnerabilities in Python packages
pip-audit
# or
safety check

# BAD — hardcoded in code
API_KEY = "sk-abc123def456"

# GOOD — use environment variables
import os
API_KEY = os.environ.get("API_KEY")

# BETTER — use a secrets manager
# AWS Secrets Manager, HashiCorp Vault, or GCP Secret Manager

# Using HashiCorp Vault from Python
import hvac

client = hvac.Client(url='https://vault.dodatech.io')
client.token = os.environ['VAULT_TOKEN']

secret = client.secrets.kv.read_secret_version(
    path='api-keys/production'
)
API_KEY = secret['data']['data']['api_key']

# .github/workflows/devsecops.yml
name: DevSecOps Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  security-checks:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4

    # 1. SAST — Static code analysis
    - name: SAST — Bandit (Python)
      run: |
        pip install bandit
        bandit -r . -f json -o bandit-report.json
    - name: Upload SAST Report
      uses: actions/upload-artifact@v4
      with:
        name: bandit-report
        path: bandit-report.json

    # 2. SCA — Dependency scanning
    - name: SCA — pip-audit
      run: |
        pip install pip-audit
        pip-audit --desc on -o pip-audit-report.json
    - name: Upload SCA Report
      uses: actions/upload-artifact@v4
      with:
        name: pip-audit-report
        path: pip-audit-report.json

    # 3. Secrets scanning
    - name: Secrets — truffleHog
      run: |
        docker run --rm -v "$(pwd):/repo" trufflesecurity/trufflehog:latest \
          filesystem /repo --json > trufflehog-report.json
    - name: Upload Secrets Report
      uses: actions/upload-artifact@v4
      with:
        name: trufflehog-report
        path: trufflehog-report.json

  build:
    needs: security-checks
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - name: Build
      run: docker build -t app:latest .

    # 4. Container scanning
    - name: Container Scan — Trivy
      run: |
        docker run --rm aquasec/trivy image \
          --severity CRITICAL,HIGH \
          --exit-code 1 \
          app:latest
    - name: Upload Container Scan
      uses: actions/upload-artifact@v4
      with:
        name: trivy-report
        path: trivy-report.json

  deploy-staging:
    needs: build
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - name: Deploy to Staging
      run: echo "Deploying to staging..."
    # 5. DAST — Dynamic scan against staging
    - name: DAST — OWASP ZAP
      run: |
        docker run --rm owasp/zap2docker-stable \
          zap-baseline.py -t https://staging.example.com -r zap-report.html
    - name: Upload ZAP Report
      uses: actions/upload-artifact@v4
      with:
        name: zap-report
        path: zap-report.html

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: detect-private-key
  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.8
    hooks:
      - id: bandit
        args: ["-r", "."]

# Generate SPDX SBOM for a container image
docker run --rm anchore/syft alpine:latest -o spdx-json > sbom.json

# devsecops_dashboard.py
# Aggregate scan results from multiple tools into one report
import json
import glob
from datetime import datetime

class DevSecOpsDashboard:
    """Aggregate security scan results into a summary."""

    def __init__(self):
        self.findings = {
            "sast": [],
            "dast": [],
            "sca": [],
            "secrets": [],
            "container": []
        }

    def load_bandit_report(self, path: str):
        """Load Bandit SAST results."""
        try:
            with open(path) as f:
                data = json.load(f)
                self.findings["sast"] = data.get("results", [])
        except (FileNotFoundError, json.JSONDecodeError):
            pass

    def summary(self) -> dict:
        """Generate a summary of all findings."""
        total = sum(len(v) for v in self.findings.values())
        by_severity = {"HIGH": 0, "MEDIUM": 0, "LOW": 0}

        for category, items in self.findings.items():
            for item in items:
                sev = item.get("issue_severity", item.get("severity", "LOW")).upper()
                if sev in by_severity:
                    by_severity[sev] += 1

        return {
            "timestamp": datetime.now().isoformat(),
            "total_findings": total,
            "by_category": {k: len(v) for k, v in self.findings.items()},
            "by_severity": by_severity,
            "pass": total == 0
        }

# Usage
dashboard = DevSecOpsDashboard()
dashboard.load_bandit_report("bandit-report.json")
print(json.dumps(dashboard.summary(), indent=2))