AI & Security Telemetry Enrichment Pipeline by Mangesh BhattacharyaAI & Security Telemetry Enrichment Pipeline by Mangesh Bhattacharya

AI & Security Telemetry Enrichment Pipeline

Mangesh Bhattacharya

Completed work

AI Engineer

Security Engineer

GitHub

Python

Cybersecurity

AI & Security Telemetry Enrichment Pipeline

A production-shaped Python pipeline that ingests raw telemetry, LLM prompt/response traces and network payload logs, redacts PII and secrets before anything leaves the process, enriches the rest with threat intelligence and prompt-injection risk scoring, and rolls it all up into an automated risk dashboard.

On top of that sits a small multi-agent workflow: pipeline agents (ingest, enrich, analyze) hand off to response agents (triage, draft an incident write-up, file a GitHub Issue) for anything that crosses a configurable severity threshold. The same workflow runs three ways, as a CLI, as a GitHub Actions job, or behind a hosted API a client system's own telemetry pipeline can POST to.

Why This Exists

Security and AI teams both generate high-volume, sensitive telemetry: SOC network logs on one side, LLM application traces on the other. Both need the same discipline before they touch a third-party API, validate the shape, strip what shouldn't leave the boundary, then enrich and analyze. This is a reference implementation of that discipline, built to be read end-to-end in one sitting: schema validation catches malformed input, a zero-trust redaction engine runs before dispatch (not after), enrichment degrades gracefully to mock data offline, and everything is covered by tests that mock every external call, so pytest never needs a live API key.

It's a portfolio project: a runnable demo of the pattern, not a service wired to production infrastructure.

Architecture

The core pipeline, validate, redact, enrich, analyze:

The agent workflow wraps that pipeline and hands escalated findings to a response layer, see Agentic Workflow below:

Threat Model

Asset, Threat, Mitigation

Raw telemetry (prompts, payloads): PII/secrets leak to a third-party enrichment API, Redaction runs before any httpx dispatch (see src/security/redaction.py); regex + optional Presidio

Threat intel / model API keys: Credential exposure in logs or source, Loaded only via python-dotenv/env vars, never hardcoded; .env is git-ignored

Enrichment API outage or rate limit: Pipeline stalls or crashes mid-batch, Exponential backoff (src/enrichment/clients.py) with a deterministic mock fallback so the batch always completes

Malformed/malicious input records: Bad data corrupts downstream analytics, Pydantic v2 schemas with extra="forbid" reject unknown shapes per-row without failing the whole batch

LLM prompts containing injection/jailbreak attempts: Downstream model manipulation, Heuristic + pluggable HuggingFace classifier flags risk before the prompt is used elsewhere (src/enrichment/prompt_classifier.py)

CI supply chain: Vulnerable dependencies or unsafe code merged, bandit SAST + ruff lint gate every PR (.github/workflows/ci.yml)

GitHub token used by the issue-filing agent: An overprivileged or leaked token could spam/modify issues beyond intent, Actions runs use the default GITHUB_TOKEN scoped to issues: write only (see agent-workflow.yml); local/service runs dry-run unless GITHUB_TOKEN+GITHUB_REPO are explicitly set

Unauthenticated hosted service endpoint: Anyone reaching the API could submit arbitrary batches or trigger issue spam, Optional bearer-token auth via SERVICE_API_KEY (src/service/app.py); a real deployment must set it or front the service with its own gateway

Repeated findings for the same record: Duplicate GitHub Issues drown out real signal (alert fatigue), GitHubIssueClient checks for an existing open issue by a finding:{record_id} label before creating a new one

An agent failing mid-workflow: Downstream agents act on incomplete or stale state, WorkflowOrchestrator stops the chain on the first agent error and records it in step_log instead of continuing blind

Repository Structure

src/
  ingestion/    # Pydantic schemas + JSON/CSV batch loader
  security/     # Redaction engine + secret/PII regex patterns
  enrichment/   # Async threat intel client + prompt-injection classifier
  analytics/    # pandas aggregation + Plotly report generation
  pipeline.py   # Straight-line CLI: ingest -> redact/enrich -> analyze
  agents/       # Agent workflow: pipeline agents + triage/incident-response/GitHub agents
    context.py            # WorkflowContext shared between agents
    pipeline_agents.py     # Ingestion / Enrichment / Analytics agents (wrap src/ingestion, enrichment, analytics)
    triage.py              # Severity banding + escalation decision
    incident_response.py   # Drafts a title/body per escalated finding
    github_client.py       # Async GitHub Issues client, dedupe + dry-run fallback
    github_agent.py         # Files (or dry-runs) an issue per draft
    orchestrator.py         # Supervisor chaining all six agents; CLI entry point
  service/      # Hosted FastAPI service exposing the same workflow over HTTP
tests/          # pytest suite, >85% coverage, all HTTP calls mocked
data/samples/   # Example telemetry batch for local runs
.github/workflows/ci.yml             # ruff + bandit + pytest-cov on every push/PR
.github/workflows/agent-workflow.yml # Runs the agent workflow as a GitHub Action, files real Issues
project_board_schema.json  # Agile epics/stories for a GitHub Project board
scripts/create_project_board.sh  # gh CLI script to populate that board

Setup

git clone https://github.com/Mangesh-Bhattacharya/ai-security-telemetry-pipeline.git
cd ai-security-telemetry-pipeline
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements-dev.txt
cp .env.example .env   # optional: fill in real API keys, or leave blank for mock mode

Run the pipeline against the bundled sample data:

python -m src.pipeline data/samples/sample_logs.json --report reports/dashboard.html

This prints an ingestion summary and a JSON risk report to stdout, then writes a standalone interactive dashboard to reports/dashboard.html, open it in any browser, no server required.

Running tests, lint, and SAST

pytest --cov=src --cov-report=term-missing
ruff check src tests
bandit -r src

Sample Analytics Output

generate_html_report renders a four-panel dashboard: risk score distribution, risk over time, prompt-injection risk labels, and an anomaly/normal split. Run the pipeline locally and open reports/dashboard.html to see it against the bundled sample data, it includes one clean prompt, one prompt-injection attempt, one leaked AWS key, and one leaked Anthropic key, so all four panels have signal.

Agentic Workflow

src/agents/orchestrator.py chains six single-purpose agents behind one shared WorkflowContext. Each pipeline agent is a thin adapter over the modules described above (src/ingestion, src/enrichment, src/analytics) rather than a reimplementation, so the straight-line CLI and the agent workflow stay in sync by construction:

Ingestion agent, loads/validates a batch (file or inline records)

Enrichment agent, redacts, then enriches with threat intel + prompt risk

Analytics agent, aggregates risk, renders the dashboard

Triage agent, bands each record into low/medium/high/critical and decides what escalates (configurable via --escalate-from)

Incident response agent, drafts a title + remediation write-up per escalated finding (template-based, no LLM call in the hot path, see the docstring in src/agents/incident_response.py for how to swap in a real model call)

GitHub issue agent, files (or dry-runs) one deduped Issue per draft

Run it locally:

python -m src.agents data/samples/sample_logs.json --report reports/agent_dashboard.html --escalate-from high

Without GITHUB_TOKEN/GITHUB_REPO set, issue filing dry-runs, you'll see Issue[dry_run] ... lines instead of real API calls, so this is safe to run anywhere, including CI (.github/workflows/ci.yml does exactly this).

Running as a GitHub Action

.github/workflows/agent-workflow.yml runs the same workflow as a scheduled (or manually dispatched) Action against a repo-hosted telemetry batch, using the default GITHUB_TOKEN (scoped to issues: write via the permissions: block) to file real Issues in the repo, no PAT setup needed to see it work.

Hosted Service

src/service/app.py is a FastAPI service exposing the same orchestrator over HTTP, the piece a client system's own telemetry pipeline would POST batches to. Run it locally:

uvicorn src.service.app:app --reload

curl -X POST http://localhost:8000/telemetry/batch \
  -H "Content-Type: application/json" \
  -d @data/samples/sample_logs.json

(data/samples/sample_logs.json is {"records": [...]}, matching the request body shape directly.) The response includes per-record findings, any GitHub issues filed, and a per-agent step log. GET /healthz is unauthenticated for use as a liveness probe; POST /telemetry/batch accepts a bearer token via SERVICE_API_KEY when that's set, leaving it unset is dev-mode-only and documented as such in .env.example. The client-facing app/UI on top of this service is a later phase; this is the API it will call.

Extending This

Real threat intel: set ABUSEIPDB_API_KEY in .env, the client switches from deterministic mock scores to live lookups with no code change.

Real prompt-injection model: set HF_PROMPT_CLASSIFIER_MODEL and swap in HFPromptInjectionClassifier from src/enrichment/prompt_classifier.py, which wraps any HuggingFace text-classification pipeline behind the same interface as the built-in heuristic classifier.

A second threat intel source: add a VirusTotalClient alongside ThreatIntelClient in src/enrichment/clients.py and blend both scores in EnrichedRecord.risk_score.

Real incident-response drafting: replace the template in _draft_body (src/agents/incident_response.py) with a call to the Claude API for richer, context-aware remediation guidance, the agent's interface doesn't change.

A different escalation sink: implement a new BaseAgent alongside GitHubIssueAgent (e.g. a PagerDuty or Slack agent) and add it to default_agents() in src/agents/orchestrator.py.

The client-facing app: src/service/app.py is the API layer; a UI on top of it (dashboard, review queue for escalated findings) is the natural next phase and wasn't in scope here.

GitHub Project Board

project_board_schema.json defines the epics and user stories shown in this project's Backlog/In Progress/Done columns. scripts/create_project_board.sh uses the GitHub CLI's gh project commands to recreate that board from scratch:

OWNER=Mangesh-Bhattacharya REPO=ai-security-telemetry-pipeline ./scripts/create_project_board.sh

License

MIT, see LICENSE .

About

Automated AI & security telemetry enrichment pipeline: PII/secret redaction, async threat intel + prompt-injection scoring, and a multi-agent workflow that files GitHub issues for escalated findings.

Like this project

Completed work

Posted Oct 28, 2026

Python pipeline that redacts PII and secrets from AI and security telemetry, adds threat intel and prompt-injection scoring, and files issues for escalations.

Likes

Views

Timeline

Sep 14, 2026 - Sep 14, 2026