

How developers can quickly check whether an email/phone appears in known breach data and integrate breach‑monitoring into their apps.
Why this matters – The CenterPoint Energy breach exposed names, addresses, SSNs, DOB, and utility‑account details for ~1 million Texans. If you handle customer data, you should be able to verify whether any of your users’ credentials have appeared in this (or other) public breach feeds and take remedial action (force‑reset passwords, notify users, etc.).
<a name="step-1-prerequisites"></a>
| Item | Why you need it | How to get it |
|---|---|---|
| Python ≥ 3.9 | Core language for the backend example | https://www.python.org/downloads/ |
| Node.js ≥ 18 (or Deno) | Runtime for the JS/TS example | https://nodejs.org/ |
| Git | Clone the sample repo (optional) | https://git-scm.com/ |
| API key for a breach‑lookup service (e.g., HaveIBeenPwned (HIBP)) | Needed to query the breach database securely | Sign up at https://haveibeenpwned.com/API/v3#Key |
| An IDE or editor (VS Code, PyCharm, WebStorm, etc.) | Write & debug code | Your preference |
| Basic knowledge of REST APIs & async/await | Understand the code flow | — |
Note: The examples use the free HIBP API (rate‑limited to 1 request per 1.5 seconds without a key; with a key you get higher limits). If you prefer a commercial breach‑intelligence feed, swap the endpoint and authentication logic – the pattern stays the same.
<a name="step-2-installation-and-setup"></a>
mkdir centerpoint-breach-check
cd centerpoint-breach-check
# Create a virtual environment (recommended)
python -m venv venv
source venv/bin/activate # on Windows: venv\Scripts\activate
# Install dependencies
pip install --upgrade pip
pip install requests python-dotenv
# Initialize a npm project
npm init -y
# Install core libraries
npm install axios dotenv
# (Optional) TypeScript support
npm install --save-dev typescript @types/node ts-node
npx tsc --init # creates tsconfig.json
Create a .env file in the project root (never commit this file!):
# .env
HIBP_API_KEY=your_haveibeenpwned_api_key_here
# Optional: adjust rate‑limit pause (seconds) if you have a paid key
HIBP_RATE_LIMIT_PAUSE=1.5
Security tip: Add
.envto your.gitignore.
<a name="step-3-basic-implementation"></a>
Below are complete, copy‑paste‑ready snippets that:
# file: hibp_check.py
"""
Simple breach‑lookup utility using HaveIBeenPwned API.
Usage:
python hibp_check.py user@example.com
"""
import os
import sys
import time
import requests
from dotenv import load_dotenv
# ----------------------------------------------------------------------
# Load environment variables from .env (if present)
# ----------------------------------------------------------------------
load_dotenv()
HIBP_API_KEY = os.getenv("HIBP_API_KEY")
# If you have a paid key you can lower the pause; otherwise keep default.
RATE_LIMIT_PAUSE = float(os.getenv("HIBP_RATE_LIMIT_PAUSE", "1.5"))
HIBP_ENDPOINT = "https://haveibeenpwned.com/api/v3/breachedaccount/{}"
HEADERS = {
"hibp-api-key": HIBP_API_KEY,
"user-agent": "ICARAX-Breach-Checker/1.0",
}
def check_breach(email: str) -> dict:
"""
Query HIBP for a single email address.
Returns a dict with keys:
- success (bool)
- breaches (list[str]) # breach names, empty if none
- error (str|None) # human‑readable message if something went wrong
"""
if not HIBP_API_KEY:
return {"success": False, "breaches": [], "error": "HIBP API key not set."}
url = HIBP_ENDPOINT.format(requests.utils.quote(email))
try:
resp = requests.get(url, headers=HEADERS, timeout=10)
# Respect HIBP rate‑limit header if present
if "retry-after" in resp.headers:
wait = int(resp.headers["retry-after"])
time.sleep(wait)
elif RATE_LIMIT_PAUSE:
time.sleep(RATE_LIMIT_PAUSE)
if resp.status_code == 200:
data = resp.json()
breach_names = [item["Name"] for item in data]
return {"success": True, "breaches": breach_names, "error": None}
elif resp.status_code == 404:
# No breaches found for this account
return {"success": True, "breaches": [], "error": None}
elif resp.status_code == 429:
# Too many requests – back off
return {"success": False, "breaches": [], "error": "Rate limit exceeded. Try again later."}
else:
return {
"success": False,
"breaches": [],
"error": f"Unexpected status {resp.status_code}: {resp.text}",
}
except requests.RequestException as exc:
return {"success": False, "breaches": [], "error": f"Network error: {exc}"}
def main():
if len(sys.argv) != 2:
print("Usage: python hibp_check.py <email-address>")
sys.exit(1)
email = sys.argv[1].strip()
result = check_breach(email)
if not result["success"]:
print(f"❌ Error: {result['error']}")
sys.exit(1)
if result["breaches"]:
print(f"⚠️ {email} appears in the following breaches:")
for b in result["breaches"]:
print(f" - {b}")
# Optional: highlight CenterPoint Energy breach if present
if any("centerpoint" in b.lower() for b in result["breaches"]):
print(" 🔎 CenterPoint Energy breach detected!")
else:
print(f"✅ Good news: {email} was not found in any known breach.")
if __name__ == "__main__":
main()
How to run
python hibp_check.py user@example.com
// file: hibpCheck.ts
/*
* Simple breach‑lookup utility using HaveIBeenPwned API.
* Compile with: npx ts-node hibpCheck.ts user@example.com
* Or run directly with Node (if you rename to .js and drop types).
*/
import * as dotenv from "dotenv";
import axios from "axios";
// Load .env file
dotenv.config();
const HIBP_API_KEY: string | undefined = process.env.HIBP_API_KEY;
const RATE_LIMIT_PAUSE: number = parseFloat(
process.env.HIBP_RATE_LIMIT_PAUSE ?? "1.5"
);
if (!HIBP_API_KEY) {
console.error("❌ HIBP_API_KEY is not set in .env");
process.exit(1);
}
const HIBP_ENDPOINT = (email: string) =>
`https://haveibeenpwned.com/api/v3/breachedaccount/${encodeURIComponent(email)}`;
const axiosInstance = axios.create({
timeout: 10_000,
headers: {
"hibp-api-key": HIBP_API_KEY,
"user-agent": "ICARAX-Breach-Checker/1.0",
},
});
/**
* Sleeps for ms milliseconds.
*/
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/**
* Query HIBP for a single email.
* Returns an object:
* { success: boolean, breaches: string[], error?: string }
*/
async function checkBreach(email: string) {
try {
const resp = await axiosInstance.get(HIBP_ENDPOINT(email));
// Respect Rate‑Limit header if present
const retryAfter = resp.headers["retry-after"];
if (retryAfter) {
await sleep(parseInt(retryAfter, 10) * 1000);
} else if (RATE_LIMIT_PAUSE > 0) {
await sleep(RATE_LIMIT_PAUSE * 1000);
}
// 200 → breaches found
if (resp.status === 200) {
const breachNames: string[] = resp.data.map((b: any) => b.Name);
return { success: true, breaches: breachNames };
}
// 404 → no breaches
if (resp.status === 404) {
return { success: true, breaches: [] };
}
// Anything else is unexpected
return {
success: false,
breaches: [],
error: `Unexpected status ${resp.status}: ${resp.statusText}`,
};
} catch (err: any) {
// Axios error handling
if (err.response) {
const { status, statusText } = err.response;
if (status === 429) {
return {
success: false,
breaches: [],
error: "Rate limit exceeded. Try again later.",
};
}
return {
success: false,
breaches: [],
error: `HTTP ${status}: ${statusText}`,
};
} else if (err.request) {
// No response received
return {
success: false,
breaches: [],
error: "Network error: no response from HIBP",
};
} else {
// Something else
return {
success: false,
breaches: [],
error: `Unexpected error: ${err.message}`,
};
}
}
}
/* -------------------- CLI entry point -------------------- */
async function main() {
const email = process.argv[2];
if (!email) {
console.error("Usage: npx ts-node hibpCheck.ts <email-address>");
process.exit(1);
}
const result = await checkBreach(email.trim());
if (!result.success) {
console.error(`❌ Error: ${result.error}`);
process.exit(1);
}
if (result.breaches.length > 0) {
console.log(`⚠️ ${email} appears in the following breaches:`);
for (const b of result.breaches) {
console.log(` - ${b}`);
}
const centerpoint = result.breaches.some((b) =>
b.toLowerCase().includes("centerpoint")
);
if (centerpoint) {
console.log(" 🔎 CenterPoint Energy breach detected!");
}
} else {
console.log(`✅ Good news: ${email} was not found in any known breach.`);
}
}
main().catch((e) => {
console.error("Unexpected error:", e);
process.exit(1);
});
How to run
# If you kept the .ts file:
npx ts-node hibpCheck.ts user@example.com
# Or compile to JS first:
npx tsc hibpCheck.ts
node hibpCheck.js user@example.com
<a name="step-4-configuration"></a>
| Variable | Description | Example |
|---|---|---|
HIBP_API_KEY | Your HaveIBeenPwned API key (required for higher rate limits). Get it at https://haveibeenpwned.com/API/v3#Key. | HIBP_API_KEY=abcdef1234567890 |
HIBP_RATE_LIMIT_PAUSE | Seconds to wait between calls when no Retry-After header is present. Adjust if you have a paid key with higher limits. | HIBP_RATE_LIMIT_PAUSE=1.5 |
LOG_LEVEL (optional) | Control verbosity (debug, info, warn, error). Not used in the snippets but handy if you plug into a logger. | LOG_LEVEL=info |
PROXY_URL (optional) | If you’re behind a corporate proxy, set HTTP_PROXY/HTTPS_PROXY or add to axios/requests config. | PROXY_URL=http://proxy.corp:3128 |
Loading the config – Both snippets use dotenv (python-dotenv for Python, dotenv for Node) which automatically reads a .env file in the working directory. Never commit .env; add it to .gitignore.
<a name="step-5-common-patterns"></a>
Below are reusable patterns you’ll see in production‑grade breach‑checking code.
# Python batch version (hibp_batch.py)
def batch_check(emails: list[str]) -> dict[str, dict]:
results = {}
for e in emails:
results[e] = check_breach(e)
# Respect rate limit between each request
time.sleep(RATE_LIMIT_PAUSE)
return results
// TS batch version (hibpBatch.ts)
async function batchCheck(emails: string[]): Promise<Record<string, any>> {
const out: Record<string, any> = {};
for (const e of emails) {
out[e] = await checkBreach(e);
await sleep(RATE_LIMIT_PAUSE * 1000);
}
return out;
}
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
def cached_check(email: str) -> dict:
key = f"hibp:{email}"
cached = r.get(key)
if cached:
return json.loads(cached)
result = check_breach(email)
# Store for 24 h (breach data changes infrequently)
r.setex(key, 86400, json.dumps(result))
return result
// Using ioredis (npm i ioredis)
import Redis from "ioredis";
const redis = new Redis();
async function cachedCheck(email: string): Promise<any> {
const key = `hibp:${email}`;
const cached = await redis.get(key);
if (cached) return JSON.parse(cached);
const result = await checkBreach(email);
await redis.setex(key, 86400, JSON.stringify(result)); // 24h TTL
return result;
}
FastAPI (Python)
# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseEmailStr, BaseModel
app = FastAPI(title="Breach Checker API")
class EmailRequest(BaseModel):
email: BaseEmailStr
@app.post("/check")
async def check(payload: EmailRequest):
res = check_breach(str(payload.email))
if not res["success"]:
raise HTTPException(status_code=502, detail=res["error"])
return {"email": payload.email, "breaches": res["breaches"]}
Run with: uvicorn api:app --host 0.0.0.0 --port 8000
Express (Node/TS)
// server.ts
import express, { Request, Response } from "express";
import { checkBreach } from "./hibpCheck";
const app = express();
app.use(express.json());
app.post("/check", async (req: Request, res: Response) => {
const { email } = req.body;
if (!email || !/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)) {
return res.status(400).json({ error: "Invalid email" });
}
const result = await checkBreach(email.trim());
if (!result.success) {
return res.status(502).json({ error: result.error });
}
res.json({ email, breaches: result.breaches });
});
const PORT = process.env.PORT ?? 3000;
app.listen(PORT, () => console.log(`🚀 Server listening on :${PORT}`));
Compile TS (npx tsc server.ts) then run node server.js.
<a name="step-6-troubleshooting"></a>
| Symptom | Likely Cause | Fix |
|---|---|---|
401 Unauthorized from HIBP | Missing or incorrect hibp-api-key | Verify HIBP_API_KEY in .env; ensure no extra spaces. |
429 Too Many Requests | Exceeded rate limit (free tier = 1 req/1.5 s) | Increase HIBP_RATE_LIMIT_PAUSE or upgrade to a paid key. |
Network error: Max retries exceeded | No internet, DNS block, or corporate proxy | Check connectivity; set PROXY_URL and pass proxies to requests or axios. |
| Empty breach list but you know the email was breached | Email address not exactly as stored (case‑sensitive, extra spaces) | Trim and lower‑case the input before sending; HIBP expects the exact email as supplied. |
502 Bad Gateway from your own API | Upstream HIBP call failed | Look at upstream error returned in JSON (error field) and act accordingly (retry, alert). |
TypeScript compilation errors (Cannot find module 'dotenv') | Missing @types/node or outdated TS config | Run npm install --save-dev @types/node and ensure "esModuleInterop": true in tsconfig.json. |
Import "dotenv" could not be resolved (Python) | python-dotenv not installed in the active venv | pip install python-dotenv inside the venv. |
Debug tip: Add print(resp.text) (Python) or console.log(resp.data) (TS) inside the error branch to see the raw HIBP payload—often it contains a helpful message like “Account not found”.
<a name="step-7-production-checklist"></a>
Before you push this breach‑checker to production, run through the following:
| ✅ Item | Why it matters |
|---|---|
| API key stored securely (e.g., AWS Secrets Manager, GCP Secret Manager, Vault) – never hard‑code. | |
Rate‑limit handling – respect Retry-After header and implement exponential back‑off for 429 responses. | |
| Input validation – validate email format, length, and reject obviously malicious strings (e.g., containing newlines or SQL). | |
| Logging & audit – log requests (hashed email, timestamp, outcome) for compliance; do not log raw emails unless required and masked. | |
Error‑response sanitization – return generic messages to clients ("Unable to check breach status"), keep detailed errors server‑side. | |
| Circuit breaker – if HIBP is consistently failing, temporarily stop calling it and serve cached results or a fallback message. | |
| Data retention – if you cache results, define a TTL (e.g., 24 h) and purge old entries to avoid storing stale breach info indefinitely. | |
| HTTPS everywhere – enforce TLS for your service and for outbound calls to HIBP. | |
Dependency scanning – run npm audit / pip check and keep libraries up‑to‑date to avoid supply‑chain risks. | |
| Monitoring & alerts – set up alerts on error rates (>5 % 5xx) or latency spikes (>2 s). | |
| Privacy notice – inform users that you’re checking their email against public breach databases and link to HIBP’s privacy policy. | |
| Load testing – simulate peak traffic (e.g., 100 rps) to verify your rate‑limit logic doesn’t cause thundering herd. | |
| Disaster recovery – document how to rotate the HIBP API key and redeploy without downtime. |
Copy the snippets above, add your own business logic (e.g., force password reset, notify security team, integrate with SIEM), and you’ll have a solid, production‑ready foundation for checking whether any of your users’ data appeared in the CenterPoint Energy breach—or any other breach catalogued by HaveIBeenPwned.
Stay safe, keep your users’ data protected, and happy hacking! 🚀
Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
