Lab 07 — Security API Integration

Mission Information

Difficulty: Intermediate → Advanced
Estimated Time: 120–180 minutes
Primary Language: Python
Security Domain: Security Automation / SOC / Cloud Security / Vulnerability Management
Environment: Local controlled lab
Automation Type: Defensive API Integration

Mission

Your task is to build a reusable Python security API client.

Instead of manually opening a security dashboard and exporting data, your script will interact with a controlled local training API.

You will learn how to:

AUTHENTICATE

SEND API REQUESTS

USE HTTP METHODS

PASS PARAMETERS

READ JSON

VALIDATE RESPONSES

HANDLE TIMEOUTS

HANDLE ERRORS

IMPLEMENT RETRIES

UNDERSTAND RATE LIMITS

NORMALIZE SECURITY DATA

GENERATE REPORTS

The final workflow will be:

SECURITY PLATFORM
       ↓
REST API
       ↓
PYTHON CLIENT
       ↓
AUTHENTICATE
       ↓
REQUEST
       ↓
VALIDATE RESPONSE
       ↓
PARSE JSON
       ↓
NORMALIZE
       ↓
ANALYZE
       ↓
REPORT

Why This Lab Matters

Modern security platforms are heavily API-driven.

Examples include:

SIEM

EDR / XDR

CLOUD SECURITY PLATFORMS

VULNERABILITY MANAGEMENT

THREAT INTELLIGENCE

IDENTITY SECURITY

TICKETING SYSTEMS

SOAR

ASSET MANAGEMENT

COMPLIANCE PLATFORMS

A security automation engineer may need to connect:

VULNERABILITY PLATFORM
        ↓
PYTHON
        ↓
TICKETING SYSTEM

or:

SIEM ALERT
   ↓
API
   ↓
THREAT INTELLIGENCE
   ↓
ENRICHED ALERT

or:

CLOUD SECURITY API
        ↓
FINDINGS
        ↓
NORMALIZATION
        ↓
SECURITY DASHBOARD

Learning Objectives

By completing this lab, you should be able to:

UNDERSTAND REST APIs

UNDERSTAND HTTP METHODS

USE API ENDPOINTS

USE QUERY PARAMETERS

USE HEADERS

USE BEARER TOKENS

PROTECT API SECRETS

SEND GET REQUESTS

SEND CONTROLLED POST REQUESTS

PARSE JSON

VALIDATE HTTP STATUS CODES

CONFIGURE TIMEOUTS

HANDLE NETWORK ERRORS

IMPLEMENT RETRIES

UNDERSTAND BACKOFF

UNDERSTAND RATE LIMITING

BUILD PAGINATION LOGIC

NORMALIZE API DATA

EXPORT SECURITY REPORTS

BUILD A REUSABLE API CLIENT

Final Architecture

             SECURITY AUTOMATION
                      ↓
              ┌─────────────┐
              │ API CLIENT  │
              └──────┬──────┘
                     ↓
                AUTHENTICATE
                     ↓
                HTTP REQUEST
                     ↓
                API ENDPOINT
                     ↓
               HTTP RESPONSE
                     ↓
          ┌──────────┼──────────┐
          ↓          ↓          ↓
       STATUS      HEADERS      JSON
          ↓          ↓          ↓
          └──────────┼──────────┘
                     ↓
                 VALIDATE
                     ↓
                 NORMALIZE
                     ↓
                  ANALYZE
                     ↓
             SECURITY REPORT

Authorization and Safety

This lab uses:

A LOCAL TRAINING API

and:

SYNTHETIC SECURITY DATA

Do not connect automation to a production security platform until you understand:

API PERMISSIONS

DATA SENSITIVITY

RATE LIMITS

CHANGE IMPACT

ERROR HANDLING

AUDIT REQUIREMENTS

For this lab, API actions remain:

READ-ONLY

except for an optional local training POST request that creates only synthetic lab records.

01 — Create the Lab Workspace

Create:

security-api-integration/
|
+-- api/
|
+-- data/
|
+-- reports/
|
+-- src/
|
+-- tests/
|
+-- README.md

Linux/macOS:

mkdir -p security-api-integration/{api,data,reports,src,tests}
cd security-api-integration

PowerShell:

mkdir security-api-integration

cd security-api-integration

mkdir api
mkdir data
mkdir reports
mkdir src
mkdir tests

02 — Verify Python

Run:

python --version

Recommended:

Python 3.10+

03 — API Concepts

Before coding, understand:

CLIENT

The application making the request.

SERVER

The application responding.

ENDPOINT

A specific API path.

Example:

/api/findings

REQUEST

What the client sends.

RESPONSE

What the server returns.

04 — REST API Mental Model

CLIENT
  ↓
HTTP REQUEST
  ↓
ENDPOINT
  ↓
SERVER
  ↓
HTTP RESPONSE
  ↓
CLIENT

05 — Understand HTTP Methods

Common methods:

Method	Typical Purpose
GET	Retrieve data
POST	Create data
PUT	Replace data
PATCH	Modify part of a resource
DELETE	Remove data

For this lab, focus mainly on:

GET

and one controlled local:

POST

06 — Understand HTTP Status Codes

Important examples:

200
Success

201
Created

400
Bad Request

401
Unauthorized

403
Forbidden

404
Not Found

429
Too Many Requests

500
Server Error

503
Service Unavailable

07 — Why Status Codes Matter

Do not write:

REQUEST COMPLETED
=
SUCCESS

The actual workflow is:

REQUEST SENT
     ↓
RESPONSE RECEIVED
     ↓
STATUS CODE?
     ↓
VALID?
     ↓
PARSE DATA

08 — Build a Local Training API

To keep the lab controlled, create a small local API.

Create:

api/training_api.py

09 — Use Python Standard Library

For the server, use:

from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
import json

This avoids requiring an external framework for the lab.

10 — Create Synthetic Security Findings

Add:

FINDINGS = [
    {
        "id": "F-1001",
        "asset": "WEB01",
        "type": "vulnerability",
        "severity": "critical",
        "status": "open",
        "score": 9.8
    },
    {
        "id": "F-1002",
        "asset": "DB01",
        "type": "vulnerability",
        "severity": "high",
        "status": "open",
        "score": 8.2
    },
    {
        "id": "F-1003",
        "asset": "APP01",
        "type": "configuration",
        "severity": "medium",
        "status": "open",
        "score": 5.4
    },
    {
        "id": "F-1004",
        "asset": "VPN01",
        "type": "configuration",
        "severity": "high",
        "status": "closed",
        "score": 7.6
    },
    {
        "id": "F-1005",
        "asset": "JUMP01",
        "type": "identity",
        "severity": "critical",
        "status": "open",
        "score": 9.1
    }
]

11 — Define a Training Token

Add:

API_TOKEN = "training-token-123"

This is acceptable only because:

THE SERVER IS LOCAL

THE TOKEN IS SYNTHETIC

THE VALUE HAS NO REAL PRIVILEGE

Do not copy this pattern for production credentials.

12 — Create the API Handler

Add:

class TrainingAPIHandler(
    BaseHTTPRequestHandler
):
    def send_json(
        self,
        data,
        status=200
    ):
        body = json.dumps(
            data
        ).encode(
            "utf-8"
        )

        self.send_response(
            status
        )

        self.send_header(
            "Content-Type",
            "application/json"
        )

        self.send_header(
            "Content-Length",
            str(
                len(body)
            )
        )

        self.end_headers()

        self.wfile.write(
            body
        )

13 — Add Authentication Checking

Add:

    def is_authorized(
        self
    ):
        authorization = (
            self.headers.get(
                "Authorization",
                ""
            )
        )

        expected = (
            f"Bearer {API_TOKEN}"
        )

        return (
            authorization
            ==
            expected
        )

14 — Why Use an Authorization Header?

A common API pattern is:

Authorization:
Bearer TOKEN

Conceptually:

CLIENT
 ↓
TOKEN
 ↓
API
 ↓
IDENTITY / PERMISSION CHECK

15 — Create the GET Handler

Add:

    def do_GET(
        self
    ):
        if not self.is_authorized():
            self.send_json(
                {
                    "error":
                        "unauthorized"
                },
                401
            )
            return

        parsed = urlparse(
            self.path
        )

        if parsed.path == "/api/findings":
            self.handle_findings(
                parsed
            )
            return

        if parsed.path == "/api/health":
            self.send_json(
                {
                    "status":
                        "ok"
                }
            )
            return

        self.send_json(
            {
                "error":
                    "not_found"
            },
            404
        )

16 — Add Query Parameter Filtering

Create:

    def handle_findings(
        self,
        parsed
    ):
        query = parse_qs(
            parsed.query
        )

        severity = query.get(
            "severity",
            [None]
        )[0]

        status = query.get(
            "status",
            [None]
        )[0]

        results = FINDINGS

        if severity:
            results = [
                item
                for item in results
                if item[
                    "severity"
                ] == severity.lower()
            ]

        if status:
            results = [
                item
                for item in results
                if item[
                    "status"
                ] == status.lower()
            ]

        self.send_json(
            {
                "count":
                    len(results),

                "results":
                    results
            }
        )

17 — Start the Training API

Add:

def main():
    server = HTTPServer(
        (
            "127.0.0.1",
            8080
        ),
        TrainingAPIHandler
    )

    print(
        "Training API listening on "
        "http://127.0.0.1:8080"
    )

    server.serve_forever()

Then:

if __name__ == "__main__":
    main()

18 — Start the Server

Run in Terminal 1:

python api/training_api.py

Expected:

Training API listening on http://127.0.0.1:8080

19 — Test the Health Endpoint

For now use Python rather than relying on additional tools.

Create:

src/test_connection.py

Add:

from urllib.request import Request, urlopen

request = Request(
    "http://127.0.0.1:8080/api/health"
)

request.add_header(
    "Authorization",
    "Bearer training-token-123"
)

with urlopen(
    request,
    timeout=5
) as response:

    print(
        response.read().decode(
            "utf-8"
        )
    )

Run in Terminal 2:

python src/test_connection.py

Expected:

{"status": "ok"}

20 — Test Unauthorized Access

Change the token temporarily to:

wrong-token

Run again.

Expected:

HTTP 401

Restore the correct synthetic token afterward.

21 — Why Authentication Errors Matter

Your script must distinguish between:

NETWORK FAILURE

and:

AUTHENTICATION FAILURE

because the troubleshooting action is different.

22 — Create the Production-Style Client

Create:

src/security_api_client.py

Start with:

from pathlib import Path
from urllib.request import Request, urlopen
from urllib.parse import urlencode
from urllib.error import HTTPError, URLError
import csv
import json
import logging
import os
import time

23 — Define Project Paths

Add:

BASE_DIR = Path(__file__).resolve().parent.parent

REPORT_DIR = BASE_DIR / "reports"

REPORT_DIR.mkdir(
    parents=True,
    exist_ok=True
)

24 — Configure Logging

Add:

logging.basicConfig(
    level=logging.INFO,
    format=(
        "%(asctime)s "
        "%(levelname)s "
        "%(message)s"
    )
)

25 — Define API Configuration

Add:

BASE_URL = (
    "http://127.0.0.1:8080"
)

DEFAULT_TIMEOUT = 5

MAX_RETRIES = 3

26 — Protect the Client Token

Do not hard-code the client token.

Use:

API_TOKEN = os.getenv(
    "SECURITY_API_TOKEN"
)

27 — Configure the Environment Variable

Linux/macOS:

export SECURITY_API_TOKEN="training-token-123"

PowerShell:

$env:SECURITY_API_TOKEN = "training-token-123"

This environment variable lasts for the relevant shell/session.

28 — Why Environment Variables?

They are better than:

API_TOKEN = "secret"

because hard-coded secrets can leak into:

GIT

BACKUPS

LOGS

SCREENSHOTS

CODE REVIEWS

29 — Environment Variables Are Not Perfect Secret Management

Production environments should consider:

SECRET MANAGERS

WORKLOAD IDENTITY

MANAGED IDENTITIES

SHORT-LIVED TOKENS

The lab uses an environment variable to teach the fundamental pattern.

30 — Validate Configuration

Create:

def validate_configuration():
    if not API_TOKEN:
        raise RuntimeError(
            "SECURITY_API_TOKEN "
            "environment variable is not set."
        )

31 — Create Headers

Create:

def build_headers():
    return {
        "Authorization":
            f"Bearer {API_TOKEN}",

        "Accept":
            "application/json",

        "User-Agent":
            "GoHackersCloud-Security-Lab/1.0"
    }

32 — Why User-Agent?

A useful API client should identify itself.

This supports:

LOGGING

TROUBLESHOOTING

AUDITING

API ANALYTICS

33 — Build URLs Safely

Create:

def build_url(
    endpoint,
    params=None
):
    url = (
        BASE_URL.rstrip("/")
        +
        "/"
        +
        endpoint.lstrip("/")
    )

    if params:
        query = urlencode(
            params
        )

        url = (
            f"{url}?{query}"
        )

    return url

34 — Why Use urlencode()?

Avoid manually constructing:

?severity=critical&status=open

because encoding rules become error-prone.

Use:

STRUCTURED PARAMETERS

35 — Build the Request Function

Create:

def send_get_request(
    endpoint,
    params=None,
    timeout=DEFAULT_TIMEOUT
):
    url = build_url(
        endpoint,
        params
    )

    request = Request(
        url,
        method="GET"
    )

    for name, value in (
        build_headers().items()
    ):
        request.add_header(
            name,
            value
        )

    with urlopen(
        request,
        timeout=timeout
    ) as response:

        return {
            "status":
                response.status,

            "headers":
                dict(
                    response.headers
                ),

            "body":
                response.read()
                .decode(
                    "utf-8"
                )
        }

36 — First API Request

Test:

response = send_get_request(
    "/api/findings"
)

print(response)

37 — Parse JSON

Create:

def parse_json_response(
    body
):
    try:
        return json.loads(
            body
        )

    except json.JSONDecodeError as error:
        raise ValueError(
            "API returned invalid JSON"
        ) from error

38 — Why JSON Validation Matters

Do not assume:

Content-Type: application/json

means the body always contains valid JSON.

Systems can return:

HTML ERROR PAGE

EMPTY BODY

TRUNCATED RESPONSE

INVALID JSON

39 — Create a Higher-Level API Function

Create:

def get_findings(
    severity=None,
    status=None
):
    params = {}

    if severity:
        params[
            "severity"
        ] = severity

    if status:
        params[
            "status"
        ] = status

    response = send_get_request(
        "/api/findings",
        params=params
    )

    if response[
        "status"
    ] != 200:
        raise RuntimeError(
            "Unexpected response status"
        )

    data = parse_json_response(
        response[
            "body"
        ]
    )

    return data

40 — Query Critical Findings

Run:

data = get_findings(
    severity="critical"
)

print(
    json.dumps(
        data,
        indent=2
    )
)

Expected:

WEB01

JUMP01

41 — Query Open Findings

Run:

data = get_findings(
    status="open"
)

42 — Combine Filters

Run:

data = get_findings(
    severity="critical",
    status="open"
)

43 — Understand Query Parameters

Your client generates:

/api/findings?severity=critical&status=open

This is different from sending filtering instructions in:

HTTP HEADERS

or:

REQUEST BODY

44 — Handle HTTP Errors

Currently:

401

404

429

500

can raise:

HTTPError

Create:

def handle_http_error(
    error
):
    if error.code == 401:
        return (
            "Authentication failed"
        )

    if error.code == 403:
        return (
            "Request forbidden"
        )

    if error.code == 404:
        return (
            "API endpoint not found"
        )

    if error.code == 429:
        return (
            "API rate limit reached"
        )

    if 500 <= error.code <= 599:
        return (
            "API server error"
        )

    return (
        f"HTTP error {error.code}"
    )

45 — Handle Network Errors

Examples include:

SERVER NOT RUNNING

DNS FAILURE

CONNECTION REFUSED

ROUTING ISSUE

These often appear as:

URLError

46 — Build a Safe Request Wrapper

Create:

def api_get(
    endpoint,
    params=None
):
    try:
        response = send_get_request(
            endpoint,
            params=params
        )

        return {
            "success":
                True,

            "status":
                response[
                    "status"
                ],

            "data":
                parse_json_response(
                    response[
                        "body"
                    ]
                ),

            "error":
                None
        }

    except HTTPError as error:
        message = (
            handle_http_error(
                error
            )
        )

        logging.error(
            message
        )

        return {
            "success":
                False,

            "status":
                error.code,

            "data":
                None,

            "error":
                message
        }

    except URLError as error:
        message = (
            f"Network error: "
            f"{error.reason}"
        )

        logging.error(
            message
        )

        return {
            "success":
                False,

            "status":
                None,

            "data":
                None,

            "error":
                message
        }

    except ValueError as error:
        logging.error(
            str(error)
        )

        return {
            "success":
                False,

            "status":
                None,

            "data":
                None,

            "error":
                str(error)
        }

47 — Test Wrong Endpoint

Call:

result = api_get(
    "/api/not-real"
)

print(result)

Expected:

404

with a clean application error rather than a full uncontrolled crash.

48 — Test Server Failure

Stop:

training_api.py

Then run the client.

Expected:

NETWORK ERROR

Restart the server afterward.

49 — Why Timeouts Matter

Never create API automation that waits forever.

Without a timeout:

REQUEST
   ↓
SERVER NEVER RESPONDS
   ↓
AUTOMATION HANGS

With a timeout:

REQUEST
   ↓
WAIT LIMITED TIME
   ↓
FAIL SAFELY

50 — Test Timeout Concept

You already configured:

DEFAULT_TIMEOUT = 5

Every request should have an explicit timeout.

51 — Understand Retry Logic

Some errors are:

TEMPORARY

such as:

503 SERVICE UNAVAILABLE

CONNECTION RESET

TRANSIENT NETWORK ISSUE

Others are not good retry candidates:

401 UNAUTHORIZED

403 FORBIDDEN

400 BAD REQUEST

52 — Retry Mental Model

REQUEST
   ↓
TEMPORARY FAILURE?
   ↓ YES
WAIT
   ↓
RETRY
   ↓
SUCCESS?

53 — Backoff

Avoid:

FAIL
↓
RETRY IMMEDIATELY
↓
FAIL
↓
RETRY IMMEDIATELY

Prefer:

FAIL
↓
WAIT 1 SECOND
↓
FAIL
↓
WAIT 2 SECONDS
↓
FAIL
↓
WAIT 4 SECONDS

54 — Build Retry Logic

Create:

def api_get_with_retry(
    endpoint,
    params=None,
    max_retries=MAX_RETRIES
):
    for attempt in range(
        max_retries
    ):
        result = api_get(
            endpoint,
            params=params
        )

        if result[
            "success"
        ]:
            return result

        status = result[
            "status"
        ]

        retryable = (
            status is None
            or
            status == 429
            or
            (
                status
                and
                500 <= status <= 599
            )
        )

        if not retryable:
            return result

        if attempt < (
            max_retries - 1
        ):
            delay = (
                2 ** attempt
            )

            logging.warning(
                f"Retrying in "
                f"{delay} seconds"
            )

            time.sleep(
                delay
            )

    return result

55 — Why Limit Retries?

Without a limit:

SERVICE FAILURE
      ↓
AUTOMATION RETRIES FOREVER

This can create:

RESOURCE EXHAUSTION

API PRESSURE

FAILED WORKFLOWS

NOISY LOGGING

56 — Rate Limiting

APIs may restrict usage to values such as:

100 REQUESTS / MINUTE

or:

1,000 REQUESTS / HOUR

57 — HTTP 429

A common rate-limit response is:

429 Too Many Requests

The API may also provide:

Retry-After

or custom headers.

58 — Rate Limit Mental Model

CLIENT
 ↓
REQUEST REQUEST REQUEST
 ↓
API LIMIT
 ↓
429
 ↓
WAIT
 ↓
CONTINUE

59 — Do Not Fight Rate Limits

Do not design:

429
 ↓
OPEN MORE THREADS
 ↓
SEND MORE REQUESTS

Respect provider limits.

60 — Pagination

Large APIs usually do not return:

100,000 FINDINGS

in one response.

Instead:

PAGE 1

PAGE 2

PAGE 3

61 — Pagination Model

REQUEST PAGE 1
      ↓
PROCESS
      ↓
MORE?
      ↓
REQUEST PAGE 2
      ↓
PROCESS

62 — Add Pagination to the Training API

Modify:

handle_findings()

to accept:

page

page_size

Add:

        page = int(
            query.get(
                "page",
                ["1"]
            )[0]
        )

        page_size = int(
            query.get(
                "page_size",
                ["2"]
            )[0]
        )

63 — Calculate Page Boundaries

Add:

        start = (
            page - 1
        ) * page_size

        end = (
            start
            +
            page_size
        )

        paged_results = (
            results[
                start:end
            ]
        )

64 — Return Pagination Metadata

Change response to:

        self.send_json(
            {
                "page":
                    page,

                "page_size":
                    page_size,

                "total":
                    len(results),

                "results":
                    paged_results
            }
        )

Restart the server.

65 — Create Pagination Client

Create:

def get_all_findings(
    page_size=2
):
    all_findings = []

    page = 1

    while True:
        result = (
            api_get_with_retry(
                "/api/findings",
                params={
                    "page":
                        page,

                    "page_size":
                        page_size
                }
            )
        )

        if not result[
            "success"
        ]:
            raise RuntimeError(
                result[
                    "error"
                ]
            )

        data = result[
            "data"
        ]

        records = data[
            "results"
        ]

        all_findings.extend(
            records
        )

        if len(
            all_findings
        ) >= data[
            "total"
        ]:
            break

        page += 1

    return all_findings

66 — Run Pagination

Test:

findings = get_all_findings()

print(
    len(findings)
)

Expected:

5

67 — Pagination Safety

Always protect against:

ENDLESS PAGINATION

caused by buggy APIs.

A mature client can enforce:

MAXIMUM PAGE COUNT

68 — Normalize API Findings

Different vendors may return different fields.

Vendor A:

{
  "machine": "WEB01",
  "risk": "CRITICAL"
}

Vendor B:

{
  "host": "WEB01",
  "severity": "critical"
}

Your internal model should be consistent.

69 — Create an Internal Finding Schema

Use:

finding_id

asset

finding_type

severity

status

score

source

70 — Normalize Training API Records

Create:

def normalize_finding(
    record
):
    required = {
        "id",
        "asset",
        "type",
        "severity",
        "status",
        "score"
    }

    missing = (
        required
        -
        set(
            record.keys()
        )
    )

    if missing:
        raise ValueError(
            "Missing fields: "
            +
            ", ".join(
                sorted(
                    missing
                )
            )
        )

    return {
        "finding_id":
            str(
                record[
                    "id"
                ]
            ),

        "asset":
            str(
                record[
                    "asset"
                ]
            ).strip().upper(),

        "finding_type":
            str(
                record[
                    "type"
                ]
            ).strip().lower(),

        "severity":
            str(
                record[
                    "severity"
                ]
            ).strip().lower(),

        "status":
            str(
                record[
                    "status"
                ]
            ).strip().lower(),

        "score":
            float(
                record[
                    "score"
                ]
            ),

        "source":
            "training-api"
    }

71 — Validate Score Range

Improve:

score = float(
    record["score"]
)

if not (
    0.0 <= score <= 10.0
):
    raise ValueError(
        "Score outside expected range"
    )

72 — Why Normalize API Data?

Without normalization:

EACH API
=
DIFFERENT DATA MODEL

With normalization:

MULTIPLE APIS
     ↓
COMMON SCHEMA
     ↓
CENTRAL ANALYTICS

73 — Normalize All Findings

Create:

def normalize_findings(
    records
):
    valid = []
    invalid = []

    for record in records:
        try:
            valid.append(
                normalize_finding(
                    record
                )
            )

        except (
            ValueError,
            TypeError
        ) as error:

            invalid.append({
                "record":
                    record,

                "reason":
                    str(error)
            })

    return valid, invalid

74 — Count Findings by Severity

Import:

from collections import Counter

Create:

def count_by_severity(
    findings
):
    return Counter(
        item[
            "severity"
        ]
        for item
        in findings
    )

75 — Open Findings

Create:

def get_open_findings(
    findings
):
    return [
        item
        for item
        in findings
        if item[
            "status"
        ] == "open"
    ]

76 — Critical Open Findings

Create:

def get_critical_open(
    findings
):
    return [
        item
        for item
        in findings
        if (
            item[
                "status"
            ] == "open"
            and
            item[
                "severity"
            ] == "critical"
        )
    ]

77 — Export CSV

Create:

def export_findings_csv(
    findings
):
    output = (
        REPORT_DIR /
        "api-findings.csv"
    )

    fields = [
        "finding_id",
        "asset",
        "finding_type",
        "severity",
        "status",
        "score",
        "source"
    ]

    with output.open(
        "w",
        encoding="utf-8",
        newline=""
    ) as file:

        writer = csv.DictWriter(
            file,
            fieldnames=fields
        )

        writer.writeheader()

        writer.writerows(
            findings
        )

78 — Export Invalid Records

Create:

def export_invalid_records(
    records
):
    output = (
        REPORT_DIR /
        "invalid-api-records.json"
    )

    with output.open(
        "w",
        encoding="utf-8"
    ) as file:

        json.dump(
            records,
            file,
            indent=2
        )

79 — Export Raw API Snapshot

A useful defensive practice is to preserve the raw source data.

Create:

def export_raw_snapshot(
    records
):
    output = (
        REPORT_DIR /
        "raw-api-response.json"
    )

    with output.open(
        "w",
        encoding="utf-8"
    ) as file:

        json.dump(
            records,
            file,
            indent=2
        )

80 — Why Preserve Raw Data?

If normalization behaves incorrectly, you can compare:

RAW API DATA

with:

NORMALIZED OUTPUT

81 — Generate Summary JSON

Create:

def generate_summary(
    findings
):
    return {
        "total":
            len(findings),

        "open":
            len(
                get_open_findings(
                    findings
                )
            ),

        "critical_open":
            len(
                get_critical_open(
                    findings
                )
            ),

        "severity_counts":
            dict(
                count_by_severity(
                    findings
                )
            )
    }

82 — Save Summary

Create:

def export_summary(
    summary
):
    output = (
        REPORT_DIR /
        "api-summary.json"
    )

    with output.open(
        "w",
        encoding="utf-8"
    ) as file:

        json.dump(
            summary,
            file,
            indent=2
        )

83 — Generate Markdown Report

Create:

def generate_markdown_report(
    findings,
    summary
):
    lines = []

    lines.append(
        "# Security API Integration Report"
    )

    lines.append("")

    lines.append(
        "## Executive Summary"
    )

    lines.append("")

    lines.append(
        f"- Total findings: "
        f"{summary['total']}"
    )

    lines.append(
        f"- Open findings: "
        f"{summary['open']}"
    )

    lines.append(
        f"- Critical open findings: "
        f"{summary['critical_open']}"
    )

    lines.append("")

    lines.append(
        "## Critical Open Findings"
    )

    lines.append("")

    critical = (
        get_critical_open(
            findings
        )
    )

    if critical:
        for item in critical:
            lines.append(
                f"- {item['finding_id']} | "
                f"{item['asset']} | "
                f"{item['finding_type']} | "
                f"Score {item['score']}"
            )

    else:
        lines.append(
            "- No critical open findings."
        )

    lines.append("")

    lines.append(
        "## Integration Notes"
    )

    lines.append("")

    lines.append(
        "The client uses a controlled training API. "
        "Production integrations should use approved "
        "credentials, least-privilege permissions, "
        "timeouts, retry controls, rate-limit handling, "
        "logging, and documented data-retention policies."
    )

    output = (
        REPORT_DIR /
        "security-api-report.md"
    )

    output.write_text(
        "\n".join(
            lines
        ),
        encoding="utf-8"
    )

84 — Create the Complete Client Workflow

Create:

def main():
    validate_configuration()

    logging.info(
        "Starting security API collection"
    )

    raw_findings = (
        get_all_findings()
    )

    export_raw_snapshot(
        raw_findings
    )

    findings, invalid = (
        normalize_findings(
            raw_findings
        )
    )

    summary = generate_summary(
        findings
    )

    export_findings_csv(
        findings
    )

    export_invalid_records(
        invalid
    )

    export_summary(
        summary
    )

    generate_markdown_report(
        findings,
        summary
    )

    logging.info(
        "Security API collection complete"
    )

    print(
        f"Reports saved to: "
        f"{REPORT_DIR}"
    )

85 — Add Entry Point

Add:

if __name__ == "__main__":
    main()

86 — Run the Complete Lab

Terminal 1:

python api/training_api.py

Terminal 2:

Linux/macOS:

export SECURITY_API_TOKEN="training-token-123"

python src/security_api_client.py

PowerShell:

$env:SECURITY_API_TOKEN = "training-token-123"

python .\src\security_api_client.py

87 — Expected Output

You should have:

reports/
|
+-- raw-api-response.json
|
+-- api-findings.csv
|
+-- invalid-api-records.json
|
+-- api-summary.json
|
+-- security-api-report.md

88 — Review the CSV

Expected structure:

finding_id,asset,finding_type,severity,status,score,source
F-1001,WEB01,vulnerability,critical,open,9.8,training-api
F-1002,DB01,vulnerability,high,open,8.2,training-api

89 — API Logging

The client should record:

START

REQUEST FAILURES

RETRIES

NORMALIZATION ERRORS

COMPLETION

Do not log:

BEARER TOKEN

90 — Sensitive Header Warning

Never write:

logging.info(
    request.headers
)

when those headers contain:

AUTHORIZATION

API TOKEN

SESSION COOKIE

91 — Add Controlled POST Support

Now learn a second HTTP method using only your local API.

The goal:

CREATE A SYNTHETIC
SECURITY NOTE

not perform a production action.

92 — Add a Notes Store to the API

At the top of:

training_api.py

add:

NOTES = []

93 — Add POST Handling

Inside the handler:

    def do_POST(
        self
    ):
        if not self.is_authorized():
            self.send_json(
                {
                    "error":
                        "unauthorized"
                },
                401
            )
            return

        parsed = urlparse(
            self.path
        )

        if parsed.path != "/api/notes":
            self.send_json(
                {
                    "error":
                        "not_found"
                },
                404
            )
            return

        content_length = int(
            self.headers.get(
                "Content-Length",
                "0"
            )
        )

        body = self.rfile.read(
            content_length
        )

        try:
            data = json.loads(
                body.decode(
                    "utf-8"
                )
            )

        except json.JSONDecodeError:
            self.send_json(
                {
                    "error":
                        "invalid_json"
                },
                400
            )
            return

        note = str(
            data.get(
                "note",
                ""
            )
        ).strip()

        if not note:
            self.send_json(
                {
                    "error":
                        "note_required"
                },
                400
            )
            return

        record = {
            "id":
                len(NOTES) + 1,

            "note":
                note
        }

        NOTES.append(
            record
        )

        self.send_json(
            record,
            201
        )

Restart the server.

94 — Create a POST Client Function

In:

security_api_client.py

create:

def send_post_request(
    endpoint,
    payload,
    timeout=DEFAULT_TIMEOUT
):
    url = build_url(
        endpoint
    )

    body = json.dumps(
        payload
    ).encode(
        "utf-8"
    )

    request = Request(
        url,
        data=body,
        method="POST"
    )

    headers = build_headers()

    headers[
        "Content-Type"
    ] = "application/json"

    for name, value in (
        headers.items()
    ):
        request.add_header(
            name,
            value
        )

    with urlopen(
        request,
        timeout=timeout
    ) as response:

        return {
            "status":
                response.status,

            "body":
                response.read()
                .decode(
                    "utf-8"
                )
        }

95 — Create a Synthetic Note

Run:

result = send_post_request(
    "/api/notes",
    {
        "note":
            "Analyst reviewed F-1001 "
            "in the training lab."
    }
)

print(result)

Expected:

201

96 — Why POST Requires More Care

GET is generally used to:

READ DATA

POST may:

CREATE STATE

In production, writes require stronger controls.

97 — Production Write Workflow

Prefer:

DETECTION
   ↓
VALIDATE
   ↓
PREVIEW
   ↓
HUMAN APPROVAL
   ↓
API WRITE
   ↓
VERIFY
   ↓
AUDIT LOG

98 — Avoid Autonomous Destructive Actions

Do not build:

ALERT
 ↓
API
 ↓
DELETE ACCOUNT

or:

IOC MATCH
 ↓
API
 ↓
BLOCK EVERYTHING

without appropriate governance and controls.

99 — Least-Privilege API Tokens

Prefer separate permissions.

Example:

REPORTING TOKEN
    ↓
READ FINDINGS

instead of:

ADMIN TOKEN
    ↓
EVERYTHING

100 — API Scope Mental Model

AUTOMATION PURPOSE
      ↓
REQUIRED ENDPOINTS
      ↓
REQUIRED METHODS
      ↓
MINIMUM PERMISSIONS
      ↓
TOKEN SCOPE

101 — Token Rotation

Production tokens should support:

ROTATION

EXPIRATION

REVOCATION

Long-lived permanent credentials increase risk.

102 — Workload Identity

Cloud environments may avoid static API secrets entirely.

Conceptually:

AUTOMATION WORKLOAD
       ↓
PLATFORM IDENTITY
       ↓
SHORT-LIVED CREDENTIAL
       ↓
API

103 — TLS

Production APIs should generally use:

HTTPS

Your lab uses:

HTTP

only because the API runs locally on:

127.0.0.1

104 — Never Disable TLS Verification Casually

Do not normalize practices such as:

VERIFY TLS = FALSE

against production APIs.

Certificate verification protects against:

SERVER IMPERSONATION

MAN-IN-THE-MIDDLE RISK

105 — API Data Classification

Before integration, determine whether API data contains:

USER DATA

INTERNAL HOSTNAMES

IP ADDRESSES

VULNERABILITIES

SECURITY ALERTS

INCIDENT INFORMATION

CREDENTIAL METADATA

106 — Data Minimization

If the report requires:

finding_id

asset

severity

do not collect:

50 EXTRA FIELDS

without a reason.

107 — Logging Correlation ID

A mature API may return:

X-Request-ID

or similar.

Store such identifiers in logs so failures can be traced.

108 — Client Request ID

You can also generate your own internal ID.

Import:

import uuid

Create:

request_id = str(
    uuid.uuid4()
)

Then log:

logging.info(
    "request_id=%s starting request",
    request_id
)

109 — Observability

Useful API automation metrics include:

REQUEST COUNT

SUCCESS RATE

ERROR RATE

401 COUNT

429 COUNT

5XX COUNT

AVERAGE RESPONSE TIME

RETRY COUNT

RECORDS PROCESSED

110 — Measure Response Time

Import:

from time import perf_counter

Concept:

start = perf_counter()

result = api_get(
    "/api/findings"
)

duration = (
    perf_counter()
    -
    start
)

logging.info(
    "Request duration: %.3f seconds",
    duration
)

111 — Cache API Results

Repeated enrichment requests may benefit from:

CACHE

Architecture:

REQUEST DATA
     ↓
CACHE HIT?
  ┌──┴──┐
 YES    NO
 ↓       ↓
USE     API
CACHE    ↓
        SAVE

112 — Cache Security Consideration

Cached API data may itself be sensitive.

Protect it with:

ACCESS CONTROL

RETENTION

ENCRYPTION WHERE REQUIRED

113 — Do Not Cache Forever

Security data can become stale.

Use:

CACHE TIMESTAMP

TTL

EXPIRATION

114 — Idempotency

An important API concept:

SAME REQUEST
EXECUTED MULTIPLE TIMES

should not unexpectedly create repeated state where idempotency is required.

This is especially important for:

TICKET CREATION

FIREWALL CHANGES

ACCOUNT CHANGES

CASE CREATION

115 — Deduplication Before POST

Suppose automation creates incidents.

Avoid:

SAME ALERT
 ↓
5 RETRIES
 ↓
5 INCIDENTS

Use:

EVENT ID

IDEMPOTENCY KEY

EXISTING CASE CHECK

116 — API Versioning

Production APIs often use paths such as:

/api/v1/findings

or:

/api/v2/findings

Do not assume API structures never change.

117 — Schema Changes

A vendor may change:

risk

to:

severity

Your normalization layer helps isolate those changes.

118 — Adapter Pattern

Conceptually:

VENDOR A API
     ↓
ADAPTER A
     ↓
COMMON MODEL

VENDOR B API
     ↓
ADAPTER B
     ↓
COMMON MODEL

119 — Why Adapters Matter

Your reporting layer should not care whether data came from:

SIEM A

SIEM B

CLOUD SECURITY TOOL

VULNERABILITY TOOL

It should receive:

NORMALIZED SECURITY DATA

120 — Build a Security API Pipeline

Final mental architecture:

SOURCE API
    ↓
AUTHENTICATE
    ↓
REQUEST
    ↓
STATUS VALIDATION
    ↓
JSON VALIDATION
    ↓
SCHEMA VALIDATION
    ↓
NORMALIZATION
    ↓
DEDUPLICATION
    ↓
ANALYSIS
    ↓
REPORT

121 — Test Case: Missing Token

Remove:

SECURITY_API_TOKEN

Run the client.

Expected:

CONFIGURATION ERROR

before API requests are sent.

122 — Test Case: Wrong Token

Set:

wrong-token

Expected:

401 Unauthorized

123 — Test Case: Server Offline

Stop the server.

Expected:

NETWORK ERROR

RETRIES

FINAL FAILURE

124 — Test Case: Invalid Endpoint

Request:

/api/unknown

Expected:

404

and:

NO RETRY

125 — Test Case: Invalid JSON

Temporarily modify the training API to return invalid text.

Confirm:

JSON VALIDATION FAILS

Then restore the valid server.

126 — Test Case: Missing Field

Remove:

severity

from one synthetic API finding.

Confirm that normalization sends the record to:

INVALID RECORDS

127 — Test Case: Bad Score

Set:

score = "invalid"

Confirm the client does not crash.

128 — Test Case: Pagination

Set:

page_size = 1

Confirm:

ALL 5 RECORDS

are still collected.

129 — Test Case: Duplicate API Records

Add a duplicate finding to the server.

Then extend your client to deduplicate using:

finding_id

130 — Add Deduplication

Create:

def deduplicate_findings(
    findings
):
    unique = []
    duplicates = []

    seen = set()

    for finding in findings:
        key = finding[
            "finding_id"
        ]

        if key in seen:
            duplicates.append(
                finding
            )
            continue

        seen.add(
            key
        )

        unique.append(
            finding
        )

    return unique, duplicates

131 — Why Deduplicate API Data?

Duplicate results can occur because of:

PAGINATION CHANGES

API BUGS

REPEATED IMPORT

EVENTUAL CONSISTENCY

RETRY LOGIC

132 — Create a Test Suite

Create:

tests/test_security_api_client.py

133 — Test URL Building

def test_build_url():
    url = build_url(
        "/api/findings",
        {
            "severity":
                "critical"
        }
    )

    assert (
        "severity=critical"
        in url
    )

134 — Test Normalization

def test_normalize_finding():
    record = {
        "id": "F-1",
        "asset": "web01",
        "type": "Vulnerability",
        "severity": "CRITICAL",
        "status": "OPEN",
        "score": 9.8
    }

    normalized = (
        normalize_finding(
            record
        )
    )

    assert (
        normalized[
            "asset"
        ]
        == "WEB01"
    )

    assert (
        normalized[
            "severity"
        ]
        == "critical"
    )

135 — Test Missing Fields

def test_missing_field():
    record = {
        "id": "F-1"
    }

    try:
        normalize_finding(
            record
        )

    except ValueError:
        assert True

    else:
        assert False

136 — Test Score Range

A score:

15.0

should be rejected when the expected range is:

0–10

137 — Test Deduplication

Provide:

F-1001

F-1001

Expected:

UNIQUE = 1

DUPLICATE = 1

138 — Challenge 01 — Add API Health Check

Before collecting:

FINDINGS

call:

/api/health

If unavailable:

STOP CLEANLY

139 — Challenge 02 — Add Config File

Instead of hard-coding:

BASE_URL

use:

data/config.json

with no secrets inside it.

140 — Challenge 03 — Add CLI Parameters

Use:

argparse

to support:

--base-url

--status

--severity

--output

141 — Challenge 04 — Add Structured Logs

Instead of human-only logs:

2026... ERROR ...

produce optional:

{
  "level": "error",
  "event": "api_request_failed",
  "status": 503
}

142 — Challenge 05 — Add Local Cache

Cache:

/api/findings

for:

60 seconds

Then compare:

API REQUEST COUNT

before and after caching.

143 — Challenge 06 — Add Retry-After Support

When:

429

is returned, inspect:

Retry-After

and wait accordingly.

144 — Challenge 07 — Add Metrics

Produce:

api-metrics.json

containing:

requests

successes

failures

retries

records_processed

duration

145 — Challenge 08 — Add SQLite Storage

Architecture:

API
 ↓
PYTHON
 ↓
NORMALIZE
 ↓
SQLITE
 ↓
SQL ANALYTICS

This connects:

Lab 05
+
Lab 07

146 — Challenge 09 — Combine Vulnerability Prioritization

Feed API results into the prioritization logic from:

Lab 06

Architecture:

SECURITY API
      ↓
FINDINGS
      ↓
NORMALIZE
      ↓
ASSET CONTEXT
      ↓
RISK PRIORITY
      ↓
REMEDIATION QUEUE

147 — Challenge 10 — Add Human Approval

For any simulated write operation:

PREVIEW ACTION
      ↓
ASK FOR APPROVAL
      ↓
POST
      ↓
VERIFY RESPONSE

148 — API Security Checklist

Before integrating a real security API, review:

• API endpoint approved
• HTTPS used
• TLS verification enabled
• Authentication method understood
• Least-privilege token used
• Secrets not hard-coded
• Token rotation supported
• Timeouts configured
• Errors handled
• Retries limited
• Backoff implemented
• 429 handling understood
• Pagination implemented
• JSON validated
• Schema validated
• Data normalized
• Duplicate handling implemented
• Sensitive fields minimized
• Logs exclude secrets
• API writes governed
• Response verified
• Data retention understood

149 — Common Security API Integration Mistakes

Avoid:

HARDCODED API KEYS

NO TIMEOUT

INFINITE RETRIES

RETRYING 401 ERRORS

IGNORING 429

NO PAGINATION

TRUSTING JSON BLINDLY

NO SCHEMA VALIDATION

NO NORMALIZATION

LOGGING AUTHORIZATION HEADERS

COLLECTING TOO MUCH DATA

USING ADMIN TOKENS FOR READ-ONLY JOBS

AUTOMATICALLY EXECUTING HIGH-IMPACT ACTIONS

NO AUDIT LOG

NO RESPONSE VERIFICATION

150 — Production Security Integration Principle

Do not think:

API
=
JUST ANOTHER DATA SOURCE

An API can be:

A PRIVILEGED CONTROL PLANE

It may be able to:

DISABLE ACCOUNTS

ISOLATE ENDPOINTS

CHANGE FIREWALLS

DELETE RESOURCES

CREATE INCIDENTS

MODIFY SECURITY POLICIES

That means API permissions require serious governance.

151 — Read Before Write

A strong security automation principle is:

READ
 ↓
UNDERSTAND
 ↓
VALIDATE
 ↓
PREVIEW
 ↓
APPROVE
 ↓
WRITE
 ↓
VERIFY

152 — API Failure Mental Model

When an API request fails:

WHAT FAILED?
     ↓
NETWORK?
     ↓
AUTH?
     ↓
PERMISSION?
     ↓
RATE LIMIT?
     ↓
SERVER?
     ↓
BAD REQUEST?
     ↓
BAD DATA?

Do not treat every failure identically.

153 — Security Automation Mental Model

REQUEST
   ↓
TIMEOUT
   ↓
STATUS
   ↓
HEADERS
   ↓
BODY
   ↓
JSON
   ↓
SCHEMA
   ↓
NORMALIZE
   ↓
ANALYZE

154 — Build Your Final Portfolio Structure

Your completed project should look like:

security-api-integration/
|
+-- api/
|   +-- training_api.py
|
+-- data/
|
+-- src/
|   +-- security_api_client.py
|
+-- tests/
|   +-- test_security_api_client.py
|
+-- reports/
|   +-- raw-api-response.json
|   +-- api-findings.csv
|   +-- invalid-api-records.json
|   +-- api-summary.json
|   +-- security-api-report.md
|
+-- README.md
|
+-- architecture.md

155 — README Structure

Include:

PROJECT OVERVIEW

SECURITY USE CASE

ARCHITECTURE

TRAINING API

AUTHENTICATION

ENVIRONMENT VARIABLES

HTTP METHODS

ERROR HANDLING

RETRY STRATEGY

RATE LIMITING

PAGINATION

NORMALIZATION

REPORTS

TESTING

SECURITY CONSIDERATIONS

LIMITATIONS

156 — Document Limitations

Your local training API does not provide:

REAL TLS

OAUTH 2.0

SHORT-LIVED TOKENS

ENTERPRISE RATE LIMITS

DISTRIBUTED PAGINATION

WEBHOOKS

PRODUCTION SECURITY DATA

HIGH AVAILABILITY

REAL RBAC

These are future learning areas.

157 — Mission Validation Checklist

Confirm:

• Lab workspace created
• Local training API created
• Synthetic findings created
• Bearer authentication implemented
• Health endpoint tested
• Unauthorized request tested
• GET request completed
• Query parameters used
• HTTP status codes reviewed
• JSON parsed
• JSON validation implemented
• Client token moved to environment variable
• Timeout configured
• HTTP errors handled
• Network errors handled
• Retry logic implemented
• Backoff implemented
• Rate limiting understood
• Pagination implemented
• All paginated records collected
• API findings normalized
• Missing fields rejected
• Invalid scores rejected
• Duplicate logic implemented
• Raw API snapshot exported
• Normalized CSV exported
• Invalid records exported
• Summary JSON exported
• Markdown report generated
• Controlled local POST tested
• No real credentials used
• Authorization headers excluded from logs
• No destructive API automation created
• README created
• Limitations documented

Mission Review

You started with:

SECURITY DATA
INSIDE AN API

You built:

PYTHON CLIENT
      ↓
AUTHENTICATION
      ↓
GET REQUEST
      ↓
TIMEOUT
      ↓
STATUS VALIDATION
      ↓
JSON PARSING
      ↓
SCHEMA VALIDATION
      ↓
NORMALIZATION
      ↓
PAGINATION
      ↓
RETRY / BACKOFF
      ↓
SECURITY ANALYSIS
      ↓
REPORT

What You Built

You now have a reusable security API integration pattern capable of handling:

AUTHENTICATION

QUERY PARAMETERS

JSON RESPONSES

STATUS CODES

TIMEOUTS

ERRORS

RETRIES

RATE LIMITS

PAGINATION

NORMALIZATION

CSV / JSON REPORTING

Key Security Lesson

The central lesson is:

A SUCCESSFUL HTTP REQUEST
DOES NOT AUTOMATICALLY MEAN
A SUCCESSFUL SECURITY WORKFLOW

You must validate:

STATUS

CONTENT

SCHEMA

DATA QUALITY

AUTHORIZATION

CONTEXT

Final Mental Model

Whenever you integrate with a security API, think:

WHAT API AM I CALLING?
        ↓
WHAT PERMISSIONS DO I NEED?
        ↓
HOW WILL I AUTHENTICATE?
        ↓
IS THE CONNECTION TRUSTED?
        ↓
WHAT IS MY TIMEOUT?
        ↓
WHAT IF IT FAILS?
        ↓
SHOULD I RETRY?
        ↓
IS THERE A RATE LIMIT?
        ↓
IS THERE PAGINATION?
        ↓
IS THE JSON VALID?
        ↓
IS THE SCHEMA VALID?
        ↓
HOW WILL I NORMALIZE IT?
        ↓
WHAT SECURITY DECISION
WILL USE THIS DATA?

The goal is not:

CALL MORE APIS

The goal is:

BUILD RELIABLE,
CONTROLLED,
AUDITABLE
SECURITY INTEGRATIONS

What’s Next?

➡️ Lab 08 — Cloud Security Configuration Auditor

The next lab takes your Python automation and API skills into cloud security.

You will build a read-only cloud security assessment workflow for an authorized lab account.

The architecture will be:

CLOUD ACCOUNT
      ↓
CLOUD API
      ↓
SECURITY AUDITOR
      ↓
IDENTITY REVIEW
      ↓
LOGGING REVIEW
      ↓
ENCRYPTION REVIEW
      ↓
PUBLIC EXPOSURE REVIEW
      ↓
SECURITY CONTROL REVIEW
      ↓
NORMALIZED FINDINGS
      ↓
SECURITY REPORT

You will learn how to turn cloud configuration data into repeatable security findings without making disruptive changes to cloud resources.