CI/CD Integration

# .github/workflows/security-scan.yml
name: Security Scan

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

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

      - name: Run Security Scanner
        id: scan
        run: |
          # Run your scanner with CSE output format
          scanner run --format cse --output findings.json

      - name: Check for Critical Findings
        run: |
          # Fail if any high-severity findings
          CRITICAL=$(jq '[.findings[] | select(.severity == "critical" or .severity == "high")] | length' findings.json)
          if [ "$CRITICAL" -gt 0 ]; then
            echo "::error::Found $CRITICAL critical/high severity findings"
            jq '.findings[] | select(.severity == "critical" or .severity == "high") | {signal_id, title: .signal.title}' findings.json
            exit 1
          fi

      - name: Upload Findings
        uses: actions/upload-artifact@v4
        with:
          name: cse-findings
          path: findings.json

# .github/workflows/compliance-gate.yml
name: Compliance Gate

on:
  pull_request:
    branches: [main]

env:
  CSE_API_KEY: ${{ secrets.CSE_API_KEY }}

jobs:
  hipaa-compliance:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run IaC Scanner
        run: |
          terraform-scanner scan ./terraform --format cse --output findings.json

      - name: Check HIPAA Compliance
        run: |
          # Extract signal IDs from findings
          SIGNALS=$(jq -r '.findings[].signal_id' findings.json | sort -u)

          # Check each signal for HIPAA mappings
          for signal in $SIGNALS; do
            HIPAA_MAPPINGS=$(curl -s "https://api.cseregistry.org/v1/signals/$signal/mappings" \
              -H "Authorization: Bearer $CSE_API_KEY" | \
              jq '[.data.mappings[] | select(.framework == "HIPAA")] | length')

            if [ "$HIPAA_MAPPINGS" -gt 0 ]; then
              echo "::warning::Signal $signal affects HIPAA controls"
            fi
          done

      - name: Generate Compliance Report
        run: |
          python scripts/generate_compliance_report.py \
            --findings findings.json \
            --framework HIPAA \
            --output compliance-report.md

      - name: Comment on PR
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const report = fs.readFileSync('compliance-report.md', 'utf8');
            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: report
            });

# .gitlab-ci.yml
stages:
  - test
  - security
  - deploy

security_scan:
  stage: security
  image: your-scanner-image:latest
  variables:
    CSE_API_KEY: $CSE_API_KEY
  script:
    - scanner run --format cse --output findings.json
    - |
      # Parse findings and check severity
      python3 << 'EOF'
      import json
      import sys

      with open('findings.json') as f:
          data = json.load(f)

      critical = [f for f in data['findings']
                  if f.get('severity') in ['critical', 'high']]

      if critical:
          print(f"Found {len(critical)} critical/high findings:")
          for f in critical:
              print(f"  - {f['signal_id']}")
          sys.exit(1)
      EOF
  artifacts:
    reports:
      dotenv: cse-findings.env
    paths:
      - findings.json
    when: always
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
    - if: $CI_COMMIT_BRANCH == "main"

compliance_report:
  stage: security
  needs: [security_scan]
  script:
    - python scripts/compliance_report.py --findings findings.json
  artifacts:
    paths:
      - compliance-report.html
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

#!/usr/bin/env python3
# scripts/check_compliance.py

import json
import sys
import os
import requests
from typing import List, Dict

CSE_API_KEY = os.environ.get("CSE_API_KEY")
CSE_API_URL = "https://api.cseregistry.org/v1"

def load_findings(path: str) -> List[Dict]:
    with open(path) as f:
        data = json.load(f)
    return data.get("findings", data)

def get_signal_mappings(signal_id: str) -> List[Dict]:
    response = requests.get(
        f"{CSE_API_URL}/signals/{signal_id}/mappings",
        headers={"Authorization": f"Bearer {CSE_API_KEY}"}
    )
    if response.status_code == 404:
        return []
    response.raise_for_status()
    return response.json()["data"]["mappings"]

def check_compliance(
    findings: List[Dict],
    framework: str = None,
    fail_on_severity: List[str] = ["critical", "high"]
) -> Dict:
    results = {
        "total_findings": len(findings),
        "blocking_findings": [],
        "framework_impacts": {},
        "passed": True
    }

    for finding in findings:
        signal_id = finding.get("signal_id")
        severity = finding.get("severity", "unknown")

        # Check severity threshold
        if severity in fail_on_severity:
            results["blocking_findings"].append({
                "signal_id": signal_id,
                "severity": severity,
                "resource": finding.get("artifact", {}).get("resource_id")
            })
            results["passed"] = False

        # Check framework impact
        if CSE_API_KEY:
            mappings = get_signal_mappings(signal_id)
            for mapping in mappings:
                fw = mapping["framework"]
                if framework and fw != framework:
                    continue

                if fw not in results["framework_impacts"]:
                    results["framework_impacts"][fw] = []

                results["framework_impacts"][fw].append({
                    "signal_id": signal_id,
                    "control_id": mapping["control_id"],
                    "relationship": mapping["relationship"]
                })

    return results

def main():
    import argparse

    parser = argparse.ArgumentParser(description="Check CSE compliance")
    parser.add_argument("--findings", required=True, help="Path to findings JSON")
    parser.add_argument("--framework", help="Filter by framework")
    parser.add_argument("--fail-on", default="critical,high",
                        help="Severities that cause failure")
    parser.add_argument("--output", help="Output report path")

    args = parser.parse_args()

    findings = load_findings(args.findings)
    fail_severities = args.fail_on.split(",")

    results = check_compliance(
        findings,
        framework=args.framework,
        fail_on_severity=fail_severities
    )

    # Output results
    print(f"Total findings: {results['total_findings']}")
    print(f"Blocking findings: {len(results['blocking_findings'])}")

    if results["blocking_findings"]:
        print("\nBlocking findings:")
        for f in results["blocking_findings"]:
            print(f"  - [{f['severity'].upper()}] {f['signal_id']}")
            if f.get("resource"):
                print(f"    Resource: {f['resource']}")

    if results["framework_impacts"]:
        print("\nFramework impacts:")
        for fw, impacts in results["framework_impacts"].items():
            controls = set(i["control_id"] for i in impacts)
            print(f"  {fw}: {len(controls)} controls affected")

    if args.output:
        with open(args.output, "w") as f:
            json.dump(results, f, indent=2)

    # Exit with appropriate code
    sys.exit(0 if results["passed"] else 1)

if __name__ == "__main__":
    main()

# .cse-policy.yaml
version: "1.0"
name: "Production Deployment Policy"

# Signals that block deployment
blocking_signals:
  - CSE-HIPAA-TECH-ENCRYPT-REST-001
  - CSE-CMMC-COMMS-UNRESTRICTED-SSH-001
  - CSE-CMMC-ACCESS-MFA-001

# Severity thresholds
severity_gates:
  critical: block
  high: block
  medium: warn
  low: allow
  info: allow

# Framework-specific rules
framework_rules:
  HIPAA:
    action: block
    message: "HIPAA violations must be resolved before deployment"

  SOC2:
    action: warn
    message: "SOC 2 findings detected - review before deployment"

# Exceptions (with expiration)
exceptions:
  - signal_id: CSE-GEN-CONFIG-DEBUG-001
    resource_pattern: "*/staging/*"
    reason: "Debug enabled in staging"
    expires: "2025-03-01"
    approved_by: "security-team"

#!/usr/bin/env python3
# scripts/enforce_policy.py

import yaml
import json
from datetime import datetime

def load_policy(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)

def check_exception(finding: dict, exceptions: list) -> bool:
    """Check if finding is covered by an exception."""
    signal_id = finding.get("signal_id")
    resource = finding.get("artifact", {}).get("resource_id", "")

    for exc in exceptions:
        if exc["signal_id"] != signal_id:
            continue

        # Check expiration
        if exc.get("expires"):
            if datetime.now() > datetime.fromisoformat(exc["expires"]):
                continue

        # Check resource pattern
        import fnmatch
        if fnmatch.fnmatch(resource, exc.get("resource_pattern", "*")):
            return True

    return False

def enforce_policy(findings: list, policy: dict) -> dict:
    results = {
        "blocked": [],
        "warnings": [],
        "allowed": [],
        "excepted": []
    }

    for finding in findings:
        signal_id = finding.get("signal_id")
        severity = finding.get("severity", "medium")

        # Check exceptions first
        if check_exception(finding, policy.get("exceptions", [])):
            results["excepted"].append(finding)
            continue

        # Check blocking signals
        if signal_id in policy.get("blocking_signals", []):
            results["blocked"].append(finding)
            continue

        # Check severity gates
        gate = policy.get("severity_gates", {}).get(severity, "allow")
        if gate == "block":
            results["blocked"].append(finding)
        elif gate == "warn":
            results["warnings"].append(finding)
        else:
            results["allowed"].append(finding)

    return results

def generate_pr_comment(findings: list, policy_results: dict) -> str:
    lines = ["## Security Scan Results\n"]

    if policy_results["blocked"]:
        lines.append("### Blocking Issues\n")
        lines.append("The following must be resolved before merge:\n")
        for f in policy_results["blocked"]:
            lines.append(f"- **{f['signal_id']}**")
            lines.append(f"  - Severity: {f.get('severity', 'unknown')}")
            lines.append(f"  - Resource: `{f.get('artifact', {}).get('resource_id', 'N/A')}`")
            lines.append("")

    if policy_results["warnings"]:
        lines.append("### Warnings\n")
        lines.append("Review these findings before merge:\n")
        for f in policy_results["warnings"]:
            lines.append(f"- {f['signal_id']}: {f.get('severity', 'unknown')}")

    if not policy_results["blocked"] and not policy_results["warnings"]:
        lines.append("No compliance issues detected.")

    lines.append("\n---")
    lines.append("*Powered by [CSE Registry](https://cseregistry.org)*")

    return "\n".join(lines)