

Developer‑focused implementation guide for monitoring, alerting, and responding to the breach using public APIs and best‑practice patterns.
<a name="step-1-prerequisites"></a>
| Item | Why you need it | Recommended version |
|---|---|---|
| Python | Core language for the breach‑monitor script | ≥ 3.9 |
| Node.js | For the JS/TS implementation (optional) | ≥ 18.x (LTS) |
| Git | Clone the example repo / manage code | any |
| API key for a breach‑lookup service (e.g., HaveIBeenPwned v3) | Needed to query whether an email/password appears in known breaches | Sign up at https://haveibeenpwned.com/API/v3 |
(Optional) Secrets manager – AWS Secrets Manager, HashiCorp Vault, or simple .env file | Store API keys & other credentials securely | – |
| IDE / editor – VS Code, PyCharm, WebStorm, etc. | Development comfort | – |
| Docker (optional) | To run the service in an isolated container | ≥ 20.10 |
Note: The HaveIBeenPwned API is free for low‑volume usage (≤ 1 request/1.5 s). For higher throughput you’ll need a paid plan or an enterprise breach‑intelligence feed.
<a name="step-2-installation-and-setup"></a>
git clone https://github.com/icarax/breach-monitor-demo.git
cd breach-monitor-demo
# Create a virtual environment
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# Install dependencies
pip install --upgrade pip
pip install requests tenacity python-dotenv loguru
# Initialize a new Node project (if you didn't clone the repo)
npm init -y
# Install core libraries
npm install axios dotenv zod p-retry p-limit winston
# TypeScript support (optional but recommended)
npm install --save-dev typescript @types/node ts-node nodemon
npx tsc --init # creates tsconfig.json
Create a .env file in the project root never commit this file.
# .env (example)
HIBP_API_KEY=your_haveibeenpwned_v3_key_here
# Optional: where to send alerts
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX/YYY/ZZZ
SMTP_HOST=smtp.example.com
SMTP_PORT=587
SMTP_USER=alerts@example.com
SMTP_PASS=your_smtp_password
FROM_EMAIL=alerts@example.com
TO_EMAILS=ops-team@example.com,sec@example.com
Security tip: In production replace
.envwith a secrets manager (AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault, etc.) and inject the values as environment variables at runtime.
<a name="step-3-basic-implementation"></a>
Below are complete, copy‑and‑paste‑ready examples that:
The code deliberately avoids storing raw breach data locally (to stay compliant with data‑protection regulations). It only keeps a hash of the email for deduplication.
# breach_monitor_py.py
"""
Pentagon Personnel Agency breach monitor – Python version.
- Uses HaveIBeenPwned API v3.
- Implements exponential back‑off with tenacity.
- Sends alerts via Slack webhook or SMTP.
- Structured logging with loguru.
"""
import os
import csv
import hashlib
from typing import List, Dict, Any
import requests
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
)
from loguru import logger
from dotenv import load_dotenv
# ----------------------------------------------------------------------
# Load environment variables
# ----------------------------------------------------------------------
load_dotenv() # reads .env file
HIBP_API_KEY = os.getenv("HIBP_API_KEY")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")
SMTP_HOST = os.getenv("SMTP_HOST")
SMTP_PORT = int(os.getenv("SMTP_PORT", "0")) if os.getenv("SMTP_PORT") else None
SMTP_USER = os.getenv("SMTP_USER")
SMTP_PASS = os.getenv("SMTP_PASS")
FROM_EMAIL = os.getenv("FROM_EMAIL")
TO_EMAILS = os.getenv("TO_EMAILS", "").split(",") if os.getenv("TO_EMAILS") else []
if not HIBP_API_KEY:
raise RuntimeError("HIBP_API_KEY is missing – set it in .env or secrets manager")
# ----------------------------------------------------------------------
# Constants
# ----------------------------------------------------------------------
HIBP_BASE_URL = "https://haveibeenpwned.com/api/v3"
HEADERS = {
"hibp-api-key": HIBP_API_KEY,
"user-agent": "ICARAX-BreachMonitor/1.0 (+https://icarax.example.com)",
"accept": "application/json",
}
RATE_LIMIT_WAIT = 1.6 # seconds (HIBP allows 1 request per 1.5 s)
# ----------------------------------------------------------------------
# Helper: hash email for dedup (SHA‑256, hex)
# ----------------------------------------------------------------------
def hash_email(email: str) -> str:
return hashlib.sha256(email.lower().strip().encode()).hexdigest()
# ----------------------------------------------------------------------
# Core: query HIBP for a single email
# ----------------------------------------------------------------------
@retry(
reraise=True,
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((requests.RequestException,)),
)
def hibp_lookup(email: str) -> List[Dict[str, Any]]:
"""
Returns a list of breach objects for the given email.
Raises requests.HTTPError for non‑2xx responses (except 404 = clean).
"""
url = f"{HIBP_BASE_URL}/breachedaccount/{requests.utils.quote(email)}?truncateResponse=false"
resp = requests.get(url, headers=HEADERS, timeout=10)
if resp.status_code == 404:
# No breaches found – treat as empty list
return []
if resp.status_code == 429:
# Rate limit – let tenacity handle back‑off after raising
resp.raise_for_status()
resp.raise_for_status()
return resp.json()
# ----------------------------------------------------------------------
# Alerting: Slack or Email
# ----------------------------------------------------------------------
def send_slack_alert(message: str) -> None:
if not SLACK_WEBHOOK:
logger.warning("SLACK_WEBHOOK_URL not set – skipping Slack alert")
return
payload = {"text": message}
try:
r = requests.post(SLACK_WEBHOOK, json=payload, timeout=5)
r.raise_for_status()
logger.info("Slack alert sent")
except Exception as exc:
logger.error(f"Failed to send Slack alert: {exc}")
def send_email_alert(subject: str, body: str) -> None:
if not all([SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS, FROM_EMAIL]):
logger.warning("SMTP settings incomplete – skipping email alert")
return
import smtplib
from email.message import EmailMessage
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = FROM_EMAIL
msg["To"] = ", ".join(filter(None, TO_EMAILS))
msg.set_content(body)
try:
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
logger.info("Email alert sent")
except Exception as exc:
logger.error(f"Failed to send email alert: {exc}")
def alert_breach(email: str, breaches: List[Dict[str, Any]]) -> None:
"""Format and dispatch an alert for a compromised email."""
lines = [f"🚨 *Breach detected* for `{email}`:"]
for b in breaches:
name = b.get("Name", "Unknown")
date = b.get("BreachDate", "???")
data_classes = ", ".join(b.get("DataClasses", []))
lines.append(f"- *{name}* ({date}) – exposed: {data_classes}")
message = "\n".join(lines)
# Try Slack first, fallback to email
if SLACK_WEBHOOK:
send_slack_alert(message)
else:
send_email_alert(
subject=f"[ALERT] Breach detected for {email}",
body=message,
)
# ----------------------------------------------------------------------
# Main workflow
# ----------------------------------------------------------------------
def load_emails_from_csv(path: str) -> List[str]:
"""Expects a single column CSV with header 'email'."""
emails = []
with open(path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
email = row.get("email", "").strip()
if email:
emails.append(email)
return emails
def main(csv_path: str = "targets.csv") -> None:
logger.add("breach_monitor_{time}.log", rotation="10 MB", retention="10 days")
logger.info("Starting breach monitor")
seen_hashes = set()
emails = load_emails_from_csv(csv_path)
logger.info(f"Loaded {len(emails)} target email(s)")
for email in emails:
email_hash = hash_email(email)
if email_hash in seen_hashes:
logger.debug(f"Skipping duplicate email: {email}")
continue
seen_hashes.add(email_hash)
logger.info(f"Checking {email}")
try:
breaches = hibp_lookup(email)
if breaches:
logger.warning(f"Breach found for {email}: {len(breaches)} incident(s)")
alert_breach(email, breaches)
else:
logger.info(f"No breaches for {email}")
except Exception as exc:
logger.error(f"Error while checking {email}: {exc}")
# Respect HIBP rate limit
import time
time.sleep(RATE_LIMIT_WAIT)
logger.info("Breach monitor finished")
if __name__ == "__main__":
main()
# 1️⃣ Prepare a CSV (targets.csv) with a header 'email' and one address per line
echo -e "email\njohn.doe@example.gov\njane.smith@example.gov" > targets.csv
# 2️⃣ Execute
python breach_monitor_py.py
// breach_monitor_ts.ts
/*
* Pentagon Personnel Agency breach monitor – TypeScript version.
* Uses axios for HTTP, p-retry for back‑off, p-limit for concurrency,
* zod for runtime validation, and winston for structured logging.
* Alerts via Slack webhook or Nodemailer (SMTP).
*/
import "dotenv/config";
import axios from "axios";
import { pRetry } from "p-retry";
import { pLimit } from "p-limit";
import { z } from "zod";
import { createLogger, format, transports } from "winston";
import nodemailer from "nodemailer";
// ----------------------------------------------------------------------
// Configuration & Constants
// ----------------------------------------------------------------------
const HIBP_API_KEY = process.env.HIBP_API_KEY;
if (!HIBP_API_KEY) {
throw new Error("Missing HIBP_API_KEY in environment");
}
const SLACK_WEBHOOK = process.env.SLACK_WEBHOOK_URL;
const SMTP_HOST = process.env.SMTP_HOST;
const SMTP_PORT = Number(process.env.SMTP_PORT) || 0;
const SMTP_USER = process.env.SMTP_USER;
const SMTP_PASS = process.env.SMTP_PASS;
const FROM_EMAIL = process.env.FROM_EMAIL;
const TO_EMAILS = process.env.TO_EMAILS?.split(",").filter(Boolean) ?? [];
const HIBP_BASE = "https://haveibeenpwned.com/api/v3";
const HEADERS = {
"hibp-api-key": HIBP_API_KEY,
"user-agent": "ICARAX-BreachMonitor/1.0 (+https://icarax.example.com)",
accept: "application/json",
};
const RATE_LIMIT_MS = 1600; // ~1 request per 1.5 s
const CONCURRENCY = 1; // HIBP is strict; keep 1 unless you have a paid plan
// ----------------------------------------------------------------------
// Logger (winston)
// ----------------------------------------------------------------------
const logger = createLogger({
level: "info",
format: format.combine(
format.timestamp(),
format.errors({ stack: true }),
format.splat(),
format.json()
),
transports: [new transports.Console(), new transports.File({ filename: "breach-monitor.log" })],
});
// ----------------------------------------------------------------------
// Zod schemas for runtime validation of HIBP response
// ----------------------------------------------------------------------
const BreachSchema = z.object({
Name: z.string(),
Title: z.string(),
Domain: z.string(),
BreachDate: z.string(),
AddedDate: z.string(),
ModifiedDate: z.string(),
PwnCount: z.number(),
Description: z.string(),
DataClasses: z.array(z.string()),
IsVerified: z.boolean(),
IsFabricated: z.boolean(),
IsSensitive: z.boolean(),
IsRetired: z.boolean(),
IsSpamList: z.boolean(),
});
type Breach = z.infer<typeof BreachSchema>;
// ----------------------------------------------------------------------
// HTTP helper with retry & rate‑limit
// ----------------------------------------------------------------------
const axiosInstance = axios.create({
baseURL: HIBP_BASE,
timeout: 10_000,
headers: HEADERS,
});
async function hibpLookup(email: string): Promise<Breach[]> {
const encoded = encodeURIComponent(email.trim().toLowerCase());
const url = `/breachedaccount/${encoded}?truncateResponse=false`;
// pRetry will retry on network errors or 429/5xx
const response = await pRetry(
async () => {
const res = await axiosInstance.get(url);
// 200 → breaches, 404 → clean
if (res.status === 200) {
// Validate each breach object
const raw = res.data as unknown[];
return raw.map((item) => BreachSchema.parse(item));
}
if (res.status === 404) {
return []; // clean
}
// Throw to trigger retry (e.g., 429, 5xx)
throw new Error(`Unexpected HIBP status ${res.status}`);
},
{
retries: 5,
factor: 2,
minTimeout: 1000,
maxTimeout: 10000,
randomize: true,
}
);
return response;
}
// ----------------------------------------------------------------------
// Alerting helpers
// ----------------------------------------------------------------------
async function sendSlackAlert(text: string): Promise<void> {
if (!SLACK_WEBHOOK) {
logger.warn("SLACK_WEBHOOK_URL not set – skipping Slack");
return;
}
try {
await axios.post(SLACK_WEBHOOK, { text });
logger.info("Slack alert sent");
} catch (err) {
logger.error({ err }, "Failed to send Slack alert");
}
}
function createSmtpTransport() {
if (!SMTP_HOST || !SMTP_PORT || !SMTP_USER || !SMTP_PASS || !FROM_EMAIL) {
return null;
}
return nodemailer.createTransport({
host: SMTP_HOST,
port: SMTP_PORT,
secure: SMTP_PORT === 465, // true for 465, false otherwise
auth: { user: SMTP_USER, pass: SMTP_PASS },
});
}
async function sendEmailAlert(subject: string, body: string): Promise<void> {
const transporter = createSmtpTransport();
if (!transporter) {
logger.warn("SMTP not fully configured – skipping email alert");
return;
}
try {
await transporter.sendMail({
from: FROM_EMAIL,
to: TO_EMAILS.join(", "),
subject,
text: body,
});
logger.info("Email alert sent");
} catch (err) {
logger.error({ err }, "Failed to send email alert");
}
}
async function alertBreach(email: string, breaches: Breach[]): Promise<void> {
const lines = [
`🚨 *Breach detected* for \`${email}\`:`,
...breaches.map(
(b) => `- *${b.Name}* (${b.BreachDate}) – exposed: ${b.DataClasses.join(", ")}`
),
];
const message = lines.join("\n");
if (SLACK_WEBHOOK) {
await sendSlackAlert(message);
} else {
await sendEmailAlert(
`[ALERT] Breach detected for ${email}`,
message,
);
}
}
// ----------------------------------------------------------------------
// CSV reader (simple, no external deps)
// ----------------------------------------------------------------------
function loadEmailsFromCsv(csvPath: string): string[] {
const fs = require("fs");
const path = require("path");
const file = fs.readFileSync(path.resolve(csvPath), "utf8");
const lines = file.trim().split("\n");
if (lines.length === 0) return [];
const header = lines[0].split(",").map((h) => h.trim());
const emailIdx = header.indexOf("email");
if (emailIdx === -1) {
throw new Error("CSV must contain an 'email' column");
}
const emails: string[] = [];
for (let i = 1; i < lines.length; i++) {
const cols = lines[i].split(",");
const email = cols[emailIdx]?.trim();
if (email) emails.push(email);
}
return emails;
}
// ----------------------------------------------------------------------
// Main orchestration
// ----------------------------------------------------------------------
async function main() {
logger.info("Starting breach monitor (TS)");
const csvPath = process.env.TARGETS_CSV ?? "targets.csv";
const emails = loadEmailsFromCsv(csvPath);
logger.info({ count: emails.length }, "Loaded target emails");
const seen = new Set<string>();
const limit = pLimit(CONCURRENCY);
for (const email of emails) {
const emailHash = require("crypto")
.createHash("sha256")
.update(email.trim().toLowerCase())
.digest("hex");
if (seen.has(emailHash)) {
logger.debug({ email }, "Skipping duplicate");
continue;
}
seen.add(emailHash);
logger.info({ email }, "Checking for breaches");
try {
const breaches = await hibpLookup(email);
if (breaches.length > 0) {
logger.warn({ email, breachCount: breaches.length }, "Breach(s) found");
await alertBreach(email, breaches);
} else {
logger.info({ email }, "No breaches found");
}
} catch (err) {
logger.error({ email, err }, "Error while querying HIBP");
}
// Respect rate limit – simple delay between each request
await new Promise((resolve) => setTimeout(resolve, RATE_LIMIT_MS));
}
logger.info("Breach monitor finished");
}
// ----------------------------------------------------------------------
// Execute
// ----------------------------------------------------------------------
main().catch((err) => {
logger.error({ err }, "Fatal error in breach monitor");
process.exit(1);
});
# 1️⃣ Install deps (if you haven't already)
npm install
# 2️⃣ Build (optional) – you can also run directly with ts-node
npx tsc # compiles to ./dist/
# 3️⃣ Execute
node dist/breach_monitor_ts.ts # or: npx ts-node breach_monitor_ts.ts
Tip: Keep
CONCURRENCY = 1unless you have a paid HaveIBeenPwned plan that allows higher request rates. Exceeding the limit results in HTTP 429 and will trigger the retry/back‑off logic.
<a name="step-4-configuration"></a>
| Variable | Required? | Description | Example |
|---|---|---|---|
HIBP_API_KEY | ✅ | HaveIBeenPwned v3 API key | abc123def456... |
SLACK_WEBHOOK_URL | ❌ | Incoming webhook for Slack alerts | https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX |
SMTP_HOST | ❌ (if using email) | SMTP server host | smtp.mailgun.org |
SMTP_PORT | ❌ (if using email) | SMTP port (usually 587) | 587 |
SMTP_USER | ❌ (if using email) | SMTP username | postmaster@sandbox123.mailgun.org |
SMTP_PASS | ❌ (if using email) | SMTP password / API key | key-1234567890abcdef |
FROM_EMAIL | ❌ (if using email) | Sender address for alerts | alerts@example.com |
TO_EMAILS | ❌ (if using email) | Comma‑separated list of recipients | ops@example.com,sec@example.com |
TARGETS_CSV | ❌ | Path to CSV containing emails to check (defaults to targets.csv) | ./data/personnel.csv |
LOG_LEVEL | ❌ | Verbosity of logs (error, warn, info, debug) | info |
Loading the config:
Both examples use dotenv (require('dotenv').config() in TS, load_dotenv() in Python) to read a .env file. In production you would replace this with your cloud provider’s secret injection mechanism (e.g., AWS Lambda environment variables, Kubernetes secrets, etc.).
<a name="step-5-common-patterns"></a>
Below are reusable snippets that you’ll see across many breach‑monitoring or security‑automation projects.
# python_utils.py
import time
import requests
from functools import wraps
def rate_limited(calls_per_period, period):
"""Decorator to enforce a max number of calls per time period."""
min_interval = period / float(calls_per_period)
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
return decorator
@rate_limited(calls_per_period=1, period=1.5) # HIBP: 1 per 1.5 s
def hibp_get(url, **kwargs):
return requests.get(url, **kwargs)
// backoff.ts
import { pRetry } from "p-retry";
export function withBackoff<T>(fn: () => Promise<T>, opts?: Parameters<typeof pRetry>[1]) {
return pRetry(fn, {
retries: 5,
factor: 2,
minTimeout: 1000,
maxTimeout: 15000,
randomize: true, // adds jitter
...opts,
});
}
# logging_setup.py
import uuid
from loguru import logger
def patch_record(record):
record["extra"]["request_id"] = record["extra"].get("request_id", str(uuid.uuid4()))
return record
logger.configure(patcher=patch_record)
# Usage:
logger.bind(request_id="abc-123").info("Started check")
# circuit.py
from pybreaker import CircuitBreaker
hibp_breaker = CircuitBreaker(fail_max=5, reset_timeout=60)
@hibp_breaker
def safe_hibp_lookup(email):
return hibp_lookup(email) # your original function
import { pLimit } from "p-limit";
const limit = pLimit(3); // max 3 concurrent jobs
async function job(email) {
return limit(() => hibpLookup(email));
}
<a name="step-6-troubleshooting"></a>
| Symptom | Likely cause | Fix / mitigation |
|---|---|---|
429 Too Many Requests from HIBP | Exceeded rate limit ( > 1 req/1.5 s ) | Increase RATE_LIMIT_WAIT (Python) or RATE_LIMIT_MS (TS). Consider upgrading to a paid HIBP plan or using a local breach dump. |
401 Unauthorized | Invalid or missing hibp-api-key | Verify HIBP_API_KEY in .env or secret store. Ensure no extra whitespace. |
502 Bad Gateway / 504 Gateway Timeout | Transient network issue or HIBP downtime | Retry logic (already present). If persistent, check status page: https://haveibeenpwned.com/status. |
| No alerts sent despite breaches found | Alert channel mis‑configured (Slack webhook URL wrong, SMTP auth failing) | Test each channel manually: curl -X POST -d '{"text":"test"}' $SLACK_WEBHOOK_URL; use telnet or openssl s_client to verify SMTP. |
| Duplicate alerts for same email | No deduplication logic or hash collision | Ensure you hash the email (SHA‑256) and keep a seen set during a run. Persist deduplication across runs if needed (e.g., Redis set). |
Program crashes with KeyError on breach data | HIBP changed response schema | Update the Zod schema (TS) or adjust dict access (Python) to use .get() with defaults. |
| Log files grow unbounded | No log rotation configured | Use loguru rotation (rotation="10 MB"), or winston daily file transport with zippedArchive. |
| High CPU usage when processing large CSVs | Reading entire file into memory | Stream CSV line‑by‑line (csv.reader in Python, fast-csv or line-reader in TS). |
Debug tip: Set environment variable LOG_LEVEL=debug (or configure the logger) to see the exact request/response payloads (strip out the API key before logging!).
<a name="step-7-production-checklist"></a>
| ✅ Item | Why it matters | How to verify |
|---|---|---|
| Secrets management | Prevent accidental API key leakage | Use AWS Secrets Manager / GCP Secret Manager / Vault; confirm .env is not in repo (git check-ignore .env). |
| HTTPS everywhere | Protect data in transit | Verify all outbound calls (axios, requests) use https://. |
| Input validation & sanitization | Avoid injection or unexpected behavior | Validate email format (zod/email-validator) before hashing. |
| Idempotent processing | Safe to re‑run without duplicate alerts | Store processed email hashes in a durable store (Redis, DynamoDB) and skip if already seen. |
| Rate limit compliance | Avoid being blocked by HIBP or other services | Measure actual request interval; add telemetry (Prometheus counter). |
| Observability | Detect failures early | Emit metrics: breaches_found, alerts_sent, api_errors. Hook logs into ELK/Datadog/Splunk. |
| Alert deduplication & throttling | Prevent alert fatigue | Combine multiple breaches for same email into one notification; add a cooldown (e.g., no repeat alert for same email within 24 h). |
| Testing | Guarantees correctness | Write unit tests for hibp_lookup (mock responses), integration test against a staging HIBP key, chaos test for network failures. |
| Dependency hygiene | Avoid supply‑chain attacks | Run npm audit / pip check regularly; lock versions (package-lock.json, requirements.txt). |
| Legal / compliance | Handling personal data lawfully | Only store hashes, never raw breach data; retain logs per your data‑retention policy; consult DPO if needed. |
| Disaster recovery | Service continuity | Deploy behind an autoscaling group or container orchestrator (ECS/EKS, Cloud Run). Have a backup region. |
| Documentation & runbooks | Enables on‑call engineers to act | Keep a README.md with deployment steps, escalation matrix, and manual verification commands. |
Copy the Python or TypeScript script that matches your stack, configure the .env (or your secret store), point TARGETS_CSV at the list of Pentagon personnel emails you need to monitor, and run the service. The implementation includes:
Feel free to extend the code—for example, push findings to a SIEM, create tickets in Jira, or trigger automated password‑reset workflows.
Stay safe, and keep those credentials protected! 🚀
Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
