DEV Community: Ayi NEDJIMI

How to implement secure file upload with validation in Go

Ayi NEDJIMI — Mon, 10 Aug 2026 10:04:01 +0000

File upload endpoints are one of the most reliable ways to compromise a server. A single misconfigured handler can let an attacker upload a web shell, exhaust disk space with a multi-gigabyte payload, or path-traverse into sensitive directories. Most Go tutorials show you how to receive a file. Almost none show you how to do it safely. This one does.

The attack surface you're actually defending against

Before writing any handler code, it helps to be explicit about what can go wrong:

MIME type spoofing: Content-Type is set by the client. It's user-controlled input. Trusting it is not validation.
Malicious filenames: ../../../etc/cron.d/backdoor, index.php, filenames with null bytes, or a 300-character Unicode mess designed to overflow a path buffer.
Oversized payloads: a 10 GB file sent to a handler that reads without limits will eat your disk — or memory — in seconds.
Web root storage: saving uploads inside your static file directory makes any uploaded script directly executable over HTTP.
Partial writes: a failed upload that leaves a half-written file at a predictable path can be exploited if cleanup is missing.

Each problem has a straightforward mitigation. The critical detail is applying them in the right order, before reading the full body.

Step 1 – Enforce size limits before anything else

The first thing your handler must do is cap how much it will read. http.MaxBytesReader wraps the request body and returns an error the moment the threshold is crossed — it never buffers the full payload.

package main

import (
    "fmt"
    "net/http"
)

const maxUploadSize = 10 << 20 // 10 MB

func uploadHandler(w http.ResponseWriter, r *http.Request) {
    // Wrap the body BEFORE calling ParseMultipartForm
    r.Body = http.MaxBytesReader(w, r.Body, maxUploadSize)

    // 2 MB held in memory, the rest spilled to temp files
    if err := r.ParseMultipartForm(2 << 20); err != nil {
        http.Error(w, "file too large or malformed", http.StatusBadRequest)
        return
    }

    file, header, err := r.FormFile("upload")
    if err != nil {
        http.Error(w, "missing upload field", http.StatusBadRequest)
        return
    }
    defer file.Close()

    fmt.Fprintf(w, "received: %s (%d bytes)\n", header.Filename, header.Size)
}

func main() {
    http.HandleFunc("/upload", uploadHandler)
    http.ListenAndServe(":8080", nil)
}

The order matters. Wrap the body first, then parse. If you call ParseMultipartForm before MaxBytesReader, Go reads the entire body to memory before your size check ever runs — the opposite of what you want.

Step 2 – Validate file type from magic bytes, not headers

HTTP Content-Type is client-supplied noise. The real signal is the first 512 bytes of the file itself. Go's net/http package ships DetectContentType, which implements the WHATWG MIME Sniffing algorithm and reads the actual byte signatures:

package main

import (
    "errors"
    "io"
    "net/http"
)

var allowedTypes = map[string]bool{
    "image/jpeg": true,
    "image/png":  true,
    "image/webp": true,
}

func detectAndValidateMIME(r io.ReadSeeker) (string, error) {
    buf := make([]byte, 512)
    n, err := r.Read(buf)
    if err != nil && err != io.EOF {
        return "", err
    }

    mimeType := http.DetectContentType(buf[:n])

    // Rewind so the full file can be read afterward
    if _, err := r.Seek(0, io.SeekStart); err != nil {
        return "", err
    }

    if !allowedTypes[mimeType] {
        return "", errors.New("file type not allowed: " + mimeType)
    }

    return mimeType, nil
}

The Seek(0, io.SeekStart) call is mandatory. Forget it, and the next reader sees an empty stream — a common bug that produces silently truncated uploads with no error.

Step 3 – Sanitize the filename

Never use header.Filename directly as a filesystem path. filepath.Base strips directory components, and a regex removes everything else that could cause trouble:

import (
    "path/filepath"
    "regexp"
    "strings"
)

var nonSafeChars = regexp.MustCompile(`[^a-zA-Z0-9._-]`)

func sanitizeFilename(name string) string {
    // Strip any directory traversal attempt
    base := filepath.Base(name)

    // Keep only alphanumeric characters, dots, dashes, and underscores
    safe := nonSafeChars.ReplaceAllString(base, "_")

    // Enforce the ext4/APFS filename length limit
    if len(safe) > 255 {
        safe = safe[:255]
    }

    return strings.ToLower(safe)
}

Even after sanitization, don't use the original filename as your storage key. Generate a random ID, store it, and record the original name in a database column. This prevents collisions, keeps storage paths unpredictable, and stops an attacker from guessing where a file landed.

Step 4 – Store outside the web root with restrictive permissions

Uploads written inside your static file directory become directly accessible over HTTP. Move them elsewhere and route downloads through a handler that enforces authorization:

import (
    "crypto/rand"
    "encoding/hex"
    "io"
    "os"
    "path/filepath"
)

// NOT inside your web root
const uploadDir = "/var/app/uploads"

func saveFile(src io.Reader, ext string) (string, error) {
    idBytes := make([]byte, 16)
    if _, err := rand.Read(idBytes); err != nil {
        return "", err
    }
    id := hex.EncodeToString(idBytes)

    dest := filepath.Join(uploadDir, id+ext)

    // O_EXCL fails if a file with this name already exists
    out, err := os.OpenFile(dest, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
    if err != nil {
        return "", err
    }
    defer out.Close()

    if _, err := io.Copy(out, src); err != nil {
        os.Remove(dest) // remove the partial write on failure
        return "", err
    }

    return id, nil
}

os.O_EXCL makes the open fail if the path already exists — a lightweight guard against the astronomically unlikely UUID collision that would otherwise silently overwrite a file. The 0600 permission means only the process owner can read or write it, limiting exposure if the directory is ever accidentally exposed.

Putting it all together

A correct secure upload flow applies these steps in strict sequence:

MaxBytesReader wraps the body (before ParseMultipartForm)
ParseMultipartForm parses with an in-memory cap
detectAndValidateMIME reads 512 bytes from the actual content
sanitizeFilename strips traversal and dangerous characters
saveFile writes to a path outside the web root using a random ID and 0600 permissions

Skipping any single step leaves a meaningful gap. These controls are designed to layer — not to be cherry-picked.

If you run a DevSecOps pipeline, consider adding a malware scanning step (ClamAV or a cloud API) between steps 3 and 5 for truly untrusted uploads. The full checklist for hardening web APIs, including upload endpoints, is part of our free security hardening checklists.

The takeaway

Secure file upload in Go is not complicated, but it requires deliberate ordering and an allowlist mindset. Size limits come first. MIME validation reads bytes, not headers. Filenames get sanitized and then discarded in favor of random storage IDs. Uploads never land inside the web root.

The Go standard library gives you every primitive you need. The patterns above add roughly 80 lines to your handler. There is no excuse for an upload endpoint that trusts the client.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

AI-Assisted Penetration Test Report Generation

Ayi NEDJIMI — Sun, 09 Aug 2026 10:02:39 +0000

Penetration test reports consume 30–40% of total engagement time. The vulnerability discovery is done — you have notes, tool outputs, screenshots — but converting that raw evidence into a structured, client-readable document is where hours disappear. Integrating a language model into this pipeline cuts report writing time significantly while keeping the critical validation work firmly in human hands.

This is not about automating your pentesting. It is about automating the prose that wraps findings you have already confirmed.

The Architecture

The pipeline runs in three stages:

Ingest — parse raw finding data from your tooling (Burp Suite exports, Nmap XML, manual notes in JSON or CSV)
Enrich — feed each finding to a language model to generate description, impact, and remediation prose
Render — merge enriched findings into a Jinja2 template to produce Markdown or DOCX output

The core constraint: the LLM generates prose only. It never changes a CVSS score, never modifies the affected host, never invents a finding. Your structured data is the source of truth; the model fills the narrative gaps around it.

Structuring Raw Findings Before Touching the LLM

Define a strict schema before you write a single prompt. This eliminates hallucination drift — if the model cannot change numeric fields, it cannot fabricate severity.

from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class RawFinding:
    id: str
    title: str
    cvss_score: float
    cvss_vector: str
    affected_host: str
    port: Optional[int]
    evidence: str          # raw tool output or tester note
    cwe_id: Optional[str]
    remediation_hint: str  # one-liner from the tester

def load_findings(path: str) -> list[RawFinding]:
    import json
    with open(path) as f:
        data = json.load(f)
    return [RawFinding(**item) for item in data]

A sample input file (findings.json):

[
  {
    "id": "FIND-001",
    "title": "Reflected XSS in search parameter",
    "cvss_score": 6.1,
    "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:C/C:L/I:L/A:N",
    "affected_host": "https://app.example.com/search",
    "port": 443,
    "evidence": "GET /search?q=<script>alert(1)</script> returned unsanitized input in response body",
    "cwe_id": "CWE-79",
    "remediation_hint": "encode output, add CSP header"
  }
]

Building the Enrichment Layer

The enrichment step sends each finding to a language model with a strict system prompt that forbids modifying structured fields. Using response_format: json_object forces the model to return parseable JSON consistently.

import json
import textwrap
from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY from environment

SYSTEM_PROMPT = textwrap.dedent("""\
    You are a senior penetration tester writing a client-facing security report.
    Given a JSON finding, produce exactly three fields:
    - "description": 2-3 sentences explaining the vulnerability technically
    - "impact": 2-3 sentences on concrete business and security impact
    - "remediation": 3-5 bullet points with actionable remediation steps

    Rules:
    - Return valid JSON only. No text outside the JSON object.
    - Do NOT change any field from the input (cvss_score, affected_host, etc.)
    - Do NOT invent additional findings or sub-findings
    - Write at a level appropriate for a technical security team
""")

def enrich_finding(finding: RawFinding) -> dict:
    payload = json.dumps(asdict(finding), indent=2)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Finding:\n{payload}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )

    enriched = json.loads(response.choices[0].message.content)

    # Enforce immutable fields after the merge
    result = {**asdict(finding), **enriched}
    for key in ("cvss_score", "cvss_vector", "affected_host", "port", "cwe_id"):
        result[key] = getattr(finding, key)
    return result


def enrich_all(findings: list[RawFinding]) -> list[dict]:
    results = []
    for f in findings:
        try:
            results.append(enrich_finding(f))
        except Exception as e:
            print(f"[WARN] Skipping {f.id}: {e}")
            results.append(asdict(f))  # fall back to raw data
    return results

The {**asdict(finding), **enriched} merge pattern lets the LLM add its prose keys, and then the explicit re-assignment locks every numeric and structured field to its original value. Even if the model attempts to return a modified cvss_score, it gets overwritten. This is not optional; it is what makes the tool safe to use in a client-deliverable workflow.

Rendering the Final Report

With enriched findings in hand, a Jinja2 template decouples content from formatting. Swap the template to target DOCX (via python-docx), PDF (via Pandoc), or any internal format without touching the enrichment logic.

from jinja2 import Environment, FileSystemLoader

SEVERITY = [
    (range(9, 11), "Critical"),
    (range(7, 9),  "High"),
    (range(4, 7),  "Medium"),
    (range(0, 4),  "Low"),
]

def get_severity(score: float) -> str:
    for r, label in SEVERITY:
        if int(score) in r:
            return label
    return "Informational"

def render_report(findings: list[dict], template_path: str, output_path: str):
    env = Environment(loader=FileSystemLoader("."))
    env.filters["severity"] = get_severity

    template = env.get_template(template_path)
    output = template.render(
        findings=sorted(findings, key=lambda f: f["cvss_score"], reverse=True)
    )
    with open(output_path, "w") as fh:
        fh.write(output)
    print(f"Report written to {output_path}")

A minimal Jinja2 template (report.md.j2) that sorts findings by severity:

# Penetration Test Report

{% for f in findings %}
## {{ f.id }} — {{ f.title }}

**Severity**: {{ f.cvss_score | severity }} (CVSS {{ f.cvss_score }})
**Host**: {{ f.affected_host }}
**CWE**: {{ f.cwe_id or 'N/A' }}

### Description
{{ f.description }}

### Impact
{{ f.impact }}

### Remediation
{{ f.remediation }}

---
{% endfor %}

Quality Checks You Cannot Skip

The LLM will occasionally produce descriptions that are generic or misaligned with the actual evidence. Two automated checks must run before every delivery.

1. Keyword overlap between title and description. If the finding title says "XSS" and the generated description discusses SQL injection, catch it before the client does.

def sanity_check(finding: dict) -> list[str]:
    warnings = []
    title_words = set(finding["title"].lower().split())
    desc_words  = set(finding["description"].lower().split())
    if not title_words & desc_words:
        warnings.append(
            f"{finding['id']}: title and description share no keywords — review manually"
        )
    if finding["cvss_score"] >= 7.0:
        impact_lower = finding["impact"].lower()
        if "high" not in impact_lower and "critical" not in impact_lower:
            warnings.append(
                f"{finding['id']}: CVSS {finding['cvss_score']} but impact text lacks severity language"
            )
    return warnings

2. Remediation bullet count. The prompt asks for 3–5 bullets. A one-liner remediation on a High-severity finding is a red flag, not a deliberate choice.

Run these checks after enrich_all() and print every warning to stdout. Do not suppress them with a try/except. These are the cases where a reviewer needs to read carefully before the report leaves the building.

Cross-referencing your recommended mitigations against your firm's internal standards is also worth the time — our security hardening checklists cover the most common remediation patterns and serve as a quick reference during that review pass.

The Takeaway

The full pipeline — ingest, enrich, render, validate — stays under 200 lines of Python and compresses report prose time from hours to under ten minutes. The human work shifts from drafting boilerplate to reviewing LLM output against raw evidence. That is the correct trade-off: your expertise goes into the validation layer, not the repetitive prose.

Three things make this safe for production use: immutable structured fields, strict JSON output mode from the model, and a sanity-check pass before every delivery. Skip any of them and you are shipping a liability, not a tool.

The Jinja2 template approach means your firm's formatting standards stay in a single file. You can maintain a DOCX-based template for enterprise clients and a Markdown template for technical ones, with no changes to the enrichment code.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building a Dependency Vulnerability Scanner for Python Projects

Ayi NEDJIMI — Sat, 08 Aug 2026 10:04:26 +0000

Most Python projects accumulate dozens of third-party packages before anyone thinks to audit them. Each pinned version is a potential CVE waiting to be exploited. Knowing which packages carry known vulnerabilities — and being able to act on that programmatically — is not optional in 2026.

This post shows you how to build a dependency vulnerability scanner from scratch: parse requirements.txt, query the OSV (Open Source Vulnerabilities) API, and generate an actionable report you can wire into any CI/CD pipeline.

Why Not Just Use pip-audit or Safety?

Tools like pip-audit and safety are reasonable starting points, but both have real friction in practice:

pip-audit resolves packages against an installed environment — useless when you want to scan a lockfile in a container build stage without installing anything first.
safety requires a paid plan for anything beyond basic checks, and its CVE database can lag behind OSV.
Both produce output in formats that resist integration into custom pipelines (Slack alerts, Jira tickets, SBOM feeds).

Building your own scanner means controlling the data source, the output format, and the integration points. The OSV API is free, maintained by Google's open-source security team, and covers the PyPI ecosystem comprehensively.

Step 1 — Parse the Requirements File

We start by reading requirements.txt and extracting package names with their pinned versions:

import re
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

@dataclass
class Dependency:
    name: str
    version: Optional[str]

def parse_requirements(filepath: str) -> list[Dependency]:
    deps = []
    for line in Path(filepath).read_text().splitlines():
        line = line.strip()
        # Skip comments, blank lines, and pip options (-r, --index-url, etc.)
        if not line or line.startswith("#") or line.startswith("-"):
            continue
        # Match name==version, name>=version, name[extra]==version, etc.
        match = re.match(
            r"^([A-Za-z0-9_\-\.]+)(?:\[[^\]]+\])?\s*([=><!\^~]+)\s*([0-9][^\s,;#]*)?",
            line
        )
        if match:
            name = match.group(1)
            op = match.group(2)
            # Only treat pinned versions (==) as authoritative for CVE matching
            version = match.group(3) if op == "==" else None
            deps.append(Dependency(name=name, version=version))
    return deps

Unpinned ranges (requests>=2.25.0) still get collected — we query OSV without a version to surface all known vulnerabilities, then flag them as "unpinned" in the report.

Step 2 — Query the OSV API

OSV exposes a simple POST endpoint that accepts a package name, ecosystem, and optionally a version:

import httpx
import time
from typing import Any

OSV_ENDPOINT = "https://api.osv.dev/v1/query"

def query_osv(dep: Dependency) -> list[dict[str, Any]]:
    payload: dict[str, Any] = {
        "package": {
            "name": dep.name,
            "ecosystem": "PyPI",
        }
    }
    if dep.version:
        payload["version"] = dep.version

    try:
        resp = httpx.post(OSV_ENDPOINT, json=payload, timeout=10)
        resp.raise_for_status()
    except httpx.HTTPStatusError as e:
        print(f"[warn] OSV returned {e.response.status_code} for {dep.name}")
        return []
    except httpx.RequestError as e:
        print(f"[warn] Network error querying {dep.name}: {e}")
        return []

    return resp.json().get("vulns", [])

def scan_dependencies(
    deps: list[Dependency], delay: float = 0.1
) -> dict[str, list[dict]]:
    results: dict[str, list[dict]] = {}
    for dep in deps:
        vulns = query_osv(dep)
        if vulns:
            results[dep.name] = vulns
        time.sleep(delay)  # stay well under OSV's public rate limit
    return results

The delay defaults to 100 ms — conservative enough for 200-package requirements files without being annoying. OSV's public rate limit sits around 10 req/s; we stay well below it.

Step 3 — Parse and Rank the Findings

Raw OSV responses are verbose. We extract CVE IDs, severity, and the earliest fixed version:

SEVERITY_ORDER = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3, "UNKNOWN": 4}

def cvss_to_severity(score: float) -> str:
    if score >= 9.0:
        return "CRITICAL"
    elif score >= 7.0:
        return "HIGH"
    elif score >= 4.0:
        return "MEDIUM"
    elif score > 0:
        return "LOW"
    return "UNKNOWN"

def extract_finding(pkg_name: str, vuln: dict, pinned: bool) -> dict:
    aliases = vuln.get("aliases", [])
    cve_ids = [a for a in aliases if a.startswith("CVE-")]

    severity = "UNKNOWN"
    for sev in vuln.get("severity", []):
        if sev.get("type") in ("CVSS_V3", "CVSS_V4"):
            try:
                numeric = float(sev.get("score", "").split("/")[0])
                severity = cvss_to_severity(numeric)
            except (ValueError, IndexError):
                pass

    fixed_versions = []
    for affected in vuln.get("affected", []):
        for rng in affected.get("ranges", []):
            for event in rng.get("events", []):
                if "fixed" in event:
                    fixed_versions.append(event["fixed"])

    return {
        "package": pkg_name,
        "pinned": pinned,
        "osv_id": vuln.get("id"),
        "cve_ids": cve_ids,
        "severity": severity,
        "summary": vuln.get("summary", ""),
        "fixed_in": sorted(set(fixed_versions)),
    }

def generate_report(
    scan_results: dict[str, list[dict]], pinned_map: dict[str, bool]
) -> list[dict]:
    findings = []
    for pkg, vulns in scan_results.items():
        for vuln in vulns:
            findings.append(extract_finding(pkg, vuln, pinned=pinned_map.get(pkg, False)))
    return sorted(findings, key=lambda f: SEVERITY_ORDER.get(f["severity"], 5))

Step 4 — CLI Entry Point with CI Gate

import json, sys, argparse

def main():
    parser = argparse.ArgumentParser(
        description="Scan a Python requirements file for known CVEs via OSV"
    )
    parser.add_argument("requirements", help="Path to requirements.txt or requirements.lock")
    parser.add_argument("--output", choices=["json", "text"], default="text")
    parser.add_argument(
        "--fail-on",
        choices=["critical", "high", "medium"],
        default=None,
        help="Exit with code 1 if any finding is at or above this severity"
    )
    args = parser.parse_args()

    deps = parse_requirements(args.requirements)
    pinned_map = {d.name: d.version is not None for d in deps}
    print(f"[*] Scanning {len(deps)} packages ({sum(pinned_map.values())} pinned)...", file=sys.stderr)

    scan_results = scan_dependencies(deps)
    report = generate_report(scan_results, pinned_map)

    if args.output == "json":
        print(json.dumps(report, indent=2))
    else:
        if not report:
            print("[+] No vulnerabilities found.")
        for f in report:
            cves = ", ".join(f["cve_ids"]) or "no CVE alias"
            fixed = ", ".join(f["fixed_in"]) or "no fix available yet"
            label = "(pinned)" if f["pinned"] else "(UNPINNED)"
            print(f"[{f['severity']}] {f['package']} {label}")
            print(f"  {f['osv_id']} | {cves}")
            print(f"  {f['summary']}")
            print(f"  Fixed in: {fixed}\n")

    if args.fail_on and report:
        threshold = {"critical": 0, "high": 1, "medium": 2}[args.fail_on]
        blocking = [f for f in report if SEVERITY_ORDER.get(f["severity"], 5) <= threshold]
        if blocking:
            print(f"[!] {len(blocking)} finding(s) at or above '{args.fail_on}'. Failing build.", file=sys.stderr)
            sys.exit(1)

if __name__ == "__main__":
    main()

Running it looks like:

$ python vuln_scan.py requirements.txt --fail-on high
[*] Scanning 47 packages (47 pinned)...
[CRITICAL] jinja2 (pinned)
  GHSA-h5c8-rqwp-cp95 | CVE-2024-56201
  Jinja2 allows an attacker to execute arbitrary code via crafted templates
  Fixed in: 3.1.5

[HIGH] cryptography (pinned)
  GHSA-3ww4-gg4f-jr7f | CVE-2024-26130
  NULL pointer dereference in PKCS12 serialization
  Fixed in: 42.0.4

[!] 2 finding(s) at or above 'high'. Failing build.

Integrating into GitHub Actions

The exit code makes CI integration a one-liner. Here is a complete job definition:

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install httpx
      - name: Export JSON report
        run: python vuln_scan.py requirements.txt --output json > vuln-report.json
      - name: Fail on HIGH or CRITICAL
        run: python vuln_scan.py requirements.txt --fail-on high
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: vuln-report
          path: vuln-report.json

The JSON artifact lets you post findings to Slack, create Jira issues, or feed a dashboard — without depending on a third-party integration. Running the scanner twice (once to export, once to gate) is intentional: it keeps the exit-code logic clean without mixing it into the report generation.

One practical note: unpinned dependencies produce noisy results because OSV returns all historical CVEs, not just ones affecting a specific version range. Lock every dependency (pip freeze > requirements.txt or use pip-compile from pip-tools). Pinning is the foundation of any dependency hygiene program — if you want a broader checklist that covers this alongside other Python application security controls, our free security hardening checklists include a dedicated Python track.

The Takeaway

The full scanner is about 130 lines of Python with a single runtime dependency (httpx). It handles the cases off-the-shelf tools miss:

Scans lockfiles without installing packages into an environment
Integrates cleanly into custom pipelines via JSON output and exit codes
Uses a free, authoritative data source (OSV) with no rate-limit anxiety at typical project sizes
Distinguishes pinned (precise CVE match) from unpinned (all-vulns, noisy) packages

Extend it with SBOM output, Slack webhooks, or a SQLite cache to avoid re-querying packages that haven't changed since the last scan. The skeleton is intentionally minimal so you can add exactly what your pipeline needs — nothing more.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Build a Log Parser That Detects Brute Force Attacks

Ayi NEDJIMI — Fri, 07 Aug 2026 10:03:01 +0000

Brute force attacks against SSH, web logins, and APIs are constant background noise on any public-facing server. The problem is not knowing they happen — your logs already record every failed attempt. The problem is catching the pattern fast enough to block the source before it succeeds, and doing it without generating so many alerts that your team starts ignoring them.

This post walks through building a practical log parser in Python that identifies brute force patterns using a sliding window, with configurable thresholds and structured JSON output you can pipe into any alerting or blocking pipeline.

What counts as brute force?

A working definition matters before writing a single line of code. In log analysis, a brute force attack typically exhibits:

Multiple failed authentication attempts from the same source IP
Within a short time window (commonly 30–120 seconds)
Targeting the same service — SSH, a login endpoint, or an API

The tunable knobs are threshold (how many failures trigger an alert) and window (over what duration). Set the threshold too low and you flag a user who mistyped their password twice. Set the window too large and you fire an alert hours after the attacker has moved on.

We will start with sshd logs (/var/log/auth.log on Debian/Ubuntu, /var/log/secure on RHEL). The same pattern extends to nginx or any structured log format.

Parsing sshd auth logs

A typical failed SSH attempt looks like:

Aug  7 04:23:11 prod-01 sshd[1842]: Failed password for invalid user admin from 203.0.113.5 port 52341 ssh2
Aug  7 04:23:12 prod-01 sshd[1842]: Failed password for root from 203.0.113.5 port 52342 ssh2

The following parser extracts the timestamp, source IP, and attempted username from every such line:

import re
from datetime import datetime
from typing import Iterator, NamedTuple

class AuthEvent(NamedTuple):
    timestamp: datetime
    source_ip: str
    username: str
    raw_line: str

# Handles both "Failed password for root" and "Failed password for invalid user admin"
SSH_PATTERN = re.compile(
    r"(\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})"           # "Aug  7 04:23:11"
    r".+?Failed password for (?:invalid user )?(\S+)"    # username
    r" from (\d{1,3}(?:\.\d{1,3}){3})"                  # IPv4 source
)

def parse_auth_log(filepath: str, year: int = 2026) -> Iterator[AuthEvent]:
    with open(filepath, encoding="utf-8", errors="replace") as fh:
        for line in fh:
            m = SSH_PATTERN.search(line)
            if not m:
                continue
            raw_ts, username, ip = m.groups()
            # auth.log omits the year — inject it
            ts = datetime.strptime(f"{year} {raw_ts.strip()}", "%Y %b %d %H:%M:%S")
            yield AuthEvent(ts, ip, username, line.rstrip())

One common gotcha: auth.log does not include the year. You inject it manually, but watch for year-boundary wrap-around if you parse rotated logs from late December.

Detecting brute force with a sliding window

The detection logic maintains a per-IP bucket of recent events and evicts anything outside the current window before each check:

from collections import defaultdict
from datetime import timedelta

def detect_brute_force(
    events: Iterator[AuthEvent],
    threshold: int = 5,
    window_seconds: int = 60,
) -> list[dict]:
    buckets: dict[str, list[AuthEvent]] = defaultdict(list)
    incidents: list[dict] = []
    fired: set[tuple] = set()  # dedup key: (ip, window_start)

    for event in events:
        ip = event.source_ip
        cutoff = event.timestamp - timedelta(seconds=window_seconds)

        # Evict stale events first, then append the new one
        buckets[ip] = [e for e in buckets[ip] if e.timestamp >= cutoff]
        buckets[ip].append(event)

        if len(buckets[ip]) >= threshold:
            window_start = buckets[ip][0].timestamp
            key = (ip, window_start)
            if key not in fired:
                fired.add(key)
                incidents.append({
                    "ip": ip,
                    "first_seen": window_start.isoformat(),
                    "last_seen": event.timestamp.isoformat(),
                    "attempt_count": len(buckets[ip]),
                    "usernames_tried": sorted({e.username for e in buckets[ip]}),
                    "window_seconds": window_seconds,
                })

    return incidents

A few design decisions worth noting:

Evict before counting. If you count then evict, you get off-by-one errors at the window boundary. Trim first, always.

Dedup by (ip, window_start). Without this, a 20-attempt burst fires 16 separate alerts — one for every event after the threshold. This approach fires exactly once per distinct burst, regardless of how large it grows.

Collect usernames. An attacker trying root, admin, ubuntu, pi, and deploy in 30 seconds tells you whether this is automated credential stuffing or something more targeted. That context matters when you are writing the incident report.

Wiring it into a CLI

import json
import sys

def main() -> None:
    log_path    = sys.argv[1] if len(sys.argv) > 1 else "/var/log/auth.log"
    threshold   = int(sys.argv[2]) if len(sys.argv) > 2 else 5
    window      = int(sys.argv[3]) if len(sys.argv) > 3 else 60

    events    = parse_auth_log(log_path)
    incidents = detect_brute_force(events, threshold=threshold, window_seconds=window)

    if not incidents:
        print("No brute force activity detected.")
        return

    print(json.dumps(incidents, indent=2))
    sys.stderr.write(f"[!] {len(incidents)} incident(s) detected\n")

if __name__ == "__main__":
    main()

Running against a real log from an exposed VPS produces output like:

[
  {
    "ip": "203.0.113.5",
    "first_seen": "2026-08-07T04:23:11",
    "last_seen": "2026-08-07T04:23:42",
    "attempt_count": 23,
    "usernames_tried": ["admin", "deploy", "pi", "root", "ubuntu"],
    "window_seconds": 60
  }
]

Connecting to blocking and alerting pipelines

The JSON output is intentionally pipeline-friendly.

Auto-block with iptables:

python bf_parser.py /var/log/auth.log | \
  jq -r '.[].ip' | \
  xargs -I{} sudo iptables -A INPUT -s {} -j DROP

Forward to a Slack or webhook endpoint:

import requests

def send_alert(incident: dict, webhook_url: str) -> None:
    text = (
        f":rotating_light: Brute force from {incident['ip']} — "
        f"{incident['attempt_count']} attempts in {incident['window_seconds']}s. "
        f"Users tried: {', '.join(incident['usernames_tried'])}"
    )
    requests.post(webhook_url, json={"text": text}, timeout=5)

If you run this across multiple servers, the in-memory buckets dict is the scaling constraint: it is per-process and per-host. Moving it to Redis with sorted sets and ZRANGEBYSCORE for window queries lets you aggregate across a fleet and survive process restarts without losing event history.

Brute force detection covers one layer. Hardening the service itself — SSH key-only auth, fail2ban tuning, firewall ingress rules, and sudoers auditing — covers the rest. The security hardening checklists we publish walk through the full SSH and Linux server surface step by step, including the specific sshd_config options that eliminate most of this log noise in the first place.

The takeaway

The core logic is around 60 lines of Python. What makes it production-useful rather than a toy:

Regex tailored to your log format — generic parsers miss edge cases like the invalid user prefix sshd adds for non-existent accounts
Sliding window with explicit eviction — naive counting over fixed time buckets misses cross-boundary bursts
Deduplication on burst identity — one burst, one alert, regardless of how many events it contains
Structured JSON output — the next tool in the chain does not need to parse your output

Tune threshold and window against real data. A fresh VPS sees its first scan within minutes of getting a public IP — run this parser on a week of logs to find natural thresholds that separate legitimate failures from attack traffic before you start auto-blocking.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Implementing TOTP Two-Factor Authentication from Scratch in Python

Ayi NEDJIMI — Thu, 06 Aug 2026 10:05:28 +0000

Most developers reach for a library when they need TOTP. That's fine for production — but if you don't understand what's underneath, you'll misconfigure the time window, miss the replay-attack problem, or ship without recovery codes. This article builds TOTP from the RFC spec so you know what every line is doing.

How TOTP Works (RFC 6238 Internals)

TOTP (Time-based One-Time Password, RFC 6238) is an extension of HOTP (RFC 4226). HOTP generates a code from an HMAC-SHA1 of a shared secret and an incrementing counter. TOTP replaces the counter with a time step derived from the current Unix timestamp:

T = floor(unix_timestamp / step)  # step = 30 seconds by default

The code-generation pipeline is:

Compute T = floor(time.time() / 30) — the current 30-second window index
Pack T as a big-endian 8-byte unsigned integer
HMAC-SHA1(secret_bytes, T_bytes) → 20-byte digest
Dynamic truncation: take the last nibble of the digest as an offset; extract 4 bytes at that offset; clear the sign bit; take the result modulo 1,000,000

The shared secret is a random byte string encoded in base32. Both the server and the authenticator app perform these steps independently and compare the result — no secret travels over the wire after enrollment.

Pure Python TOTP from Scratch

Here's a complete implementation using only Python's standard library — no external dependencies:

import hmac
import hashlib
import struct
import time
import base64
import os

def generate_secret(length: int = 20) -> str:
    # Generate a random base32-encoded TOTP secret (160-bit).
    return base64.b32encode(os.urandom(length)).decode("utf-8")

def _hotp(secret: str, counter: int) -> int:
    # Compute HOTP per RFC 4226, Section 5.3.
    key = base64.b32decode(secret.upper(), casefold=True)
    msg = struct.pack(">Q", counter)           # big-endian unsigned 64-bit
    digest = hmac.new(key, msg, hashlib.sha1).digest()

    offset = digest[-1] & 0x0F                # dynamic offset
    code = struct.unpack(">I", digest[offset:offset + 4])[0]
    code &= 0x7FFFFFFF                         # clear the sign bit
    return code % 1_000_000

def generate_totp(secret, timestamp=None, step=30):
    # Return a zero-padded 6-digit TOTP code for the current time window.
    t = int((timestamp or time.time()) / step)
    return str(_hotp(secret, t)).zfill(6)

def verify_totp(secret, code, timestamp=None, step=30, window=1):
    # Verify a TOTP code, accepting codes from adjacent time windows.
    # window=1 tolerates +/-30 seconds of clock skew between client and server.
    # Use hmac.compare_digest to prevent timing-based side-channel attacks.
    t = int((timestamp or time.time()) / step)
    for delta in range(-window, window + 1):
        expected = str(_hotp(secret, t + delta)).zfill(6)
        if hmac.compare_digest(expected, code):
            return True
    return False

A few implementation decisions worth calling out:

hmac.compare_digest is mandatory. Plain == comparison leaks timing information. An attacker measuring response times can narrow down the correct code digit by digit. compare_digest runs in constant time regardless of how many characters match.
os.urandom, never random. Python's random module is a PRNG seeded from a deterministic state. os.urandom pulls from the OS's cryptographically secure entropy pool.
window=1 is the RFC recommendation. It accepts the previous, current, and next 30-second code — covering ±30s of clock drift. Widening to 2 or more doubles or triples the effective brute-force window.

Generating QR Codes for Authenticator Apps

Users provision TOTP by scanning a QR code that encodes an otpauth:// URI. The format is standardized across Google Authenticator, Aegis, Bitwarden, and every RFC 6238-compatible app:

from urllib.parse import quote

def build_otpauth_uri(secret, account, issuer="MyApp"):
    label = quote(f"{issuer}:{account}", safe="")
    params = (
        f"secret={secret}"
        f"&issuer={quote(issuer)}"
        f"&algorithm=SHA1"
        f"&digits=6"
        f"&period=30"
    )
    return f"otpauth://totp/{label}?{params}"

def print_qr_to_terminal(uri):
    import qrcode  # pip install qrcode[pil]
    qr = qrcode.QRCode(border=1)
    qr.add_data(uri)
    qr.make(fit=True)
    qr.print_ascii()

# Enrollment flow
secret = generate_secret()
uri = build_otpauth_uri(secret, account="alice@example.com", issuer="MyApp")
print_qr_to_terminal(uri)
# Save `secret` encrypted to the database; display QR once, never again

FastAPI Integration with Replay Attack Prevention

A minimal REST API with two routes — enroll and verify. The important addition here is the replay-attack guard: without it, a valid code intercepted by a man-in-the-middle can be reused within the same 30-second window.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# Replace with encrypted DB storage in production
_secrets = {}
_used_codes = {}

class EnrollResponse(BaseModel):
    otpauth_uri: str
    secret: str  # display once at enrollment; never again

class VerifyRequest(BaseModel):
    user_id: str
    code: str

@app.post("/totp/enroll/{user_id}", response_model=EnrollResponse)
def enroll(user_id: str):
    secret = generate_secret()
    _secrets[user_id] = secret
    _used_codes[user_id] = set()
    return EnrollResponse(
        otpauth_uri=build_otpauth_uri(secret, account=user_id),
        secret=secret,
    )

@app.post("/totp/verify")
def verify(req: VerifyRequest):
    secret = _secrets.get(req.user_id)
    if not secret:
        raise HTTPException(404, "User not enrolled")

    used = _used_codes.setdefault(req.user_id, set())
    if req.code in used:
        raise HTTPException(400, "Code already used")  # replay blocked

    if not verify_totp(secret, req.code):
        raise HTTPException(400, "Invalid or expired code")

    used.add(req.code)
    return {"status": "ok"}

Note the two layers of protection: the TOTP algorithm itself (valid for at most 90 seconds with window=1) plus the explicit per-code deduplication that blocks within-window replay.

Common Mistakes Worth Avoiding

No enrollment confirmation. Store the secret only after the user successfully verifies their first code. This catches QR scan errors before you commit a broken secret to the database and lock the user out immediately.

No recovery codes. If a device is lost, the user needs a way back in. Generate 8–10 single-use recovery codes at enrollment, store them bcrypt-hashed (not SHA-256 — short codes need a slow hash), and let users redeem them one at a time.

Logging TOTP codes. Short-lived credentials still qualify as credentials. Make sure your request logging middleware strips the code field before it reaches your log aggregator or error tracker.

Storing secrets in plaintext. TOTP secrets deserve the same treatment as passwords: encrypt at rest with a key managed separately from the database, such as envelope encryption via a KMS.

For a structured checklist covering TOTP enrollment flows, brute-force lockout, session binding, and secure logout, the free security hardening checklists at AYI NEDJIMI Consultants are a solid starting point.

The Takeaway

The TOTP algorithm itself is about 30 lines of Python. The security properties come from the surrounding decisions: constant-time comparison, replay prevention, proper entropy for secrets, and recovery paths for lost devices.

Using a library like pyotp in production is reasonable — it's well-tested and saves maintenance overhead. But having walked through the spec, you'll recognize when a configuration option actually matters and when a tutorial is quietly skipping the parts that get users locked out.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building a DMARC Report Parser and Dashboard in Python

Ayi NEDJIMI — Wed, 05 Aug 2026 10:02:41 +0000

DMARC aggregate reports land in your mailbox every day as gzip-compressed XML attachments. Most teams configure DMARC once and forget it exists. That's a problem: those reports tell you who's sending email on behalf of your domain, what fraction is passing authentication, and whether someone is actively spoofing you.

This article walks through building a parser and terminal dashboard in Python — no third-party libraries beyond the standard library.

What's Inside a DMARC Aggregate Report

DMARC aggregate reports follow RFC 7489. Each report is a ZIP or gzip archive wrapping an XML file. The relevant structure:

<feedback>
  <report_metadata>
    <org_name>Google</org_name>
    <date_range>
      <begin>1720742400</begin>
      <end>1720828799</end>
    </date_range>
  </report_metadata>
  <policy_published>
    <domain>example.com</domain>
    <p>reject</p>
  </policy_published>
  <record>
    <row>
      <source_ip>209.85.220.41</source_ip>
      <count>47</count>
      <policy_evaluated>
        <disposition>none</disposition>
        <dkim>pass</dkim>
        <spf>pass</spf>
      </policy_evaluated>
    </row>
    <identifiers>
      <header_from>example.com</header_from>
    </identifiers>
  </record>
</feedback>

Each <record> represents a batch of messages from a single source IP. You want IP, count, SPF result, DKIM result, and disposition for every record.

Building the Parser

Start with a parser that handles both ZIP and gzip formats — Google, Microsoft, and Yahoo all use slightly different packaging.

import gzip
import zipfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import socket
from datetime import datetime


@dataclass
class DmarcRecord:
    source_ip: str
    count: int
    disposition: str
    dkim: str
    spf: str
    header_from: str
    hostname: Optional[str] = None


@dataclass
class DmarcReport:
    org_name: str
    domain: str
    begin: datetime
    end: datetime
    policy: str
    records: list[DmarcRecord] = field(default_factory=list)


def _extract_xml(path: Path) -> str:
    if path.suffix == ".zip":
        with zipfile.ZipFile(path) as zf:
            xml_name = next(n for n in zf.namelist() if n.endswith(".xml"))
            return zf.read(xml_name).decode("utf-8")
    elif path.suffix in (".gz", ".gzip"):
        with gzip.open(path, "rb") as f:
            return f.read().decode("utf-8")
    else:
        return path.read_text()


def parse_report(path: Path, resolve_hostnames: bool = False) -> DmarcReport:
    xml_content = _extract_xml(path)
    root = ET.fromstring(xml_content)

    meta = root.find("report_metadata")
    policy = root.find("policy_published")

    report = DmarcReport(
        org_name=meta.findtext("org_name", "unknown"),
        domain=policy.findtext("domain", "unknown"),
        begin=datetime.utcfromtimestamp(int(meta.findtext("date_range/begin", "0"))),
        end=datetime.utcfromtimestamp(int(meta.findtext("date_range/end", "0"))),
        policy=policy.findtext("p", "none"),
    )

    for record in root.findall("record"):
        row = record.find("row")
        idents = record.find("identifiers")
        evaluated = row.find("policy_evaluated")

        ip = row.findtext("source_ip", "")
        hostname = None
        if resolve_hostnames and ip:
            try:
                hostname = socket.gethostbyaddr(ip)[0]
            except socket.herror:
                hostname = None

        report.records.append(
            DmarcRecord(
                source_ip=ip,
                count=int(row.findtext("count", "0")),
                disposition=evaluated.findtext("disposition", "none"),
                dkim=evaluated.findtext("dkim", "unknown"),
                spf=evaluated.findtext("spf", "unknown"),
                header_from=idents.findtext("header_from", ""),
                hostname=hostname,
            )
        )

    return report

The resolve_hostnames flag is off by default. Reverse DNS adds latency when processing hundreds of reports in a batch.

Aggregating Reports and Building the Dashboard

A single report covers one day and one reporting organization. You'll receive several per domain per day. This function aggregates them into a unified pass/fail summary:

from collections import defaultdict


def aggregate_reports(paths: list[Path]) -> dict:
    totals: dict = defaultdict(lambda: {"pass": 0, "fail": 0, "ips": defaultdict(int)})

    for path in paths:
        try:
            report = parse_report(path)
        except Exception as e:
            print(f"[WARN] Could not parse {path.name}: {e}")
            continue

        for rec in report.records:
            domain = report.domain
            both_pass = rec.dkim == "pass" and rec.spf == "pass"

            if both_pass:
                totals[domain]["pass"] += rec.count
            else:
                totals[domain]["fail"] += rec.count

            totals[domain]["ips"][rec.source_ip] += rec.count

    return dict(totals)


def print_dashboard(aggregated: dict) -> None:
    print(f"\n{'Domain':<30} {'Pass':>8} {'Fail':>8} {'Pass%':>8}  Top Sender IP")
    print("-" * 75)
    for domain, stats in sorted(aggregated.items()):
        total = stats["pass"] + stats["fail"]
        pct = (stats["pass"] / total * 100) if total else 0.0
        top_ip = max(stats["ips"], key=stats["ips"].get, default="—")
        flag = "  ⚠ low pass rate" if pct < 95 else ""
        print(f"{domain:<30} {stats['pass']:>8} {stats['fail']:>8} {pct:>7.1f}%  {top_ip}{flag}")
    print()

The ⚠ threshold at 95% is a starting point. Adjust to your domain's established baseline once you have a few weeks of data.

Pulling Reports Directly from Your Inbox

DMARC reports arrive as email attachments. Point them to a dedicated folder in your mailbox via an email filter, then pull them with this IMAP snippet:

import imaplib
import email
import tempfile
import os

IMAP_HOST = os.environ["IMAP_HOST"]
IMAP_USER = os.environ["IMAP_USER"]
IMAP_PASS = os.environ["IMAP_PASS"]
IMAP_FOLDER = os.environ.get("IMAP_FOLDER", "DMARC")


def fetch_attachments(dest_dir: Path) -> list[Path]:
    saved: list[Path] = []
    with imaplib.IMAP4_SSL(IMAP_HOST) as imap:
        imap.login(IMAP_USER, IMAP_PASS)
        imap.select(IMAP_FOLDER)
        _, data = imap.search(None, "UNSEEN")
        for uid in data[0].split():
            _, msg_data = imap.fetch(uid, "(RFC822)")
            msg = email.message_from_bytes(msg_data[0][1])
            for part in msg.walk():
                fn = part.get_filename()
                if fn and (fn.endswith(".zip") or fn.endswith(".gz")):
                    dest = dest_dir / fn
                    dest.write_bytes(part.get_payload(decode=True))
                    saved.append(dest)
    return saved


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as tmp:
        tmpdir = Path(tmp)
        attachments = fetch_attachments(tmpdir)
        print(f"Fetched {len(attachments)} report(s).")
        aggregated = aggregate_reports(attachments)
        print_dashboard(aggregated)

Four environment variables, one cron job, and you have nightly DMARC visibility with zero external dependencies.

What Patterns Actually Matter

Once the parser is running, focus on these signals:

Sudden fail spike from a known IP. A legitimate email service provider that previously passed DKIM and SPF starts failing. This almost always means a key rotation or header misconfiguration on their side — your DMARC report catches it before customers see quarantined mail.

Unknown source IPs at volume. An IP you don't recognize is sending 500+ messages per day claiming your domain as sender. That's either shadow IT using an unlisted ESP, or active spoofing at scale. Either way, you need to know.

Disposition mismatch. Your policy is p=reject but messages with both DKIM and SPF failures show disposition none. This usually means the reporting organization applied a local override (forwarding scenarios are common). Understand the override before assuming your policy is enforced end-to-end.

Gradual policy ramp-up. Most production deployments start at p=none, move to p=quarantine, then p=reject over 60–90 days. The dashboard lets you track the fail rate trend and move confidently between stages.

For organizations setting up email authentication from scratch, the email and domain security hardening checklist from AYI NEDJIMI Consultants covers SPF flattening, DKIM key sizes, DMARC ramp-up timelines, and a monitoring cadence you can adapt to this parser.

The Takeaway

DMARC aggregate reports are free, daily, and machine-readable. Most teams configure the DNS record and never look at the data again. A 250-line Python script built entirely on the standard library turns that ignored stream into a daily authentication health check: who's sending on your behalf, what's failing, and whether someone is spoofing your domain.

Run it as a cron job. Pipe the output to a Slack webhook or write the fail counts to a time-series metric. The infrastructure footprint is essentially zero.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Build a Canary Token System for Breach Detection

Ayi NEDJIMI — Tue, 04 Aug 2026 10:14:45 +0000

Attackers who breach your systems rarely announce themselves. They move laterally, exfiltrate data quietly, and by the time your SIEM fires an alert, the damage is done. Canary tokens flip this dynamic: you plant fake credentials, URLs, or files in tempting locations, and when an intruder accesses them, you get an immediate alert. No signatures, no heuristics -- just a tripwire that fires precisely when someone is where they should not be.

What Is a Canary Token?

A canary token is a resource that has no legitimate use inside your environment. It could be:

A URL embedded in a credentials.json file on a shared drive
A fake AWS access key that triggers an alert on first use
A DNS token embedded in a document that beacons home when opened

The key property: no legitimate process or user should ever access it. Any hit is a signal, full stop.

Services like canarytokens.org provide hosted versions, but running your own gives you control over the data, no rate limits, and custom alerting logic. Here is how to build a minimal but production-viable version in Python.

Setting Up the Token Server

We need an HTTP server that accepts requests on unique token URLs, logs the hit (IP, user-agent, timestamp), and fires an alert. FastAPI handles this cleanly.

# canary_server.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import httpx, sqlite3, os
from datetime import datetime

app = FastAPI()
DB_PATH = 'canary.db'
TELEGRAM_TOKEN = os.environ['TELEGRAM_BOT_TOKEN']
TELEGRAM_CHAT_ID = os.environ['TELEGRAM_CHAT_ID']

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        'CREATE TABLE IF NOT EXISTS hits '
        '(id INTEGER PRIMARY KEY AUTOINCREMENT, token TEXT NOT NULL, '
        'ip TEXT, user_agent TEXT, headers TEXT, timestamp TEXT)'
    )
    conn.commit()
    conn.close()

def log_hit(token: str, ip: str, user_agent: str, headers: str):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        'INSERT INTO hits (token, ip, user_agent, headers, timestamp) VALUES (?, ?, ?, ?, ?)',
        (token, ip, user_agent, headers, datetime.utcnow().isoformat()),
    )
    conn.commit()
    conn.close()

async def send_alert(token: str, ip: str, user_agent: str):
    msg = f'Canary token fired!\nToken: {token}\nIP: {ip}\nUA: {user_agent}'
    async with httpx.AsyncClient() as client:
        await client.post(
            f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage',
            json={'chat_id': TELEGRAM_CHAT_ID, 'text': msg},
        )

@app.on_event('startup')
def startup():
    init_db()

@app.get('/t/{token}')
async def canary_hit(token: str, request: Request):
    ip = request.client.host
    ua = request.headers.get('user-agent', '')
    log_hit(token, ip, ua, str(dict(request.headers)))
    await send_alert(token, ip, ua)
    return JSONResponse({'status': 'ok'})

Run it with uvicorn canary_server:app --host 0.0.0.0 --port 8080 behind a reverse proxy with TLS. Use a generic-looking domain so the server is not identifiable as a canary.

Generating Tokens for Different Trap Types

A UUID per token is enough for uniqueness. The interesting part is where and how you embed it.

# canary_gen.py
import uuid, json

BASE_URL = 'https://canary.yourdomain.com/t'

def new_token() -> str:
    return uuid.uuid4().hex[:16]

def fake_aws_credentials(token: str) -> str:
    key_id = 'AKIA' + token.upper()[:16]
    secret = uuid.uuid4().hex + uuid.uuid4().hex
    return f'[internal-backup]\naws_access_key_id = {key_id}\naws_secret_access_key = {secret}\n'

def fake_api_config(token: str) -> dict:
    return {
        'api_key': f'sk-internal-{token}',
        'endpoint': f'{BASE_URL}/{token}',
        'environment': 'staging',
    }

def web_bug_html(token: str) -> str:
    return f'<img src="{BASE_URL}/{token}" width="1" height="1" style="display:none" />'

if __name__ == '__main__':
    t = new_token()
    print(f'Token: {t}')
    print(fake_aws_credentials(t))
    print(json.dumps(fake_api_config(t), indent=2))
    print(web_bug_html(t))

Each trap targets a different threat:

Fake AWS credentials planted in an S3 bucket or backup archive catch credential harvesters
JSON config files in old git branches catch attackers who clone archived repositories
Web bugs embedded in sensitive exported PDFs catch anyone who opens the document

Where to Plant Tokens

Placement strategy matters more than the token format. High-value locations:

Location	What it catches
`~/.aws/` backup archives on S3	Server compromise, backup exfiltration
Old git branches named `config-backup`	Internal repository scraping
Internal wiki API Keys or Credentials page	Insider threat or phishing pivot
Docker image build args in CI logs	CI/CD pipeline breach
`/etc/shadow` backup files in storage buckets	Privileged credential hunting

One token per location. Register each one with name and date so that when a token fires, you know exactly which asset was accessed -- this matters for incident scoping.

For teams that want a structured approach to pairing canary tokens with other detection controls, our free security hardening checklists cover this alongside network segmentation and secrets management.

Handling Alert Noise Without Missing Real Hits

A single token firing at 3 AM is urgent. A thousand tokens firing in 10 seconds is a scanner or a misconfigured monitoring job. Add deduplication:

import time
from collections import defaultdict

_recent_alerts: dict[str, float] = defaultdict(float)
DEBOUNCE_SECONDS = 60

async def maybe_alert(token: str, ip: str, user_agent: str):
    now = time.time()
    if now - _recent_alerts[token] < DEBOUNCE_SECONDS:
        return
    _recent_alerts[token] = now
    await send_alert(token, ip, user_agent)

Replace send_alert in the route handler with maybe_alert. For production, use Redis with SET NX EX instead of an in-process dict so deduplication survives restarts.

Enrich each alert before sending: a GeoIP lookup tells you whether the hit came from a Tor exit node, a known cloud provider range, or your office VPN. That context changes how fast you escalate.

The Takeaway

A canary token system has three properties that make it worth the few hours to set up:

Zero false positives by design. No legitimate process accesses them, so every hit is meaningful. This is genuinely rare in detection work.

Cheap to maintain. Once planted, tokens sit idle until triggered. Unlike SIEM rules that need constant tuning, a canary just waits.

Works when everything else fails. If your perimeter is compromised and your logging pipeline is blinded, a canary that beacons to an external server still fires. It operates independently of your internal observability stack.

The main operational risk is losing track of what you planted. Keep a register, review it when systems are decommissioned, and make sure newly onboarded infrastructure gets fresh tokens. Start with three tokens in your three most sensitive locations. Expand from there.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists -- PDF and Excel.

Implementing Content Security Policy (CSP) Generation with Go

Ayi NEDJIMI — Mon, 03 Aug 2026 10:05:17 +0000

Content Security Policy headers are one of those things every web team knows they should have but often skips — the syntax is verbose, the directives are easy to misconfigure, and a wrong policy breaks the app silently in production. Writing a small generator in Go removes the human error and makes CSP a first-class part of your build.

Why CSP matters and where it fails

CSP is an HTTP response header that tells the browser which resources — scripts, styles, images, fonts — are allowed to load and from where. A well-crafted policy defeats cross-site scripting (XSS) attacks even when an attacker has already injected content into your HTML, because the browser simply refuses to execute disallowed scripts.

The problem is maintenance. Teams start with a permissive policy (default-src 'self'), then add exceptions for every CDN and vendor widget until the header looks like an allow-list for the entire internet. At that point the policy is longer than it is useful.

Building a generator means you describe intent — "scripts from our CDN only, no inline styles, report violations to this endpoint" — and get a validated, reproducible header string every time.

Designing the CSP struct

A CSP is a collection of directives. Each directive has a name (script-src, style-src, img-src, …) and one or more source values. In Go, the natural representation is a struct with typed fields:

package csp

import (
    "fmt"
    "strings"
)

type Policy struct {
    DefaultSrc              []string
    ScriptSrc               []string
    StyleSrc                []string
    ImgSrc                  []string
    FontSrc                 []string
    ConnectSrc              []string
    FrameSrc                []string
    ObjectSrc               []string
    ReportURI               string
    UpgradeInsecureRequests bool
}

func (p Policy) Build() string {
    var parts []string

    add := func(name string, values []string) {
        if len(values) > 0 {
            parts = append(parts, fmt.Sprintf("%s %s", name, strings.Join(values, " ")))
        }
    }

    add("default-src", p.DefaultSrc)
    add("script-src", p.ScriptSrc)
    add("style-src", p.StyleSrc)
    add("img-src", p.ImgSrc)
    add("font-src", p.FontSrc)
    add("connect-src", p.ConnectSrc)
    add("frame-src", p.FrameSrc)
    add("object-src", p.ObjectSrc)

    if p.UpgradeInsecureRequests {
        parts = append(parts, "upgrade-insecure-requests")
    }
    if p.ReportURI != "" {
        parts = append(parts, "report-uri "+p.ReportURI)
    }

    return strings.Join(parts, "; ")
}

This gives you a Build() method that emits a valid header value. Fields you leave nil simply do not appear in the output — no empty directives to confuse the browser.

Wiring it into a Go HTTP middleware

The generator is only useful if the header is actually sent. Here is a middleware for the standard net/http library that takes a Policy and attaches it to every response:

package csp

import "net/http"

func Middleware(p Policy) func(http.Handler) http.Handler {
    header := p.Build()
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            w.Header().Set("Content-Security-Policy", header)
            next.ServeHTTP(w, r)
        })
    }
}

Usage in main.go:

policy := csp.Policy{
    DefaultSrc:              []string{"'self'"},
    ScriptSrc:               []string{"'self'", "https://cdn.example.com"},
    StyleSrc:                []string{"'self'", "'unsafe-inline'"},
    ImgSrc:                  []string{"'self'", "data:", "https://images.example.com"},
    ObjectSrc:               []string{"'none'"},
    ReportURI:               "https://csp-report.example.com/collect",
    UpgradeInsecureRequests: true,
}

mux := http.NewServeMux()
mux.HandleFunc("/", homeHandler)

http.ListenAndServe(":8080", csp.Middleware(policy)(mux))

The header value is computed once at startup, not on every request. At scale, that matters.

Handling nonces for inline scripts

'unsafe-inline' in script-src wipes out most of CSP's XSS protection. The correct alternative is nonces: a per-request cryptographically random value that must appear both in the header and on each <script> tag. The browser rejects any inline script that does not carry the matching nonce.

Nonces require the header to be generated per-request rather than once at startup:

package csp

import (
    "context"
    "crypto/rand"
    "encoding/base64"
    "fmt"
    "net/http"
)

type contextKey struct{}

func generateNonce() (string, error) {
    b := make([]byte, 16)
    if _, err := rand.Read(b); err != nil {
        return "", err
    }
    return base64.StdEncoding.EncodeToString(b), nil
}

func NonceMiddleware(p Policy) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            nonce, err := generateNonce()
            if err != nil {
                http.Error(w, "internal error", http.StatusInternalServerError)
                return
            }

            adjusted := p
            adjusted.ScriptSrc = append(
                append([]string{}, p.ScriptSrc...),
                fmt.Sprintf("'nonce-%s'", nonce),
            )

            w.Header().Set("Content-Security-Policy", adjusted.Build())

            ctx := context.WithValue(r.Context(), contextKey{}, nonce)
            next.ServeHTTP(w, r.WithContext(ctx))
        })
    }
}

// NonceFromCtx retrieves the nonce for use in HTML templates.
func NonceFromCtx(ctx context.Context) string {
    v, _ := ctx.Value(contextKey{}).(string)
    return v
}

In your HTML template, the nonce goes on each inline script tag:

<script nonce="{{ .Nonce }}">
  // initialization code that must stay inline
</script>

With this in place you can drop 'unsafe-inline' from script-src entirely, which is the single most impactful change you can make to an existing CSP.

Testing the output

Before shipping, verify the generated header against known valid strings:

package csp_test

import (
    "strings"
    "testing"
)

func TestPolicyBuild(t *testing.T) {
    p := Policy{
        DefaultSrc: []string{"'self'"},
        ObjectSrc:  []string{"'none'"},
    }
    got := p.Build()
    if !strings.Contains(got, "default-src 'self'") {
        t.Errorf("missing default-src: %q", got)
    }
    if !strings.Contains(got, "object-src 'none'") {
        t.Errorf("missing object-src: %q", got)
    }
}

func TestEmptyFieldsOmitted(t *testing.T) {
    p := Policy{DefaultSrc: []string{"'self'"}}
    got := p.Build()
    if strings.Contains(got, "script-src") {
        t.Errorf("expected script-src to be omitted: %q", got)
    }
}

Run with go test ./.... You can also paste the generated header into Google's CSP Evaluator to catch gaps — it flags missing object-src 'none', which would otherwise allow plugin injection even in modern browsers.

For a broader view of HTTP security headers and hardening steps that complement CSP, our free security hardening checklists cover CSP alongside HSTS, CORS, authentication controls, and more.

The takeaway

Writing CSP by hand does not scale. A typed Go struct enforces that every directive is deliberate, lets you unit-test the output, and makes policy reviews part of code review rather than an afterthought buried in Nginx config.

The nonce approach is non-trivial to wire up but pays off immediately: you remove 'unsafe-inline' from script-src, which is the most common reason CSP fails to prevent XSS in practice.

If you maintain multiple services, the next logical step is externalizing the policy definition to a JSON or YAML file loaded at startup — one source of truth, all services consistent, policy diffs visible in git.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building an AI-powered phishing URL detector

Ayi NEDJIMI — Sun, 02 Aug 2026 10:06:41 +0000

Phishing remains one of the most effective attack vectors in 2026 — not because defenders are incompetent, but because attackers have gotten very good at making malicious URLs look legitimate. A URL like secure-login.paypal-account-verification.com/oauth passes a casual glance. Your users won't catch it. A well-built detector can.

This post walks through building a practical phishing URL detector in Python: feature engineering, training a classifier, and adding a language model escalation layer for ambiguous cases.

Why URL-based detection works

Most phishing URLs share structural patterns that legitimate URLs avoid. The domain has an unusual character distribution. The path is unusually long. The hostname contains brand names (paypal, netflix, google) in odd positions — a subdomain or path component rather than the registrable domain itself.

These signals aren't perfect individually, but they compose well into a classifier. The goal isn't 100% accuracy — it's adding a reliable, low-latency layer to your security stack that flags URLs worth inspecting before a user clicks.

Feature engineering

Before touching any model, decide what you're measuring. Here are the features that consistently show up in practice:

Domain length: phishing domains tend to be longer
Dot count in hostname: mail.secure.paypal.verification.com has more dots than paypal.com
Digit ratio: more digits often signals auto-generated hostnames
Presence of IP address: http://192.168.1.1/login is a classic tell
Use of URL shorteners (bit.ly, tinyurl, etc.)
Suspicious keywords in path/query: login, verify, account, secure, update
Entropy of the domain: high Shannon entropy suggests algorithmically generated names

import re
import math
from urllib.parse import urlparse
from collections import Counter

SUSPICIOUS_KEYWORDS = {"login", "verify", "account", "secure", "update", "confirm", "bank"}
URL_SHORTENERS = {"bit.ly", "tinyurl.com", "t.co", "goo.gl", "ow.ly"}

def shannon_entropy(s: str) -> float:
    if not s:
        return 0.0
    counts = Counter(s)
    total = len(s)
    return -sum((c / total) * math.log2(c / total) for c in counts.values())

def extract_features(url: str) -> dict:
    parsed = urlparse(url if url.startswith("http") else f"https://{url}")
    hostname = parsed.hostname or ""
    path = parsed.path or ""
    full = url.lower()
    digits = sum(c.isdigit() for c in hostname)
    return {
        "url_length": len(url),
        "hostname_length": len(hostname),
        "dot_count": hostname.count("."),
        "digit_ratio": digits / len(hostname) if hostname else 0,
        "has_ip": bool(re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", hostname)),
        "is_shortener": hostname in URL_SHORTENERS,
        "keyword_count": sum(kw in full for kw in SUSPICIOUS_KEYWORDS),
        "entropy": shannon_entropy(hostname),
        "path_length": len(path),
        "has_at_sign": "@" in url,
        "subdomain_count": max(0, len(hostname.split(".")) - 2),
        "uses_https": parsed.scheme == "https",
    }

Run this against a few URLs to sanity-check — paypal.com should score very differently from secure-paypal-login.verification-center.com.

Training a gradient boosting classifier

A gradient boosting model (scikit-learn's HistGradientBoostingClassifier) works well here: it handles the mix of binary and continuous features, trains fast, and is interpretable enough that you can explain why it flagged a URL.

For training data, the PhishTank dataset (phishing URLs) and the Tranco top-1M (legitimate URLs) are the standard sources. Here's the training scaffold:

import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import joblib

def load_dataset(phishing_path: str, legit_path: str) -> tuple[pd.DataFrame, pd.Series]:
    phishing_urls = pd.read_csv(phishing_path, names=["url"])["url"].dropna().tolist()
    legit_urls = pd.read_csv(legit_path, names=["url"])["url"].dropna().tolist()
    records = []
    for url in phishing_urls:
        feats = extract_features(url)
        feats["label"] = 1
        records.append(feats)
    for url in legit_urls:
        feats = extract_features(url)
        feats["label"] = 0
        records.append(feats)
    df = pd.DataFrame(records)
    return df.drop("label", axis=1), df["label"]

def train(phishing_path: str, legit_path: str, model_path: str = "phishing_model.joblib"):
    X, y = load_dataset(phishing_path, legit_path)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    clf = HistGradientBoostingClassifier(max_iter=200, learning_rate=0.05, max_depth=6)
    clf.fit(X_train, y_train)
    print(classification_report(y_test, clf.predict(X_test)))
    joblib.dump(clf, model_path)
    print(f"Model saved to {model_path}")

On a balanced dataset of ~100k URLs, this typically reaches 96-98% accuracy with a false positive rate under 2%. That is a reasonable starting point.

Using a language model as a second opinion

The classifier is fast and cheap — it should be your first filter. But for borderline cases (score between 0.3 and 0.7), you can escalate to a language model for contextual analysis.

An LLM can reason about things the feature extractor cannot: does the domain plausibly belong to the brand it is impersonating? Does the path structure match known phishing kit patterns? The key is structured output — a verdict and a reasoning string, not free-form text:

import json

def build_prompt(url: str) -> str:
    return (
        "You are a security analyst. Analyze this URL for phishing indicators.\n"
        f"URL: {url}\n\n"
        "Return JSON with these fields:\n"
        '  verdict: "phishing" | "legitimate" | "suspicious"\n'
        "  confidence: float 0.0-1.0\n"
        "  reasoning: one sentence\n"
        "  indicators: list of strings\n"
        "Return only valid JSON, no other text."
    )

def llm_assess_url(url: str, client) -> dict:
    response = client.chat(build_prompt(url), max_tokens=256)
    try:
        return json.loads(response.text)
    except json.JSONDecodeError:
        return {"verdict": "suspicious", "confidence": 0.5, "reasoning": "parse error", "indicators": []}

def classify_url(url: str, clf, client, threshold_low=0.3, threshold_high=0.7) -> dict:
    features = extract_features(url)
    feature_df = pd.DataFrame([features])
    score = float(clf.predict_proba(feature_df)[0][1])
    if score >= threshold_high:
        return {"verdict": "phishing", "score": score, "method": "classifier"}
    elif score <= threshold_low:
        return {"verdict": "legitimate", "score": score, "method": "classifier"}
    else:
        result = llm_assess_url(url, client)
        result["score"] = score
        result["method"] = "llm_escalation"
        return result

This hybrid approach keeps costs low — the language model only sees the roughly 20% of URLs that land in the ambiguous middle band.

Operationalizing the detector

A few things matter when you move from notebook to production:

Latency: the feature extractor runs in microseconds. The classifier adds a few milliseconds. Reserve LLM calls for borderline cases and set a hard timeout (500ms is reasonable for a synchronous security check).

Domain age: WHOIS lookups add latency but are worth it for high-confidence verdicts. Domains registered in the last 30 days are disproportionately malicious. Cache results aggressively.

Feedback loop: every false positive costs user trust. Build a reporting endpoint from day one and use the collected corrections to retrain monthly.

What the model will not catch: homograph attacks, zero-day phishing kits not yet in training data, and legitimate services that just happen to look suspicious. A threat intelligence feed — checking URLs against known-bad lists — fills the gaps a classifier alone cannot cover. See our security hardening checklists for a structured reference of additional defensive layers worth stacking alongside URL detection.

The takeaway

Phishing URL detection is a well-scoped problem with measurable real-world impact. A feature-based gradient boosting classifier gets you to ~97% accuracy with minimal infrastructure. Adding a language model escalation layer handles the ambiguous cases that sit on the decision boundary. Together they make a solid, explainable, cheap-to-run first line of defense.

Start with the feature extractor. Train on PhishTank + Tranco. Measure your false positive rate against your own traffic before deploying in blocking mode. Iterate from there.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

How to Scrape and Parse CVE Data from NVD API v2

Ayi NEDJIMI — Sat, 01 Aug 2026 10:04:05 +0000

The NVD (National Vulnerability Database) changed its public API in 2023, and many existing scrapers broke without warning. If you're building a vulnerability tracker, a security dashboard, or an automated patch prioritization tool, you need to work with API v2 — the old v1 endpoints are permanently shut down. This guide covers authentication, pagination, safe JSON parsing, and the rate limit traps most tutorials skip.

What Changed in NVD API v2

The old API (v1) required no authentication and returned a relatively flat JSON structure. API v2 is different in three important ways:

API key required for usable rate limits. Without a key, you're capped at 5 requests per 30 seconds. With a free NVD API key you get 50 requests per 30 seconds — a 10x difference that matters when you're backfilling months of CVE history. Register at nvd.nist.gov/developers/request-an-api-key.

Pagination is mandatory. The API returns at most 2000 results per call. For any real query you must use startIndex to walk through pages.

The JSON schema changed. CVE data now lives under vulnerabilities[].cve rather than result.CVE_Items. Fields like CVSS scores are split across three separate metric versions (v2, v3.0, v3.1) and any of them may be absent for recently published CVEs.

Fetching CVEs: Building the Request

The base URL is https://services.nvd.nist.gov/rest/json/cves/2.0. The API key goes in a header (apiKey), not a query parameter. Key filter parameters:

keywordSearch — full-text search across CVE descriptions
pubStartDate / pubEndDate — publication date range; must use ISO 8601 with explicit timezone offset: 2024-01-01T00:00:00.000+00:00
cvssV3Severity — filter by LOW, MEDIUM, HIGH, or CRITICAL
lastModStartDate / lastModEndDate — for incremental syncs, fetch only CVEs modified since your last pull

import time
import requests
from datetime import datetime

NVD_BASE = "https://services.nvd.nist.gov/rest/json/cves/2.0"
API_KEY = "your-nvd-api-key"  # set to None to test without a key

def fetch_cves(
    keyword: str | None = None,
    severity: str | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    start_index: int = 0,
    results_per_page: int = 500,
) -> dict:
    headers = {"apiKey": API_KEY} if API_KEY else {}
    params: dict = {
        "startIndex": start_index,
        "resultsPerPage": min(results_per_page, 2000),
    }
    if keyword:
        params["keywordSearch"] = keyword
    if severity:
        params["cvssV3Severity"] = severity.upper()
    if start_date:
        params["pubStartDate"] = start_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00")
    if end_date:
        params["pubEndDate"] = end_date.strftime("%Y-%m-%dT%H:%M:%S.000+00:00")

    resp = requests.get(NVD_BASE, headers=headers, params=params, timeout=30)
    resp.raise_for_status()
    return resp.json()

Parsing the Response Without Breaking

Naively accessing data["vulnerabilities"][0]["cve"]["metrics"]["cvssMetricV31"][0]["cvssData"]["baseScore"] will blow up on roughly 30% of CVEs. CVSS data is optional, versioned, and sometimes an empty list even when the key exists.

from dataclasses import dataclass

@dataclass
class CVERecord:
    cve_id: str
    description: str
    published: str
    severity: str
    cvss_score: float | None
    references: list[str]

def parse_cve(raw: dict) -> CVERecord:
    cve = raw["cve"]

    # prefer English, fall back to first available
    descriptions = cve.get("descriptions", [])
    description = next(
        (d["value"] for d in descriptions if d.get("lang") == "en"),
        descriptions[0]["value"] if descriptions else "",
    )

    # try CVSS v3.1 -> v3.0 -> v2.0 in order of preference
    metrics = cve.get("metrics", {})
    cvss_score: float | None = None
    severity = "UNKNOWN"
    for metric_key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"):
        entries = metrics.get(metric_key, [])
        if entries:
            data = entries[0].get("cvssData", {})
            cvss_score = data.get("baseScore")
            severity = data.get("baseSeverity", "UNKNOWN")
            break

    references = [r["url"] for r in cve.get("references", []) if "url" in r]

    return CVERecord(
        cve_id=cve["id"],
        description=description,
        published=cve.get("published", ""),
        severity=severity,
        cvss_score=cvss_score,
        references=references,
    )

The key pattern: iterate over metric versions in descending preference order and break on the first non-empty list. This gives you the most precise score available without ever crashing on a freshly published CVE that has not been scored yet.

Pagination and Rate Limit Handling

A search for "OpenSSL" or "Apache" returns thousands of CVEs. You need to paginate and sleep between requests to stay within the rate limit ceiling.

def fetch_all_cves(
    keyword: str,
    severity: str | None = None,
    max_results: int | None = None,
) -> list[CVERecord]:
    sleep_s = 0.7 if API_KEY else 6.1  # respect rate limits
    page_size = 500
    start = 0
    results: list[CVERecord] = []

    while True:
        data = fetch_cves(
            keyword=keyword,
            severity=severity,
            start_index=start,
            results_per_page=page_size,
        )
        total = data["totalResults"]
        batch = [parse_cve(v) for v in data.get("vulnerabilities", [])]
        results.extend(batch)
        print(f"  {len(results)} / {total}")

        if max_results and len(results) >= max_results:
            break
        if start + page_size >= total:
            break

        start += page_size
        time.sleep(sleep_s)

    return results

For production pipelines, do not re-fetch everything from scratch on every run. Use lastModStartDate set to your last successful sync timestamp. The NVD updates existing CVE records — re-scoring, adding references, changing status — so you need this to catch edits to records you already ingested, not just new ones.

Caching in SQLite for Team Use

If more than one person needs CVE data, add a local SQLite cache so queries do not hit the NVD on every call. Index on severity and published to keep filtering fast.

import sqlite3, json

def init_db(path: str = "cves.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS cves ("
        "cve_id TEXT PRIMARY KEY, description TEXT, published TEXT,"
        "severity TEXT, cvss_score REAL, references_json TEXT,"
        "fetched_at TEXT DEFAULT (datetime('now')))"
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_severity ON cves(severity)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_published ON cves(published)")
    conn.commit()
    return conn

def upsert_cves(conn: sqlite3.Connection, records: list) -> int:
    rows = [
        (r.cve_id, r.description, r.published, r.severity,
         r.cvss_score, json.dumps(r.references))
        for r in records
    ]
    conn.executemany(
        "INSERT OR REPLACE INTO cves"
        " (cve_id, description, published, severity, cvss_score, references_json)"
        " VALUES (?, ?, ?, ?, ?, ?)",
        rows,
    )
    conn.commit()
    return len(rows)

With this in place, run a nightly job that fetches the previous day's CVEs using lastModStartDate and upserts them. Your team can query the local database — filtering by severity, CVSS score range, or description keyword — without every lookup costing a network roundtrip to the NVD.

One caveat before wiring CVE data to automated alerting: CVSS base scores do not encode your environment. A CRITICAL vulnerability in a library you do not ship matters less than a MEDIUM in your authentication layer. A structured triage process that maps CVEs to your actual attack surface is what turns raw data into actionable work. We cover that workflow step by step in our security hardening checklists.

The Takeaway

NVD API v2 gives you reliable, structured access to the full CVE catalog, but it requires more defensive coding than the old API. Three things that will save you time:

Get an API key first — 10x rate limit for free, takes two minutes.
Never trust CVSS fields to exist — fall back gracefully across metric versions.
Use lastModStartDate for incremental syncs — otherwise you re-fetch tens of thousands of records to find a few hundred changes.

From here you can wire CVE records to your SBOM, fire Slack alerts when a new CRITICAL lands on software in your stack, or build a scoring layer on top that factors in your actual exposure context.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Building a Password Strength Validator with Entropy Calculation

Ayi NEDJIMI — Fri, 31 Jul 2026 10:05:21 +0000

Most password rules lie to users. "At least 8 characters, one uppercase, one number, one special character" — that regex check produces passwords like Password1! which scores 100% on every consumer-facing strength meter but cracks in minutes with a dictionary attack. Entropy-based validation gives you a number that actually correlates with cracking resistance.

What Is Password Entropy and Why It Matters

Entropy, measured in bits, tells you how many guesses an attacker needs in the worst case. The formula:

H = L × log₂(N)

Where L is password length and N is the size of the character pool in use.

A password using only lowercase letters (N = 26) at 8 characters gives H = 8 × log₂(26) ≈ 37.6 bits. That means at most 2^37.6 ≈ 200 billion possible values — sounds large until you consider a GPU cluster doing 10 billion guesses per second. With full ASCII printable characters (N = 95) at 8 characters, you get ≈ 52.6 bits. Double the length to 16 characters: ≈ 105 bits. Practically unbreakable with current hardware.

The core insight: length beats complexity rules every time.

Detecting Character Pools

Before calculating entropy, determine which character sets the password draws from. More pools used means a larger N.

import math
import re

CHARSETS = {
    "lowercase": (re.compile(r'[a-z]'), 26),
    "uppercase": (re.compile(r'[A-Z]'), 26),
    "digits":    (re.compile(r'[0-9]'), 10),
    "special":   (re.compile(r'[^a-zA-Z0-9]'), 32),
}

def detect_charset_size(password: str) -> int:
    size = 0
    for _, (pattern, pool_size) in CHARSETS.items():
        if pattern.search(password):
            size += pool_size
    return size if size > 0 else 1

def calculate_entropy(password: str) -> float:
    n = detect_charset_size(password)
    length = len(password)
    if length == 0:
        return 0.0
    return length * math.log2(n)

This returns theoretical maximum entropy — what you would get if the characters were drawn randomly from those pools. Real passwords are not random, so raw entropy overestimates strength. You need additional heuristic checks on top.

Building the Full Validator

Entropy alone misses practical weaknesses: common dictionary words, keyboard walk patterns (qwerty, asdf), repetitive characters. A production validator layers several signals.

from typing import NamedTuple

COMMON_PASSWORDS = {
    "password", "123456", "password1", "qwerty", "abc123",
    "letmein", "iloveyou", "admin", "welcome", "monkey",
    "dragon", "master", "sunshine", "princess", "football",
    "shadow", "superman", "michael", "jessica", "password123",
}

KEYBOARD_WALKS = ["qwerty", "asdf", "zxcv", "qazwsx", "1234", "abcd"]


class PasswordReport(NamedTuple):
    entropy: float
    strength: str        # "weak" | "fair" | "strong" | "very_strong"
    score: int           # 0-100
    issues: list[str]
    suggestions: list[str]


def validate_password(password: str) -> PasswordReport:
    issues: list[str] = []
    suggestions: list[str] = []
    penalty = 0

    entropy = calculate_entropy(password)

    if len(password) < 8:
        issues.append("Too short (minimum 8 characters)")
        suggestions.append("Use at least 12 characters for meaningful security")
        penalty += 40

    if password.lower() in COMMON_PASSWORDS:
        issues.append("This password appears in breach databases")
        suggestions.append("Avoid dictionary words and known patterns")
        penalty += 60

    for walk in KEYBOARD_WALKS:
        if walk in password.lower():
            issues.append(f"Contains keyboard walk pattern '{walk}'")
            suggestions.append("Avoid sequential keyboard patterns")
            penalty += 20
            break

    if len(set(password)) / max(len(password), 1) < 0.5:
        issues.append("Too many repeated characters")
        suggestions.append("Use a wider variety of characters")
        penalty += 15

    if entropy < 28:
        base_score = 10
    elif entropy < 36:
        base_score = 30
    elif entropy < 50:
        base_score = 55
    elif entropy < 70:
        base_score = 75
    elif entropy < 90:
        base_score = 90
    else:
        base_score = 100

    score = max(0, min(100, base_score - penalty))

    if score < 25:
        strength = "weak"
        if not suggestions:
            suggestions.append("Consider a passphrase: 4 random words beat most complex short passwords")
    elif score < 50:
        strength = "fair"
    elif score < 75:
        strength = "strong"
    else:
        strength = "very_strong"

    return PasswordReport(
        entropy=round(entropy, 2),
        strength=strength,
        score=score,
        issues=issues,
        suggestions=suggestions,
    )

A few representative outputs:

"password" → 37.6 bits entropy, hit in common list → score 0, weak
"P@ssw0rd1!" → ~65 bits, no blocklist match → score 75, strong
"correct-horse-battery-staple" → 131+ bits, no issues → score 100, very_strong

That last one illustrates why passphrases win. The entropy formula rewards length above all else.

Exposing This as an API Endpoint

If you are validating passwords at registration, here is a minimal FastAPI endpoint:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class PasswordCheckRequest(BaseModel):
    password: str

class PasswordCheckResponse(BaseModel):
    entropy_bits: float
    strength: str
    score: int
    issues: list[str]
    suggestions: list[str]

@app.post("/validate-password", response_model=PasswordCheckResponse)
def check_password(req: PasswordCheckRequest) -> PasswordCheckResponse:
    report = validate_password(req.password)
    return PasswordCheckResponse(
        entropy_bits=report.entropy,
        strength=report.strength,
        score=report.score,
        issues=report.issues,
        suggestions=report.suggestions,
    )

Two hard rules for this endpoint: never log the password itself (only the score and metadata), and always run this check server-side even if you also run it in the browser. Client-side validation is UX; server-side validation is security.

For production, complement the local blocklist with a Have I Been Pwned API lookup using the k-anonymity endpoint: send the first 5 hex characters of the SHA-1 hash, then check if the full hash appears in the returned list. That covers breached passwords your static list does not know about.

Passphrases: Counting Words, Not Characters

The character-based entropy formula underestimates passphrase strength. A 4-word Diceware passphrase drawn from a 7776-word list has log2(7776^4) ≈ 51.7 bits regardless of character count. Detect phrases and score them separately:

def passphrase_entropy(phrase: str) -> float:
    words = phrase.strip().split()
    if len(words) < 3:
        return 0.0
    # Conservative estimate: ~7776-word Diceware list
    return len(words) * math.log2(7776)

def smart_entropy(password: str) -> float:
    words = password.split()
    if len(words) >= 3:
        return passphrase_entropy(password)
    return calculate_entropy(password)

"correct horse battery staple" yields 4 × 12.9 ≈ 51.7 bits — comparable to a strong random character password but far easier to remember and type correctly. If your UI currently bans spaces in passwords, remove that restriction.

The Takeaway

Entropy calculation gives you an honest measure of password strength — one tied to actual attack cost rather than checkbox compliance. The implementation above handles:

Theoretical entropy from character pool detection
Common password blocklist with penalty scoring
Keyboard walk pattern detection
Repetition penalty
Passphrase detection with word-level entropy
A clean 0–100 score with human-readable suggestions

If you are hardening an authentication system more broadly, check the security hardening checklists we publish — free PDF and Excel versions covering auth, API security, and container hardening.

The most impactful change most teams can make today: stop enforcing arbitrary complexity rules and set a minimum entropy threshold of 50 bits instead. Users get more flexibility, attackers face higher costs, and you stop training people to write Password1!.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Implementing IP Reputation Scoring in a Go API

Ayi NEDJIMI — Thu, 30 Jul 2026 10:02:21 +0000

Your API logs show a flood of requests from a single IP — 400 attempts in 30 seconds. By the time your rate limiter kicks in, your database has already taken 400 unnecessary hits. IP reputation scoring lets you make a risk decision before a request ever touches your business logic.

This is not about blocking entire countries or trusting commercial blocklists blindly. It is about building a lightweight, composable scoring layer that combines signals: past behavior, known threat lists, and ASN metadata.

What goes into a reputation score

A reputation score is a number (0–100, higher = riskier) aggregated from several signals:

Recent failed requests: failed auth attempts, 429s, 4xx errors from this IP in the last hour
Threat intelligence feeds: lists like AbuseIPDB, Emerging Threats, or Spamhaus
ASN/organization: requests from hosting providers (AS14061 DigitalOcean) carry more weight than residential ISPs
Behavioral velocity: requests-per-minute, unique endpoints hit, unusual timing patterns

Each signal contributes a sub-score. The final score is a weighted sum. You tune the weights based on what a false positive costs your business.

Data structures and scoring logic

Here is a Go struct and scoring function that implements the basics:

package reputation

import (
    "math"
    "time"
)

type IPRecord struct {
    IP             string
    FailedAuths    int
    Total4xx       int
    RequestsPerMin float64
    InThreatFeed   bool
    IsHostingASN   bool
    LastSeen       time.Time
}

type Score struct {
    Value  int    // 0-100, higher = riskier
    Reason string
}

func Calculate(r IPRecord) Score {
    score := 0.0
    var reasons []string

    // Failed auth signals (max 40 points)
    if r.FailedAuths > 0 {
        authScore := math.Min(float64(r.FailedAuths)*4, 40)
        score += authScore
        if r.FailedAuths >= 5 {
            reasons = append(reasons, "repeated_auth_failures")
        }
    }

    // 4xx rate (max 20 points)
    if r.Total4xx > 10 {
        score += math.Min(float64(r.Total4xx-10)*0.5, 20)
    }

    // Velocity (max 20 points)
    if r.RequestsPerMin > 60 {
        score += math.Min((r.RequestsPerMin-60)*0.5, 20)
        reasons = append(reasons, "high_velocity")
    }

    // Threat feed match (30 points flat)
    if r.InThreatFeed {
        score += 30
        reasons = append(reasons, "threat_feed_match")
    }

    // Hosting ASN multiplier
    if r.IsHostingASN {
        score *= 1.3
        reasons = append(reasons, "hosting_asn")
    }

    finalScore := int(math.Min(score, 100))
    reason := "ok"
    if len(reasons) > 0 {
        reason = reasons[0]
    }

    return Score{Value: finalScore, Reason: reason}
}

The weights are opinionated — adjust them to your traffic profile. A failed auth from a hosting ASN that also appears in a threat feed can easily hit 90+. A residential IP with a few stray 404s stays below 20.

Building the middleware

The scorer is most useful as HTTP middleware that short-circuits requests above a threshold:

package middleware

import (
    "fmt"
    "net"
    "net/http"
    "strings"

    "yourapp/reputation"
    "yourapp/store"
)

const BlockThreshold     = 75
const ChallengeThreshold = 50

func ReputationCheck(s store.IPStore) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ip := extractIP(r)
            record, err := s.Get(ip)
            if err != nil {
                // Fail open — never block on infrastructure errors
                next.ServeHTTP(w, r)
                return
            }

            score := reputation.Calculate(record)
            w.Header().Set("X-Reputation-Score", fmt.Sprintf("%d", score.Value))

            switch {
            case score.Value >= BlockThreshold:
                http.Error(w, "forbidden", http.StatusForbidden)
                return
            case score.Value >= ChallengeThreshold:
                // Upstream can decide to require CAPTCHA or step-up auth
                r.Header.Set("X-Reputation-Challenge", "true")
            }

            next.ServeHTTP(w, r)
        })
    }
}

func extractIP(r *http.Request) string {
    // Only trust X-Forwarded-For if a controlled proxy sets it
    if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
        parts := strings.Split(xff, ",")
        return strings.TrimSpace(parts[0])
    }
    ip, _, _ := net.SplitHostPort(r.RemoteAddr)
    return ip
}

One critical detail in extractIP: only trust X-Forwarded-For if a trusted reverse proxy sets it. If your app is directly exposed, a client can forge any IP in that header and bypass scoring entirely.

Feeding the store and integrating threat feeds

The IP store accumulates signals as requests complete. A Redis-backed store with TTL-expiring counters works well at scale:

// Run this after next.ServeHTTP returns so you have the final status code
func recordOutcome(ip string, statusCode int, s store.IPStore) {
    const ttl1Hour   = 3600
    const ttl1Minute = 60

    if statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden {
        s.IncrField(ip, "failed_auths", 1, ttl1Hour)
    }
    if statusCode >= 400 {
        s.IncrField(ip, "total_4xx", 1, ttl1Hour)
    }
    s.IncrField(ip, "request_count", 1, ttl1Minute)
}

For threat feed lookups, AbuseIPDB offers a free tier. Query it asynchronously on first sight of an IP, then cache the result with a 24-hour TTL — you do not want a synchronous external call on every request:

func checkAbuseIPDB(ip, apiKey string) (bool, error) {
    req, _ := http.NewRequest("GET", "https://api.abuseipdb.com/api/v2/check", nil)
    q := req.URL.Query()
    q.Add("ipAddress", ip)
    q.Add("maxAgeInDays", "30")
    req.URL.RawQuery = q.Encode()
    req.Header.Set("Key", apiKey)
    req.Header.Set("Accept", "application/json")

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return false, err
    }
    defer resp.Body.Close()

    var result struct {
        Data struct {
            AbuseConfidenceScore int `json:"abuseConfidenceScore"`
        } `json:"data"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return false, err
    }
    return result.Data.AbuseConfidenceScore > 25, nil
}

For ASN lookups, MaxMind GeoLite2-ASN ships as a local .mmdb file and resolves in microseconds — no network round-trip needed.

Tuning and avoiding false positives

Before flipping the middleware on in production:

Run in shadow mode first: compute and log scores but do not block anything. Collect a week of data.
Histogram your scores: if 30% of legitimate traffic scores above 50, your weights are off.
Allowlist known IPs: your CI runner, monitoring probes, partner services — seed them with score 0 in the store.
Alert on block rate spikes: a sudden jump in blocks often means a signal source is misbehaving — a threat feed returning stale bulk data or a counter TTL bug.

For a broader hardening checklist covering authentication, headers, and rate limiting, our free security hardening checklists cover these patterns in structured PDF and Excel format.

The takeaway

IP reputation scoring is not a silver bullet. A motivated attacker rotates IPs; a residential botnet scores low on ASN checks. But as a first-pass filter it cuts noise significantly — automated scanners almost always trigger your highest-scoring thresholds within the first 10 requests.

The pattern scales cleanly: start with in-memory counters, graduate to Redis, then layer in external feeds as traffic volume justifies the operational complexity. Keep the fail-open behavior in the middleware. The cost of a false positive — blocking a legitimate user — is almost always higher than the cost of letting one more bot through.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.