Export audit events
Aiden keeps a cursor-paginated audit trail (tool calls, policy decisions, break-glass on/off, and more). The first page alone is not enough when you need a full window for compliance or a SIEM. The SDK pages for you and writes one JSON object per line.
This sample exports the last N days to a local NDJSON file, with optional filters for event type and OCSF shape.
When you need this
- Hand a window to an external auditor. Write native or OCSF events for a date range to a file you can attach to a ticket.
- Feed a SIEM. Pass
format: "ocsf"so each line is OCSF API Activity (class_uid 6003). - Spot-check break-glass. Filter
event_type=break_glass_activated(orbreak_glass_deactivated) after an outage.
How to run
export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
pip install stackgen-sdk==0.4.0
python examples/automation/audit-export.py --days 7 --output audit.ndjsonexport STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
npm install stackgen-sdk@0.4.0 tsx
npx tsx examples/automation/audit-export.ts --days 7 --output audit.ndjsonexport STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
go run ./examples/go/automation/audit-export.go --days 7 --output audit.ndjsonOptional flags: --event-type, --format ocsf, --max-events (dry run).
Minimal
from datetime import datetime, timedelta, timezone
from stackgen import StackgenClient, StackgenConfig
client = StackgenClient(
StackgenConfig(
base_url="https://app.stackgen.com",
api_token="stackgen_…",
workspace_id="<workspace-uuid>",
)
)
now = datetime.now(timezone.utc)
start = now - timedelta(days=7)
result = client.aiden.export_audit_events(
"audit.ndjson",
**{
"from": start.strftime("%Y-%m-%dT%H:%M:%SZ"),
"to": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
"format": "ocsf", # or omit / "native"
},
)
print(result.count, result.output_path)
# Stream without a file:
for event in client.aiden.iter_audit_events(event_type="break_glass_activated"):
print(event["id"], event.get("actor_login"))import { StackgenClient, StackgenConfig } from "stackgen-sdk";
const client = new StackgenClient(
new StackgenConfig({
baseUrl: "https://app.stackgen.com",
apiToken: "stackgen_…",
workspaceId: "<workspace-uuid>",
}),
);
const now = new Date();
const start = new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000);
const result = await client.aiden.exportAuditEvents("audit.ndjson", {
from: start.toISOString(),
to: now.toISOString(),
format: "ocsf", // or omit / "native"
});
console.log(result.count, result.outputPath);
// Stream without a file:
for await (const event of client.aiden.iterAuditEvents({
event_type: "break_glass_activated",
})) {
console.log(event.id, event.actor_login);
}now := time.Now().UTC()
start := now.Add(-7 * 24 * time.Hour)
result, err := client.Aiden.ExportAuditEvents(ctx, "audit.ndjson", aiden.AuditQuery{
From: start.Format(time.RFC3339),
To: now.Format(time.RFC3339),
Format: "ocsf", // or omit / "native"
})
if err != nil { return err }
fmt.Println(result.Count, result.OutputPath)
events, err := client.Aiden.IterAuditEvents(ctx, aiden.AuditQuery{
EventType: "break_glass_activated",
})
if err != nil { return err }
for _, event := range events {
fmt.Println(event["id"], event["actor_login"])
}
return nilNotes
list_audit_events/listAuditEventsstill return one page only. Useiter_*orexport_*when you need the full window.- Native list responses omit large payload fields unless you pass
include_payload=true.get_audit_event(id)always returns the full payload. - OCSF export (
format=ocsf) expects afrom/towindow on the server; the sample always sets both. - Pass
max_events/maxEventsto stop early while you dry-run a filter.
Complete scripts
- Python:
examples/automation/audit-export.py - TypeScript:
examples/automation/audit-export.ts - Go:
examples/go/automation/audit-export.go
#!/usr/bin/env python3
"""Export Aiden audit events to a local NDJSON file.
Pages past the server's per-page cap automatically. Use ``--format ocsf`` when
handing the file to a SIEM or external auditor.
Usage:
export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
python examples/automation/audit-export.py --days 7 --output audit.ndjson
python examples/automation/audit-export.py --days 30 --event-type break_glass_activated
python examples/automation/audit-export.py --days 7 --format ocsf --output audit-ocsf.ndjson
"""
from __future__ import annotations
import argparse
import os
from datetime import datetime, timedelta, timezone
from stackgen import StackgenClient, StackgenConfig
def require_env(name: str, *fallbacks: str) -> str:
for key in (name, *fallbacks):
value = (os.environ.get(key) or "").strip()
if value:
return value
raise SystemExit(f"missing env {name}")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--days",
type=int,
default=7,
help="how many days back to export (default: 7)",
)
parser.add_argument(
"--output",
default="audit.ndjson",
help="local NDJSON path (default: audit.ndjson)",
)
parser.add_argument(
"--event-type",
default=None,
help="optional event_type filter (e.g. break_glass_activated)",
)
parser.add_argument(
"--format",
choices=("native", "ocsf"),
default="native",
help="native AuditEvent objects, or OCSF API Activity for SIEM hand-off",
)
parser.add_argument(
"--max-events",
type=int,
default=None,
help="optional cap for a dry run",
)
args = parser.parse_args()
if args.days < 1:
raise SystemExit("--days must be >= 1")
client = StackgenClient(
StackgenConfig(
base_url=require_env("STACKGEN_URL"),
api_token=require_env("STACKGEN_TOKEN"),
workspace_id=require_env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
)
)
now = datetime.now(timezone.utc)
start = now - timedelta(days=args.days)
params: dict[str, object] = {
"from": start.strftime("%Y-%m-%dT%H:%M:%SZ"),
"to": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
"format": args.format,
}
if args.event_type:
params["event_type"] = args.event_type
if args.max_events is not None:
params["max_events"] = args.max_events
result = client.aiden.export_audit_events(args.output, **params)
print(f"wrote {result.count} events to {result.output_path}")
if __name__ == "__main__":
main()/**
* Export Aiden audit events to a local NDJSON file.
*
* Pages past the server's per-page cap automatically. Use `--format ocsf` when
* handing the file to a SIEM or external auditor.
*
* Usage:
* export STACKGEN_URL=https://app.stackgen.com
* export STACKGEN_TOKEN=stackgen_…
* export STACKGEN_WORKSPACE_ID=<workspace-uuid>
* npx tsx examples/automation/audit-export.ts --days 7 --output audit.ndjson
* npx tsx examples/automation/audit-export.ts --days 30 --event-type break_glass_activated
* npx tsx examples/automation/audit-export.ts --days 7 --format ocsf --output audit-ocsf.ndjson
*/
import { parseArgs } from "node:util";
import { StackgenClient, StackgenConfig } from "stackgen-sdk";
function requireEnv(name: string, ...fallbacks: string[]): string {
for (const key of [name, ...fallbacks]) {
const value = (process.env[key] || "").trim();
if (value) {
return value;
}
}
throw new Error(`missing env ${name}`);
}
async function main(): Promise<void> {
const { values } = parseArgs({
options: {
days: { type: "string", default: "7" },
output: { type: "string", default: "audit.ndjson" },
"event-type": { type: "string" },
format: { type: "string", default: "native" },
"max-events": { type: "string" },
},
});
const days = Number(values.days);
if (!Number.isFinite(days) || days < 1) {
throw new Error("--days must be >= 1");
}
const format = values.format ?? "native";
if (format !== "native" && format !== "ocsf") {
throw new Error('--format must be "native" or "ocsf"');
}
const client = new StackgenClient(
new StackgenConfig({
baseUrl: requireEnv("STACKGEN_URL"),
apiToken: requireEnv("STACKGEN_TOKEN"),
workspaceId: requireEnv("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
}),
);
const now = new Date();
const start = new Date(now.getTime() - days * 24 * 60 * 60 * 1000);
const params: Record<string, unknown> = {
from: start.toISOString(),
to: now.toISOString(),
format,
};
if (values["event-type"]) {
params.event_type = values["event-type"];
}
if (values["max-events"]) {
params.maxEvents = Number(values["max-events"]);
}
const result = await client.aiden.exportAuditEvents(values.output ?? "audit.ndjson", params);
console.log(`wrote ${result.count} events to ${result.outputPath}`);
}
main().catch((err) => {
console.error(err);
process.exit(1);
});// Export Aiden audit events to a local NDJSON file.
//
// Documented at https://appcd-dev.github.io/stackgen-sdk/code-samples/audit-export/
//
// go run ./examples/go/automation/audit-export.go --days 7 --output audit.ndjson
package main
import (
"context"
"flag"
"fmt"
"os"
"strings"
"time"
stackgen "github.com/appcd-dev/stackgen-sdk/clients/go"
"github.com/appcd-dev/stackgen-sdk/clients/go/aiden"
)
// envOr returns the first non-empty environment variable among keys.
func envOr(keys ...string) string {
for _, key := range keys {
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v
}
}
return ""
}
func main() {
days := flag.Int("days", 7, "how many days back to export")
output := flag.String("output", "audit.ndjson", "local NDJSON path")
eventType := flag.String("event-type", "", "optional event_type filter")
format := flag.String("format", "native", "native or ocsf")
maxEvents := flag.Int("max-events", 0, "optional cap for a dry run (0 means no cap)")
flag.Parse()
if *days < 1 {
fmt.Fprintln(os.Stderr, "--days must be >= 1")
os.Exit(2)
}
if *format != "native" && *format != "ocsf" {
fmt.Fprintln(os.Stderr, "--format must be native or ocsf")
os.Exit(2)
}
base, token, org := envOr("STACKGEN_URL"), envOr("STACKGEN_TOKEN"), envOr("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT")
if base == "" || token == "" || org == "" {
fmt.Fprintln(os.Stderr, "set STACKGEN_URL, STACKGEN_TOKEN, and STACKGEN_WORKSPACE_ID")
os.Exit(2)
}
client, err := stackgen.NewClient(stackgen.Config{BaseURL: base, APIToken: token, WorkspaceID: org})
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
now := time.Now().UTC()
start := now.AddDate(0, 0, -*days)
result, err := client.Aiden.ExportAuditEvents(context.Background(), *output, aiden.AuditQuery{
EventType: *eventType,
From: start.Format("2006-01-02T15:04:05Z"),
To: now.Format("2006-01-02T15:04:05Z"),
Format: *format,
MaxEvents: *maxEvents,
})
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
fmt.Printf("wrote %d events to %s\n", result.Count, result.OutputPath)
}