Boost secrets management in OWASP vs Falco: What You Need to Know

#!/usr/bin/env python3
"""
OWASP ZAP CI Secret Scanner
Scans a target GitHub repository for hardcoded secrets (API keys, AWS creds, etc.)
in CI pipelines. Fails the build if high-severity secrets are found.
Version: 1.0.0
Dependencies: zapv2>=0.0.21, requests>=2.31.0
OWASP ZAP: https://github.com/zaproxy/zaproxy
"""

import os
import sys
import time
import logging
from typing import List, Dict
import requests
import zapv2

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Benchmark config (from our 2024 test on GitHub Actions 8 vCPU runners)
ZAP_API_URL = "http://localhost:8080"  # ZAP running in CI container
ZAP_API_KEY = os.getenv("ZAP_API_KEY", "ci-zap-key-1234")
TARGET_REPO = os.getenv("TARGET_REPO", "https://github.com/example/my-service")
SCAN_TIMEOUT = 300  # 5 minutes max scan time
HIGH_SEVERITY_THRESHOLD = 1  # Fail build if >=1 high severity secret found

def init_zap_client() -> zapv2.ZAPv2:
    """Initialize ZAP API client with error handling."""
    try:
        zap = zapv2.ZAPv2(apikey=ZAP_API_KEY, proxies={"http": ZAP_API_URL, "https": ZAP_API_URL})
        zap.core.set_option_timeout(ZAP_API_URL, SCAN_TIMEOUT)
        logger.info(f"Connected to ZAP at {ZAP_API_URL}")
        return zap
    except Exception as e:
        logger.error(f"Failed to connect to ZAP: {str(e)}")
        sys.exit(1)

def scan_repo_for_secrets(zap: zapv2.ZAPv2, repo_url: str) -> List[Dict]:
    """Scan target repo for hardcoded secrets using ZAP's passive scan."""
    try:
        # Start spider scan to crawl repo contents (static files only)
        spider_id = zap.spider.scan(repo_url, maxchildren=10, recurse=True)
        logger.info(f"Started ZAP spider scan: {spider_id}")

        # Wait for spider to complete
        while int(zap.spider.status(spider_id)) < 100:
            logger.info(f"Spider progress: {zap.spider.status(spider_id)}%")
            time.sleep(5)

        # Start passive scan for secrets (no active attacks to keep CI fast)
        zap.pscan.enable_all_scanners()
        zap.pscan.set_scan_only_in_scope(True)
        logger.info("Started passive secret scan")

        # Wait for passive scan to complete
        while int(zap.pscan.records_to_scan()) > 0:
            logger.info(f"Passive scan remaining: {zap.pscan.records_to_scan()}")
            time.sleep(5)

        # Get all alerts (filtered to secret-related)
        alerts = zap.core.alerts(baseurl=repo_url)
        secret_alerts = [
            a for a in alerts if "secret" in a["name"].lower() or "credential" in a["name"].lower()
        ]
        logger.info(f"Found {len(secret_alerts)} secret-related alerts")
        return secret_alerts
    except Exception as e:
        logger.error(f"Scan failed: {str(e)}")
        sys.exit(1)

def evaluate_results(alerts: List[Dict]) -> None:
    """Evaluate scan results and fail build if high-severity secrets found."""
    high_severity = [a for a in alerts if a["risk"] == "High"]
    if len(high_severity) >= HIGH_SEVERITY_THRESHOLD:
        logger.error(f"Build failed: {len(high_severity)} high-severity secrets found")
        for alert in high_severity:
            logger.error(f"Secret: {alert['name']} | URL: {alert['url']} | Confidence: {alert['confidence']}")
        sys.exit(1)
    else:
        logger.info("No high-severity secrets found. Build passed.")

if __name__ == "__main__":
    logger.info(f"Starting OWASP ZAP secret scan for {TARGET_REPO}")
    zap_client = init_zap_client()
    secret_alerts = scan_repo_for_secrets(zap_client, TARGET_REPO)
    evaluate_results(secret_alerts)
    zap_client.core.shutdown()

#!/usr/bin/env python3
"""
Falco Runtime Secret Exfiltration Monitor
Consumes Falco gRPC events to detect secret exfiltration attempts,
sends alerts to Slack, and kills compromised pods via Kubernetes API.
Version: 1.0.0
Dependencies: grpcio>=1.60.0, slack_sdk>=3.27.0, kubernetes>=28.1.0
Falco: https://github.com/falcosecurity/falco
"""

import os
import sys
import logging
import grpc
import time
from typing import Optional
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from kubernetes import client, config
from falcosecurity import falco_output_pb2, falco_output_pb2_grpc

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Config from environment
FALCO_GRPC_URL = os.getenv("FALCO_GRPC_URL", "localhost:5060")
SLACK_TOKEN = os.getenv("SLACK_TOKEN", "")
SLACK_CHANNEL = os.getenv("SLACK_CHANNEL", "#security-alerts")
K8S_NAMESPACE = os.getenv("K8S_NAMESPACE", "default")
FALCO_RULE_FILTER = "SecretExfiltration"  # Falco rule name for secret leaks

# Initialize clients
slack_client = WebClient(token=SLACK_TOKEN) if SLACK_TOKEN else None
try:
    config.load_incluster_config()  # Runs inside K8S pod
    k8s_client = client.CoreV1Api()
except Exception as e:
    logger.warning(f"Failed to load K8S config: {str(e)}. Pod kill disabled.")
    k8s_client = None

def init_falco_channel() -> falco_output_pb2_grpc.OutputServiceStub:
    """Initialize gRPC channel to Falco with error handling."""
    try:
        channel = grpc.insecure_channel(FALCO_GRPC_URL)
        stub = falco_output_pb2_grpc.OutputServiceStub(channel)
        logger.info(f"Connected to Falco gRPC at {FALCO_GRPC_URL}")
        return stub
    except Exception as e:
        logger.error(f"Failed to connect to Falco: {str(e)}")
        sys.exit(1)

def send_slack_alert(alert: falco_output_pb2.FalcoOutput) -> None:
    """Send exfiltration alert to Slack."""
    if not slack_client:
        logger.warning("Slack token not set. Skipping alert.")
        return
    try:
        message = (
            f"🚨 *Secret Exfiltration Detected* 🚨\n"
            f"*Pod*: {alert.output_fields.get('k8s.pod.name', 'unknown')}\n"
            f"*Namespace*: {alert.output_fields.get('k8s.namespace.name', 'unknown')}\n"
            f"*Rule*: {alert.rule}\n"
            f"*Time*: {time.ctime(alert.time)}"
        )
        slack_client.chat_postMessage(channel=SLACK_CHANNEL, text=message)
        logger.info(f"Sent Slack alert for rule {alert.rule}")
    except SlackApiError as e:
        logger.error(f"Failed to send Slack alert: {e.response['error']}")

def kill_compromised_pod(pod_name: str, namespace: str) -> None:
    """Kill compromised pod via K8S API."""
    if not k8s_client:
        logger.warning("K8S client not initialized. Skipping pod kill.")
        return
    try:
        k8s_client.delete_namespaced_pod(name=pod_name, namespace=namespace)
        logger.info(f"Killed pod {pod_name} in namespace {namespace}")
    except Exception as e:
        logger.error(f"Failed to kill pod {pod_name}: {str(e)}")

def process_falco_events(stub: falco_output_pb2_grpc.OutputServiceStub) -> None:
    """Stream Falco events and process secret exfiltration alerts."""
    try:
        # Subscribe to all Falco events
        request = falco_output_pb2.SubscribeRequest()
        event_stream = stub.Subscribe(request)
        logger.info("Subscribed to Falco event stream")

        for event in event_stream:
            # Filter for secret exfiltration rules only
            if FALCO_RULE_FILTER not in event.rule:
                continue
            logger.info(f"Detected secret exfiltration: {event.rule}")

            # Extract pod details
            pod_name = event.output_fields.get("k8s.pod.name")
            namespace = event.output_fields.get("k8s.namespace.name", K8S_NAMESPACE)

            # Send alert and kill pod
            send_slack_alert(event)
            if pod_name:
                kill_compromised_pod(pod_name, namespace)
    except grpc.RpcError as e:
        logger.error(f"gRPC error: {e.details()}")
        time.sleep(5)
        process_falco_events(stub)  # Retry on error
    except Exception as e:
        logger.error(f"Event processing failed: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    logger.info("Starting Falco secret exfiltration monitor")
    falco_stub = init_falco_channel()
    process_falco_events(falco_stub)

#!/usr/bin/env python3
"""
OWASP Dependency-Check to Falco Rule Generator
Scans a Java project for dependencies with known secret leaks using OWASP Dependency-Check,
then generates a custom Falco rule to detect those dependencies at runtime.
Version: 1.0.0
Dependencies: requests>=2.31.0, xmltodict>=0.13.0
OWASP Dependency-Check: https://github.com/jeremylong/DependencyCheck
"""

import os
import sys
import logging
import subprocess
import xmltodict
from typing import List, Dict
import json

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Config
DEP_CHECK_PATH = os.getenv("DEP_CHECK_PATH", "/opt/dependency-check/bin/dependency-check.sh")
PROJECT_PATH = os.getenv("PROJECT_PATH", "./java-service")
REPORT_PATH = os.getenv("REPORT_PATH", "./dep-check-report.xml")
FALCO_RULE_PATH = os.getenv("FALCO_RULE_PATH", "./falco-secret-deps.yaml")
SECRET_CWE_FILTER = ["CWE-312", "CWE-522", "CWE-798"]  # Credentials in code CWEs

def run_dependency_check(project_path: str, report_path: str) -> None:
    """Run OWASP Dependency-Check on target project with error handling."""
    try:
        cmd = [
            DEP_CHECK_PATH,
            "--project", "java-service",
            "--scan", project_path,
            "--format", "XML",
            "--out", report_path,
            "--suppress", "./suppress.xml",  # Suppress known false positives
            "--failOnCVSS", "7"  # Fail on high CVSS (secret leaks)
        ]
        logger.info(f"Running Dependency-Check: {' '.join(cmd)}")
        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
        logger.info(f"Dependency-Check completed. Report: {report_path}")
    except subprocess.CalledProcessError as e:
        logger.error(f"Dependency-Check failed: {e.stderr}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"Failed to run Dependency-Check: {str(e)}")
        sys.exit(1)

def parse_secret_dependencies(report_path: str) -> List[Dict]:
    """Parse Dependency-Check XML report for dependencies with secret-related CWEs."""
    try:
        with open(report_path, "r") as f:
            report = xmltodict.parse(f.read())
        dependencies = report.get("analysis", {}).get("dependencies", {}).get("dependency", [])
        if not isinstance(dependencies, list):
            dependencies = [dependencies]

        secret_deps = []
        for dep in dependencies:
            vulnerabilities = dep.get("vulnerabilities", {}).get("vulnerability", [])
            if not isinstance(vulnerabilities, list):
                vulnerabilities = [vulnerabilities]
            for vuln in vulnerabilities:
                cwe = vuln.get("cwe", "")
                if any(secret_cwe in cwe for secret_cwe in SECRET_CWE_FILTER):
                    secret_deps.append({
                        "name": dep.get("@name", "unknown"),
                        "version": dep.get("version", "unknown"),
                        "cwe": cwe,
                        "cvss": vuln.get("cvssScore", "0")
                    })
        logger.info(f"Found {len(secret_deps)} dependencies with secret leaks")
        return secret_deps
    except Exception as e:
        logger.error(f"Failed to parse report: {str(e)}")
        sys.exit(1)

def generate_falco_rule(secret_deps: List[Dict], output_path: str) -> None:
    """Generate custom Falco rule to detect secret-containing dependencies at runtime."""
    try:
        # Build Falco rule YAML
        rule = {
            "rules": [
                {
                    "rule": "SecretDependencyDetected",
                    "desc": "Detect runtime use of dependencies with known secret leaks",
                    "condition": "k8s.pod.name != \"\" and (",
                    "output": "Secret dependency detected: %k8s.pod.name %k8s.namespace.name %evt.type",
                    "priority": "CRITICAL",
                    "tags": ["secrets", "dependencies"]
                }
            ]
        }

        # Add conditions for each secret dependency
        conditions = []
        for dep in secret_deps:
            dep_name = dep["name"].split(":")[-1]  # Extract artifact ID
            conditions.append(f"proc.cmdline contains \"{dep_name}\"")
        rule["rules"][0]["condition"] += " or ".join(conditions) + ")"

        # Write to file
        with open(output_path, "w") as f:
            json.dump(rule, f, indent=2)
        logger.info(f"Generated Falco rule: {output_path}")
    except Exception as e:
        logger.error(f"Failed to generate Falco rule: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    logger.info(f"Starting OWASP Dependency-Check to Falco rule generator for {PROJECT_PATH}")
    run_dependency_check(PROJECT_PATH, REPORT_PATH)
    secret_deps = parse_secret_dependencies(REPORT_PATH)
    if secret_deps:
        generate_falco_rule(secret_deps, FALCO_RULE_PATH)
        logger.info(f"Generated Falco rule for {len(secret_deps)} secret dependencies")
    else:
        logger.info("No secret dependencies found. No Falco rule generated.")

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: owasp-zap-secret-scan
        name: OWASP ZAP Secret Scan
        entry: zap-cli -v -a passive-scan --scan-only-secret-leaks
        language: system
        files: \.(java|go|py|yaml|json|env)$
        exclude: test/

# falco-secret-exfiltration.yaml
rule: SecretExfiltration
_desc: Detect unauthorized secret exfiltration via network egress
condition: >
  k8s.pod.name != "" and
  evt.type = connect and
  fd.sip != "127.0.0.1" and
  fd.sport = 443 and
  not proc.name in ("curl", "wget") and
  not fd.sip in ("s3.amazonaws.com", "config.my-company.com")
output: "Secret exfiltration detected: %k8s.pod.name (ns: %k8s.namespace.name) connected to %fd.sip"
 priority: CRITICAL
tags: [secrets, network]

  org.owasp
  dependency-check-maven
  8.4.3



        check


        7
        CWE-312,CWE-522,CWE-798