DEV Community: H3l!0s_T3k

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

H3l!0s_T3k — Wed, 05 Aug 2026 20:15:36 +0000

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

INTERPOL's 2026 Report: AI Now Powers Over Half of Africa's Cybercrime

H3l!0s_T3k — Mon, 03 Aug 2026 20:18:27 +0000

The rapid digital transformation sweeping across the African continent has brought with it a surge in cybercrime that threatens financial institutions, government bodies, and critical digital infrastructure. The INTERPOL African Cyberthreat Assessment Report 2026 reveals a qualitative shift in the nature of these crimes. Powered increasingly by artificial intelligence, cyberattacks have become faster, more scalable, and harder to detect. This report is a serious wake-up call, demanding a fundamental rethink of security and legislative strategies to combat cybercrime that has evolved into an industrialized, borderless ecosystem.

Key Findings from the Report

1. AI Dominates the Cybercrime Landscape

The report's most striking revelation is that 55% of reported cybercrimes in Africa are AI-enabled. This figure reflects a radical transformation in digital criminal methods. Perpetrators no longer rely solely on individual human expertise; they now invest in sophisticated AI tools that allow them to execute massive attacks with high efficiency and relatively low cost.

2. Online Scams Lead the Pack

Online scams top the list of the most frequently reported forms of cybercrime across African nations. AI amplifies the effectiveness of these scams by enabling criminals to create convincing fraudulent messages, simulate real identities, and precisely target victims based on data analysis.

3. Hardest-Hit Sectors

Attacks have primarily concentrated on three critical sectors:

Financial Services: Banks, financing companies, and digital payment platforms.
Telecommunications: The primary gateway to internet and digital services.
Government Institutions: Posing a direct threat to sensitive data and public services.

4. Financial Losses Have More Than Doubled

Financial losses from cybercrime have skyrocketed, more than doubling since 2024 to reach USD 484 million. This staggering increase reflects not only the growing volume of attacks but also their evolving sophistication and enhanced ability to breach security systems.

5. The Proliferation of Scam Centres

72% of surveyed countries reported the presence of scam centres on their soil, with the highest concentration in Southern and West Africa. These centres represent the practical embodiment of the "industrialization" of cybercrime, where criminal operations are organized with professional efficiency—akin to factories—with clear division of labour and specialized roles.

6. Bypassing Biometric Security

Among the most alarming criminal innovations documented in the report is the ability of criminals to bypass biometric checks by creating AI-generated synthetic identities. These identities blend stolen real personal data with fabricated elements, allowing them to pass through advanced digital verification systems undetected.

Analyzing the Shift to "Industrialized Cybercrime"

The report describes contemporary cybercrime as having evolved into an "industrialized, borderless criminal ecosystem." This description carries deep implications:

Industrialization: Cybercrime is no longer a series of random, individual acts. It has transformed into organized operations using mass-production techniques, with specialized assembly lines for each stage of the crime—data harvesting, fraudulent content creation, system breaches, and money laundering.
Borderlessness: Thanks to the internet and digital technologies, criminals can launch attacks from one country against victims in another, exploiting legislative gaps and the difficulty of international investigative cooperation.
AI as the Accelerator: Artificial intelligence is the pivotal factor in this transformation, providing criminals with:
- Speed: Executing thousands of attacks in a short time.
- Scalability: The ability to replicate operations on a massive scale without requiring large human resources.
- Stealth: Making attacks difficult for victims and security platforms to detect.

Challenges and Implications

Security Challenges

AI-generated synthetic identities pose an existential challenge to digital verification systems that rely on biometric data. When criminals can forge fingerprints, facial recognition data, and identity information, digital gateways lose much of their effectiveness in protecting access.

Economic Impact

The USD 484 million in losses represents a devastating blow to African economies, especially since it targets the most vital sectors. These figures may not even reflect the true scale of losses, as many incidents go unreported due to fears of exposing security vulnerabilities or losing customer trust.

Social Ramifications

The proliferation of scam centres reflects a social spread of organized crime, where young people are recruited into fraudulent operations under the guise of "job opportunities." This contributes to a complex social phenomenon that requires comprehensive intervention beyond mere security measures.

Recommendations and Strategic Proposals

Based on the report's findings, the following strategic recommendations can be drawn:

Strengthen International Cooperation: Develop mechanisms for information exchange and coordination between African and international law enforcement agencies, including facilitating the extradition of digital criminals across borders.
Modernize Legal Frameworks: Update local laws to criminalize the use of artificial intelligence in cybercrimes and establish strict standards for protecting biometric data.
Invest in Defensive AI: Countering AI-powered threats requires defensive tools that leverage machine learning to detect fraudulent patterns in real time.
Build Human and Institutional Capacity: Launch widespread awareness programs for citizens about the risks of online scams, and train security and financial personnel on the latest attack and defence techniques.
Target Scam Centre Operations: Conduct targeted intelligence and security campaigns to uncover and dismantle scam centres, with a focus on the high-concentration regions of Southern and West Africa.

Conclusion

INTERPOL's 2026 report is an alarm bell revealing a painful truth: despite AI's transformative potential for development, it is increasingly being weaponized by organized crime. Cybercrime in Africa is no longer merely a technical threat; it has become a cross-border industry draining economies and threatening national security. Confronting this challenge requires more than just technical solutions—it demands a comprehensive response combining international cooperation, modern legislation, and investment in human and technical capabilities to ensure that artificial intelligence remains in service of humanity, not against it.

> Source: INTERPOL African Cyberthreat Assessment Report 2026

> 🔗 Read the full report

What are your thoughts on the role of AI in cybercrime? How can the tech community contribute to building more resilient digital defences? Share your insights in the comments below!

ArCode Standard: An Open Source Standard for Software Architecture, Project Structure, and Modern Development

H3l!0s_T3k — Mon, 03 Aug 2026 01:21:10 +0000

Every year, thousands of software projects are created by developers, startups, and organizations.

However, many teams face the same problem:

They spend valuable time rebuilding the same foundations again and again.

How should the project structure look?

Where should documentation live?

What naming conventions should be followed?

How should developers organize workflows, testing, security, and collaboration?

Most teams create their own answers from scratch.

This creates fragmentation across the software ecosystem.

Different projects use different structures, different conventions, and different development practices, making collaboration and maintenance more difficult.

This is the problem ArCode Standard aims to solve.

What is ArCode Standard?

ArCode Standard is an open-source initiative focused on creating a practical and modern standard for building software projects.

It provides a structured foundation that helps developers create projects that are:

More organized
Easier to maintain
Easier to understand
Easier to scale
Easier to collaborate on

ArCode Standard is not a programming language.

It is not a framework.

It is not designed to replace existing technologies.

Instead, it is a set of principles, structures, and practices that developers can apply across different programming languages, frameworks, and environments.

Why Do We Need Software Standards?

Modern software development is becoming increasingly complex.

A single project may include:

Multiple programming languages
Frontend and backend systems
APIs
Databases
Cloud infrastructure
Automated testing
CI/CD pipelines
Security requirements
Artificial intelligence tools

Without clear standards, projects can quickly become difficult to manage.

Common problems include:

Inconsistent project structures
Poor documentation
Difficult onboarding for new contributors
Different coding practices between teams
Lack of clear workflows
Difficult project migration and maintenance

A developer joining a new project should spend time building features, not trying to understand every team's unique organization system.

How Is ArCode Standard Different?

Many existing software concepts solve specific problems.

For example:

Clean Architecture focuses on designing software architecture.

SOLID principles focus on object-oriented design practices.

Domain-Driven Design (DDD) focuses on modeling complex business logic.

Project templates provide initial structures.

All of these are valuable.

However, software projects need more than one concept.

They need a unified approach that considers:

Project organization
Development workflow
Documentation standards
Collaboration practices
Scalability
Security considerations
Future AI-assisted development

ArCode Standard aims to bring these ideas together into an open and community-driven standard.

The Core Vision

We believe that open standards are as important as open-source software.

Open source allows developers to share code.

Open standards allow developers to share a common way of building.

When projects follow clear conventions:

Developers collaborate faster
New contributors understand projects more easily
Teams reduce repeated work
Software becomes easier to maintain
Tools can better understand project structures

The goal of ArCode Standard is not to control how developers build software.

The goal is to provide a strong foundation that developers can adapt and improve.

Why AI-Ready Software Development Matters

Artificial intelligence is changing the way software is created.

Developers are increasingly using AI assistants for:

Code generation
Debugging
Documentation
Testing
Project analysis

However, AI tools work better when projects have:

Clear structures
Good documentation
Consistent naming
Predictable organization

A well-structured project is easier for both humans and intelligent tools to understand.

ArCode Standard aims to encourage practices that make modern software development more efficient, including future AI-assisted workflows.

Who Can Contribute?

ArCode Standard is built for the community.

We welcome contributions from:

Software Engineers
Open Source Developers
AI Engineers
Security Researchers
DevOps Engineers
Cloud Engineers
UI/UX Designers
Technical Writers
Computer Science Students
Researchers interested in software engineering

You do not need to be an expert.

You can contribute through:

Code
Documentation
Ideas
Discussions
Reviews
Testing
Suggestions

Every contribution helps shape the future of the project.

Future Roadmap

The long-term vision of ArCode Standard includes:

Core Standard Development

Creating and improving the fundamental specifications.

Documentation System

Building clear guidelines and examples for developers.

Developer Tools

Creating tools that help developers validate and apply the standard.

AI Integration

Exploring how standardized project structures can improve AI-assisted development workflows.

Community Governance

Building an open community where contributors help evolve the standard.

Join ArCode Standard

ArCode Standard is still at an early stage.

This is the best moment to participate, share ideas, and help shape the direction of the project.

GitHub Repository:

https://github.com/ArCode-Standard/ArCode-Core

Project Website:

https://arcode-standard.github.io/ArCode-Core/

You can contribute by:

⭐ Starring the repository
Opening discussions
Reporting issues
Suggesting improvements
Contributing code or documentation
Sharing the project with other developers

Open standards are created by communities.

Let's build a better foundation for modern software development together.

I Built a Browser-Based YouTube Frame Extractor (No FFmpeg Required)

H3l!0s_T3k — Thu, 09 Jul 2026 01:04:23 +0000

If you've ever wanted to extract frames from a YouTube video, you've probably downloaded the video first and used FFmpeg. It works—but it's not always the fastest or simplest solution.
I wanted a tool that works entirely in the browser, with no downloads, no CLI, and no servers.
So I built YTFR: YT Fr4m3 R1pp3r, a Tampermonkey userscript that captures frames directly from the YouTube player in real time.

How it works

Instead of downloading the video again, the script captures frames that the browser is already rendering.

YouTube Video
↓
Capture Frames
↓
Store as Blobs
↓
Generate ZIP
↓
Download
Everything runs locally on your machine.

Features

Automatic frame capture at custom intervals
Manual frame capture
ZIP export with:
Extracted frames
Contact sheets
JSON reports
CSV frame metadata
Draggable UI
Automatic fallback for browsers without requestVideoFrameCallback()
Configurable memory limit to prevent crashes

Challenges

The biggest challenges were:

Handling YouTube's Single Page Application (SPA) navigation.
Supporting browsers without requestVideoFrameCallback().
Managing memory while capturing hundreds of images.
Building a movable interface that remembers its position.

What's next?

I'm planning to add:

Timestamp-based extraction
Frame comparison
Playlist batch processing

If you have ideas or feedback, feel free to open an issue or contribute.

GitHub

DEV Community: H3l!0s_T3k

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

Table of Contents

OSINT vs Threat Intelligence

Behind the Scenes: How Breach Monitoring Services Work

1. Data Collection

2. Data Processing

3. Indexing and Search

4. Application Programming Interfaces (APIs)

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?

1. Map of API Locations in the Project

2. The Golden Rule

3. How to Change the HIBP Key / URL?

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

5. HIBP Endpoints and Their Location

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

7. FAQ

Data Normalization Layer

Storage Layer

Alerting System

The Scheduler and Main Program

Docker Deployment

GitHub Actions for Scheduled Scans

Tests

Professional Improvements

1. Using Async/Await for Better Performance

2. Adding Redis for Caching

Security and Ethical Considerations

Guiding Principles

Best Practices

Development Roadmap

Phase 1: Core (complete) ✅

Phase 2: Enhancements

Phase 3: Full platform

Phase 4: Intelligence

Conclusion

What We Learned

Next Step

References

INTERPOL's 2026 Report: AI Now Powers Over Half of Africa's Cybercrime

Key Findings from the Report

1. AI Dominates the Cybercrime Landscape

2. Online Scams Lead the Pack

3. Hardest-Hit Sectors

4. Financial Losses Have More Than Doubled

5. The Proliferation of Scam Centres

6. Bypassing Biometric Security

Analyzing the Shift to "Industrialized Cybercrime"

Challenges and Implications

Security Challenges

Economic Impact

Social Ramifications

Recommendations and Strategic Proposals

Conclusion

ArCode Standard: An Open Source Standard for Software Architecture, Project Structure, and Modern Development

What is ArCode Standard?

Why Do We Need Software Standards?

How Is ArCode Standard Different?

The Core Vision

Why AI-Ready Software Development Matters

Who Can Contribute?

Future Roadmap

Core Standard Development

Documentation System

Developer Tools

AI Integration

Community Governance

Join ArCode Standard

I Built a Browser-Based YouTube Frame Extractor (No FFmpeg Required)