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

# .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

# 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

# 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));
}