H3l!0s_T3k

Posted on Aug 5

How to Build an Advanced Data Breach Monitoring System with OSINT and Python — LeakRadar

#python #opensource #security #osint

In the fast-paced world of cybersecurity, waiting until a breach happens is no longer an acceptable strategy. Modern organizations rely on Threat Intelligence and Open Source Intelligence (OSINT) to spot risks before they become real security incidents.

One of the most practical applications of this concept is Breach Monitoring — continuously checking whether your organization's email addresses, usernames, or domains have appeared in newly released breach databases.

In this article, we will build a complete breach monitoring system called LeakRadar using Python, with real, tested, runnable code — and reveal the technical secrets behind these systems.

OSINT vs Threat Intelligence
Behind the Scenes: How Breach Monitoring Services Work
What Will the Tool Do?
System Architecture
Technologies Used
Project Structure
Protecting API Secrets
Data Models
The API Layer
Detailed Report: Where Are the APIs and How Do You Change Them?
Data Normalization Layer
Storage Layer
Alerting System
The Scheduler and Main Program
Docker Deployment
GitHub Actions for Scheduled Scans
Tests
Professional Improvements
Security and Ethical Considerations
Development Roadmap
Conclusion
References

OSINT vs Threat Intelligence

Many people confuse these two terms, but the core difference is:

OSINT (Open Source Intelligence) is the process of collecting information from open, publicly available sources, such as:

Breach databases (Have I Been Pwned, DeHashed, BreachDirectory)
Public GitHub repositories
Specialized search engines (Shodan, Censys)
Public forums and the Dark Web
Paste sites like Pastebin
Social networks

Threat Intelligence is the process of analyzing this data and connecting it to a security context that helps make proactive defensive decisions.

In other words:

OSINT provides the raw data
Threat Intelligence turns it into actionable knowledge

The tool we are building relies on OSINT sources to provide a Credential Exposure Monitoring service.

Behind the Scenes: How Breach Monitoring Services Work

Before we start building, let's understand how services like Have I Been Pwned work:

1. Data Collection

Security researchers gather breaches from multiple sources:

Dark Web Markets — markets like BreachForums (formerly RaidForums)
Paste Sites — sites like Pastebin and Ghostbin
Leak Databases — databases leaked from previous breaches
Honeypots — decoy systems for collecting leaked data

2. Data Processing

Raw data is cleaned and normalized:

# Example of how data is processed
def normalize_breach_data(raw_data):
    """
    Convert raw data into a standardized format
    """
    normalized = []
    for entry in raw_data:
        normalized.append({
            'email': entry.get('email', '').lower().strip(),
            'username': entry.get('username', '').lower().strip(),
            'domain': extract_domain(entry.get('email', '')),
            'source': entry.get('source', 'Unknown'),
            'date': parse_date(entry.get('date')),
            'data_types': entry.get('data_types', []),
            'password_hash': entry.get('password_hash'),
            'is_verified': verify_breach(entry)
        })
    return normalized

3. Indexing and Search

Data is stored in databases optimized for fast search:

Elasticsearch for fast full-text search
Redis for caching
PostgreSQL for structured data

4. Application Programming Interfaces (APIs)

Most services provide APIs for querying:

Have I Been Pwned: limited free API + paid for commercial use
DeHashed: paid API with more detailed data
Hunter.io: for email validation
LeakCheck: comprehensive leak API

What Will the Tool Do?

The system will be responsible for:

Scheduled scanning — monitor one or more emails every hour
Smart queries — use trusted APIs like HIBP
Normalization — convert data to a standardized format
Deduplication — ignore previously detected breaches
Instant alerts — send notifications via Telegram, Discord, and email
Logging — save all events in log files
Reports — generate weekly reports

System Architecture

┌─────────────────────────────────────────────────────────────┐
│                      Scheduler (Cron)                       │
│                   Runs every 1 hour                         │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────────┐
│              Breach Intelligence APIs Layer                 │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │     HIBP     │  │   DeHashed   │  │  BreachDir   │      │
│  │     API      │  │     API      │  │     API      │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────────┐
│                   Data Normalizer                           │
│  - Cleans data                                              │
│  - Standardizes formats                                     │
│  - Extracts domains                                         │
│  - Classifies exposed data types                            │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────────┐
│              Duplicate Check & Storage                      │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │   SQLite     │  │    Redis     │  │  PostgreSQL  │      │
│  │   Cache      │  │    Cache     │  │  (Optional)  │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
└──────────────────────┬──────────────────────────────────────┘
                       │
            New Leak?  │
          ┌────────────┴────────────┐
          │                         │
         No                        Yes
          │                         │
          ▼                         ▼
   Wait Next Scan           ┌─────────────────┐
                            │  Alert Engine   │
                            └────────┬────────┘
                                     │
        ┌─────────────────────────────┼─────────────────────────────┐
        ▼                             ▼                             ▼
  ┌───────────┐               ┌──────────────┐             ┌──────────────┐
  │ Telegram  │               │   Discord    │             │    Email     │
  │   Bot     │               │  Webhook     │             │    SMTP      │
  └───────────┘               └──────────────┘             └──────────────┘

Technologies Used

Core:
  - Python 3.11+
  - async/await for asynchronous operations

Libraries:
  - requests (HTTP queries)
  - aiohttp (async queries)
  - python-dotenv (environment variables)
  - pydantic (data validation)
  - loguru (advanced logging)
  - schedule (scheduled tasks)

Storage:
  - SQLite (local data)
  - Redis (caching)

Notifications:
  - Telegram Bot API
  - Discord Webhook
  - SMTP for email

DevOps:
  - Docker
  - GitHub Actions
  - Docker Compose

Project Structure

leakradar/
├── src/
│   ├── __init__.py
│   ├── config.py              # System configuration
│   ├── models.py              # Data models (Breach, BreachCheckResult)
│   ├── breach_api.py          # HIBP / DeHashed API clients
│   ├── async_breach_api.py    # Async API client (optional)
│   ├── normalizer.py          # Data normalization & risk scoring
│   ├── storage.py             # SQLite storage layer
│   ├── cache.py               # Redis caching layer (optional)
│   ├── alerting.py            # Telegram / Discord / Email alerts
│   ├── scheduler.py           # Scheduling & orchestration
│   ├── security.py            # Ownership validation & audit logging
│   └── main.py                # Entry point
├── tests/
│   ├── test_breach_api.py
│   ├── test_normalizer.py
│   └── test_storage.py
├── data/
│   ├── leaks.db               # SQLite database
│   └── logs/                  # Log files
├── .env.example               # Example environment file
├── .gitignore
├── requirements.txt
├── requirements-dev.txt
├── Dockerfile
├── docker-compose.yml
├── .github/
│   └── workflows/
│       └── scheduled-scan.yml
└── README.md

Protecting API Secrets

One of the most common mistakes is pushing API keys to GitHub. Use a .env file:

.env.example:

# Email(s) to monitor (comma-separated)
TARGET_EMAILS=user1@company.com,user2@company.com

# Have I Been Pwned API
HIBP_API_KEY=your_hibp_api_key_here
HIBP_API_URL=https://haveibeenpwned.com/api/v3

# Telegram Bot
TELEGRAM_BOT_TOKEN=your_bot_token
TELEGRAM_CHAT_ID=your_chat_id

# Discord Webhook
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...

# Email SMTP
SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your_email@gmail.com
SMTP_PASSWORD=your_app_password
SMTP_FROM=your_email@gmail.com
SMTP_TO=admin@company.com

# Database
DATABASE_URL=sqlite:///data/leaks.db

# Logging
LOG_LEVEL=INFO
LOG_FILE=data/logs/leakradar.log

# Scan Settings
SCAN_INTERVAL_HOURS=1
MAX_RETRIES=3
REQUEST_TIMEOUT=30

Code that loads the variables:

# src/config.py
"""Application configuration loaded from environment variables."""
import os
from typing import List

from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()


class Config(BaseModel):
    """Configuration model with basic validation."""

    # Targets
    target_emails: List[str]

    # HIBP
    hibp_api_key: str
    hibp_api_url: str = "https://haveibeenpwned.com/api/v3"

    # Telegram
    telegram_bot_token: str | None = None
    telegram_chat_id: str | None = None

    # Discord
    discord_webhook_url: str | None = None

    # SMTP
    smtp_server: str | None = None
    smtp_port: int = 587
    smtp_username: str | None = None
    smtp_password: str | None = None
    smtp_from: str | None = None
    smtp_to: str | None = None

    # Database
    database_url: str = "sqlite:///data/leaks.db"

    # Logging
    log_level: str = "INFO"
    log_file: str = "data/logs/leakradar.log"

    # Scan settings
    scan_interval_hours: int = 1
    max_retries: int = 3
    request_timeout: int = 30


def _as_int(value: str | None, default: int) -> int:
    try:
        return int(value) if value else default
    except ValueError:
        return default


def load_config() -> Config:
    """Load the configuration from environment variables."""
    return Config(
        target_emails=[
            email.strip()
            for email in os.getenv("TARGET_EMAILS", "").split(",")
            if email.strip()
        ],
        hibp_api_key=os.getenv("HIBP_API_KEY", ""),
        hibp_api_url=os.getenv(
            "HIBP_API_URL", "https://haveibeenpwned.com/api/v3"
        ),
        telegram_bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
        telegram_chat_id=os.getenv("TELEGRAM_CHAT_ID"),
        discord_webhook_url=os.getenv("DISCORD_WEBHOOK_URL"),
        smtp_server=os.getenv("SMTP_SERVER"),
        smtp_port=_as_int(os.getenv("SMTP_PORT"), 587),
        smtp_username=os.getenv("SMTP_USERNAME"),
        smtp_password=os.getenv("SMTP_PASSWORD"),
        smtp_from=os.getenv("SMTP_FROM"),
        smtp_to=os.getenv("SMTP_TO"),
        database_url=os.getenv("DATABASE_URL", "sqlite:///data/leaks.db"),
        log_level=os.getenv("LOG_LEVEL", "INFO"),
        log_file=os.getenv("LOG_FILE", "data/logs/leakradar.log"),
        scan_interval_hours=_as_int(os.getenv("SCAN_INTERVAL_HOURS"), 1),
        max_retries=_as_int(os.getenv("MAX_RETRIES"), 3),
        request_timeout=_as_int(os.getenv("REQUEST_TIMEOUT"), 30),
    )


config = load_config()

Data Models

# src/models.py
"""Pydantic data models for LeakRadar."""
from datetime import datetime
from enum import Enum
from typing import List, Optional

from pydantic import BaseModel, Field


class BreachSeverity(str, Enum):
    """Severity classification for a breach."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class DataType(str, Enum):
    """Types of exposed data."""

    EMAIL = "email"
    PASSWORD = "password"
    USERNAME = "username"
    PHONE = "phone"
    ADDRESS = "address"
    IP_ADDRESS = "ip_address"
    CREDIT_CARD = "credit_card"
    SSN = "ssn"
    OTHER = "other"


class Breach(BaseModel):
    """A single data breach record."""

    name: str = Field(..., description="Breach identifier")
    title: str = Field(..., description="Breach display title")
    domain: str = Field(..., description="Affected domain")
    breach_date: datetime = Field(..., description="When the breach occurred")
    added_date: datetime = Field(..., description="When the breach was indexed")
    modified_date: Optional[datetime] = Field(None, description="Last modification date")
    pwn_count: int = Field(..., description="Number of affected accounts")
    description: str = Field(..., description="Breach description")
    logo_path: Optional[str] = Field(None, description="Logo URL")
    data_classes: List[str] = Field(..., description="Exposed data types")
    is_verified: bool = Field(..., description="Is the breach verified?")
    is_fabricated: bool = Field(False, description="Is the breach fabricated?")
    is_sensitive: bool = Field(False, description="Is the breach sensitive?")
    is_retired: bool = Field(False, description="Is the breach retired?")
    is_spam_list: bool = Field(False, description="Is it a spam list?")

    @property
    def severity(self) -> BreachSeverity:
        """Compute severity based on exposed data types."""
        sensitive_types = {"password", "credit card", "ssn", "phone", "address"}
        data_lower = [d.lower() for d in self.data_classes]

        def contains_any(keywords) -> bool:
            return any(kw in text for text in data_lower for kw in keywords)

        if contains_any(["password", "credit card", "ssn"]):
            return BreachSeverity.CRITICAL
        if contains_any(sensitive_types):
            return BreachSeverity.HIGH
        if len(data_lower) > 3:
            return BreachSeverity.MEDIUM
        return BreachSeverity.LOW

    def to_dict(self) -> dict:
        """Convert the breach to a dictionary."""
        return self.model_dump()


class BreachCheckResult(BaseModel):
    """The result of checking a single email address."""

    email: str
    breaches: List[Breach]
    check_date: datetime = Field(default_factory=datetime.now)
    is_new: bool = Field(..., description="Are there new breaches?")

    @property
    def breach_count(self) -> int:
        return len(self.breaches)

Important note: When classifying severity we use substring matching (kw in text) and not exact list matching. The reason is that HIBP returns data types like "Passwords" or "Credit cards", and without substring matching the system would fail to detect passwords and classify every breach as low severity.

The API Layer

# src/breach_api.py
"""API client layer for breach intelligence sources."""
import time
from datetime import datetime
from typing import List, Optional

import requests
from loguru import logger

from src.config import config
from src.models import Breach


class BreachAPIError(Exception):
    """Custom exception for API errors."""

    pass


class HaveIBeenPwnedAPI:
    """Client for the Have I Been Pwned (HIBP) v3 API."""

    def __init__(self, api_key: str, base_url: str = "https://haveibeenpwned.com/api/v3"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update(
            {
                "hibp-api-key": api_key,
                "user-agent": "LeakRadar/1.0",
                "Content-Type": "application/json",
            }
        )

    def _make_request(self, endpoint: str, params: Optional[dict] = None) -> dict | list:
        """Perform an HTTP request with retries and error handling."""
        url = f"{self.base_url}/{endpoint}"

        for attempt in range(config.max_retries):
            try:
                logger.debug(f"Request attempt {attempt + 1} to {url}")

                response = self.session.get(
                    url, params=params, timeout=config.request_timeout
                )

                if response.status_code == 200:
                    return response.json()
                if response.status_code == 404:
                    return []
                if response.status_code == 401:
                    raise BreachAPIError("Invalid API key")
                if response.status_code == 403:
                    raise BreachAPIError("Access forbidden - check your API key")
                if response.status_code == 429:
                    wait_time = int(response.headers.get("Retry-After", 60))
                    logger.warning(f"Rate limit reached, waiting {wait_time}s")
                    time.sleep(wait_time)
                    continue
                if response.status_code == 503:
                    logger.warning(f"Service unavailable, attempt {attempt + 1}")
                    time.sleep(2**attempt)
                    continue
                raise BreachAPIError(f"HTTP {response.status_code}: {response.text}")

            except requests.exceptions.Timeout:
                logger.error(f"Request timeout on attempt {attempt + 1}")
                if attempt == config.max_retries - 1:
                    raise BreachAPIError("Request timeout after all retries")
                time.sleep(2**attempt)
            except requests.exceptions.ConnectionError:
                logger.error(f"Connection error on attempt {attempt + 1}")
                if attempt == config.max_retries - 1:
                    raise BreachAPIError("Connection failed after all retries")
                time.sleep(2**attempt)
            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                if attempt == config.max_retries - 1:
                    raise BreachAPIError(f"Unexpected error: {e}")
                time.sleep(2**attempt)

        raise BreachAPIError("Failed after all retries")

    @staticmethod
    def _to_breach(item: dict) -> Breach:
        def _parse(value: str | None) -> datetime:
            if not value:
                return datetime(2000, 1, 1)
            return datetime.fromisoformat(value.replace("Z", "+00:00"))

        return Breach(
            name=item.get("Name", ""),
            title=item.get("Title", ""),
            domain=item.get("Domain", ""),
            breach_date=_parse(item.get("BreachDate")),
            added_date=_parse(item.get("AddedDate")),
            modified_date=_parse(item.get("ModifiedDate")) if item.get("ModifiedDate") else None,
            pwn_count=item.get("PwnCount", 0),
            description=item.get("Description", ""),
            logo_path=item.get("LogoPath"),
            data_classes=item.get("DataClasses", []),
            is_verified=item.get("IsVerified", False),
            is_fabricated=item.get("IsFabricated", False),
            is_sensitive=item.get("IsSensitive", False),
            is_retired=item.get("IsRetired", False),
            is_spam_list=item.get("IsSpamList", False),
        )

    def get_breached_account(self, email: str) -> List[Breach]:
        """Check whether an email address has been involved in a breach."""
        logger.info(f"Checking breaches for email: {email}")

        endpoint = f"breachedaccount/{email}"
        params = {"truncateResponse": "false", "includeUnverified": "false"}
        response_data = self._make_request(endpoint, params)

        if not response_data:
            logger.info(f"No breaches found for {email}")
            return []

        breaches = [self._to_breach(item) for item in response_data]
        logger.info(f"Found {len(breaches)} breaches for {email}")
        return breaches

    def get_all_breaches(self) -> List[Breach]:
        """Fetch a list of all known breaches."""
        logger.info("Fetching all breaches")

        response_data = self._make_request("breaches")
        breaches = [self._to_breach(item) for item in response_data]

        logger.info(f"Fetched {len(breaches)} breaches")
        return breaches

    def get_pastes_for_email(self, email: str) -> List[dict]:
        """Search for pastes that contain the given email address."""
        logger.info(f"Checking pastes for email: {email}")

        response_data = self._make_request(f"pasteaccount/{email}")

        if not response_data:
            logger.info(f"No pastes found for {email}")
            return []

        logger.info(f"Found {len(response_data)} pastes for {email}")
        return response_data


class DeHashedAPI:
    """Client for the DeHashed API (optional)."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.dehashed.com"
        self.session = requests.Session()
        self.session.headers.update(
            {"Accept": "application/json", "Authorization": f"Basic {api_key}"}
        )

    def search(self, query: str) -> dict:
        """Search the DeHashed database."""
        logger.info(f"Searching DeHashed for: {query}")

        params = {"query": query, "page": 1, "size": 100}

        try:
            response = self.session.get(
                f"{self.base_url}/search",
                params=params,
                timeout=config.request_timeout,
            )

            if response.status_code == 200:
                return response.json()
            raise BreachAPIError(f"HTTP {response.status_code}: {response.text}")

        except Exception as e:
            logger.error(f"DeHashed API error: {e}")
            raise

Detailed Report: Where Are the APIs and How Do You Change Them?

Many developers search the code for the API endpoints and cannot find them because they are spread across multiple layers. Here is the complete report.

1. Map of API Locations in the Project

#	Component	Variable / Place	File	Line	Current Value
1	HIBP key	`HIBP_API_KEY`	`.env`	—	your personal key
2	HIBP URL	`HIBP_API_URL`	`.env`	7	`https://haveibeenpwned.com/api/v3`
3	HIBP default URL	`hibp_api_url`	`src/config.py`	19	`https://haveibeenpwned.com/api/v3`
4	Key header in request	`hibp-api-key`	`src/breach_api.py`	28	read from `config`
5	User-Agent	`user-agent`	`src/breach_api.py`	29	`LeakRadar/1.0`
6	HIBP endpoints	`breachedaccount/{email}` etc.	`src/breach_api.py`	112, 128, 138	endpoints
7	DeHashed URL	`self.base_url`	`src/breach_api.py`	153	`https://api.dehashed.com`
8	DeHashed key	`Authorization: Basic`	`src/breach_api.py`	156	passed in constructor
9	Async URL	`self.base_url`	`src/async_breach_api.py`	16	`https://haveibeenpwned.com/api/v3`
10	Request timeout	`request_timeout`	`.env` / `src/config.py`	45	30 seconds
11	Retries	`max_retries`	`.env` / `src/config.py`	44	3 attempts

2. The Golden Rule

Keys and secrets → only in .env (never put them in the code).
URLs you want to change without redeploying → in .env via HIBP_API_URL.
Hardcoded URLs → inside the code, they require an edit and a restart:
- DeHashed URL in breach_api.py
- Async URL in async_breach_api.py

3. How to Change the HIBP Key / URL?

Do not touch the code at all, just edit .env:

# Before
HIBP_API_KEY=old_key
HIBP_API_URL=https://haveibeenpwned.com/api/v3

# After
HIBP_API_KEY=new_key
HIBP_API_URL=https://haveibeenpwned.com/api/v3/   # or another URL

Then restart the program. config.py reads from the environment on every run:

hibp_api_key=os.getenv("HIBP_API_KEY", ""),
hibp_api_url=os.getenv("HIBP_API_URL", "https://haveibeenpwned.com/api/v3"),

4. How to Change the DeHashed URL? (Hardcoded)

Currently the DeHashed URL is hardcoded, so to change it edit src/breach_api.py:

class DeHashedAPI:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.dehashed.com"   # change this line

Better: make it configurable from .env like the rest of the settings:

# in src/breach_api.py
def __init__(self, api_key: str, base_url: str = "https://api.dehashed.com"):
    self.api_key = api_key
    self.base_url = base_url.rstrip("/")

And add to .env and config.py:

# .env
DEHASHED_API_URL=https://api.dehashed.com

# config.py
dehashed_api_url: str = "https://api.dehashed.com"

5. HIBP Endpoints and Their Location

Each function has its own endpoint inside HaveIBeenPwnedAPI in src/breach_api.py:

Function	Endpoint	Usage
`get_breached_account`	`breachedaccount/{email}`	Is the email in a breach?
`get_all_breaches`	`breaches`	All known breaches
`get_pastes_for_email`	`pasteaccount/{email}`	Pastes containing the email

To change an endpoint (e.g., after an API update) edit the line inside the function:

endpoint = f"breachedaccount/{email}"   # change it here

6. How to Add a New API Provider Step by Step?

We'll add LeakCheckAPI as an example:

Step 1 — Add the class in src/breach_api.py (following the DeHashedAPI pattern):

class LeakCheckAPI:
    """Client for the LeakCheck API (example)."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.leakcheck.io"
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": api_key})

    def search(self, email: str) -> dict:
        """Search LeakCheck for an email."""
        logger.info(f"Searching LeakCheck for: {email}")
        try:
            response = self.session.get(
                f"{self.base_url}/v2/search/email/{email}",
                timeout=config.request_timeout,
            )
            if response.status_code == 200:
                return response.json()
            raise BreachAPIError(f"HTTP {response.status_code}: {response.text}")
        except Exception as e:
            logger.error(f"LeakCheck API error: {e}")
            raise

Step 2 — Add the key to .env and .env.example:

LEAKCHECK_API_KEY=your_key_here

Step 3 — Add the field to src/config.py:

leakcheck_api_key: str | None = None

And define it in load_config():

leakcheck_api_key=os.getenv("LEAKCHECK_API_KEY"),

Step 4 — Use it in src/scheduler.py inside scan_email:

def __init__(self):
    self.api = HaveIBeenPwnedAPI(config.hibp_api_key, config.hibp_api_url)
    self.storage = BreachStorage(config.database_url)
    self.alerter = AlertEngine()
    self.leakcheck = LeakCheckAPI(config.leakcheck_api_key) if config.leakcheck_api_key else None

def scan_email(self, email: str) -> None:
    ...
    if self.leakcheck:
        extra = self.leakcheck.search(email)   # additional query
        logger.info(f"LeakCheck results for {email}: {len(extra)}")

Step 5 — Add from src.breach_api import LeakCheckAPI at the top of scheduler.py.

7. FAQ

Q: Why isn't the key in the code?
A: Because pushing keys to GitHub means an immediate leak. .env is excluded in .gitignore and never uploaded.

Q: I changed the API URL and it didn't work?
A: Make sure you edited .env and not .env.example, then restart the program. If the provider is DeHashed, you edited the hardcoded line in breach_api.py.

Q: How do I know the key works?
A: Run a manual scan on your email and watch the log: if you see 401 the key is wrong; if you see 200 it works.

Q: What about Rate Limiting?
A: The system automatically waits for Retry-After on 429, and you can increase REQUEST_TIMEOUT and MAX_RETRIES from .env.

Q: Can I use multiple providers together?
A: Yes, the system is designed with an extensible API layer — add each provider as a class in breach_api.py and call it in scan_email.

Data Normalization Layer

# src/normalizer.py
"""Data normalization and enrichment utilities."""
from datetime import datetime
from typing import List

from loguru import logger

from src.models import Breach


class DataNormalizer:
    """Utilities to normalize and enrich breach data."""

    @staticmethod
    def normalize_email(email: str) -> str:
        """Normalize an email address."""
        return email.lower().strip()

    @staticmethod
    def extract_domain(email: str) -> str:
        """Extract the domain from an email address."""
        try:
            return email.split("@")[1].lower()
        except (IndexError, AttributeError):
            return ""

    @staticmethod
    def classify_data_types(data_classes: List[str]) -> dict:
        """Classify exposed data types into categories."""
        categories = {
            "credentials": [],
            "personal_info": [],
            "financial": [],
            "other": [],
        }

        credential_keywords = ["password", "username", "email"]
        personal_keywords = ["name", "phone", "address", "date of birth", "ssn"]
        financial_keywords = ["credit card", "bank account", "ip address"]

        for data_type in data_classes:
            data_lower = data_type.lower()

            if any(keyword in data_lower for keyword in credential_keywords):
                categories["credentials"].append(data_type)
            elif any(keyword in data_lower for keyword in personal_keywords):
                categories["personal_info"].append(data_type)
            elif any(keyword in data_lower for keyword in financial_keywords):
                categories["financial"].append(data_type)
            else:
                categories["other"].append(data_type)

        return categories

    @staticmethod
    def calculate_risk_score(breach: Breach) -> int:
        """Compute a 0-100 risk score for a breach."""
        score = 0
        data_lower = [d.lower() for d in breach.data_classes]

        def contains_any(keywords) -> bool:
            return any(kw in text for text in data_lower for kw in keywords)

        if contains_any(["password"]):
            score += 40
        if contains_any(["credit card"]):
            score += 30
        if contains_any(["ssn", "social security"]):
            score += 30
        if contains_any(["phone"]):
            score += 10
        if contains_any(["address"]):
            score += 10

        if breach.pwn_count > 1_000_000:
            score += 20
        elif breach.pwn_count > 100_000:
            score += 10
        elif breach.pwn_count > 10_000:
            score += 5

        if breach.is_verified:
            score += 10

        days_old = (datetime.now() - breach.breach_date).days
        if days_old > 365 * 2:
            score -= 20
        elif days_old > 365:
            score -= 10

        return max(0, min(100, score))

    @staticmethod
    def generate_summary(breaches: List[Breach]) -> dict:
        """Generate a summary for a list of breaches."""
        if not breaches:
            return {
                "total_breaches": 0,
                "critical_breaches": 0,
                "high_breaches": 0,
                "medium_breaches": 0,
                "low_breaches": 0,
                "total_accounts_affected": 0,
                "data_types_found": [],
            }

        summary = {
            "total_breaches": len(breaches),
            "critical_breaches": sum(1 for b in breaches if b.severity.value == "critical"),
            "high_breaches": sum(1 for b in breaches if b.severity.value == "high"),
            "medium_breaches": sum(1 for b in breaches if b.severity.value == "medium"),
            "low_breaches": sum(1 for b in breaches if b.severity.value == "low"),
            "total_accounts_affected": sum(b.pwn_count for b in breaches),
            "data_types_found": [],
        }

        data_types: set = set()
        for breach in breaches:
            data_types.update(breach.data_classes)
        summary["data_types_found"] = sorted(data_types)

        logger.debug(f"Generated summary: {summary}")
        return summary

Storage Layer

# src/storage.py
"""SQLite storage layer for breach data."""
import json
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timedelta
from pathlib import Path
from typing import Iterator, List

from loguru import logger

from src.config import config
from src.models import Breach, BreachCheckResult
from src.normalizer import DataNormalizer


@contextmanager
def _db_connection(db_path: str) -> Iterator[sqlite3.Connection]:
    """Open a SQLite connection and guarantee it is closed afterwards."""
    conn = sqlite3.connect(db_path)
    try:
        with conn:
            yield conn
    finally:
        conn.close()


class BreachStorage:
    """Manage persistence of breach data in SQLite."""

    def __init__(self, db_url: str = "sqlite:///data/leaks.db"):
        self.db_path = db_url.replace("sqlite:///", "")
        self._ensure_directory()
        self._init_database()

    def _ensure_directory(self) -> None:
        db_file = Path(self.db_path)
        db_file.parent.mkdir(parents=True, exist_ok=True)

        log_dir = Path(config.log_file).parent
        log_dir.mkdir(parents=True, exist_ok=True)

    def _init_database(self) -> None:
        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()

            cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS breaches (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    email TEXT NOT NULL,
                    breach_name TEXT NOT NULL,
                    breach_title TEXT NOT NULL,
                    domain TEXT NOT NULL,
                    breach_date TEXT NOT NULL,
                    added_date TEXT NOT NULL,
                    pwn_count INTEGER NOT NULL,
                    description TEXT,
                    data_classes TEXT,
                    is_verified BOOLEAN,
                    severity TEXT,
                    risk_score INTEGER,
                    first_seen TEXT NOT NULL,
                    last_seen TEXT NOT NULL,
                    UNIQUE(email, breach_name)
                )
                """
            )

            cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS check_results (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    email TEXT NOT NULL,
                    check_date TEXT NOT NULL,
                    breach_count INTEGER NOT NULL,
                    new_breaches INTEGER NOT NULL,
                    summary TEXT
                )
                """
            )

            cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS alerts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    email TEXT NOT NULL,
                    breach_name TEXT NOT NULL,
                    alert_date TEXT NOT NULL,
                    alert_type TEXT NOT NULL,
                    sent BOOLEAN DEFAULT 0,
                    sent_date TEXT
                )
                """
            )

            conn.commit()
            logger.info(f"Database initialized at {self.db_path}")

    def save_breaches(self, email: str, breaches: List[Breach]) -> List[Breach]:
        """Persist breaches and return only the newly detected ones."""
        new_breaches: List[Breach] = []

        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()

            for breach in breaches:
                try:
                    cursor.execute(
                        """
                        SELECT id FROM breaches
                        WHERE email = ? AND breach_name = ?
                        """,
                        (email, breach.name),
                    )

                    if cursor.fetchone():
                        cursor.execute(
                            """
                            UPDATE breaches
                            SET last_seen = ?
                            WHERE email = ? AND breach_name = ?
                            """,
                            (datetime.now().isoformat(), email, breach.name),
                        )
                        continue

                    cursor.execute(
                        """
                        INSERT INTO breaches
                        (email, breach_name, breach_title, domain, breach_date,
                         added_date, pwn_count, description, data_classes,
                         is_verified, severity, risk_score, first_seen, last_seen)
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                        (
                            email,
                            breach.name,
                            breach.title,
                            breach.domain,
                            breach.breach_date.isoformat(),
                            breach.added_date.isoformat(),
                            breach.pwn_count,
                            breach.description,
                            json.dumps(breach.data_classes),
                            breach.is_verified,
                            breach.severity.value,
                            DataNormalizer.calculate_risk_score(breach),
                            datetime.now().isoformat(),
                            datetime.now().isoformat(),
                        ),
                    )

                    new_breaches.append(breach)
                    logger.info(f"New breach saved: {breach.name} for {email}")

                except Exception as e:
                    logger.error(f"Error saving breach {breach.name}: {e}")

            conn.commit()

        return new_breaches

    def save_check_result(self, result: BreachCheckResult) -> None:
        """Persist a check result."""
        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            summary = DataNormalizer.generate_summary(result.breaches)

            cursor.execute(
                """
                INSERT INTO check_results
                (email, check_date, breach_count, new_breaches, summary)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    result.email,
                    result.check_date.isoformat(),
                    result.breach_count,
                    len([b for b in result.breaches if result.is_new]),
                    json.dumps(summary),
                ),
            )

            conn.commit()

    def get_all_breaches_for_email(self, email: str) -> List[Breach]:
        """Retrieve all breaches recorded for an email address."""
        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM breaches WHERE email = ? ORDER BY first_seen DESC",
                (email,),
            )
            rows = cursor.fetchall()

        breaches: List[Breach] = []
        for row in rows:
            breaches.append(
                Breach(
                    name=row[2],
                    title=row[3],
                    domain=row[4],
                    breach_date=datetime.fromisoformat(row[5]),
                    added_date=datetime.fromisoformat(row[6]),
                    pwn_count=row[7],
                    description=row[8] or "",
                    data_classes=json.loads(row[9]),
                    is_verified=bool(row[10]),
                )
            )

        return breaches

    def get_recent_breaches(self, days: int = 7) -> List[tuple]:
        """Retrieve breaches first seen within the last `days` days."""
        cutoff_date = datetime.now() - timedelta(days=days)

        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                """
                SELECT * FROM breaches
                WHERE first_seen >= ?
                ORDER BY first_seen DESC
                """,
                (cutoff_date.isoformat(),),
            )
            return cursor.fetchall()

    def mark_alert_sent(self, email: str, breach_name: str, alert_type: str) -> None:
        """Record that an alert has been sent."""
        with _db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                """
                INSERT INTO alerts
                (email, breach_name, alert_date, alert_type, sent, sent_date)
                VALUES (?, ?, ?, ?, 1, ?)
                """,
                (
                    email,
                    breach_name,
                    datetime.now().isoformat(),
                    alert_type,
                    datetime.now().isoformat(),
                ),
            )
            conn.commit()

Important note: We used _db_connection instead of with sqlite3.connect(...) directly. In Python, the with sqlite3.connect(...) statement does not close the connection (it only commits/rolls back), which leaks connections and keeps the database file locked on Windows, causing PermissionError when trying to delete or copy it.

Alerting System

# src/alerting.py
"""Multi-channel alerting engine (Telegram, Discord, Email)."""
import smtplib
from datetime import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import List

import requests
from loguru import logger

from src.config import config
from src.models import Breach, BreachCheckResult
from src.normalizer import DataNormalizer

SEVERITY_EMOJI = {
    "critical": "\U0001F534",
    "high": "\U0001F7E0",
    "medium": "\U0001F7E1",
    "low": "\U0001F7E2",
}


class AlertEngine:
    """Send breach alerts across multiple channels."""

    def __init__(self):
        self.telegram_enabled = bool(config.telegram_bot_token and config.telegram_chat_id)
        self.discord_enabled = bool(config.discord_webhook_url)
        self.email_enabled = bool(config.smtp_server and config.smtp_username)

    def send_alerts(self, result: BreachCheckResult) -> None:
        """Send alerts via all configured channels."""
        if not result.is_new:
            logger.info("No new breaches, skipping alerts")
            return

        breaches = result.breaches
        logger.info(f"Sending alerts for {len(breaches)} new breaches")

        if self.telegram_enabled:
            try:
                self._send_telegram_alert(result.email, breaches)
                logger.info("Telegram alert sent successfully")
            except Exception as e:
                logger.error(f"Failed to send Telegram alert: {e}")

        if self.discord_enabled:
            try:
                self._send_discord_alert(result.email, breaches)
                logger.info("Discord alert sent successfully")
            except Exception as e:
                logger.error(f"Failed to send Discord alert: {e}")

        if self.email_enabled:
            try:
                self._send_email_alert(result.email, breaches)
                logger.info("Email alert sent successfully")
            except Exception as e:
                logger.error(f"Failed to send email alert: {e}")

    def _build_summary_lines(self, breaches: List[Breach]) -> str:
        summary = DataNormalizer.generate_summary(breaches)
        return (
            f"- Total breaches: {summary['total_breaches']}\n"
            f"- Critical: {summary['critical_breaches']}\n"
            f"- High: {summary['high_breaches']}\n"
            f"- Medium: {summary['medium_breaches']}\n"
            f"- Accounts affected: {summary['total_accounts_affected']:,}"
        )

    def _send_telegram_alert(self, email: str, breaches: List[Breach]) -> None:
        """Send an alert via the Telegram Bot API."""
        url = f"https://api.telegram.org/bot{config.telegram_bot_token}/sendMessage"

        message = "\U0001F6A8 *New Data Breach Alert!*\n\n"
        message += f"\U0001F4E7 *Email:* `{email}`\n"
        message += f"\U0001F4C5 *Date:* {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
        message += f"\U0001F4CA *Summary:*\n{self._build_summary_lines(breaches)}\n\n"
        message += "\U0001F513 *New breaches:*\n"

        for breach in breaches[:5]:
            emoji = SEVERITY_EMOJI.get(breach.severity.value, "\u26AA")
            message += (
                f"\n{emoji} *{breach.title}*\n"
                f"  Domain: {breach.domain}\n"
                f"  Date: {breach.breach_date.strftime('%Y-%m-%d')}\n"
                f"  Accounts: {breach.pwn_count:,}\n"
                f"  Data: {', '.join(breach.data_classes[:3])}\n"
            )

        if len(breaches) > 5:
            message += f"\n... and {len(breaches) - 5} more breaches\n"

        message += "\n\U000026A0 *Recommended actions:*\n"
        message += "- Change passwords immediately\n"
        message += "- Enable two-factor authentication\n"
        message += "- Monitor financial accounts\n"
        message += "- Watch for phishing emails\n"

        payload = {"chat_id": config.telegram_chat_id, "text": message, "parse_mode": "Markdown"}
        response = requests.post(url, json=payload, timeout=10)

        if response.status_code != 200:
            raise Exception(f"Telegram API error: {response.text}")

    def _send_discord_alert(self, email: str, breaches: List[Breach]) -> None:
        """Send an alert via a Discord webhook."""
        summary = DataNormalizer.generate_summary(breaches)

        embed = {
            "title": "\U0001F6A8 New Data Breach Alert!",
            "description": f"New breaches detected for email: `{email}`",
            "color": 15158332,
            "fields": [
                {
                    "name": "\U0001F4CA Summary",
                    "value": (
                        f"**Total breaches:** {summary['total_breaches']}\n"
                        f"**Critical:** {summary['critical_breaches']}\n"
                        f"**High:** {summary['high_breaches']}"
                    ),
                    "inline": False,
                }
            ],
            "timestamp": datetime.now().isoformat(),
            "footer": {"text": "LeakRadar"},
        }

        for breach in breaches[:5]:
            embed["fields"].append(
                {
                    "name": f"\U0001F513 {breach.title}",
                    "value": (
                        f"**Domain:** {breach.domain}\n"
                        f"**Date:** {breach.breach_date.strftime('%Y-%m-%d')}\n"
                        f"**Accounts:** {breach.pwn_count:,}"
                    ),
                    "inline": False,
                }
            )

        payload = {"username": "LeakRadar", "embeds": [embed]}
        response = requests.post(config.discord_webhook_url, json=payload, timeout=10)

        if response.status_code not in [200, 204]:
            raise Exception(f"Discord webhook error: {response.text}")

    def _send_email_alert(self, email: str, breaches: List[Breach]) -> None:
        """Send an alert via SMTP email."""
        summary = DataNormalizer.generate_summary(breaches)

        html = f"""<html>
        <body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333;">
            <div style="max-width: 600px; margin: 0 auto; padding: 20px;">
                <h1 style="color: #e74c3c;">\U0001F6A8 New Data Breach Alert!</h1>
                <p>New breaches detected for email: <strong>{email}</strong></p>
                <p>Date: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
                <h2 style="color: #3498db;">\U0001F4CA Summary</h2>
                <ul>
                    <li><strong>Total breaches:</strong> {summary['total_breaches']}</li>
                    <li><strong>Critical:</strong> {summary['critical_breaches']}</li>
                    <li><strong>High:</strong> {summary['high_breaches']}</li>
                    <li><strong>Medium:</strong> {summary['medium_breaches']}</li>
                    <li><strong>Accounts affected:</strong> {summary['total_accounts_affected']:,}</li>
                </ul>
                <h2 style="color: #3498db;">\U0001F513 New Breaches</h2>"""

        for breach in breaches[:5]:
            html += f"""
                <div style="border-left: 4px solid #e74c3c; padding-left: 15px; margin: 15px 0;">
                    <h3 style="color: #e74c3c;">{breach.title}</h3>
                    <p><strong>Domain:</strong> {breach.domain}</p>
                    <p><strong>Date:</strong> {breach.breach_date.strftime('%Y-%m-%d')}</p>
                    <p><strong>Accounts affected:</strong> {breach.pwn_count:,}</p>
                    <p><strong>Exposed data:</strong> {', '.join(breach.data_classes)}</p>
                </div>"""

        html += """
                <h2 style="color: #3498db;">\u26A0 Recommended Actions</h2>
                <ul>
                    <li>Change passwords immediately</li>
                    <li>Enable two-factor authentication (2FA)</li>
                    <li>Monitor financial accounts</li>
                    <li>Watch for phishing emails</li>
                    <li>Use a password manager</li>
                </ul>
                <hr style="margin: 30px 0;">
                <p style="color: #7f8c8d; font-size: 12px;">
                    This email was sent automatically by LeakRadar.
                </p>
            </div>
        </body>
        </html>"""

        msg = MIMEMultipart("alternative")
        msg["Subject"] = f"\U0001F6A8 Data Breach Alert - {email}"
        msg["From"] = config.smtp_from
        msg["To"] = config.smtp_to
        msg.attach(MIMEText(html, "html"))

        try:
            server = smtplib.SMTP(config.smtp_server, config.smtp_port)
            server.starttls()
            server.login(config.smtp_username, config.smtp_password)
            server.send_message(msg)
            server.quit()
        except Exception as e:
            raise Exception(f"SMTP error: {e}")

The Scheduler and Main Program

# src/scheduler.py
"""Main scheduling and orchestrator for LeakRadar."""
import time
from datetime import datetime

import schedule
from loguru import logger

from src.alerting import AlertEngine
from src.breach_api import HaveIBeenPwnedAPI
from src.config import config
from src.models import BreachCheckResult
from src.storage import BreachStorage


class LeakRadar:
    """Main LeakRadar orchestrator."""

    def __init__(self):
        self.api = HaveIBeenPwnedAPI(config.hibp_api_key, config.hibp_api_url)
        self.storage = BreachStorage(config.database_url)
        self.alerter = AlertEngine()

        logger.info("LeakRadar initialized")

    def scan_email(self, email: str) -> None:
        """Scan a single email address for breaches."""
        logger.info(f"Starting scan for {email}")

        try:
            breaches = self.api.get_breached_account(email)
            new_breaches = self.storage.save_breaches(email, breaches)

            result = BreachCheckResult(
                email=email,
                breaches=new_breaches,
                is_new=len(new_breaches) > 0,
            )

            self.storage.save_check_result(result)

            if result.is_new:
                logger.warning(f"Found {len(new_breaches)} new breaches for {email}")
                self.alerter.send_alerts(result)
            else:
                logger.info(f"No new breaches found for {email}")

        except Exception as e:
            logger.error(f"Error scanning {email}: {e}")

    def scan_all_emails(self) -> None:
        """Scan all configured email addresses."""
        logger.info("=" * 60)
        logger.info(f"Starting scan cycle at {datetime.now()}")
        logger.info("=" * 60)

        for email in config.target_emails:
            email = email.strip()
            if email:
                self.scan_email(email)
                time.sleep(2)

        logger.info("=" * 60)
        logger.info("Scan cycle completed")
        logger.info("=" * 60)

    def run_scheduled(self) -> None:
        """Run scheduled monitoring in a loop."""
        logger.info(
            f"Starting scheduled monitoring every {config.scan_interval_hours} hours"
        )

        self.scan_all_emails()
        schedule.every(config.scan_interval_hours).hours.do(self.scan_all_emails)

        try:
            while True:
                schedule.run_pending()
                time.sleep(60)
        except KeyboardInterrupt:
            logger.info("Shutting down LeakRadar")
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            raise

# src/main.py
"""Entry point for the LeakRadar application."""
import sys
from pathlib import Path

from loguru import logger

from src.config import config
from src.scheduler import LeakRadar


def setup_logging() -> None:
    """Configure console and file logging."""
    logger.remove()

    logger.add(
        sys.stderr,
        format=(
            "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        ),
        level=config.log_level,
    )

    log_path = Path(config.log_file)
    log_path.parent.mkdir(parents=True, exist_ok=True)

    logger.add(
        config.log_file,
        rotation="10 MB",
        retention="30 days",
        compression="zip",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
        level=config.log_level,
    )


def main() -> None:
    """Application entry point."""
    setup_logging()

    logger.info("=" * 60)
    logger.info("LeakRadar Starting")
    logger.info("=" * 60)
    logger.info(f"Monitoring {len(config.target_emails)} email(s)")
    logger.info(f"Scan interval: {config.scan_interval_hours} hour(s)")
    logger.info(f"Database: {config.database_url}")
    logger.info("=" * 60)

    monitor = LeakRadar()
    monitor.run_scheduled()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        logger.info("Shutdown requested by user")
    except Exception as e:
        logger.critical(f"Fatal error: {e}")
        sys.exit(1)

Docker Deployment

Dockerfile:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY src/ ./src/

# Create directories for runtime data
RUN mkdir -p /app/data/logs

# Environment
ENV PYTHONUNBUFFERED=1
ENV PYTHONPATH=/app

# Run the application
CMD ["python", "-m", "src.main"]

docker-compose.yml:

version: '3.8'

services:
  leakradar:
    build: .
    container_name: leakradar
    restart: unless-stopped
    env_file:
      - .env
    volumes:
      - ./data:/app/data
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  # Optional Redis cache for faster lookups
  redis:
    image: redis:7-alpine
    container_name: leakradar-redis
    restart: unless-stopped
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

volumes:
  redis_data:

requirements.txt:

requests==2.31.0
aiohttp==3.9.1
python-dotenv==1.0.0
pydantic==2.5.0
loguru==0.7.2
schedule==1.2.1

requirements-dev.txt:

-r requirements.txt
redis==5.0.1
fastapi==0.109.0
uvicorn[standard]==0.27.0
pytest==7.4.4
pytest-asyncio==0.23.4

GitHub Actions for Scheduled Scans

.github/workflows/scheduled-scan.yml:

name: Scheduled Breach Scan

on:
  schedule:
    # Run every hour
    - cron: '0 * * * *'
  workflow_dispatch: # Allow manual triggering

jobs:
  scan:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Create data directory
        run: mkdir -p data/logs

      - name: Run breach scan
        env:
          TARGET_EMAILS: ${{ secrets.TARGET_EMAILS }}
          HIBP_API_KEY: ${{ secrets.HIBP_API_KEY }}
          HIBP_API_URL: ${{ secrets.HIBP_API_URL }}
          TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }}
          TELEGRAM_CHAT_ID: ${{ secrets.TELEGRAM_CHAT_ID }}
          DISCORD_WEBHOOK_URL: ${{ secrets.DISCORD_WEBHOOK_URL }}
          SMTP_SERVER: ${{ secrets.SMTP_SERVER }}
          SMTP_PORT: ${{ secrets.SMTP_PORT }}
          SMTP_USERNAME: ${{ secrets.SMTP_USERNAME }}
          SMTP_PASSWORD: ${{ secrets.SMTP_PASSWORD }}
          SMTP_FROM: ${{ secrets.SMTP_FROM }}
          SMTP_TO: ${{ secrets.SMTP_TO }}
          DATABASE_URL: sqlite:///data/leaks.db
          LOG_LEVEL: INFO
          LOG_FILE: data/logs/leakradar.log
          SCAN_INTERVAL_HOURS: 1
          MAX_RETRIES: 3
          REQUEST_TIMEOUT: 30
        run: |
          python -m src.main

      - name: Upload logs
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: leakradar-logs
          path: data/logs/
          retention-days: 7

      - name: Upload database
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: leakradar-db
          path: data/leaks.db
          retention-days: 7

Tests

tests/test_breach_api.py:

"""Tests for the HIBP API client."""
from unittest.mock import Mock, patch

import pytest

from src.breach_api import HaveIBeenPwnedAPI
from src.models import Breach


@pytest.fixture
def api():
    return HaveIBeenPwnedAPI(api_key="test_key")


@patch("requests.Session.get")
def test_get_breached_account_success(mock_get, api):
    """Test successfully fetching breaches for an email."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = [
        {
            "Name": "TestBreach",
            "Title": "Test Breach",
            "Domain": "example.com",
            "BreachDate": "2023-01-01",
            "AddedDate": "2023-01-02T00:00:00Z",
            "PwnCount": 1000,
            "Description": "Test description",
            "DataClasses": ["Email addresses", "Passwords"],
            "IsVerified": True,
        }
    ]
    mock_get.return_value = mock_response

    breaches = api.get_breached_account("test@example.com")

    assert len(breaches) == 1
    assert isinstance(breaches[0], Breach)
    assert breaches[0].name == "TestBreach"
    assert breaches[0].domain == "example.com"
    assert breaches[0].pwn_count == 1000
    assert breaches[0].severity.value == "critical"


@patch("requests.Session.get")
def test_get_breached_account_no_breaches(mock_get, api):
    """Test the 404 / no-breach case."""
    mock_response = Mock()
    mock_response.status_code = 404
    mock_get.return_value = mock_response

    breaches = api.get_breached_account("clean@example.com")

    assert len(breaches) == 0


@patch("requests.Session.get")
def test_api_error_handling(mock_get, api):
    """Test invalid API key error handling."""
    mock_response = Mock()
    mock_response.status_code = 401
    mock_response.text = "Unauthorized"
    mock_get.return_value = mock_response

    from src.breach_api import BreachAPIError

    with pytest.raises(BreachAPIError):
        api.get_breached_account("test@example.com")

tests/test_normalizer.py:

"""Tests for the data normalizer."""
from datetime import datetime

from src.models import Breach
from src.normalizer import DataNormalizer


def test_normalize_email():
    assert DataNormalizer.normalize_email("  TEST@Example.COM  ") == "test@example.com"
    assert DataNormalizer.normalize_email("user@domain.com") == "user@domain.com"


def test_extract_domain():
    assert DataNormalizer.extract_domain("user@example.com") == "example.com"
    assert DataNormalizer.extract_domain("invalid-email") == ""


def test_calculate_risk_score():
    breach = Breach(
        name="Test",
        title="Test Breach",
        domain="example.com",
        breach_date=datetime.now(),
        added_date=datetime.now(),
        pwn_count=100000,
        description="Test",
        data_classes=["Passwords", "Email addresses"],
        is_verified=True,
    )

    score = DataNormalizer.calculate_risk_score(breach)

    assert 0 <= score <= 100
    assert score > 50


def test_classify_data_types():
    data_classes = [
        "Email addresses",
        "Passwords",
        "Phone numbers",
        "Credit cards",
        "IP addresses",
    ]

    categories = DataNormalizer.classify_data_types(data_classes)

    assert "Email addresses" in categories["credentials"]
    assert "Passwords" in categories["credentials"]
    assert "Phone numbers" in categories["personal_info"]
    assert "Credit cards" in categories["financial"]


def test_generate_summary_empty():
    summary = DataNormalizer.generate_summary([])

    assert summary["total_breaches"] == 0
    assert summary["critical_breaches"] == 0


def test_severity_classification():
    breach = Breach(
        name="Critical",
        title="Critical Breach",
        domain="example.com",
        breach_date=datetime.now(),
        added_date=datetime.now(),
        pwn_count=5000,
        description="Test",
        data_classes=["Email addresses", "Passwords"],
        is_verified=True,
    )

    assert breach.severity.value == "critical"

tests/test_storage.py:

"""Tests for the SQLite storage layer."""
import tempfile
from datetime import datetime
from pathlib import Path

from src.models import Breach
from src.storage import BreachStorage


def _make_breach(name: str, domain: str = "example.com", pwn_count: int = 100) -> Breach:
    now = datetime.now()
    return Breach(
        name=name,
        title=name.title(),
        domain=domain,
        breach_date=now,
        added_date=now,
        pwn_count=pwn_count,
        description="Test breach",
        data_classes=["Email addresses"],
        is_verified=True,
    )


def _storage(tmp_dir: str) -> BreachStorage:
    return BreachStorage(db_url=f"sqlite:///{tmp_dir}/test.db")


def test_save_and_retrieve_breaches():
    with tempfile.TemporaryDirectory() as tmp_dir:
        storage = _storage(tmp_dir)
        email = "test@example.com"

        storage.save_breaches(email, [_make_breach("BreachA"), _make_breach("BreachB")])
        breaches = storage.get_all_breaches_for_email(email)

        assert len(breaches) == 2
        assert {b.name for b in breaches} == {"BreachA", "BreachB"}


def test_save_breaches_is_idempotent():
    with tempfile.TemporaryDirectory() as tmp_dir:
        storage = _storage(tmp_dir)
        email = "test@example.com"

        first = storage.save_breaches(email, [_make_breach("BreachA")])
        second = storage.save_breaches(email, [_make_breach("BreachA")])

        assert len(first) == 1
        assert len(second) == 0
        assert len(storage.get_all_breaches_for_email(email)) == 1


def test_get_recent_breaches():
    with tempfile.TemporaryDirectory() as tmp_dir:
        storage = _storage(tmp_dir)
        email = "test@example.com"

        storage.save_breaches(email, [_make_breach("BreachA")])
        recent = storage.get_recent_breaches(days=7)

        assert len(recent) == 1
        assert recent[0][2] == "BreachA"


def test_mark_alert_sent():
    with tempfile.TemporaryDirectory() as tmp_dir:
        storage = _storage(tmp_dir)
        storage.mark_alert_sent("test@example.com", "BreachA", "telegram")

        db_file = Path(tmp_dir) / "test.db"
        assert db_file.exists()

To run the tests:

pip install -r requirements-dev.txt
pytest

All tests pass: 13 passed.

Professional Improvements

1. Using Async/Await for Better Performance

# src/async_breach_api.py
"""Async API client for parallel breach checks."""
import asyncio
from typing import List

import aiohttp
from loguru import logger

from src.config import config


class AsyncBreachAPI:
    """Asynchronous variant of the breach API client."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://haveibeenpwned.com/api/v3"

    async def check_multiple_emails(self, emails: List[str]) -> dict:
        """Check several emails in parallel."""
        async with aiohttp.ClientSession() as session:
            tasks = [self._check_email_async(session, email) for email in emails]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            return {
                email: result
                for email, result in zip(emails, results)
                if not isinstance(result, Exception)
            }

    async def _check_email_async(self, session: aiohttp.ClientSession, email: str):
        """Check a single email asynchronously."""
        url = f"{self.base_url}/breachedaccount/{email}"
        headers = {"hibp-api-key": self.api_key}
        params = {"truncateResponse": "false"}

        try:
            async with session.get(
                url, headers=headers, params=params, timeout=config.request_timeout
            ) as response:
                if response.status == 200:
                    return await response.json()
                if response.status == 404:
                    return []
                logger.error(f"Error checking {email}: {response.status}")
                return None
        except Exception as e:
            logger.error(f"Async error checking {email}: {e}")
            return None

2. Adding Redis for Caching

# src/cache.py
"""Optional Redis-backed caching layer."""
import json
from typing import List, Optional

from loguru import logger

from src.models import Breach


class BreachCache:
    """Redis-backed cache for breach lookups."""

    def __init__(self, host: str = "localhost", port: int = 6379):
        import redis

        self.redis_client = redis.Redis(
            host=host, port=port, db=0, decode_responses=True
        )
        self.ttl = 3600

    def get_breaches(self, email: str) -> Optional[List[dict]]:
        """Retrieve cached breaches for an email, if any."""
        cached = self.redis_client.get(f"breaches:{email}")
        if cached:
            return json.loads(cached)
        return None

    def set_breaches(self, email: str, breaches: List[Breach]) -> None:
        """Cache breaches for an email with a TTL."""
        self.redis_client.setex(
            f"breaches:{email}",
            self.ttl,
            json.dumps([b.model_dump() for b in breaches]),
        )

    def invalidate(self, email: str) -> None:
        """Remove cached breaches for an email."""
        self.redis_client.delete(f"breaches:{email}")


class NullCache:
    """No-op cache used when Redis is unavailable."""

    def get_breaches(self, email: str) -> Optional[List[dict]]:
        return None

    def set_breaches(self, email: str, breaches: List[Breach]) -> None:
        pass

    def invalidate(self, email: str) -> None:
        pass


def get_cache() -> BreachCache | NullCache:
    """Return a cache instance, falling back to NullCache when Redis is off."""
    try:
        return BreachCache()
    except Exception as e:
        logger.warning(f"Redis unavailable, using NullCache: {e}")
        return NullCache()

Bonus: when Redis is unavailable, the system falls back to NullCache (a no-op object) instead of crashing, so it keeps working even without Redis.

Security and Ethical Considerations

Guiding Principles

Legality — only scan accounts you own or are explicitly authorized to monitor.
Consent — never use this tool to collect data on others without explicit permission.
Secret hygiene — never commit API keys or upload them to public repositories.
Terms of service — respect the terms of all API providers.
Compliance — follow data-protection laws such as GDPR and CCPA.

Best Practices

# src/security.py
"""Security utilities: ownership validation and audit logging."""
from functools import wraps
from typing import List

from loguru import logger


def validate_email_ownership(email: str, allowed_domains: List[str]) -> bool:
    """Validate that an email belongs to an allowed domain."""
    domain = email.split("@")[1] if "@" in email else ""

    if domain not in allowed_domains:
        logger.warning(f"Unauthorized domain: {domain}")
        return False

    return True


def audit_log(action: str, email: str, user: str = "system") -> None:
    """Log an auditable action."""
    logger.info(f"AUDIT: {action} | Email: {email} | User: {user}")


def audit(action: str):
    """Decorator that records an audit entry for the wrapped function."""

    def decorator(func):
        @wraps(func)
        def wrapper(email: str, *args, **kwargs):
            audit_log(action, email)
            return func(email, *args, **kwargs)

        return wrapper

    return decorator


class SecurityManager:
    """Manage authorization for scan operations."""

    def __init__(self, allowed_domains: List[str]):
        self.allowed_domains = allowed_domains

    def authorize_scan(self, email: str) -> bool:
        """Check whether an email is authorized to be scanned."""
        if not validate_email_ownership(email, self.allowed_domains):
            audit_log("UNAUTHORIZED_SCAN_ATTEMPT", email)
            return False

        audit_log("AUTHORIZED_SCAN", email)
        return True

Development Roadmap

Phase 1: Core (complete) ✅

[x] Email monitoring
[x] HIBP API integration
[x] Alerting system
[x] Local storage
[x] Docker
[x] Automated tests

Phase 2: Enhancements

[ ] Add additional OSINT sources (DeHashed, LeakCheck)
[ ] Monitor GitHub repositories for leaked secrets
[ ] GitLab and Gists support
[ ] Automatic triage of new breaches
[ ] Async performance improvements

Phase 3: Full platform

[ ] Advanced web dashboard
[ ] Multi-user support with roles
[ ] Multi-domain management
[ ] Periodic PDF reports
[ ] Advanced risk scoring
[ ] SIEM integrations (Splunk, ELK)

Phase 4: Intelligence

[ ] Sentiment analysis of leaks
[ ] Breach prediction
[ ] Automatic threat classification
[ ] Intelligent remediation recommendations

Conclusion

You don't need a Security Operations Center (SOC) with a huge budget to build tools that help you monitor whether your data has been exposed. Using Python, some APIs, and OSINT principles, you can build a simple system that provides continuous visibility into your data appearing in open sources.

This system is not a final product — it's a starting point that can gradually evolve into a full Threat Intelligence platform supporting:

Corporate security teams
Independent security researchers
Small and medium businesses
Individuals concerned about their digital privacy

What We Learned

The difference between OSINT and Threat Intelligence — OSINT provides the data, Threat Intelligence turns it into actionable knowledge
How breach monitoring services work — from data collection to indexing and search
Building a complete system — from API connections to alerts
Best practices — security, performance, scalability
Deployment and operations — Docker, GitHub Actions, monitoring
Testing your code — automated tests catch bugs before release (they are what revealed the fixes in this system)

Next Step

The feature that should be the top priority for this system is monitoring GitHub repositories for accidentally leaked secrets. This is a common and serious problem, as developers accidentally publish API keys and passwords in public repositories.

You can use the GitHub API to search for:

Leaked AWS keys
GitHub tokens
Database passwords
Encryption keys

This would add an extra layer of protection that complements traditional breach monitoring.

Which feature do you think should be the next step for such a system? Share your thoughts in the comments! 🚀

References

📌GitHub Repository

Table of Contents