Code Examples

Production-ready code examples for integrating with the Cert-IX Scan API. Each example includes retry logic, error handling, and webhook signature verification.

cURL Examples

Submit a Scan

curl -X POST https://api.cert-ix.com/scan-api/api/v1/scans \
  -H "X-API-Key: $CERTIX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "target": "example.com",
    "scanType": "nmap",
    "name": "Network audit",
    "priority": "normal",
    "options": {
      "ports": "1-1024",
      "timing": "T3",
      "serviceDetection": true
    }
  }'

Check Scan Status

curl -X GET "https://api.cert-ix.com/scan-api/api/v1/scans/$SCAN_ID?scanType=nmap" \
  -H "X-API-Key: $CERTIX_API_KEY"

Retrieve Results

curl -X GET "https://api.cert-ix.com/scan-api/api/v1/scans/$SCAN_ID/results?scanType=nmap" \
  -H "X-API-Key: $CERTIX_API_KEY"

List Scans with Filters

curl -X GET "https://api.cert-ix.com/scan-api/api/v1/scans?scanType=nmap&status=completed&page=1&limit=10" \
  -H "X-API-Key: $CERTIX_API_KEY"

Cancel a Scan

curl -X POST "https://api.cert-ix.com/scan-api/api/v1/scans/$SCAN_ID/cancel?scanType=nmap" \
  -H "X-API-Key: $CERTIX_API_KEY"

Create a Webhook

curl -X POST https://api.cert-ix.com/scan-api/api/v1/webhooks \
  -H "X-API-Key: $CERTIX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Scan notifications",
    "url": "https://your-server.example.com/webhooks/certix",
    "events": ["scan.completed", "scan.failed"]
  }'

Create a Scan Template

curl -X POST https://api.cert-ix.com/scan-api/api/v1/scan-templates \
  -H "X-API-Key: $CERTIX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Container CI Scan",
    "scanType": "trivy",
    "priority": "high",
    "options": {
      "scanners": ["vuln", "misconfig", "secret"],
      "severity": ["CRITICAL", "HIGH"]
    },
    "tags": ["ci", "container"]
  }'

Launch a Scan from a Template

curl -X POST "https://api.cert-ix.com/scan-api/api/v1/scan-templates/$TEMPLATE_ID/launch" \
  -H "X-API-Key: $CERTIX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "target": "registry.example.com/myapp:latest"
  }'

---

Python Client

Installation

pip install requests

Full Client Class

"""
Cert-IX Scan API Python Client
Production-ready with retry logic, error handling, and webhook verification.
"""

import hashlib
import hmac
import os
import time
from typing import Any, Dict, List, Optional

import requests


class CertIXScanClient:
    """Cert-IX Scan API client with built-in retry and error handling."""

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        # Read API key from parameter or environment variable
        self.api_key = api_key or os.environ.get("CERTIX_API_KEY")
        if not self.api_key:
            raise ValueError("API key required. Set CERTIX_API_KEY or pass api_key.")

        self.base_url = (
            base_url
            or os.environ.get("CERTIX_BASE_URL")
            or "https://api.cert-ix.com/scan-api/api/v1"
        )
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": self.api_key,
            "Content-Type": "application/json",
        })
        # Maximum retry attempts for transient errors
        self.max_retries = 3

    def _request(self, method: str, path: str, **kwargs) -> Dict[str, Any]:
        """Execute an HTTP request with automatic retry on transient errors."""
        url = f"{self.base_url}{path}"

        for attempt in range(self.max_retries):
            response = self.session.request(method, url, **kwargs)

            # Success — return parsed response
            if response.status_code < 400:
                return response.json()

            # Rate limited — wait and retry
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(retry_after)
                continue

            # Server error — retry with backoff
            if response.status_code >= 500:
                time.sleep(2 ** attempt)
                continue

            # Client error — do not retry, raise immediately
            data = response.json()
            raise CertIXAPIError(
                status_code=response.status_code,
                code=data.get("code", "UNKNOWN"),
                message=data.get("error", "Unknown error"),
            )

        raise CertIXAPIError(429, "RATE_LIMIT_EXCEEDED", "Max retries exceeded")

    # ── Scan Operations ─────────────────────────────────────────────────

    def create_scan(
        self,
        target: str,
        scan_type: str,
        name: Optional[str] = None,
        priority: str = "normal",
        options: Optional[Dict] = None,
        tags: Optional[List[str]] = None,
        timeout: Optional[int] = None,
    ) -> Dict[str, Any]:
        """Submit a new scan request."""
        payload = {"target": target, "scanType": scan_type, "priority": priority}
        if name:
            payload["name"] = name
        if options:
            payload["options"] = options
        if tags:
            payload["tags"] = tags
        if timeout:
            payload["timeout"] = timeout
        result = self._request("POST", "/scans", json=payload)
        return result["data"]

    def get_scan(self, scan_id: str, scan_type: str = "nmap") -> Dict[str, Any]:
        """Get scan status and details."""
        result = self._request("GET", f"/scans/{scan_id}", params={"scanType": scan_type})
        return result["data"]

    def get_scan_results(self, scan_id: str, scan_type: str = "nmap") -> Any:
        """Retrieve scan results (scan must be completed)."""
        result = self._request("GET", f"/scans/{scan_id}/results", params={"scanType": scan_type})
        return result["data"]

    def cancel_scan(self, scan_id: str, scan_type: str = "nmap") -> Dict[str, Any]:
        """Cancel a queued or running scan."""
        result = self._request("POST", f"/scans/{scan_id}/cancel", params={"scanType": scan_type})
        return result["data"]

    def wait_for_scan(self, scan_id: str, scan_type: str = "nmap", timeout: int = 3600) -> Dict[str, Any]:
        """Poll until the scan reaches a terminal status."""
        start = time.time()
        terminal = {"completed", "failed", "cancelled"}

        while time.time() - start < timeout:
            scan = self.get_scan(scan_id, scan_type)
            if scan["status"] in terminal:
                return scan

            # Adaptive polling interval
            elapsed = time.time() - start
            if elapsed < 120:
                interval = 5
            elif elapsed < 600:
                interval = 15
            else:
                interval = 30
            time.sleep(interval)

        raise TimeoutError(f"Scan {scan_id} did not complete within {timeout}s")

    # ── Template Operations ─────────────────────────────────────────────

    def create_template(self, name: str, scan_type: str, **kwargs) -> Dict[str, Any]:
        """Create a reusable scan template."""
        payload = {"name": name, "scanType": scan_type}
        payload.update(kwargs)
        result = self._request("POST", "/scan-templates", json=payload)
        return result["data"]

    def launch_template(self, template_id: str, target: str, **overrides) -> Dict[str, Any]:
        """Launch a scan from a saved template."""
        payload = {"target": target}
        payload.update(overrides)
        result = self._request("POST", f"/scan-templates/{template_id}/launch", json=payload)
        return result["data"]

    # ── Webhook Operations ──────────────────────────────────────────────

    def create_webhook(self, name: str, url: str, events: Optional[List[str]] = None) -> Dict[str, Any]:
        """Register a webhook for scan event notifications."""
        payload = {"name": name, "url": url}
        if events:
            payload["events"] = events
        result = self._request("POST", "/webhooks", json=payload)
        return result["data"]

    # ── Usage Analytics ─────────────────────────────────────────────────

    def get_usage_summary(self, period: str = "30d") -> Dict[str, Any]:
        """Get aggregated usage summary."""
        result = self._request("GET", "/usage/summary", params={"period": period})
        return result["data"]

    # ── Webhook Verification ────────────────────────────────────────────

    @staticmethod
    def verify_webhook_signature(body: bytes, signature_header: str, secret: str) -> bool:
        """Verify HMAC-SHA256 webhook signature."""
        if not signature_header.startswith("sha256="):
            return False
        received = signature_header[7:]
        expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        return hmac.compare_digest(received, expected)


class CertIXAPIError(Exception):
    """Exception raised for Cert-IX API errors."""
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message
        super().__init__(f"[{status_code}] {code}: {message}")

Usage Example

from certix_client import CertIXScanClient

# Initialize client (reads CERTIX_API_KEY from environment)
client = CertIXScanClient()

# Submit a scan and wait for results
scan = client.create_scan(
    target="example.com",
    scan_type="nmap",
    name="Python SDK test",
    options={"ports": "1-1024", "timing": "T3"},
)
print(f"Scan created: {scan['id']}")

# Wait for completion
final = client.wait_for_scan(scan["id"], "nmap")
print(f"Status: {final['status']}")

# Retrieve results
if final["status"] == "completed":
    results = client.get_scan_results(scan["id"], "nmap")
    print(f"Results: {results}")

---

Go Client

Full Client Package

package certix

import (
    "bytes"
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "net/url"
    "os"
    "strings"
    "time"
)

// Client is the Cert-IX Scan API client.
type Client struct {
    apiKey     string
    baseURL    string
    httpClient *http.Client
    maxRetries int
}

// NewClient creates a new Cert-IX API client.
func NewClient(apiKey, baseURL string) (*Client, error) {
    if apiKey == "" {
        apiKey = os.Getenv("CERTIX_API_KEY")
    }
    if apiKey == "" {
        return nil, fmt.Errorf("API key required: set CERTIX_API_KEY or pass apiKey")
    }
    if baseURL == "" {
        baseURL = "https://api.cert-ix.com/scan-api/api/v1"
    }
    return &Client{
        apiKey:     apiKey,
        baseURL:    strings.TrimRight(baseURL, "/"),
        httpClient: &http.Client{Timeout: 30 * time.Second},
        maxRetries: 3,
    }, nil
}

// APIResponse is the standard response envelope.
type APIResponse struct {
    Success bool            `json:"success"`
    Data    json.RawMessage `json:"data,omitempty"`
    Error   string          `json:"error,omitempty"`
    Code    string          `json:"code,omitempty"`
}

// ScanRequest is the request body for creating a scan.
type ScanRequest struct {
    Target   string                 `json:"target"`
    ScanType string                 `json:"scanType"`
    Name     string                 `json:"name,omitempty"`
    Priority string                 `json:"priority,omitempty"`
    Options  map[string]interface{} `json:"options,omitempty"`
    Tags     []string               `json:"tags,omitempty"`
}

// ScanResponse is the scan status response.
type ScanResponse struct {
    ID          string `json:"id"`
    ScanType    string `json:"scanType"`
    Name        string `json:"name"`
    Target      string `json:"target"`
    Status      string `json:"status"`
    Progress    int    `json:"progress"`
    ResultCount int    `json:"resultCount,omitempty"`
    DurationMs  int64  `json:"durationMs,omitempty"`
    Error       string `json:"error,omitempty"`
}

// APIError represents a Cert-IX API error.
type APIError struct {
    StatusCode int
    Code       string
    Message    string
}

func (e *APIError) Error() string {
    return fmt.Sprintf("[%d] %s: %s", e.StatusCode, e.Code, e.Message)
}

// doRequest executes an HTTP request with retry logic.
func (c *Client) doRequest(method, path string, body interface{}, params map[string]string) (*APIResponse, error) {
    for attempt := 0; attempt < c.maxRetries; attempt++ {
        reqURL := c.baseURL + path
        if len(params) > 0 {
            q := url.Values{}
            for k, v := range params {
                q.Set(k, v)
            }
            reqURL += "?" + q.Encode()
        }

        var bodyReader io.Reader
        if body != nil {
            data, err := json.Marshal(body)
            if err != nil {
                return nil, fmt.Errorf("failed to marshal request: %w", err)
            }
            bodyReader = bytes.NewReader(data)
        }

        req, err := http.NewRequest(method, reqURL, bodyReader)
        if err != nil {
            return nil, fmt.Errorf("failed to create request: %w", err)
        }
        req.Header.Set("X-API-Key", c.apiKey)
        req.Header.Set("Content-Type", "application/json")

        resp, err := c.httpClient.Do(req)
        if err != nil {
            time.Sleep(time.Duration(1<<attempt) * time.Second)
            continue
        }
        defer resp.Body.Close()

        respBody, _ := io.ReadAll(resp.Body)
        var apiResp APIResponse
        json.Unmarshal(respBody, &apiResp)

        if resp.StatusCode < 400 {
            return &apiResp, nil
        }
        if resp.StatusCode == 429 || resp.StatusCode >= 500 {
            time.Sleep(time.Duration(1<<attempt) * time.Second)
            continue
        }
        return nil, &APIError{StatusCode: resp.StatusCode, Code: apiResp.Code, Message: apiResp.Error}
    }
    return nil, &APIError{StatusCode: 429, Code: "RATE_LIMIT_EXCEEDED", Message: "max retries exceeded"}
}

// CreateScan submits a new scan request.
func (c *Client) CreateScan(req *ScanRequest) (*ScanResponse, error) {
    resp, err := c.doRequest("POST", "/scans", req, nil)
    if err != nil {
        return nil, err
    }
    var scan ScanResponse
    json.Unmarshal(resp.Data, &scan)
    return &scan, nil
}

// GetScan retrieves scan status and details.
func (c *Client) GetScan(scanID, scanType string) (*ScanResponse, error) {
    resp, err := c.doRequest("GET", "/scans/"+scanID, nil, map[string]string{"scanType": scanType})
    if err != nil {
        return nil, err
    }
    var scan ScanResponse
    json.Unmarshal(resp.Data, &scan)
    return &scan, nil
}

// GetScanResults retrieves results for a completed scan.
func (c *Client) GetScanResults(scanID, scanType string) (json.RawMessage, error) {
    resp, err := c.doRequest("GET", "/scans/"+scanID+"/results", nil, map[string]string{"scanType": scanType})
    if err != nil {
        return nil, err
    }
    return resp.Data, nil
}

// WaitForScan polls until the scan reaches a terminal status.
func (c *Client) WaitForScan(scanID, scanType string, timeout time.Duration) (*ScanResponse, error) {
    start := time.Now()
    terminal := map[string]bool{"completed": true, "failed": true, "cancelled": true}

    for time.Since(start) < timeout {
        scan, err := c.GetScan(scanID, scanType)
        if err != nil {
            return nil, err
        }
        if terminal[scan.Status] {
            return scan, nil
        }
        elapsed := time.Since(start)
        var interval time.Duration
        switch {
        case elapsed < 2*time.Minute:
            interval = 5 * time.Second
        case elapsed < 10*time.Minute:
            interval = 15 * time.Second
        default:
            interval = 30 * time.Second
        }
        time.Sleep(interval)
    }
    return nil, fmt.Errorf("scan %s did not complete within %v", scanID, timeout)
}

// VerifyWebhookSignature validates an HMAC-SHA256 webhook signature.
func VerifyWebhookSignature(body []byte, signatureHeader, secret string) bool {
    if !strings.HasPrefix(signatureHeader, "sha256=") {
        return false
    }
    received := signatureHeader[7:]
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(received), []byte(expected))
}

Usage Example

package main

import (
    "fmt"
    "log"
    "time"
    certix "your-module/certix"
)

func main() {
    client, err := certix.NewClient("", "")
    if err != nil {
        log.Fatal(err)
    }

    scan, err := client.CreateScan(&certix.ScanRequest{
        Target:   "example.com",
        ScanType: "nmap",
        Name:     "Go SDK test",
        Options:  map[string]interface{}{"ports": "1-1024", "timing": "T3"},
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Scan created: %s\n", scan.ID)

    final, err := client.WaitForScan(scan.ID, "nmap", 10*time.Minute)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Status: %s\n", final.Status)
}

---

Node.js Client

Installation

npm install node-fetch@3
# or use Node.js 18+ built-in fetch

Full Client Class

/**
 * Cert-IX Scan API Node.js Client
 * Production-ready with retry logic, error handling, and webhook verification.
 */

const crypto = require("crypto");

class CertIXScanClient {
  constructor(apiKey, baseUrl) {
    this.apiKey = apiKey || process.env.CERTIX_API_KEY;
    if (!this.apiKey) throw new Error("API key required. Set CERTIX_API_KEY or pass apiKey.");
    this.baseUrl = baseUrl || process.env.CERTIX_BASE_URL || "https://api.cert-ix.com/scan-api/api/v1";
    this.maxRetries = 3;
  }

  async _request(method, path, { body, params } = {}) {
    let url = `${this.baseUrl}${path}`;
    if (params) url += `?${new URLSearchParams(params).toString()}`;

    for (let attempt = 0; attempt < this.maxRetries; attempt++) {
      const options = {
        method,
        headers: { "X-API-Key": this.apiKey, "Content-Type": "application/json" },
      };
      if (body) options.body = JSON.stringify(body);

      const response = await fetch(url, options);

      if (response.status < 400) return response.json();

      if (response.status === 429) {
        const retryAfter = parseInt(response.headers.get("Retry-After") || 2 ** attempt, 10);
        await new Promise((r) => setTimeout(r, retryAfter * 1000));
        continue;
      }
      if (response.status >= 500) {
        await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
        continue;
      }

      const data = await response.json();
      throw new CertIXAPIError(response.status, data.code || "UNKNOWN", data.error || "Unknown error");
    }
    throw new CertIXAPIError(429, "RATE_LIMIT_EXCEEDED", "Max retries exceeded");
  }

  // ── Scan Operations ──────────────────────────────────────────────────

  async createScan({ target, scanType, name, priority = "normal", options, tags, timeout }) {
    const payload = { target, scanType, priority };
    if (name) payload.name = name;
    if (options) payload.options = options;
    if (tags) payload.tags = tags;
    if (timeout) payload.timeout = timeout;
    const result = await this._request("POST", "/scans", { body: payload });
    return result.data;
  }

  async getScan(scanId, scanType = "nmap") {
    const result = await this._request("GET", `/scans/${scanId}`, { params: { scanType } });
    return result.data;
  }

  async getScanResults(scanId, scanType = "nmap") {
    const result = await this._request("GET", `/scans/${scanId}/results`, { params: { scanType } });
    return result.data;
  }

  async cancelScan(scanId, scanType = "nmap") {
    const result = await this._request("POST", `/scans/${scanId}/cancel`, { params: { scanType } });
    return result.data;
  }

  async waitForScan(scanId, scanType = "nmap", timeoutMs = 3600000) {
    const start = Date.now();
    const terminal = new Set(["completed", "failed", "cancelled"]);

    while (Date.now() - start < timeoutMs) {
      const scan = await this.getScan(scanId, scanType);
      if (terminal.has(scan.status)) return scan;

      const elapsed = Date.now() - start;
      const interval = elapsed < 120_000 ? 5_000 : elapsed < 600_000 ? 15_000 : 30_000;
      await new Promise((r) => setTimeout(r, interval));
    }
    throw new Error(`Scan ${scanId} did not complete within ${timeoutMs}ms`);
  }

  // ── Template Operations ──────────────────────────────────────────────

  async createTemplate({ name, scanType, priority = "normal", options, tags, ...rest }) {
    const payload = { name, scanType, priority, ...rest };
    if (options) payload.options = options;
    if (tags) payload.tags = tags;
    const result = await this._request("POST", "/scan-templates", { body: payload });
    return result.data;
  }

  async launchTemplate(templateId, target, overrides = {}) {
    const payload = { target, ...overrides };
    const result = await this._request("POST", `/scan-templates/${templateId}/launch`, { body: payload });
    return result.data;
  }

  // ── Webhook Verification ─────────────────────────────────────────────

  static verifyWebhookSignature(body, signatureHeader, secret) {
    if (!signatureHeader.startsWith("sha256=")) return false;
    const received = signatureHeader.slice(7);
    const expected = crypto.createHmac("sha256", secret).update(body).digest("hex");
    return crypto.timingSafeEqual(Buffer.from(received), Buffer.from(expected));
  }
}

class CertIXAPIError extends Error {
  constructor(statusCode, code, message) {
    super(`[${statusCode}] ${code}: ${message}`);
    this.statusCode = statusCode;
    this.code = code;
  }
}

module.exports = { CertIXScanClient, CertIXAPIError };

Usage Example

const { CertIXScanClient } = require("./certix-client");

async function main() {
  const client = new CertIXScanClient();

  const scan = await client.createScan({
    target: "example.com",
    scanType: "nmap",
    name: "Node.js SDK test",
    options: { ports: "1-1024", timing: "T3" },
  });
  console.log(`Scan created: ${scan.id}`);

  const final = await client.waitForScan(scan.id, "nmap", 600_000);
  console.log(`Status: ${final.status}`);

  if (final.status === "completed") {
    const results = await client.getScanResults(scan.id, "nmap");
    console.log("Results:", JSON.stringify(results, null, 2));
  }
}

main().catch(console.error);

Express Webhook Handler

const express = require("express");
const { CertIXScanClient } = require("./certix-client");

const app = express();
const WEBHOOK_SECRET = process.env.CERTIX_WEBHOOK_SECRET;

app.post("/webhooks/certix", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.headers["x-webhook-signature"] || "";
  if (!CertIXScanClient.verifyWebhookSignature(req.body, signature, WEBHOOK_SECRET)) {
    return res.status(401).send("Invalid signature");
  }

  const payload = JSON.parse(req.body);
  console.log(`Event: ${payload.event}`);

  switch (payload.event) {
    case "scan.completed":
      console.log(`Scan ${payload.data.scanId} completed with ${payload.data.resultCount} findings`);
      break;
    case "scan.failed":
      console.error(`Scan ${payload.data.scanId} failed: ${payload.data.error}`);
      break;
  }

  res.status(200).send("OK");
});

app.listen(3000, () => console.log("Webhook server listening on port 3000"));

---

CI/CD Integration — GitHub Actions

# .github/workflows/security-scan.yml
name: Security Scan
on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

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

      - name: Build container image
        run: docker build -t myapp:${{ github.sha }} .

      - name: Run Cert-IX container scan
        env:
          CERTIX_API_KEY: ${{ secrets.CERTIX_API_KEY }}
        run: |
          python3 - <<'EOF'
          import os, json, sys
          sys.path.insert(0, ".")
          from certix_client import CertIXScanClient

          client = CertIXScanClient()

          # Submit container image scan
          scan = client.create_scan(
              target=f"ghcr.io/${{ github.repository }}:${{ github.sha }}",
              scan_type="trivy",
              name=f"CI Scan - {os.environ.get('GITHUB_SHA', '')[:8]}",
              priority="high",
              options={
                  "scanners": ["vuln", "misconfig", "secret"],
                  "severity": ["CRITICAL", "HIGH"],
                  "ignoreUnfixed": True,
              },
          )
          print(f"Scan submitted: {scan['id']}")

          # Wait for completion (max 10 minutes)
          result = client.wait_for_scan(scan["id"], "trivy", timeout=600)

          if result["status"] == "failed":
              print(f"Scan failed: {result.get('error')}")
              exit(1)

          # Check for critical findings
          findings = client.get_scan_results(scan["id"], "trivy")
          summary = findings.get("summary", {})
          critical = summary.get("critical", 0)
          high = summary.get("high", 0)

          print(f"Findings: {critical} critical, {high} high")

          if critical > 0:
              print("CRITICAL vulnerabilities found — blocking deployment")
              exit(1)

          print("Security scan passed")
          EOF

---

Next Steps:

• Getting Started →
• Submitting Scans →
• Webhooks →