

ICARAX Tech Blog – Practical guide for developers
Why this matters – The AdaptHealth breach exposed 4.1 million records containing names, addresses, Social Security numbers, and health‑plan information. Developers can help mitigate the impact by building utilities that:
- Detect whether a user’s credentials appear in known breach dumps (e.g., HaveIBeenPwned).
- Protect any PHI (Protected Health Information) they store or transmit with strong encryption.
- Alert stakeholders quickly and securely.
The following sections give you a complete, copy‑and‑paste‑ready implementation in both Python and JavaScript/TypeScript, plus configuration, patterns, troubleshooting, and a production checklist.
<a name="step-1-prerequisites"></a>
| Item | Why you need it | Recommended version |
|---|---|---|
| Python | Core language for the Python example | >=3.9 |
| Node.js | Runtime for the JS/TS example | >=14 LTS |
| HaveIBeenPwned (HIBP) API key (optional but recommended) | Allows authenticated requests → higher rate limit (1 request/1.5 s vs 1 request/10 s anonymously) | Sign up at https://haveibeenpwned.com/API/v3 |
| AWS account (or any KMS‑compatible service) | Demo encryption of PHI using a managed key | Free tier is sufficient |
| Git | To clone the example repo (optional) | Any recent version |
| IDE / Editor | VS Code, PyCharm, WebStorm, etc. | — |
| curl (for quick testing) | Verify API reachability | — |
Tip: If you don’t want to provision AWS KMS, the code also works with a locally generated symmetric key (for demo only – never use this in production).
<a name="step-2-installation--setup"></a>
git clone https://github.com/icarax/adapthealth-breach-toolkit.git
cd adapthealth-breach-toolkit
# 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 python-dotenv cryptography
# Initialize a new npm project (if not already done)
npm init -y
# Install core libraries
npm install axios dotenv
# Install TypeScript and typings (dev dependencies)
npm install --save-dev typescript @types/node ts-node
npx tsc --init # creates tsconfig.json (default is fine)
Create a .env file in the project root (both language examples read from it):
# .env
# HaveIBeenPwned API key – get it from https://haveibeenpwned.com/API/v3#Key
HIBP_API_KEY=your_hibp_api_key_here
# AWS KMS key ARN (or alias) used to encrypt/decrypt PHI
# Example: arn:aws:kms:us-east-1:123456789012:key/abcd1234-a123-4a12-a12b-a123b4cd56ef
KMS_KEY_ARN=arn:aws:kms:us-east-1:123456789012:key/abcd1234-a123-4a12-a12b-a123b4cd56ef
# AWS region (if using KMS)
AWS_REGION=us-east-1
# Optional: enable debug logging
DEBUG=true
Never commit
.envto source control. Add it to.gitignore.
<a name="step-3-basic-implementation"></a>
Below are two self‑contained scripts that:
Both scripts expose a simple CLI interface so you can test them instantly.
breach_checker.py)#!/usr/bin/env python3
"""
breach_checker.py
-----------------
A tiny utility that:
1. Queries HaveIBeenPwned for breaches associated with an email.
2. Demonstrates encryption/decryption of PHI using AWS KMS (with a local fallback).
Author: ICARAX DevRel
"""
import os
import sys
import base64
import json
import logging
from typing import List, Dict, Optional
import requests
from dotenv import load_dotenv
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.exceptions import InvalidSignature
# ----------------------------------------------------------------------
# Setup
# ----------------------------------------------------------------------
load_dotenv() # loads .env into os.environ
logging.basicConfig(
level=logging.DEBUG if os.getenv("DEBUG") == "true" else logging.INFO,
format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)
log = logging.getLogger("breach_checker")
HIBP_API_KEY = os.getenv("HIBP_API_KEY")
KMS_KEY_ARN = os.getenv("KMS_KEY_ARN")
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
HIBP_ENDPOINT = "https://haveibeenpwned.com/api/v3/breachedaccount/{}"
HIBP_HEADERS = {
"api-version": "3",
"User-Agent": "ICARAX-BreachChecker/1.0",
}
if HIBP_API_KEY:
HIBP_HEADERS["hibp-api-key"] = HIBP_API_KEY
# ----------------------------------------------------------------------
# Helper: HTTP request with retry & back‑off
# ----------------------------------------------------------------------
def hibp_get(url: str, max_retries: int = 3) -> Optional[requests.Response]:
"""GET with exponential back‑off for 429 (rate limit) and transient errors."""
import time
for attempt in range(1, max_retries + 1):
try:
resp = requests.get(url, headers=HIBP_HEADERS, timeout=10)
if resp.status_code == 429:
wait = int(resp.headers.get("retry-after", "2")) * attempt
log.warning(f"Rate limited. Waiting {wait}s (attempt {attempt})")
time.sleep(wait)
continue
if 500 <= resp.status_code < 600:
wait = 2 ** attempt
log.warning(f"Server error {resp.status_code}. Retrying in {wait}s")
time.sleep(wait)
continue
resp.raise_for_status()
return resp
except requests.RequestException as exc:
log.error(f"Request failed (attempt {attempt}): {exc}")
if attempt == max_retries:
return None
time.sleep(2 ** attempt)
return None
# ----------------------------------------------------------------------
# 1️⃣ Breach lookup
# ----------------------------------------------------------------------
def check_breaches(email: str) -> List[Dict]:
"""
Returns a list of breach dicts (as JSON) for the supplied email.
Empty list => no known breaches.
"""
email_encoded = requests.utils.quote(email)
url = HIBP_ENDPOINT.format(email_encoded)
log.info(f"Querying HIBP for {email}")
resp = hibp_get(url)
if resp is None:
log.error("Unable to get a response from HIBP after retries")
return []
# HIBP returns 404 when the email is not found in any breach
if resp.status_code == 404:
log.info("No breaches found for this email.")
return []
try:
data = resp.json()
if not isinstance(data, list):
data = [data]
log.info(f"Found {len(data)} breach(es)")
return data
except json.JSONDecodeError:
log.exception("Failed to decode JSON from HIBP")
return []
# ----------------------------------------------------------------------
# 2️⃣ PHI encryption demo (AWS KMS + local fallback)
# ----------------------------------------------------------------------
def _get_kms_client():
"""Lazy‑load boto3 KMS client only if credentials are present."""
try:
import boto3
return boto3.client("kms", region_name=AWS_REGION)
except Exception as e:
log.debug(f"boto3 not available or mis‑configured: {e}")
return None
def encrypt_phi(plaintext: str) -> str:
"""
Encrypts a UTF‑8 string.
- If KMS is configured & reachable → uses AWS KMS (envelope encryption).
- Else → uses a deterministic local AES‑GCM key derived from a hard‑coded secret
(FOR DEMO ONLY – never use in production).
Returns base64‑encoded ciphertext.
"""
kms = _get_kms_client()
if kms and KMS_KEY_ARN:
log.info("Encrypting with AWS KMS")
# KMS encrypt returns ciphertext blob; we base64‑encode for easy storage
resp = kms.encrypt(KeyId=KMS_KEY_ARN, Plaintext=plaintext.encode("utf-8"))
ciphertext_blob = resp["CiphertextBlob"]
return base64.b64encode(ciphertext_blob).decode("utf-8")
# ---- Fallback (demo) -------------------------------------------------
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
# Derive a 256‑bit key from a static secret – **NOT** safe for prod!
secret = b"ICARAX-DEMO-STATIC-32BYTE-KEY!!" # 32 bytes
aesgcm = AESGCM(secret)
nonce = os.urandom(12) # GCM recommended nonce size
ct = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
# Prepend nonce so we can decrypt later
bundled = nonce + ct
return base64.b64encode(bundled).decode("utf-8")
def decrypt_phi(token_b64: str) -> str:
"""
Decrypts a token produced by `encrypt_phi`.
Raises ValueError if decryption fails.
"""
raw = base64.b64decode(token_b64)
kms = _get_kms_client()
if kms and KMS_KEY_ARN:
log.info("Decrypting with AWS KMS")
plaintext = kms.decrypt(CiphertextBlob=raw)["Plaintext"]
return plaintext.decode("utf-8")
# ---- Fallback (demo) -------------------------------------------------
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
secret = b"ICARAX-DEMO-STATIC-32BYTE-KEY!!"
aesgcm = AESGCM(secret)
nonce, ct = raw[:12], raw[12:]
plaintext = aesgcm.decrypt(nonce, ct, None)
return plaintext.decode("utf-8")
# ----------------------------------------------------------------------
# CLI entry point
# ----------------------------------------------------------------------
def _print_breaches(breaches: List[Dict]):
if not breaches:
print("✅ No known breaches for this email.")
return
print("\n🚨 Breaches found:")
for b in breaches:
name = b.get("Name", "Unknown")
domain = b.get("Domain", "N/A")
breach_date = b.get("BreachDate", "N/A")
pwn_count = b.get("PwnCount", "N/A")
print(f" - {name} ({domain}) – {breach_date} – {pwn_count:,} accounts")
def main():
if len(sys.argv) != 2:
print("Usage: python breach_checker.py <email-address>")
sys.exit(1)
email = sys.argv[1].strip()
breaches = check_breaches(email)
_print_breaches(breaches)
# Demo encryption/decryption of a fake PHI field
sample_phi = f"PatientID={email};SSN=123-45-6789;DOB=1970-01-01"
print("\n🔐 Sample PHI:", sample_phi)
encrypted = encrypt_phi(sample_phi)
print("🔒 Encrypted (base64):", encrypted[:80] + ("..." if len(encrypted) > 80 else ""))
decrypted = decrypt_phi(encrypted)
print("🔓 Decrypted:", decrypted)
assert decrypted == sample_phi, "Encryption round‑trip failed!"
if __name__ == "__main__":
main()
How to run
# Make sure .env is present with your HIBP API key (optional) and KMS ARN
python breach_checker.py user@example.com
Output (truncated for readability):
2025-09-25 12:34:56,789 INFO breach_checker - Querying HIBP for user@example.com
2025-09-25 12:34:57,012 INFO breach_checker - Found 2 breach(es)
🚨 Breaches found:
- Adobe (adobe.com) – 2013-10-04 – 152,445,165 accounts
- LinkedIn (linkedin.com) – 2021-06-20 – 705,449,032 accounts
🔐 Sample PHI: PatientID=user@example.com;SSN=123-45-6789;DOB=1970-01-01
🔒 Encrypted (base64): AQICAHhi...
🔓 Decrypted: PatientID=user@example.com;SSN=123-45-6789;DOB=1970-01-01
breachChecker.ts)#!/usr/bin/env node
/**
* breachChecker.ts
* ----------------
* Node/TS counterpart of the Python utility:
* • Checks HaveIBeenPwned for an email.
* • Demonstrates encryption/decryption of PHI using AWS KMS (with a local fallback).
*
* Requires: Node >=14, axios, dotenv, @aws-sdk/client-kms (optional)
*/
import axios from "axios";
import * as dotenv from "dotenv";
import { readFileSync } from "fs";
import { resolve } from "path";
// Load .env
dotenv.config({ path: resolve(process.cwd(), ".env") });
const HIBP_API_KEY = process.env.HIBP_API_KEY ?? "";
const KMS_KEY_ARN = process.env.KMS_KEY_ARN ?? "";
const AWS_REGION = process.env.AWS_REGION ?? "us-east-1";
const HIBP_BASE = "https://haveibeenpwned.com/api/v3/breachedaccount/";
const HIBP_HEADERS = {
"api-version": "3",
"User-Agent": "ICARAX-BreachChecker/1.0",
...(HIBP_API_KEY ? { "hibp-api-key": HIBP_API_KEY } : {}),
};
// ---------------------------------------------------------------------
// Helper: exponential back‑off axios wrapper
// ---------------------------------------------------------------------
async function hibpGet(url: string, maxRetries = 3): Promise<any> {
for (let attempt = 1; attempt <= maxRetries; attempt++) {
try {
const resp = await axios.get(url, {
headers: HIBP_HEADERS,
timeout: 10_000,
});
return resp.data;
} catch (err: any) {
if (err.response?.status === 429) {
const wait = Number(err.response.headers["retry-after"] ?? "2") * attempt;
console.warn(`[HIBP] Rate limited. Waiting ${wait}s (attempt ${attempt})`);
await new Promise((r) => setTimeout(r, wait * 1000));
continue;
}
if (err.response?.status >= 500 && err.response.status < 600) {
const wait = 2 ** attempt;
console.warn(`[HIBP] Server error ${err.response.status}. Retrying in ${wait}s`);
await new Promise((r) => setTimeout(r, wait * 1000));
continue;
}
// Non‑retryable
if (err.response?.status === 404) {
console.info("[HIBP] No breaches found for this email.");
return []; // treat as empty list
}
console.error(`[HIBP] Request failed (attempt ${attempt}):`, err.message);
if (attempt === maxRetries) throw err;
await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
}
}
throw new Error("Unreachable");
}
// ---------------------------------------------------------------------
// 1️⃣ Breach lookup
// ---------------------------------------------------------------------
export async function checkBreaches(email: string): Promise<any[]> {
const encoded = encodeURIComponent(email);
const url = `${HIBP_BASE}${encoded}`;
console.info(`[HIBP] Querying ${url}`);
try {
const data = await hibpGet(url);
// HIBP returns an array; if it's a single object we wrap it
return Array.isArray(data) ? data : [data];
} catch (e) {
console.error("[HIBP] Error while fetching breaches:", e);
return [];
}
}
// ---------------------------------------------------------------------
// 2️⃣ PHI encryption demo (AWS KMS + local fallback)
// ---------------------------------------------------------------------
let kmsClient: any | null = null;
async function getKmsClient() {
if (kmsClient !== null) return kmsClient;
if (!KMS_KEY_ARN) {
console.warn("[KMS] No KMS_KEY_ARN set – using local demo encryption.");
kmsClient = null;
return null;
}
try {
const { KMSClient, EncryptCommand, DecryptCommand } = await import(
"@aws-sdk/client-kms"
);
kmsClient = new KMSClient({ region: AWS_REGION });
console.info("[KMS] Client initialized");
return kmsClient;
} catch (e) {
console.error("[KMS] Failed to init AWS SDK:", e);
kmsClient = null;
return null;
}
}
/**
* Encrypts a UTF‑8 string.
* Returns base64‑encoded ciphertext.
*/
export async function encryptPhi(plaintext: string): Promise<string> {
const kms = await getKmsClient();
if (kms && KMS_KEY_ARN) {
const { EncryptCommand } = await import("@aws-sdk/client-kms");
const input = {
KeyId: KMS_KEY_ARN,
Plaintext: Buffer.from(plaintext, "utf-8"),
};
const command = new EncryptCommand(input);
const { CiphertextBlob } = await kms.send(command);
return Buffer.from(CiphertextBlob).toString("base64");
}
// ---- Local fallback (demo) ---------------------------------------
// NOTE: This uses a static key – ONLY for demonstration!
const crypto = await import("crypto");
const secret = Buffer.from(
"ICARAX-DEMO-STATIC-32BYTE-KEY!!",
"utf-8"
); // 32 bytes
const iv = crypto.randomBytes(12); // GCM nonce
const cipher = crypto.createCipheriv("aes-256-gcm", secret, iv);
let ciphertext = cipher.update(plaintext, "utf-8");
ciphertext = Buffer.concat([ciphertext, cipher.final()]);
const tag = cipher.getAuthTag();
const bundle = Buffer.concat([iv, tag, ciphertext]); // iv(12) + tag(16) + ct
return bundle.toString("base64");
}
/**
* Decrypts a base64 string produced by encryptPhi.
*/
export async function decryptPhi(tokenB64: string): Promise<string> {
const raw = Buffer.from(tokenB64, "base64");
const kms = await getKmsClient();
if (kms && KMS_KEY_ARN) {
const { DecryptCommand } = await import("@aws-sdk/client-kms");
const input = {
CiphertextBlob: raw,
};
const command = new DecryptCommand(input);
const { Plaintext } = await kms.send(command);
return Plaintext.toString("utf-8");
}
// ---- Local fallback (demo) ---------------------------------------
const crypto = await import("crypto");
const secret = Buffer.from(
"ICARAX-DEMO-STATIC-32BYTE-KEY!!",
"utf-8"
);
const iv = raw.slice(0, 12);
const tag = raw.slice(12, 28);
const ciphertext = raw.slice(28);
const decipher = crypto.createDecipheriv("aes-256-gcm", secret, iv);
decipher.setAuthTag(tag);
let plaintext = decipher.update(ciphertext, undefined, "utf-8");
plaintext += decipher.final("utf-8");
return plaintext;
}
// ---------------------------------------------------------------------
// CLI
// ---------------------------------------------------------------------
async function main() {
const email = process.argv[2];
if (!email) {
console.error("Usage: npx ts-node breachChecker.ts <email-address>");
process.exit(1);
}
const breaches = await checkBreaches(email);
if (breaches.length === 0) {
console.log("✅ No known breaches for this email.");
} else {
console.log("\n🚨 Breaches found:");
breaches.forEach((b) => {
console.log(
` - ${b.Name} (${b.Domain}) – ${b.BreachDate} – ${b.PwnCount.toLocaleString()} accounts`
);
});
}
// Demo PHI encryption/decryption
const samplePhi = `PatientID=${email};SSN=123-45-6789;DOB=1970-01-01`;
console.log("\n🔐 Sample PHI:", samplePhi);
const encrypted = await encryptPhi(samplePhi);
console.log("🔒 Encrypted (base64):", encrypted.substring(0, 80) + "...");
const decrypted = await decryptPhi(encrypted);
console.log("🔓 Decrypted:", decrypted);
if (decrypted !== samplePhi) {
console.error("❌ Encryption round‑trip failed!");
process.exit(1);
} else {
console.log("✅ Encryption round‑trip succeeded.");
}
}
main().catch((err) => {
console.error("❌ Fatal error:", err);
process.exit(1);
});
How to run
# Install deps (once)
npm install
# Run with ts-node (or compile first)
npx ts-node breachChecker.ts user@example.com
You’ll see output analogous to the Python version.
<a name="step-4-configuration"></a>
| Variable | Description | Required? | Example |
|---|---|---|---|
HIBP_API_KEY | Your HaveIBeenPwned API key (higher rate limit). Leave empty for anonymous (10 s delay). | No (optional) | abcdef1234567890 |
KMS_KEY_ARN | ARN or alias of the AWS KMS key used to encrypt/decrypt PHI. | No (fallback to local demo) | arn:aws:kms:us-east-1:123456789012:key/abcd1234-a123-4a12-a12b-a123b4cd56ef |
AWS_REGION | AWS region for the KMS client. | No (defaults to us-east-1) | eu-west-1 |
DEBUG | Set to "true" to enable verbose logging. | No | true |
AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY | Standard AWS credentials (only needed if using real KMS). | Yes when KMS_KEY_ARN is set | AKIA… / wJalrXUtnFEMI/K7MDENG/bPxRfiCY… |
Best practice: Store these in a secret manager (AWS Secrets Manager, HashiCorp Vault, GCP Secret Manager) and inject them as environment variables at runtime—never hard‑code.
<a name="step-5-common-patterns"></a>
Retry-After header when present.Encrypt/Decrypt).pino for Node, logging for Python) with JSON output for easy ingestion into ELK/Datadog.validator.js / email-validator).hibpGet, encryptPhi) in its own module.interface HibpBreach {
Name: string;
Title: string;
Domain: string;
BreachDate: string; // ISO 8601
PwnCount: number;
// …other fields as needed
}
dotenv for local dev; in production rely on the platform’s env var injection (ECS task role, Lambda env, Kubernetes secret, etc.).<a name="step-6-troubleshooting"></a>
| Symptom | Likely Cause | Fix |
|---|---|---|
403 Forbidden from HIBP | Missing or invalid hibp-api-key or you’ve hit the anonymous rate limit (1 req/10 s). | Add a valid API key to .env or increase delay between requests. |
404 Not Found from HIBP | Email not present in any breach (expected). | No action needed – treat as clean. |
Error: Missing credentials in config when using KMS | AWS SDK cannot find credentials. | Ensure AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY are set, or run on an EC2/ECS/EKS instance with an IAM role that has kms:Encrypt and kms:Decrypt permissions on the key. |
InvalidKeyException: Key ARN is not valid | The ARN/alias is malformed or the key is in a different region. | Double‑check KMS_KEY_ARN matches the region in AWS_REGION. Use the full ARN, not just an alias unless the SDK version supports it. |
Error: unable to verify the first certificate (Node) | Corporate proxy or custom CA intercepting TLS. | Set NODE_OPTIONS=--openssl-legacy-provider or add the CA to NODE_EXTRA_CA_CERTS. |
ModuleNotFoundError: No module named 'boto3' (Python) | boto3 not installed (only needed for real KMS). | Install with pip install boto3 or rely on the local demo fallback (remove KMS config). |
| Encryption/decryption mismatch | Using different keys/secrets between encrypt and decrypt calls. | Ensure the same KMS_KEY_ARN (or same local secret) is used for both operations. Verify environment variables are loaded correctly. |
| Program crashes on non‑ASCII email | Email contains characters not UTF‑8 encodable. | Validate/normalize email to ASCII (punycode) before passing to HIBP; HIBP expects the raw email string as‑is, but most IDs are ASCII. |
Debug tip: Set DEBUG=true in .env to see verbose logs (including request URLs, headers, and retry attempts).
<a name="step-7-production-checklist"></a>
| ✅ Item | Why it matters | How to verify |
|---|---|---|
| Transport Security | All API calls (HIBP, KMS) must be over TLS 1.2+. | Use curl -v https://haveibeenpwned.com – look for * TLSv1.2 (OUT), ... |
| Secret Management | API keys and KMS credentials must never be stored in source code. | Scan repo with git-secrets or trivy; confirm .env is in .gitignore. |
| Least‑Privilege IAM | The IAM role/user used for KMS should only have kms:Encrypt and kms:Decrypt on the specific key. | Review IAM policy: "Resource": "arn:aws:kms:…:key/…". |
| Rate‑Limit Compliance | Respect HIBP’s anonymous limit (1 req/10 s) or your paid limit. | Observe logs for 429 responses; ensure back‑off logic triggers. |
| Input Validation | Prevent injection or malformed emails that could cause unexpected behavior. | Use a library like email-validator (Python) or validator.isEmail (TS). |
| Audit Logging | Record who queried which email and when (without storing the raw email if privacy‑sensitive). | Log a hash (SHA‑256) of the email + timestamp to an immutable store. |
| Data Minimization | Only retain the minimum PHI needed for the operation; encrypt at rest. | Verify that any stored PHI is ciphertext; plaintext never touches disk/logs. |
| Error Handling & Monitoring | Uncaught exceptions should not leak stack traces to users. | Centralize error handling; send exceptions to monitoring (Sentry, Datadog). |
| Dependency Hygiene | Keep libraries up‑to‑date to avoid known vulnerabilities. | Run npm audit / pip check regularly; use Dependabot or Renovate. |
| Testing | Unit tests for encryption round‑trip, breach‑lookup mocking, and error paths. | Achieve >80% coverage; run CI on every PR. |
| Versioning & Deployments | Use immutable artifacts (Docker images, Lambda layers) and blue/green or canary releases. | Tag images with git SHA; verify rollback procedure works. |
| Compliance | If handling health data, ensure HIPAA/GDPR relevance (BAA with AWS, data‑subject rights). | Sign AWS BAA; document data‑flow diagram; conduct periodic risk assessment. |
| Load Testing | Verify the service can handle expected query volume without exhausting rate limits. | Use k6 or locust to simulate traffic; monitor latency and error rates. |
| Documentation | Provide a clear README, API spec (OpenAPI/Swagger), and runbook for on‑call engineers. | Keep README.md up‑to‑date; host Swagger UI at /docs. |
When all of the above are satisfied, you can confidently move the breach‑checking utility from a demo script to a production‑grade micro‑service or CLI tool used by your organization’s security operations team.
You now have:
Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
