OWASP ZAP Automation via REST API

Problem Statement

In modern software development, security testing is often treated as an afterthought—executed manually right before production releases or skipped entirely due to tight deadlines. Manual Dynamic Application Security Testing (DAST) creates bottleneck points in release pipelines, increases human error, and delays feedback to developers when security flaws are introduced.

Automating dynamic security testing within continuous integration and continuous deployment (CI/CD) pipelines solves this issue by ensuring that every staging deployment undergoes automated vulnerability assessment. By programmatic orchestration of OWASP Zed Attack Proxy (ZAP) through its REST API, security teams and DevOps engineers can execute automated spidering and active scanning against test environments, block builds on critical security regressions, and generate compliance reports automatically.

How It Works

The script acts as an orchestration client that interacts with a running OWASP ZAP instance (daemon or container) via ZAP's HTTP REST API. It follows a controlled sequential workflow: API authentication, target registration, passive crawling (spidering), active scanning, and report generation.

+------------------+         HTTP REST API         +---------------------+
|                  | ----------------------------> |                     |
|  zap_scan.py     |   1. Start Spider             |  OWASP ZAP Daemon   |
|  (CLI / Pipeline)| <---------------------------- |  (Docker / Service) |
|                  |   2. Poll Spider Status       |                     |
|                  |                               |                     |
|                  | ----------------------------> |                     |
|                  |   3. Start Active Scan        |                     |
|                  | <---------------------------- |                     |
|                  |   4. Poll Active Scan Status  |                     |
|                  |                               |                     |
|                  | ----------------------------> |                     |
|                  |   5. Fetch HTML Report        |                     |
|                  | <---------------------------- |                     |
+------------------+                               +---------------------+

Design patterns and key implementation details include:

• Stateless REST Client: Uses Python's requests.Session for connection pooling and header management (passing the X-ZAP-API-Key header with every request).
• Polling Loops with Backoff: Employs synchronous polling loops that query scan progress percentage (0-100%) at regular intervals until completion or timeout.
• Fail-Fast Error Handling: Validates ZAP connectivity and target reachability prior to triggering long-running scan operations.

Full Script

#!/usr/bin/env python3
"""
OWASP ZAP Automation via REST API

Automates Web Application Security Testing using OWASP ZAP REST API.
Executes Spidering and Active Scanning workflows and generates reports for CI/CD integration.
"""

import argparse
import os
import sys
import time
from typing import Any, Dict, Optional
import requests


class ZAPClientError(Exception):
    """Custom exception raised when interactions with OWASP ZAP API fail."""
    pass


class ZAPAutomator:
    """Wrapper class to manage OWASP ZAP REST API operations."""

    def __init__(self, base_url: str, api_key: str, timeout: int = 10) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.timeout = timeout
        self.session = requests.Session()
        if self.api_key:
            self.session.headers.update({"X-ZAP-API-Key": self.api_key})

    def _request(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Executes a GET request against the ZAP REST API."""
        url = f"{self.base_url}{endpoint}"
        try:
            response = self.session.get(url, params=params, timeout=self.timeout)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as err:
            raise ZAPClientError(f"API request failed for {url}: {err}") from err

    def check_connection(self) -> bool:
        """Verifies connection to OWASP ZAP API."""
        try:
            res = self._request("/JSON/core/view/version/")
            version = res.get("version", "Unknown")
            print(f"[+] Connected to OWASP ZAP version: {version}")
            return True
        except ZAPClientError as err:
            print(f"[-] Connection check failed: {err}", file=sys.stderr)
            return False

    def start_spider(self, target_url: str) -> str:
        """Triggers the ZAP Spider on the target URL."""
        print(f"[+] Starting ZAP Spider on target: {target_url}")
        res = self._request("/JSON/spider/action/scan/", params={"url": target_url})
        scan_id = res.get("scan")
        if not scan_id:
            raise ZAPClientError("Failed to retrieve scan ID for Spider task.")
        return str(scan_id)

    def wait_for_spider(self, scan_id: str, poll_interval: int = 5) -> None:
        """Polls the Spider progress until it reaches 100%."""
        while True:
            res = self._request("/JSON/spider/view/status/", params={"scanId": scan_id})
            progress = int(res.get("status", 0))
            print(f"[*] Spider progress: {progress}%")
            if progress >= 100:
                break
            time.sleep(poll_interval)
        print("[+] Spider completed successfully.")

    def start_active_scan(self, target_url: str) -> str:
        """Triggers the ZAP Active Scanner on the target URL."""
        print(f"[+] Starting ZAP Active Scan on target: {target_url}")
        res = self._request("/JSON/ascan/action/scan/", params={"url": target_url, "recurse": "true"})
        scan_id = res.get("scan")
        if not scan_id:
            raise ZAPClientError("Failed to retrieve scan ID for Active Scan task.")
        return str(scan_id)

    def wait_for_active_scan(self, scan_id: str, poll_interval: int = 10) -> None:
        """Polls the Active Scan progress until it reaches 100%."""
        while True:
            res = self._request("/JSON/ascan/view/status/", params={"scanId": scan_id})
            progress = int(res.get("status", 0))
            print(f"[*] Active Scan progress: {progress}%")
            if progress >= 100:
                break
            time.sleep(poll_interval)
        print("[+] Active Scan completed successfully.")

    def export_html_report(self, output_file_path: str) -> None:
        """Fetches the HTML security report from ZAP and saves it to disk."""
        url = f"{self.base_url}/OTHER/core/other/htmlreport/"
        print(f"[+] Downloading HTML report from {url}...")
        try:
            response = self.session.get(url, timeout=30)
            response.raise_for_status()
            with open(output_file_path, "w", encoding="utf-8") as file_handle:
                file_handle.write(response.text)
            print(f"[+] Report successfully saved to: {output_file_path}")
        except (requests.RequestException, IOError) as err:
            raise ZAPClientError(f"Failed to export HTML report: {err}") from err


def parse_arguments() -> argparse.Namespace:
    """Parses command-line arguments."""
    parser = argparse.ArgumentParser(description="Automate OWASP ZAP security scans via REST API.")
    parser.add_argument("--target", required=True, help="Target application URL (e.g., https://staging.example.com)")
    parser.add_argument("--spider", action="store_true", help="Execute passive spider crawling")
    parser.add_argument("--active-scan", action="store_true", help="Execute active vulnerability scanning")
    parser.add_argument("--report", default="zap_report.html", help="Path to write the HTML output report")
    return parser.parse_args()


def main() -> None:
    """Main execution function."""
    args = parse_arguments()

    # Read config from environment or use defaults
    zap_url = os.environ.get("ZAP_URL", "http://127.0.0.1:8080")
    zap_api_key = os.environ.get("ZAP_API_KEY", "")

    client = ZAPAutomator(base_url=zap_url, api_key=zap_api_key)

    if not client.check_connection():
        sys.exit(1)

    try:
        if args.spider:
            spider_id = client.start_spider(args.target)
            client.wait_for_spider(spider_id)

        if args.active_scan:
            ascan_id = client.start_active_scan(args.target)
            client.wait_for_active_scan(ascan_id)

        if args.spider or args.active_scan:
            client.export_html_report(args.report)
        else:
            print("[!] No scan option selected. Specify --spider and/or --active-scan.")

    except ZAPClientError as error:
        print(f"[-] Execution error: {error}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

Configuration

The script prioritizes environment variables for sensitive or environment-specific values, falling back to sensible defaults when not provided:

To configure ZAP API authentication in local shell session or CI environments:

export ZAP_URL="http://zap.internal.local:8080"
export ZAP_API_KEY="my-secret-zap-api-key"

Usage Examples

1. Run Spider Crawl Only

Fast discovery of web application endpoints and structure:

python3 zap_scan.py --target https://staging.example.com --spider

Expected Output:

[+] Connected to OWASP ZAP version: 2.14.0
[+] Starting ZAP Spider on target: https://staging.example.com
[*] Spider progress: 15%
[*] Spider progress: 65%
[*] Spider progress: 100%
[+] Spider completed successfully.
[+] Downloading HTML report from http://127.0.0.1:8080/OTHER/core/other/htmlreport/...
[+] Report successfully saved to: zap_report.html

2. Full Security Scan (Spider + Active Scan)

Performs complete crawl followed by active security rule checks:

python3 zap_scan.py --target https://staging.example.com --spider --active-scan --report staging_scan.html

Expected Output:

[+] Connected to OWASP ZAP version: 2.14.0
[+] Starting ZAP Spider on target: https://staging.example.com
[*] Spider progress: 40%
[*] Spider progress: 100%
[+] Spider completed successfully.
[+] Starting ZAP Active Scan on target: https://staging.example.com
[*] Active Scan progress: 10%
[*] Active Scan progress: 55%
[*] Active Scan progress: 100%
[+] Active Scan completed successfully.
[+] Downloading HTML report from http://127.0.0.1:8080/OTHER/core/other/htmlreport/...
[+] Report successfully saved to: staging_scan.html

Error Handling

The script includes defensive failure strategies:

1. Network & Connectivity Failures: ZAPClientError captures standard requests.RequestException errors (such as connection refusal or timeouts) when OWASP ZAP is unreachable or unresponsive.
2. Invalid API Responses: If ZAP returns invalid data or non-200 HTTP statuses (e.g., HTTP 403 Forbidden due to a missing or invalid API key), the exception handler extracts details and exits with status code 1.
3. File I/O Protection: Captures disk permission errors (IOError) when writing the final HTML security report to prevent unhandled script crashes.

Extending Ideas

• Build Pipeline Gating: Parse JSON alerts (/JSON/core/view/alerts/) instead of HTML reports, and fail the pipeline script if high-risk or critical alerts exceed a defined threshold.
• Context & Authentication Injection: Add API calls to set up ZAP Contexts, enabling authenticated scans using session cookies or bearer tokens.
• CI/CD Service Containers: Run OWASP ZAP as a background service container in GitHub Actions or GitLab CI (owasp/zap2docker-stable), triggering this Python script as a post-deployment test step.
• Slack / Webhook Alerts: Send dynamic summary notifications to Slack, Teams, or Jira via webhooks upon scan completion.