Configuring Splunk Integration

AttackForge configuration

1. Create a dedicated service user for Splunk.

1. Grant API access explicitly: Users > (select user) > Access > Self-Service RESTful API > Add Access. Grant GetApplicationAuditLogs.

1. Generate that user's API Key under user settings.

1. The key goes in the X-SSAPI-KEY header, and all calls must be HTTPS.

Splunk configuration

1. Create the App Directory Structure

Navigate to your Heavy Forwarder’s app directory and build this structure:

$SPLUNK_HOME/etc/apps/attackforge_app/
├── bin/
│   └── af_auditlogs.py
└── default/
    └── inputs.conf

1. Write the Script (bin/af_auditlogs.py)

#!/usr/bin/env python3
import json, os, sys
import requests

BASE       = os.environ["AF_BASE_URL"]          # https://<tenant>.attackforge.com
API_KEY    = os.environ["AF_SSAPI_KEY"]
LIMIT      = 5000                                # max supported
CHECKPOINT = os.path.join(
    os.environ.get("SPLUNK_DB", "/opt/splunk/var/lib/splunk"),
    "modinputs", "af_auditlogs", "offset.json"
)

def read_offset():
    try:
        with open(CHECKPOINT) as fh:
            return int(json.load(fh)["skip"])
    except (OSError, ValueError, KeyError):
        return 0                                 # first run: start from the beginning

def write_offset(skip):
    os.makedirs(os.path.dirname(CHECKPOINT), exist_ok=True)
    tmp = CHECKPOINT + ".tmp"
    with open(tmp, "w") as fh:
        json.dump({"skip": skip}, fh)
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp, CHECKPOINT)                  # atomic; never leaves a half-written file

def main():
    skip    = read_offset()
    session = requests.Session()
    session.headers.update({
        "X-SSAPI-KEY": API_KEY,
        "Content-Type": "application/json",
        "Connection": "close",
    })

    while True:
        resp = session.get(
            f"{BASE}/api/ss/auditlogs",
            params={
                "skip": skip,
                "limit": LIMIT,
                "include_request_body": "true",
            },
            timeout=120,
        )
        resp.raise_for_status()

        logs = resp.json().get("logs", [])
        if not logs:
            break

        for entry in logs:
            sys.stdout.write(json.dumps(entry) + "\n")
        sys.stdout.flush()                       # get events out before advancing the offset

        skip += len(logs)
        write_offset(skip)

        if len(logs) < LIMIT:
            break                                # caught up; wait for the next interval

if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        # stderr from a scripted input lands in splunkd.log — offset stays put, so the
        # next run re-requests the same page rather than skipping it
        sys.stderr.write(f"ERROR af_auditlogs: {exc}\n")
        sys.exit(1)

1. Configure the Input (default/inputs.conf)

[script://$SPLUNK_HOME/etc/apps/attackforge_app/bin/af_auditlogs.py]
disabled = 1
interval = 300
index = security
sourcetype = attackforge:auditlog
source = attackforge:ssapi
python.version = python3