DEV Community: Ayi NEDJIMI

Building an AI-powered code vulnerability scanner

Ayi NEDJIMI — Sun, 30 Aug 2026 10:02:07 +0000

Your security team can't review every pull request by hand. Static analysis tools like Semgrep and Bandit catch known patterns, but they miss logic flaws, insecure data flows, and context-dependent vulnerabilities. Layering a language model on top of those tools — using LLM reasoning to triage and explain findings — bridges that gap without replacing what already works.

Why Static Analysis Alone Falls Short

Traditional SAST tools operate on AST patterns and regex. They're fast, deterministic, and great for known issues — hardcoded secrets, obvious SQL injections, unvalidated inputs. But they fail at:

Business logic flaws — a function that looks safe in isolation but is dangerous in context
Chained vulnerabilities — where vuln A + vuln B = full compromise
False positive noise — teams disable rules when the signal-to-noise ratio degrades

A language model can read a function, understand its intent, and reason about what an attacker could do with it. That reasoning is what you want in a scanner.

Architecture Overview

The scanner has three layers:

AST extraction — parse Python source files into their raw structure
Pattern pre-filter — run Bandit to surface known bad patterns cheaply
LLM analysis — send suspicious code chunks to a language model with a structured prompt; get back findings with severity, CWE, and remediation advice

We keep the LLM calls targeted. You don't want to send 10,000 lines of boilerplate to an API — you want to send the 200 lines that Bandit flagged or that heuristics marked as risky.

Building the Core Scanner

import ast
import subprocess
import json
import os
from pathlib import Path
from openai import OpenAI  # any OpenAI-compatible endpoint works

client = OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=os.environ.get("LLM_BASE_URL"))

SYSTEM_PROMPT = """You are a security code reviewer. Analyze the provided Python function 
for vulnerabilities. Return a JSON object with:
- vulnerabilities: list of {cwe, severity (low/medium/high/critical), description, line_hint, remediation}
- safe: boolean (true if no significant issues found)

Be precise. Only flag real issues. Do not invent problems."""

def extract_functions(filepath: str) -> list[dict]:
    """Parse a Python file and extract all function definitions with their source."""
    source = Path(filepath).read_text()
    tree = ast.parse(source)
    lines = source.splitlines()
    functions = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            start = node.lineno - 1
            end = node.end_lineno
            func_source = "\n".join(lines[start:end])
            functions.append({
                "name": node.name,
                "lineno": node.lineno,
                "source": func_source,
            })
    return functions

def run_bandit(filepath: str) -> set[int]:
    """Run Bandit and return line numbers with findings."""
    result = subprocess.run(
        ["bandit", "-f", "json", "-q", filepath],
        capture_output=True, text=True
    )
    try:
        data = json.loads(result.stdout)
        return {r["line_number"] for r in data.get("results", [])}
    except json.JSONDecodeError:
        return set()

def analyze_function(func: dict) -> dict | None:
    """Send a function to the LLM for vulnerability analysis."""
    prompt = f"File function `{func['name']}` (line {func['lineno']}):\n\n```
{% endraw %}
python\n{func['source']}\n
{% raw %}
```"

    response = client.chat.completions.create(
        model="gpt-4o-mini",  # swap for any model you deploy
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )

    result = json.loads(response.choices[0].message.content)
    if not result.get("safe") and result.get("vulnerabilities"):
        return {"function": func["name"], "line": func["lineno"], **result}
    return None

def scan_file(filepath: str) -> list[dict]:
    """Full scan pipeline: extract → pre-filter → LLM analyze."""
    flagged_lines = run_bandit(filepath)
    functions = extract_functions(filepath)

    findings = []
    for func in functions:
        # Only send to LLM if Bandit flagged it OR function touches risky operations
        risky_keywords = {"eval", "exec", "subprocess", "pickle", "yaml.load", "os.system"}
        is_risky = (
            func["lineno"] in flagged_lines
            or any(kw in func["source"] for kw in risky_keywords)
        )
        if not is_risky:
            continue

        result = analyze_function(func)
        if result:
            findings.append(result)

    return findings

The pre-filter keeps LLM costs sane. On a typical 5,000-line Python service, you might send 15–25 functions to the model instead of 200+. At gpt-4o-mini pricing, that's under $0.01 per scan.

Integrating into CI/CD

Wrap the scanner in a CLI that exits non-zero on critical findings — this is what your CI pipeline will react to:

#!/usr/bin/env python3
import sys
import argparse
from pathlib import Path

def main():
    parser = argparse.ArgumentParser(description="AI code vulnerability scanner")
    parser.add_argument("paths", nargs="+", help="Files or directories to scan")
    parser.add_argument("--fail-on", default="high", choices=["low","medium","high","critical"])
    parser.add_argument("--output", default="text", choices=["text","json"])
    args = parser.parse_args()

    severity_order = {"low": 0, "medium": 1, "high": 2, "critical": 3}
    threshold = severity_order[args.fail_on]

    all_findings = []
    files = []
    for p in args.paths:
        path = Path(p)
        if path.is_dir():
            files.extend(path.rglob("*.py"))
        else:
            files.append(path)

    for filepath in files:
        findings = scan_file(str(filepath))
        for f in findings:
            f["file"] = str(filepath)
            all_findings.append(f)

    if args.output == "json":
        import json
        print(json.dumps(all_findings, indent=2))
    else:
        for f in all_findings:
            for vuln in f.get("vulnerabilities", []):
                sev = vuln["severity"].upper()
                print(f"[{sev}] {f['file']}:{f['line']} ({f['function']}) — {vuln['cwe']}")
                print(f"  {vuln['description']}")
                print(f"  Fix: {vuln['remediation']}\n")

    worst = max(
        (severity_order.get(v["severity"], 0) for f in all_findings for v in f.get("vulnerabilities", [])),
        default=-1
    )
    if worst >= threshold:
        print(f"❌ Scan failed: findings at or above '{args.fail_on}' severity found.")
        sys.exit(1)

    print(f"✅ Scan passed ({len(all_findings)} findings below threshold).")

if __name__ == "__main__":
    main()

Add this to your GitHub Actions workflow:

- name: AI vulnerability scan
  run: |
    pip install openai bandit
    python scanner.py src/ --fail-on high --output json > scan-results.json
  env:
    LLM_API_KEY: ${{ secrets.LLM_API_KEY }}

Tuning Signal Quality

A few things that matter in practice:

Chunking strategy — don't send entire files. Function-level chunks give the model enough context without overwhelming it with irrelevant code. If you have class methods that reference class state, include the class attributes in the prompt.

Prompt precision — vague prompts get vague answers. The system prompt above specifies the exact output schema and tells the model to be conservative. "Only flag real issues" cuts false positives significantly compared to "find all security issues."

Model selection — for code analysis, a smaller model fine-tuned on code often outperforms a larger general-purpose model. Run a benchmark against your own codebase before committing to a model. Keep a set of known-vulnerable functions as a test fixture and measure recall.

Rate limiting and cost — cache LLM results keyed by a hash of the function source. If the function didn't change between commits, reuse the previous finding rather than re-querying.

You can find a ready-to-use security hardening checklist that covers secure code review criteria — useful as a reference when tuning your model's prompt.

The Takeaway

An LLM-powered scanner doesn't replace Bandit or Semgrep — it augments them. The pattern-based tools are your first pass: fast, cheap, zero API cost. The model handles the nuanced cases those tools miss.

The key discipline is keeping LLM calls targeted. Scan everything with patterns, invoke the model only on suspicious code, cache aggressively. That approach keeps scan time under 30 seconds for most codebases and cost well under a dollar per day in CI.

The code above is production-ready in the sense that it handles errors, respects cost, and integrates with standard CI. You'll want to add retry logic, structured logging, and a proper config file for the threshold settings — but the core pipeline is solid.

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

How to Implement Function Calling with Structured Outputs

Ayi NEDJIMI — Sat, 29 Aug 2026 10:02:37 +0000

Language models are great at generating text. They are less predictable when you need them to return machine-readable data every single time. Function calling — sometimes called tool use — solves this by letting you describe functions in JSON Schema format, then having the model decide when and how to call them. Paired with structured output constraints, you get deterministic, parse-safe responses instead of hoping the model formats its reply correctly.

This matters in production. A model that occasionally returns "price": "twelve dollars" instead of "price": 12.0 breaks downstream parsing and requires fragile regex fallbacks. Function calling eliminates that class of bug.

What function calling actually does

When you send a request with function definitions, the model does not call anything. It returns a structured payload saying "I would call this function with these arguments." Your code then executes whatever that means — a database lookup, an API call, a local computation — and optionally feeds the result back to the model for a final natural-language response.

The key insight: function calling is a contract between you and the model about output format, not an execution mechanism. The model is a router, not an executor.

Defining a function schema

Major provider SDKs use JSON Schema for function definitions. Here is a minimal Python example:

import json
import openai

client = openai.OpenAI()  # uses OPENAI_API_KEY from env

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g. 'Paris'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "celsius"
                    }
                },
                "required": ["city"],
                "additionalProperties": False
            },
            "strict": True
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What's the weather in Lyon?"}],
    tools=tools,
    tool_choice="auto"
)

message = response.choices[0].message

if message.tool_calls:
    call = message.tool_calls[0]
    args = json.loads(call.function.arguments)
    print(f"Function: {call.function.name}")
    print(f"Arguments: {args}")
    # -> {'city': 'Lyon', 'unit': 'celsius'}

The "strict": True flag forces the model to conform exactly to your schema — no extra keys, no missing required fields. Enable it. Without it, the model may omit optional fields or add unexpected properties that break your deserialization code.

Handling the tool call loop

In a real agent, you feed the result back to continue the conversation. The pattern is always the same: execute the tool, append the result, call the API again, repeat until no tool calls remain.

def call_tool(name: str, args: dict) -> str:
    """Execute the tool and return a string result."""
    if name == "get_weather":
        # Replace with a real weather API call
        return json.dumps({"city": args["city"], "temp_c": 18, "condition": "partly cloudy"})
    raise ValueError(f"Unknown tool: {name}")


def run_agent(user_message: str, max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": user_message}]
    iterations = 0

    while iterations < max_iterations:
        iterations += 1
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        msg = response.choices[0].message
        finish_reason = response.choices[0].finish_reason

        if finish_reason == "length":
            raise RuntimeError("Response truncated — increase max_tokens or shorten schema")

        messages.append(msg)  # append assistant turn to history

        if not msg.tool_calls:
            return msg.content  # done

        for call in msg.tool_calls:
            result = call_tool(
                call.function.name,
                json.loads(call.function.arguments)
            )
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result
            })

    raise RuntimeError(f"Agent did not terminate after {max_iterations} iterations")


print(run_agent("What's the weather in Lyon and should I bring a jacket?"))

Two things worth noting: always cap iterations, and always check finish_reason before parsing arguments. A truncated response mid-JSON will fail silently without that guard.

Structured outputs for extraction

Function calling is designed for "the model decides what to do next." If you just need to extract structured data from unstructured text, the Structured Outputs approach is cleaner — you pass a Pydantic model and get back a typed, validated object:

from pydantic import BaseModel
from openai import OpenAI

client = OpenAI()


class Invoice(BaseModel):
    vendor: str
    total_eur: float
    due_date: str  # ISO 8601
    line_items: list[str]


completion = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Extract invoice data from the text."},
        {
            "role": "user",
            "content": (
                "Invoice from Acme Corp, 1250 EUR due 2026-09-30. "
                "Line items: consulting 10h, travel expenses."
            )
        }
    ],
    response_format=Invoice,
)

invoice = completion.choices[0].message.parsed
print(invoice.vendor)       # Acme Corp
print(invoice.total_eur)    # 1250.0
print(invoice.due_date)     # 2026-09-30
print(invoice.line_items)   # ['consulting 10h', 'travel expenses']

The model either returns a valid Invoice or raises a RefusalError. No try/except around json.loads, no regex fallbacks, no post-processing.

The decision rule is simple: use structured outputs for deterministic extraction, use function calling when the model needs to decide which action to take.

Common pitfalls

Over-engineering the schema. Keep schemas flat where possible. Deeply nested schemas with many optional fields increase the chance the model makes wrong choices. If you find yourself at three levels of nesting, split the task into two sequential calls.

Not handling refusals. Models can refuse to fill a field — especially for sensitive or ambiguous content. Always check message.refusal before accessing message.parsed:

if completion.choices[0].message.refusal:
    raise ValueError(f"Model refused: {completion.choices[0].message.refusal}")

Putting too many tools in scope. Don't expose twenty functions at once and expect the model to pick correctly. For complex agents, gate which tools are available based on conversational state. More tools means more ambiguity, which means more routing errors.

Trusting arguments blindly. The model constructs tool arguments from user input. A user who types something adversarial could end up with unexpected values in your function arguments. Treat every argument as untrusted input — validate, sanitize, and check authorization before execution.

The takeaway

Function calling and structured outputs are complementary, not competing. Tool use handles agentic decision-making: which action, in what order, with what parameters. Structured outputs handle data extraction: get me a typed object, every time, without parsing brittle text.

For anything touching user data, access controls, or external services, apply defense-in-depth to every function the model can invoke. The model does not know your business logic — you do. Treat the LLM's function arguments exactly as you would treat user-supplied HTTP query parameters: validate types, enforce bounds, check permissions. A solid security hardening checklist applied at the tool layer prevents the most common agentic vulnerabilities before they reach production.

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

Prompt caching strategies to cut LLM costs by 70%

Ayi NEDJIMI — Fri, 28 Aug 2026 10:08:59 +0000

If you're running LLM-powered features in production, your token bill is probably higher than it should be. Most teams feed the same system prompt, tool definitions, or retrieval context with every request — paying full price to process tokens they've already processed. Prompt caching changes that equation significantly, and it requires almost no refactoring to implement.

What is prompt caching and how does it work

Prompt caching lets you mark a prefix of your prompt as cacheable — system instructions, tool schemas, static documents. The provider stores the attention KV state for those tokens on their side. When your next request starts with the exact same prefix, processing is skipped: you pay only for cache read tokens, which are priced at roughly 1/10th of regular input tokens.

Both major providers support this at the API level. One uses a cache_control block in the request body; the other exposes cache_read_input_tokens in billing data. The mechanics differ slightly, but the principle is identical.

The cost math is straightforward. If you're sending a 10,000-token system prompt with every request and you process 1,000 requests per day, that's 10M input tokens daily. With caching, the first write is slightly more expensive (typically 1.25× input price), but each subsequent read is 10× cheaper. Over 1,000 requests, you pay for 1 write and 999 reads — a 70–80% cost reduction on that prefix.

The prefix pinning pattern

Caching only works on exact prefix matches. One changed character in the cached portion invalidates the cache for that prefix. This constraint shapes how you must structure your prompts: stable content at the top, dynamic content at the bottom.

import anthropic

client = anthropic.Anthropic()

# Build stable content once -- this gets cached across requests
SYSTEM_PROMPT = (
    "You are a security analyst assistant."
    # ... 5000 tokens of static instructions, rules, and context ...
)

TOOL_DEFINITIONS = []  # your function/tool schemas -- also stable

def query_llm(user_message: str, session_context: dict) -> str:
    """
    Cache the stable prefix; dynamic data goes at the bottom in messages[].
    """
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},  # mark for caching
            }
        ],
        tools=TOOL_DEFINITIONS,
        messages=[
            {
                "role": "user",
                # Dynamic per-request context goes here, NOT in system
                "content": f"Context: {session_context}\n\nQuestion: {user_message}"
            }
        ],
    )

    usage = response.usage
    print(f"Input: {usage.input_tokens} | Cache read: {usage.cache_read_input_tokens}")
    return response.content[0].text

What kills cache hit rates in practice:

Injecting a timestamp or request ID into the system prompt
Building system prompts with f-strings that include session-specific data
Appending per-user context to a shared instruction block

If your system prompt is constructed dynamically, refactor it into a fixed static string. Pass per-user data as a user turn message instead.

Multi-level caching for conversation history

In multi-turn applications, conversation history grows with each exchange. You want to cache the stable parts (system prompt, tools) but also checkpoint conversation history so you're not re-processing past turns from scratch.

The pattern: use two cache markers — one on the system prompt, one at a fixed depth into the conversation history.

def build_messages_with_cache(
    system_prompt: str,
    conversation_history: list[dict],
    new_user_message: str,
    history_cache_depth: int = 10,
) -> tuple[list, list]:
    """
    Returns (system, messages) with cache markers at stable breakpoints.
    Caches the system prompt + last N turns of history.
    """
    system = [
        {
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"},
        }
    ]

    messages = []
    history_len = len(conversation_history)

    for i, msg in enumerate(conversation_history):
        msg_copy = dict(msg)
        # Place a cache marker at the depth boundary
        if i == history_len - history_cache_depth and history_len >= history_cache_depth:
            if isinstance(msg_copy["content"], str):
                msg_copy["content"] = [
                    {
                        "type": "text",
                        "text": msg_copy["content"],
                        "cache_control": {"type": "ephemeral"},
                    }
                ]
        messages.append(msg_copy)

    # New message appended without cache marker -- it's the dynamic part
    messages.append({"role": "user", "content": new_user_message})

    return system, messages

The cache marker at position history_len - N tells the provider: compute and store KV state up to this point. The next request reuses that stored state if its prefix matches exactly. This works well for chatbots and agents with long sessions — the bulk of the conversation history stops being re-processed after the first time.

Measuring actual cache effectiveness

Before optimizing, instrument. Most providers return cache usage in the API response — make it part of your standard logging from day one.

import json
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class LLMCallMetrics:
    timestamp: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int
    cache_write_tokens: int
    estimated_cost_usd: float

    @property
    def cache_hit_rate(self) -> float:
        total = self.input_tokens + self.cache_read_tokens
        return self.cache_read_tokens / total if total > 0 else 0.0

def log_llm_call(
    response,
    model: str,
    input_price: float,
    cache_read_price: float,
    output_price: float,
) -> LLMCallMetrics:
    usage = response.usage

    cost = (
        (usage.input_tokens / 1_000_000) * input_price
        + (getattr(usage, "cache_read_input_tokens", 0) / 1_000_000) * cache_read_price
        + (usage.output_tokens / 1_000_000) * output_price
    )

    metrics = LLMCallMetrics(
        timestamp=datetime.utcnow().isoformat(),
        model=model,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cache_read_tokens=getattr(usage, "cache_read_input_tokens", 0),
        cache_write_tokens=getattr(usage, "cache_creation_input_tokens", 0),
        estimated_cost_usd=cost,
    )

    print(json.dumps(asdict(metrics)))
    print(f"  -> Cache hit rate: {metrics.cache_hit_rate:.1%}")
    return metrics

A cache hit rate below 60% on a system-prompt-heavy workload signals that your prefix is changing between requests. Track this metric over a rolling window. If it drops after a deploy, something in your prompt construction changed.

What to cache vs. what not to

Good candidates:

System instructions: role definition, output format, guardrails
Tool and function definitions — especially large schemas with many parameters
Static RAG documents or knowledge base chunks injected once per session
Few-shot examples shared across all users

Do not try to cache:

Per-user personalization mixed into the system prompt
Retrieved chunks that change per query
Anything including a timestamp, request ID, or random seed

If you're building compliance or security tooling — say, an assistant that checks configurations against a fixed policy document — load your reference material once, cache it, then vary only the user's specific question. This maps well to the pattern of pre-loading security hardening checklists as static context: the document never changes between users, so it caches perfectly.

For standard chat where context is mostly conversation history, realistic savings are 30–50% depending on turn length and session depth. The 70% figure applies when static prefixes dominate — RAG with fixed corpora, assistants with large tool schemas, or compliance tools with extensive rule sets.

The takeaway

Prompt caching is one of the highest-ROI optimizations available for production LLM workloads: near-zero implementation cost, immediate cost impact, no model quality tradeoff. The main discipline is structural — stable content at the top, dynamic content at the bottom, no "just add a timestamp" shortcuts.

Start with a single cache marker on your system prompt. Log cache_read_input_tokens for every response. If your hit rate is above 80% after a few hundred requests, add multi-level caching for conversation history. If it's below 50%, audit your prefix — something is changing that shouldn't be.

The pricing asymmetry makes this a no-brainer: cache reads cost roughly 1/10th of input tokens. Write once, read many, pay little.

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

Building a Local LLM API Server with Ollama and FastAPI

Ayi NEDJIMI — Thu, 27 Aug 2026 10:02:46 +0000

Running a language model locally and exposing it as a proper REST API has real practical value: no rate limits, no per-token billing, no data leaving your infrastructure. The problem is that Ollama's native HTTP interface is intentionally minimal — it works, but you'll want authentication, structured request validation, streaming support, and proper error handling before using it in any real workload.

This guide builds a thin FastAPI wrapper around Ollama that you'd actually deploy.

Prerequisites

You'll need Ollama installed and running, with at least one model pulled:

# Install Ollama (Linux)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull a model
ollama pull llama3.2:3b

# Verify
ollama list

For the FastAPI layer:

pip install fastapi uvicorn httpx python-dotenv

The Architecture

Ollama runs its own HTTP server on localhost:11434. The strategy here is simple:

FastAPI handles incoming requests from clients
It validates the payload with Pydantic
Forwards the request to Ollama
Returns the response — either full JSON or a streaming SSE response

This keeps the Ollama process as a dumb backend and lets you layer authentication, logging, and request transformation at the FastAPI level without touching Ollama's configuration.

Building the Server

# server.py
import os
import httpx
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
API_KEY = os.getenv("API_KEY", "change-me-in-production")

app = FastAPI(title="Local LLM API", version="1.0.0")


class ChatMessage(BaseModel):
    role: str = Field(..., pattern="^(system|user|assistant)$")
    content: str


class ChatRequest(BaseModel):
    model: str = Field(default="llama3.2:3b")
    messages: list[ChatMessage]
    stream: bool = False
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=1024, ge=1, le=8192)


def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key


async def stream_ollama_response(payload: dict) -> AsyncGenerator[str, None]:
    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream(
            "POST",
            f"{OLLAMA_BASE_URL}/api/chat",
            json=payload,
        ) as response:
            async for line in response.aiter_lines():
                if line:
                    yield f"data: {line}\n\n"


@app.post("/v1/chat")
async def chat(request: ChatRequest, _: str = Depends(verify_api_key)):
    payload = {
        "model": request.model,
        "messages": [m.model_dump() for m in request.messages],
        "stream": request.stream,
        "options": {
            "temperature": request.temperature,
            "num_predict": request.max_tokens,
        },
    }

    if request.stream:
        return StreamingResponse(
            stream_ollama_response(payload),
            media_type="text/event-stream",
        )

    async with httpx.AsyncClient(timeout=120) as client:
        try:
            response = await client.post(
                f"{OLLAMA_BASE_URL}/api/chat",
                json=payload,
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise HTTPException(status_code=502, detail=f"Ollama error: {e.response.text}")
        except httpx.ConnectError:
            raise HTTPException(status_code=503, detail="Ollama is not running")

    return response.json()


@app.get("/v1/models")
async def list_models(_: str = Depends(verify_api_key)):
    async with httpx.AsyncClient(timeout=10) as client:
        try:
            resp = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
            resp.raise_for_status()
            return resp.json()
        except httpx.ConnectError:
            raise HTTPException(status_code=503, detail="Ollama is not running")


@app.get("/health")
async def health():
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            resp = await client.get(f"{OLLAMA_BASE_URL}/api/tags")
        return {"status": "ok", "ollama": resp.status_code == 200}
    except httpx.ConnectError:
        return {"status": "degraded", "ollama": False}

Start the server:

API_KEY=my-secret-key uvicorn server:app --host 0.0.0.0 --port 8000

Test it:

curl -X POST http://localhost:8000/v1/chat \
  -H "x-api-key: my-secret-key" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "llama3.2:3b",
    "messages": [{"role": "user", "content": "What is a JWT?"}]
  }'

Handling Streaming Responses

Ollama returns newline-delimited JSON when streaming. The stream_ollama_response generator wraps each chunk as a Server-Sent Event. Here's a client that consumes it:

# client_stream.py
import json
import httpx


def stream_chat(prompt: str, api_key: str, base_url: str = "http://localhost:8000"):
    payload = {
        "model": "llama3.2:3b",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    with httpx.Client(timeout=120) as client:
        with client.stream(
            "POST",
            f"{base_url}/v1/chat",
            json=payload,
            headers={"x-api-key": api_key},
        ) as response:
            for line in response.iter_lines():
                if line.startswith("data: "):
                    chunk = json.loads(line[6:])
                    if not chunk.get("done"):
                        print(chunk["message"]["content"], end="", flush=True)
    print()


if __name__ == "__main__":
    stream_chat("Explain TLS handshake in 3 sentences.", "my-secret-key")

The output appears token by token — same behavior you'd expect from any hosted API.

Production Considerations

A few things that matter before this goes anywhere beyond localhost.

Model validation. The current code accepts any model string and lets Ollama fail with a 500. Better to fetch available models at startup and validate against that set on each request.

Concurrency control. Ollama processes one request at a time by default (unless your GPU supports true parallelism). Concurrent requests queue internally, but they can pile up fast under load. Cap concurrent inference calls with asyncio.Semaphore:

_sem = asyncio.Semaphore(2)  # max 2 concurrent Ollama requests

@app.post("/v1/chat")
async def chat(request: ChatRequest, _: str = Depends(verify_api_key)):
    async with _sem:
        # ... forward to Ollama
        pass

Timeouts. timeout=120 is a ceiling, not a guarantee. A 13B model generating a long response can blow past 2 minutes on consumer hardware. Tune per model, or expose timeout as an optional request field.

Logging. Add structured logging at the request level: model name, input token count (Ollama returns this), latency, and status code. Without it, debugging slow responses is painful.

Security headers. If this API is behind nginx, add X-Content-Type-Options, X-Frame-Options, and a narrow Content-Security-Policy. For a structured API hardening reference, this security checklist covers the gaps that tend to get overlooked at deployment time.

TLS. Do not put this directly on a public interface without TLS termination. Use Caddy or nginx in front of it, even for internal use.

The Takeaway

Running a language model behind a FastAPI layer costs you about 100 lines of code and gives you a self-hostable inference endpoint with proper validation, auth, and streaming. The key tradeoffs versus a hosted API: you own the hardware and the latency profile, model size is bounded by VRAM, and you're on the hook for updates and reliability.

For internal tooling, air-gapped environments, or cost-sensitive workloads where you're hitting the API thousands of times a day, this stack is genuinely competitive. Where it breaks down is high concurrent throughput — at that point, move to vLLM instead of Ollama, which is purpose-built around throughput over single-request latency.

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

Agentic AI Security: Sandboxing LLM Tool Calls in Production

Ayi NEDJIMI — Wed, 26 Aug 2026 10:05:12 +0000

When you give a language model the ability to call tools — run code, query databases, browse the web — you've created an autonomous execution surface. Most tutorials skip the part where that surface gets exploited.

This post covers practical steps for sandboxing LLM tool calls before they reach production. No theory: concrete code patterns that limit the blast radius when something goes wrong.

Why Tool Calls Are a Different Threat Surface

A standard LLM integration is relatively contained: input goes in, text comes out. The worst case is a model generating harmful content.

Agent architectures change the calculus entirely. The loop looks like this:

User prompt → LLM → tool call decision → tool execution → result → LLM → ...

At the "tool execution" step, the model's output directly drives system behavior. A prompt injection in a document the agent reads can redirect it to exfiltrate data. An unchecked shell tool lets the model run arbitrary commands. An HTTP tool without domain restrictions can trigger SSRF against your internal network.

These aren't hypothetical. They're the same attack classes that plagued web applications for decades — now applied to any agentic system you build.

The Minimal Sandbox: An Allowlist-First Tool Registry

The single most effective control is a strict allowlist of what tools exist and what parameters they accept. Don't let the model invent tool calls — only allow predefined, validated schemas.

import pathlib
import jsonschema
from dataclasses import dataclass
from typing import Any, Callable


@dataclass
class ToolSpec:
    name: str
    description: str
    param_schema: dict  # JSON Schema for validation
    handler: Callable
    requires_confirmation: bool = False


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec):
        self._tools[spec.name] = spec

    def call(self, tool_name: str, params: dict) -> Any:
        if tool_name not in self._tools:
            raise ValueError(
                f"Unknown tool: {tool_name!r}. Allowed: {list(self._tools)}"
            )
        spec = self._tools[tool_name]
        jsonschema.validate(params, spec.param_schema)
        return spec.handler(**params)


# Example: a file-read tool restricted to one directory
SAFE_DIR = pathlib.Path("/var/app/data").resolve()


def safe_read_file(path: str) -> str:
    target = (SAFE_DIR / path).resolve()
    if not str(target).startswith(str(SAFE_DIR)):
        raise PermissionError(f"Path traversal blocked: {path!r}")
    return target.read_text()


registry = ToolRegistry()
registry.register(ToolSpec(
    name="read_file",
    description="Read a file from the data directory",
    param_schema={
        "type": "object",
        "properties": {
            "path": {"type": "string", "pattern": r"^[\w\-/\.]+$"}
        },
        "required": ["path"],
        "additionalProperties": False,
    },
    handler=safe_read_file,
))

Two independent layers: the registry rejects unknown tool names outright, validates params against a JSON Schema before any execution, and the handler itself re-checks path resolution to block traversal.

Subprocess Sandboxing: Don't Give Agents a Raw Shell

If your agent needs to execute code — and many do — never use subprocess.Popen(shell=True) with model-generated content. The model controls the string, and shell=True hands it command injection on a plate.

For Python code execution, run it in a child process with tightly bounded resources:

import os
import resource
import subprocess
import tempfile


def run_python_sandbox(code: str, timeout: int = 5) -> str:
    # Run untrusted Python in a restricted subprocess.
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
        f.write(code)
        tmpfile = f.name

    try:
        result = subprocess.run(
            ["python3", "-E", "-S", tmpfile],  # -E: ignore env vars, -S: no site
            capture_output=True,
            text=True,
            timeout=timeout,
            preexec_fn=_set_resource_limits,
        )
        if result.returncode != 0:
            return f"Error: {result.stderr[:500]}"
        return result.stdout[:2000]
    except subprocess.TimeoutExpired:
        return "Execution timed out"
    finally:
        os.unlink(tmpfile)


def _set_resource_limits():
    # 50 MB address space
    resource.setrlimit(resource.RLIMIT_AS, (50 * 1024 * 1024, 50 * 1024 * 1024))
    # 10 seconds CPU time
    resource.setrlimit(resource.RLIMIT_CPU, (10, 10))
    # Max 10 open file descriptors (no network sockets)
    resource.setrlimit(resource.RLIMIT_NOFILE, (10, 10))

On Linux, pair this with a seccomp filter. At the container level, run agent workloads with dropped capabilities:

docker run \
  --security-opt seccomp=/etc/docker/seccomp-restricted.json \
  --security-opt no-new-privileges \
  --cap-drop ALL \
  --read-only \
  --tmpfs /tmp \
  agent-sandbox:latest

For higher-assurance workloads — untrusted user-supplied code, multi-tenant setups — use gVisor (runsc) or Firecracker microVMs. The subprocess pattern above is a floor, not a ceiling.

Rate Limiting at the Agent Level

The tool registry controls what an agent can do. Rate limiting controls how much it can do before a human should review it.

import threading
from collections import defaultdict
from datetime import datetime, timedelta


class AgentBudget:
    def __init__(self, max_calls: int = 20, window_seconds: int = 300):
        self.max_calls = max_calls
        self.window = timedelta(seconds=window_seconds)
        self._calls: dict[str, list[datetime]] = defaultdict(list)
        self._lock = threading.Lock()

    def check_and_consume(self, session_id: str, tool_name: str) -> bool:
        now = datetime.utcnow()
        key = f"{session_id}:{tool_name}"

        with self._lock:
            self._calls[key] = [
                t for t in self._calls[key] if now - t < self.window
            ]
            if len(self._calls[key]) >= self.max_calls:
                return False
            self._calls[key].append(now)
            return True

When a tool call fails the budget check, return an error to the model — not to the user directly. The model reports that it cannot complete the task, which is the correct outcome. Don't silently swallow the limit or retry automatically.

Pair budget enforcement with capability scoping: an agent handling customer support queries should never receive access to database write tools, even if those tools exist in the system. Instantiate the registry with only the tools relevant to the task at hand.

Audit Logging: The Prerequisite for Everything Else

None of the above controls are verifiable without logs. Every tool call should emit a structured record:

import json
import logging
from datetime import datetime
from typing import Any

logger = logging.getLogger("agent.audit")


def audit_tool_call(
    session_id: str,
    tool_name: str,
    params: dict,
    result: Any = None,
    error: str | None = None,
):
    record = {
        "ts": datetime.utcnow().isoformat() + "Z",
        "session": session_id,
        "tool": tool_name,
        "params": params,       # sanitize sensitive fields before this point
        "success": error is None,
        "error": error,
        "result_bytes": len(str(result)) if result is not None else 0,
    }
    logger.info(json.dumps(record))

Log before execution with status: "attempting" and after with the outcome. If the process is killed mid-call, you still have a record of intent.

Store audit logs in append-only storage — S3 with Object Lock, write-once Kafka topics, or a WORM-capable log backend. An agent that is compromised should not be able to clean up after itself by truncating its own log file.

The Takeaway

Agentic systems are code paths driven by model output. The security controls are the same as for any external input: validate, restrict, rate limit, log.

The difference is that model outputs are less predictable than typed user input, which makes defense-in-depth more critical. An allowlist registry, a restricted execution environment, and append-only audit logs give you three independent layers — any single bypass still hits the next one.

For a structured checklist of what to harden in AI-connected and web infrastructure, see the free security hardening checklists we publish.

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

How to Implement Retrieval-Augmented Generation (RAG) with Re-Ranking

Ayi NEDJIMI — Tue, 25 Aug 2026 10:05:41 +0000

RAG solves a real problem: language models hallucinate when they don't have relevant context, and they can't access data beyond their training cutoff. But naive RAG — embed a query, retrieve the top-K chunks, feed them to the model — has a precision problem. The chunks you retrieve are often related to the query, not relevant. Re-ranking fixes this by scoring retrieved passages a second time, with a model that understands semantic relationships at a finer grain.

This article walks through building a RAG pipeline with re-ranking in Python, from embedding to final answer generation.

Why Naive RAG Falls Short

The standard retrieval step uses a bi-encoder: embed the query and all documents into the same vector space, then find the nearest neighbors by cosine similarity. This is fast enough to run at scale, but bi-encoders compress meaning into a fixed-size vector — they can't compare query and passage directly.

The result: you retrieve chunks that are topically similar but not actually useful. Ask "what are the access control requirements for SOC 2?" and you might get chunks that discuss HIPAA or ISO 27001 instead. Nearby in vector space, useless in practice.

Re-ranking addresses this with a cross-encoder: a model that takes the query and each candidate passage together as input and produces a relevance score. Cross-encoders are slower (you can't pre-compute embeddings), but they're significantly more accurate.

The standard pipeline becomes:

Retrieve top-N candidates using a fast bi-encoder (N = 20–50)
Re-rank those candidates with a cross-encoder
Select top-K re-ranked passages (K = 3–5) and pass them to the language model

Setting Up the Environment

pip install sentence-transformers faiss-cpu openai

We'll use sentence-transformers for both embedding and re-ranking, FAISS for the vector index, and any language model API for generation.

Building the Retrieval Layer

Index your documents with a bi-encoder. For this example, we're indexing a collection of security policy documents — the kind of structured knowledge base where precision matters.

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

bi_encoder = SentenceTransformer("BAAI/bge-small-en-v1.5")

documents = [
    "Access control policies must enforce least privilege for all users.",
    "Encryption at rest is required for all data classified as sensitive.",
    "Multi-factor authentication is mandatory for administrative accounts.",
    "Log retention must cover a minimum of 12 months for SOC 2 compliance.",
    "Vulnerability scans must be conducted at least quarterly.",
    # ... your actual documents
]

doc_embeddings = bi_encoder.encode(documents, normalize_embeddings=True)
dimension = doc_embeddings.shape[1]

index = faiss.IndexFlatIP(dimension)  # Inner product = cosine sim when normalized
index.add(doc_embeddings.astype(np.float32))

def retrieve(query: str, top_n: int = 20) -> list[tuple[str, float]]:
    q_emb = bi_encoder.encode([query], normalize_embeddings=True)
    scores, indices = index.search(q_emb.astype(np.float32), top_n)
    return [(documents[i], float(scores[0][j])) for j, i in enumerate(indices[0])]

This retrieves 20 candidates quickly. Now we re-rank them.

Adding the Cross-Encoder Re-Ranker

The cross-encoder takes (query, passage) pairs and scores each one directly:

from sentence_transformers import CrossEncoder

cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(
    query: str,
    candidates: list[tuple[str, float]],
    top_k: int = 5
) -> list[str]:
    pairs = [[query, doc] for doc, _ in candidates]
    scores = cross_encoder.predict(pairs)

    ranked = sorted(
        zip([doc for doc, _ in candidates], scores),
        key=lambda x: x[1],
        reverse=True,
    )
    return [doc for doc, _ in ranked[:top_k]]

ms-marco-MiniLM-L-6-v2 is a solid default: 22M parameters, fast on CPU, and trained on the MS MARCO passage ranking dataset.

Full Pipeline: Retrieve → Re-Rank → Generate

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def rag_with_reranking(query: str) -> str:
    # Step 1: retrieve broad candidates
    candidates = retrieve(query, top_n=20)

    # Step 2: re-rank to a precise top-K
    top_passages = rerank(query, candidates, top_k=4)

    # Step 3: build grounded context and generate
    context = "\n\n".join(
        f"[{i+1}] {passage}" for i, passage in enumerate(top_passages)
    )
    prompt = (
        "Answer using only the provided context. "
        "If the answer is not there, say so explicitly.\n\n"
        f"Context:\n{context}\n\nQuestion: {query}"
    )
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
    )
    return response.choices[0].message.content

print(rag_with_reranking("What are the log retention requirements?"))

Temperature is set to 0.1 deliberately — low creativity keeps the model grounded in retrieved content rather than extrapolating.

Measuring the Gain

Re-ranking adds latency: roughly 50–80ms on CPU for 20 candidates with the MiniLM model. Whether that trade-off is worth it depends on your context. In practice, re-ranking improves hit rate by 15–30% on most retrieval benchmarks.

A minimal evaluation harness:

def evaluate_pipeline(test_pairs: list[dict]) -> dict:
    """test_pairs: [{"query": str, "expected": str}]"""
    hits_base, hits_reranked = 0, 0

    for pair in test_pairs:
        query, expected = pair["query"], pair["expected"]

        # Baseline: top-5 from bi-encoder only
        base = [doc for doc, _ in retrieve(query, top_n=5)]
        hits_base += any(expected in doc for doc in base)

        # With re-ranking: retrieve 20, re-rank to 5
        candidates = retrieve(query, top_n=20)
        reranked = rerank(query, candidates, top_k=5)
        hits_reranked += any(expected in doc for doc in reranked)

    n = len(test_pairs)
    return {
        "hit_rate_baseline": hits_base / n,
        "hit_rate_reranked": hits_reranked / n,
    }

Build a test set of 50–100 query/answer pairs from your actual data. Even a small evaluation corpus will surface where your chunking strategy breaks down.

Practical Notes

Chunk size affects re-ranking quality. Chunks shorter than 100 tokens lose context; longer than 512 and the cross-encoder degrades. 200–300 tokens with 20% overlap is a reasonable starting point.

Model alternatives. For multilingual corpora, swap bge-small-en-v1.5 for BAAI/bge-m3. For higher re-ranking accuracy at the cost of 4× latency, try cross-encoder/ms-marco-electra-base.

Don't strip metadata. If your documents have categories, dates, or source identifiers, include them in the chunk. Cross-encoders can use structural signals to break ties between semantically equivalent passages.

Where this pattern really pays off. Precision-critical retrieval — compliance documentation, access control policies, medical protocols — is where re-ranking earns its latency cost. If your RAG pipeline is querying a corpus of security hardening guidelines (the kind you'd find in a structured security checklist library), a hallucinated answer is actively dangerous. Get retrieval right before you trust the model's output.

The Takeaway

Retrieve many, re-rank to precision, generate from the best. The bi-encoder handles scale; the cross-encoder handles accuracy. You don't have to choose.

If your RAG pipeline produces plausible-but-imprecise answers, add a cross-encoder re-ranker before you try anything else — it's the highest-leverage improvement you can make without changing your architecture.

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

Building a Self-Correcting AI Agent with Reflection Loops in Python

Ayi NEDJIMI — Sun, 23 Aug 2026 10:03:11 +0000

Language models produce wrong answers. Not occasionally — regularly. When you deploy an LLM to automate tasks, you need a way to catch and fix those errors without human intervention. Reflection loops are one practical answer: the model checks its own output, flags problems, and retries until it meets a quality bar you define in code.

What Is a Reflection Loop?

A reflection loop is a control flow pattern where an agent runs a task, evaluates the result, then decides whether to retry. The simplest form is a two-step cycle: generate and critique. The critique step is usually a second LLM call with a different prompt, but it can also be a deterministic check — a JSON parser, a unit test runner, or a schema validator.

The key insight is that critique is easier than generation. A language model will often miss edge cases in a first pass but correctly identify them when asked “what is wrong with this output?” This asymmetry is what makes the pattern work at all.

A Minimal Working Implementation

Here is a Python class that wraps any LLM call in a reflection loop:

import json
from typing import Any, Callable
import httpx

MAX_ITERATIONS = 4


def call_llm(prompt: str, system: str = "") -> str:
    response = httpx.post(
        "https://api.example-llm.com/v1/messages",
        headers={"x-api-key": "YOUR_KEY", "Content-Type": "application/json"},
        json={
            "model": "your-model",
            "max_tokens": 1024,
            "system": system,
            "messages": [{"role": "user", "content": prompt}],
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["content"][0]["text"]


class ReflectionAgent:
    def __init__(
        self,
        task_prompt: str,
        critique_prompt: str,
        validator: Callable[[str], tuple[bool, str]],
    ):
        self.task_prompt = task_prompt
        self.critique_prompt = critique_prompt
        self.validator = validator

    def run(self, user_input: str) -> dict[str, Any]:
        history: list[dict] = []
        attempt = 0

        while attempt < MAX_ITERATIONS:
            attempt += 1
            context = f"User input: {user_input}"
            if history:
                last = history[-1]
                context += (
                    f"\n\nPrevious attempt:\n{last['output']}"
                    f"\nCritique:\n{last['critique']}"
                )

            output = call_llm(f"{self.task_prompt}\n\n{context}")
            ok, critique = self.validator(output)

            history.append(
                {"attempt": attempt, "output": output, "critique": critique, "passed": ok}
            )

            if ok:
                return {"success": True, "output": output, "iterations": attempt, "history": history}

            llm_critique = call_llm(
                f"{self.critique_prompt}\n\nOutput to critique:\n{output}",
                system="Be specific about what is wrong. Do not repeat the corrected version.",
            )
            history[-1]["critique"] = llm_critique

        return {"success": False, "output": history[-1]["output"], "iterations": attempt, "history": history}

The validator is a plain Python callable that returns (passed: bool, message: str). Keeping it separate from the LLM logic means you can unit-test it independently and swap it without touching the agent loop.

Adding a JSON Schema Validator

Malformed or schema-violating JSON is the most common failure mode in structured LLM output tasks. Here is a concrete validator that enforces a fixed schema:

import jsonschema

EXPECTED_SCHEMA = {
    "type": "object",
    "required": ["summary", "severity", "cve_ids"],
    "properties": {
        "summary": {"type": "string", "minLength": 10},
        "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
        "cve_ids": {
            "type": "array",
            "items": {"type": "string", "pattern": "^CVE-\\d{4}-\\d+$"},
        },
    },
}


def validate_security_report(text: str) -> tuple[bool, str]:
    clean = text.strip()
    if clean.startswith("```

"):
        clean = "\n".join(clean.split("\n")[1:])
    if clean.endswith("

```"):
        clean = "\n".join(clean.split("\n")[:-1])

    try:
        data = json.loads(clean)
    except json.JSONDecodeError as exc:
        return False, f"Invalid JSON: {exc}"

    try:
        jsonschema.validate(data, EXPECTED_SCHEMA)
    except jsonschema.ValidationError as exc:
        return False, f"Schema violation: {exc.message}"

    return True, "OK"


agent = ReflectionAgent(
    task_prompt=(
        "Analyze the following security advisory and return a JSON object "
        "with keys 'summary' (string), 'severity' (low/medium/high/critical), "
        "and 'cve_ids' (array of CVE strings). Return only the JSON, no prose."
    ),
    critique_prompt=(
        "Review this JSON output for accuracy, completeness, and schema compliance. "
        "List each specific violation and explain why it fails."
    ),
    validator=validate_security_report,
)

result = agent.run(
    "CVE-2024-3094: backdoor found in XZ Utils 5.6.0 and 5.6.1 affecting liblzma, "
    "allowing remote code execution on affected Linux distributions."
)
print(json.dumps(result, indent=2))

The deterministic schema check runs before the LLM critique. If JSON parsing fails, the error message is already precise and actionable. The LLM critique step is reserved for semantic problems that a schema cannot express.

Controlling Cost and Preventing Runaway Loops

Reflection loops without a hard cap on iterations will generate runaway costs in production. Three rules to enforce:

1. Hard iteration limit. Never exceed 4–5 cycles per request. If the model has not produced valid output after four attempts, the problem is almost always the prompt, not a fixable runtime error.

2. Track token spend across iterations. Add a budget guard:

COST_PER_1K_TOKENS = 0.003  # adjust to your model's pricing


class BudgetedReflectionAgent(ReflectionAgent):
    def __init__(self, *args, max_cost_usd: float = 0.10, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_cost_usd = max_cost_usd
        self._total_cost = 0.0

    def _check_budget(self, prompt: str) -> None:
        # Real usage: read token counts from the API response body
        estimated_tokens = len(prompt.split()) * 1.3
        projected_cost = (estimated_tokens / 1000) * COST_PER_1K_TOKENS
        if self._total_cost + projected_cost > self.max_cost_usd:
            raise RuntimeError(
                f"Budget exceeded: ${self._total_cost:.4f} spent, "
                f"limit ${self.max_cost_usd}"
            )
        self._total_cost += projected_cost

3. Log every iteration to a structured store. A simple SQLite table with (task_id, attempt, input_tokens, passed, critique_text) is enough to identify which task types consistently fail and need prompt improvements.

Where Reflection Loops Break Down

The pattern fails in predictable ways:

Self-reinforcing errors. The model accepts a wrong assumption from iteration 1 in iteration 2. Fix: add to the system prompt — "Do not treat facts from the previous output as verified. Re-derive from the original input."
Vague critique prompts. Asking “is this correct?” returns “yes” more often than not. Ask “list each specific constraint the output violates and explain why.”
No ground truth. Reflection only works when the validator has a clear, programmatic success criterion. Subjective tasks — tone, persuasiveness, brand alignment — need human review, not another LLM iteration.

Security-sensitive use cases, where the agent takes real actions based on its output, need sandboxing on top of reflection. Tool calls should be scoped to least-privilege, outputs sanitized before use, and every action logged with rollback support. We maintain free hardening checklists for agentic LLM deployments covering tool isolation, output validation, and incident recovery.

The Takeaway

Reflection loops improve output reliability without fine-tuning or swapping models. The pattern has three components: a generation step, a deterministic validator, and an LLM critique that feeds the next attempt. Keep the iteration count low (≤4), track costs per request, and log every iteration. The goal is not to make the model infallible — it is to catch errors programmatically before they propagate downstream.

Start with the schema validator version above. It handles 80% of structured output failures and requires zero prompt engineering on the critique side. Add LLM self-critique only when deterministic checks cannot express the quality bar you need.

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

Implementing Zero-Trust Network Access (ZTNA) Concepts in Code

Ayi NEDJIMI — Sat, 22 Aug 2026 10:05:03 +0000

Traditional perimeter security operates on a dangerous assumption: if you're inside the network, you're trusted. This model worked when "inside" meant a physical office with a hardware firewall at the door. Today, with remote work, cloud infrastructure, SaaS tools, and contractor access scattered across the internet, the perimeter has dissolved — and attackers exploit that gap every day. Zero-trust network access (ZTNA) replaces implicit trust with continuous, contextual verification. Here's how to implement its core concepts in actual code.

What Zero Trust Means in Engineering Terms

Vendor marketing has diluted the term, but the underlying principles are concrete and implementable:

Never trust, always verify — every request must be authenticated and authorized, regardless of which network it originates from
Least privilege — users and services get the minimum access required, scoped to the specific resource they need
Assume breach — design your system for the scenario where an attacker already holds valid credentials

These three principles translate to specific engineering choices: short-lived tokens, per-resource policy evaluation, contextual trust scoring, mandatory step-up authentication for sensitive operations, and continuous session validation.

Building an Identity-Aware Middleware in Python

The foundation of ZTNA is moving authentication checks from "once at the perimeter" to "on every single request." Here is a FastAPI middleware that validates JWT tokens and computes a contextual trust score on each call:

import ipaddress
import jwt
from datetime import datetime, timezone
from fastapi import Request, HTTPException

SECRET_KEY = "your-secret-key"
KNOWN_CIDRS = ["10.0.0.0/8", "192.168.0.0/16", "172.16.0.0/12"]

def is_known_network(ip: str) -> bool:
    try:
        addr = ipaddress.ip_address(ip)
        return any(addr in ipaddress.ip_network(cidr) for cidr in KNOWN_CIDRS)
    except ValueError:
        return False

def compute_trust_score(payload: dict, request: Request) -> float:
    score = 1.0

    # Penalize stale tokens — fresh auth signals higher confidence
    issued_at = datetime.fromtimestamp(payload["iat"], tz=timezone.utc)
    age_hours = (datetime.now(timezone.utc) - issued_at).total_seconds() / 3600
    if age_hours > 8:
        score -= 0.5
    elif age_hours > 4:
        score -= 0.3

    # Penalize requests from unrecognized networks
    client_ip = request.client.host
    if not is_known_network(client_ip):
        score -= 0.25

    # Penalize missing user agents (common in automated scanners)
    if not request.headers.get("user-agent", ""):
        score -= 0.2

    return max(score, 0.0)

async def ztna_middleware(request: Request, call_next):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")

    token = auth.split(" ", 1)[1]
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    trust_score = compute_trust_score(payload, request)
    if trust_score < 0.4:
        raise HTTPException(
            status_code=403,
            detail=f"Request blocked: trust score {trust_score:.2f} below threshold"
        )

    request.state.user = payload
    request.state.trust_score = trust_score
    return await call_next(request)

The trust score approach is more nuanced than binary allow/deny. Different resources can require different minimum scores, and you can tune thresholds per environment without changing your auth infrastructure.

Per-Resource Policy Enforcement

Authentication is necessary but not sufficient. Authorization must be evaluated at the resource level, not at the network edge. Here is a minimal policy engine that enforces role and trust requirements per endpoint:

from dataclasses import dataclass
from typing import Optional

@dataclass
class AccessPolicy:
    resource: str
    allowed_roles: list[str]
    min_trust_score: float = 0.6
    require_recent_auth: bool = False  # token must be < 1 hour old

POLICIES: dict[str, AccessPolicy] = {
    "/api/admin": AccessPolicy(
        resource="/api/admin",
        allowed_roles=["admin"],
        min_trust_score=0.9,
        require_recent_auth=True,
    ),
    "/api/reports": AccessPolicy(
        resource="/api/reports",
        allowed_roles=["admin", "analyst"],
        min_trust_score=0.65,
    ),
    "/api/health": AccessPolicy(
        resource="/api/health",
        allowed_roles=["admin", "analyst", "readonly"],
        min_trust_score=0.2,
    ),
}

def evaluate_access(
    path: str,
    user_role: str,
    trust_score: float,
    token_age_hours: float,
) -> tuple[bool, Optional[str]]:
    policy = POLICIES.get(path)
    if not policy:
        return False, f"No policy defined for '{path}'"

    if user_role not in policy.allowed_roles:
        return False, f"Role '{user_role}' not permitted on '{path}'"

    if trust_score < policy.min_trust_score:
        return False, (
            f"Trust score {trust_score:.2f} below required "
            f"{policy.min_trust_score} for '{path}'"
        )

    if policy.require_recent_auth and token_age_hours > 1.0:
        return False, "Step-up authentication required — please re-authenticate"

    return True, None

The require_recent_auth flag implements step-up authentication: sensitive operations force fresh credentials even when a valid session already exists. This is essential for admin endpoints, write operations on financial data, or any irreversible action.

Short-Lived Tokens and Contextual Binding

Long-lived sessions are fundamentally incompatible with zero trust. If an attacker steals a token, the blast radius is bounded by how long that token remains valid. Enforce short TTLs and embed contextual claims at issuance time:

import secrets
import jwt
from datetime import datetime, timezone, timedelta

SECRET_KEY = "your-secret-key"

def generate_token_id() -> str:
    return secrets.token_hex(16)

def issue_access_token(user_id: str, role: str, client_ip: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "role": role,
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(minutes=30)).timestamp()),
        "bound_ip": client_ip,      # contextual binding for anomaly detection
        "jti": generate_token_id(), # unique ID for revocation tracking
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

Practical TTL targets: access tokens at 15–30 minutes, refresh tokens as single-use (rotated on every refresh), admin-scoped tokens capped at 15 minutes with no refresh path — require full re-authentication.

The bound_ip claim enables a useful anomaly signal: if a token issued to 10.0.1.15 arrives from 185.220.101.x three minutes later, that warrants investigation even when the signature validates. Store jti values in Redis with a TTL matching the token expiry so you can instantly revoke individual tokens without touching your auth infrastructure.

Continuous Validation: Trust Is Not a One-Time Decision

Zero trust does not stop at request authentication. Sessions need continuous health checks:

Impossible travel: token issued in Paris, used five minutes later from Singapore
Behavioral drift: a read-only account issuing bulk DELETE requests
Device posture changes: certificate revoked or OS vulnerability detected mid-session

Publish auth and access events to a queue — Redis Streams works well for this at moderate scale, Kafka if you are processing millions of events per day. A background worker consumes events, correlates signals, and adds jti values to the revocation list when anomalies cross a threshold.

One anomaly is noise. Three signals within ten minutes warrant automatic session termination. This is where your ZTNA implementation intersects with your SIEM and incident response process. For a structured approach to what events to capture and how to wire alerts, the security hardening checklists we publish cover the event taxonomy and correlation rules used in production deployments.

The Takeaway

Zero trust is not a product you buy — it is a set of engineering patterns applied at every layer. The building blocks are: contextual trust scoring on every request, per-resource policy evaluation, token binding, short TTLs with instant revocation, and anomaly signal correlation.

Start with the gap that matters most in your current architecture. For most teams, that is moving from "authenticated at login" to "validated on every request." The middleware above is a half-day implementation that meaningfully reduces your attack surface. Step-up auth, session revocation, and anomaly detection layer on top of that foundation — and each one closes a different category of attack.

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

Building a TLS Fingerprinting Tool with Go (JA3/JA4)

Ayi NEDJIMI — Fri, 21 Aug 2026 10:04:25 +0000

Every TLS client leaves a fingerprint. The combination of cipher suites, extensions, and elliptic curves advertised in the ClientHello message is unique enough to distinguish a Python requests call from a curl command from a Cobalt Strike beacon — even when the payload is fully encrypted. JA3 and its successor JA4 turn that fingerprint into a reproducible hash you can act on, and Go is a natural fit for building tools around it.

What TLS Fingerprinting Is and Why It Matters

When a TLS handshake begins, the client sends a ClientHello message that lists:

TLS version
Supported cipher suites (in order)
Supported extensions (in order)
Supported elliptic curve groups
Elliptic curve point formats

This data is always unencrypted and is essentially fixed for a given client library and version. A production bot scraping your API will produce the same ClientHello on every request, regardless of destination. JA3 concatenates specific fields from that message and MD5-hashes them into a 32-character string. JA4 improves on this by sorting fields before hashing, making fingerprints stable across minor client differences and human-readable at a glance.

On the defensive side, a WAF or reverse proxy that fingerprints incoming connections can flag scrapers mimicking a browser, catch C2 framework check-ins before they reach your application, or identify hosts still running outdated TLS stacks. This is a cheap, passive signal — no payload decryption, no latency overhead. If you want a broader look at layered TLS hardening, our free security hardening checklists cover certificate policies, cipher suite selection, and HSTS alongside fingerprinting.

The JA3 Hash Formula

JA3 concatenates five fields with a predictable separator scheme:

TLSVersion,Ciphers,Extensions,EllipticCurves,EllipticCurvePointFormats

Each list is dash-separated. GREASE values (0x0a0a, 0x1a1a, ... per RFC 8701) are stripped from all lists before hashing. The result is the MD5 of that string.

A concrete raw string looks like:

771,4866-4867-4865-49196-49200,0-23-65281-10-11-35-16-5-13,29-23-24,0

MD5 of that gives 9b0a1e74c72f9ab20e6379dffde7a5de

JA4 changes the format to a structured prefix plus two truncated SHA-256 hashes:

t13d1516h2_8daaf6152771_b1ff8ab2d16f

The prefix t13d1516h2 encodes: TLS (t), version 1.3 (13), domain (d), 15 cipher suites, 16 extensions, HTTP/2 ALPN (h2). You can read the client type without decoding anything.

Parsing a ClientHello in Go

The standard library does not expose raw TLS handshake bytes, so you parse them yourself — either from gopacket captures or from a buffered listener (shown in the next section). Here is the core hash logic as a standalone package:

package fingerprint

import (
    "crypto/md5"
    "encoding/hex"
    "fmt"
    "strings"
)

// greaseValues lists GREASE constants (RFC 8701) that must be filtered before hashing.
var greaseValues = map[uint16]bool{
    0x0a0a: true, 0x1a1a: true, 0x2a2a: true, 0x3a3a: true,
    0x4a4a: true, 0x5a5a: true, 0x6a6a: true, 0x7a7a: true,
    0x8a8a: true, 0x9a9a: true, 0xaaaa: true, 0xbaba: true,
    0xcaca: true, 0xdada: true, 0xeaea: true, 0xfafa: true,
}

// ClientHello holds the fields relevant to TLS fingerprinting.
type ClientHello struct {
    Version         uint16
    CipherSuites    []uint16
    Extensions      []uint16
    SupportedGroups []uint16
    PointFormats    []uint8
    SNI             string
    ALPNProtocols   []string
}

// JA3String builds the raw concatenated string before hashing.
func JA3String(h *ClientHello) string {
    return fmt.Sprintf("%d,%s,%s,%s,%s",
        h.Version,
        joinU16(filterGREASE(h.CipherSuites), "-"),
        joinU16(filterGREASE(h.Extensions), "-"),
        joinU16(filterGREASE(h.SupportedGroups), "-"),
        joinU8(h.PointFormats, "-"),
    )
}

// JA3Hash returns the MD5 fingerprint of a ClientHello.
func JA3Hash(h *ClientHello) string {
    sum := md5.Sum([]byte(JA3String(h)))
    return hex.EncodeToString(sum[:])
}

func filterGREASE(in []uint16) []uint16 {
    out := make([]uint16, 0, len(in))
    for _, v := range in {
        if !greaseValues[v] {
            out = append(out, v)
        }
    }
    return out
}

func joinU16(s []uint16, sep string) string {
    parts := make([]string, len(s))
    for i, v := range s {
        parts[i] = fmt.Sprintf("%d", v)
    }
    return strings.Join(parts, sep)
}

func joinU8(s []uint8, sep string) string {
    parts := make([]string, len(s))
    for i, v := range s {
        parts[i] = fmt.Sprintf("%d", v)
    }
    return strings.Join(parts, sep)
}

This is pure and stateless — give it a ClientHello, get a deterministic hash. Testing it is straightforward: capture a few handshakes with Wireshark, extract the ClientHello fields manually, and compare against the salesforce/ja3 reference dataset.

Building a Fingerprinting Listener

The listener approach fingerprints every connection at the TCP level before TLS negotiates. The trick is to wrap net.Conn so the bytes we peek at are replayed to crypto/tls unchanged:

package main

import (
    "crypto/tls"
    "log"
    "net"
    "net/http"

    "fingerprint"
)

// peekConn replays buffered bytes so the TLS stack sees the complete handshake.
type peekConn struct {
    net.Conn
    buf []byte
}

func (c *peekConn) Read(b []byte) (int, error) {
    if len(c.buf) > 0 {
        n := copy(b, c.buf)
        c.buf = c.buf[n:]
        return n, nil
    }
    return c.Conn.Read(b)
}

type fpListener struct{ inner net.Listener }

func (l *fpListener) Accept() (net.Conn, error) {
    conn, err := l.inner.Accept()
    if err != nil {
        return nil, err
    }
    raw := make([]byte, 1024)
    n, err := conn.Read(raw)
    if err != nil {
        conn.Close()
        return nil, err
    }
    raw = raw[:n]
    if hello, err := fingerprint.ParseClientHello(raw); err == nil {
        log.Printf("addr=%s ja3=%s sni=%q",
            conn.RemoteAddr(), fingerprint.JA3Hash(hello), hello.SNI)
    }
    return &peekConn{Conn: conn, buf: raw}, nil
}

func (l *fpListener) Close() error   { return l.inner.Close() }
func (l *fpListener) Addr() net.Addr { return l.inner.Addr() }

func main() {
    cert, err := tls.LoadX509KeyPair("server.crt", "server.key")
    if err != nil {
        log.Fatal(err)
    }
    raw, err := net.Listen("tcp", ":8443")
    if err != nil {
        log.Fatal(err)
    }
    tl := tls.NewListener(&fpListener{raw}, &tls.Config{
        Certificates: []tls.Certificate{cert},
    })
    mux := http.NewServeMux()
    mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
        w.Write([]byte("ok"))
    })
    log.Println("listening on :8443")
    log.Fatal(http.Serve(tl, mux))
}

ParseClientHello (not shown here) walks the raw TLS record bytes to populate the ClientHello struct. The byte layout is well-documented: 5-byte record header, 4-byte handshake header, then the ClientHello body. The Go standard library crypto/tls source is the cleanest reference for the exact offsets.

Every accepted connection now emits a structured log line with the remote address, JA3 hash, and SNI before any HTTP layer sees it. Pipe that into your log aggregator and you have a passive fingerprinting layer with near-zero overhead.

JA4: What Changed and Why

JA4 was proposed by FoxIO in 2023 to address JA3's main weaknesses.

Sorted before hashing. JA3 hashes fields in the order the client sends them. Two clients advertising identical capabilities in a different extension order get different JA3 hashes. JA4 sorts cipher suites and extensions before hashing, so the fingerprint reflects what the client supports rather than the arbitrary order of its implementation.

Human-readable prefix. The prefix t13d1516h2 encodes TLS version, SNI presence, cipher count, extension count, and ALPN — all readable without decoding the hash itself. A security analyst can parse a JA4 fingerprint in a log at a glance.

Truncated SHA-256. Two 12-character truncated SHA-256 hashes replace a single MD5. The two-part structure separates cipher fingerprint from extension fingerprint, which is useful when matching partial client profiles across different TLS versions.

The Go implementation follows the same parsing path as JA3. The difference is in the hash step: sort the filtered cipher and extension lists before feeding them to SHA-256, then take the first 12 hex characters of each. The driftnet/ja4 package is a clean reference implementation if you want to skip the wire format details.

The Takeaway

JA3 and JA4 give you a passive, encryption-transparent signal on every TLS connection. The ClientHello is always in the clear, and computing its hash adds microseconds of overhead per connection. In Go, the implementation maps directly onto the language's net abstractions: a custom listener buffers the handshake bytes, a parser extracts the fields, and a deterministic hash function does the rest.

Concrete uses that deliver immediate value:

Log JA3 hashes alongside request IDs in your reverse proxy — instant timeline correlation during security incidents
Build a fingerprint baseline from 48 hours of legitimate traffic and alert on deviations
Cross-reference hashes against threat intel lists (Emerging Threats publishes JA3 indicators)
Extend to JA4S (the server-side equivalent) to audit what TLS stack your servers expose to clients

The fingerprint alone is not a firewall rule. A determined attacker can spoof ClientHello fields in seconds. But commodity tooling — scrapers, scanners, most malware — does not bother to randomize its TLS profile. As one layer in a defense stack, fingerprinting catches a meaningful share of noise before it reaches your application logic.

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

How to Build an AI Model That Classifies Phishing Emails

Ayi NEDJIMI — Thu, 20 Aug 2026 10:05:09 +0000

Phishing emails account for over 90% of confirmed breaches according to the Verizon DBIR. The core problem with rule-based and blacklist-based filters is that they are reactive — you have to have seen an attack before you can block it. An ML-based classifier trained on email content generalizes to new campaigns based on linguistic and structural patterns, not signatures.

This post walks through building a working phishing email classifier in Python: feature engineering, a TF-IDF baseline you can ship in an afternoon, and a fine-tuned language model for when precision really matters. All code is runnable.

What Makes an Email "Phishy"

Before writing a line of code, understand your feature space. Phishing emails cluster around a handful of detectable patterns:

Urgency and fear language: "Your account will be suspended in 24 hours", "Immediate action required"
Impersonation signals: sender display name doesn't match the actual domain, lookalike domains (paypa1.com, amazon-support.com)
Credential harvesting intent: requests for passwords, social security numbers, card details
Suspicious URLs: IP-based links, URL shorteners, domains with suspicious TLDs (.tk, .ml, .ga)
Generic salutations: "Dear Customer", "Dear User" rather than a real name

A model trained on enough labeled examples will learn these patterns without you explicitly coding each one — but knowing the signal space helps you evaluate whether the model has learned the right things.

Dataset Selection

For this project, combine publicly available datasets:

CEAS 2008 phishing corpus (~5k phishing emails)
SpamAssassin public corpus (~6k emails — keep the legitimate subset)
Enron email dataset (~30k legitimate corporate emails)

The class imbalance matters. Real inboxes have roughly 0.5–2% phishing, so train with class_weight="balanced" and measure recall on the phishing class specifically. A model that achieves 99% accuracy by predicting everything as legitimate is worthless in practice.

Baseline: TF-IDF + Logistic Regression

A TF-IDF pipeline is fast to train, interpretable, and often good enough for internal tooling. Start here before reaching for transformers.

import re
import email
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report


def extract_text(raw_email: str) -> str:
    # Extract subject and body text from a raw email string.
    try:
        msg = email.message_from_string(raw_email)
        subject = msg.get('Subject', '') or ''
        body = ''
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == 'text/plain':
                    payload = part.get_payload(decode=True)
                    body += payload.decode('utf-8', errors='ignore') if payload else ''
        else:
            payload = msg.get_payload(decode=True)
            body = payload.decode('utf-8', errors='ignore') if payload else ''
        return f'{subject} {body}'
    except Exception:
        return raw_email


def url_heuristics(text: str) -> str:
    # Append URL-derived pseudo-tokens that signal phishing patterns.
    import re
    urls = re.findall(r'https?://[^\s<>"\']]+', text)
    domains = [re.sub(r'^https?://', '', u).split('/')[0] for u in urls]
    tokens = []
    if any(re.match(r'\d+\.\d+\.\d+\.\d+', d) for d in domains):
        tokens.append('FEAT_IP_URL')
    if any(d.endswith(('.tk', '.ml', '.ga', '.cf')) for d in domains):
        tokens.append('FEAT_SUSPICIOUS_TLD')
    if any(d in {'bit.ly', 't.co', 'goo.gl', 'ow.ly', 'tinyurl.com'} for d in domains):
        tokens.append('FEAT_URL_SHORTENER')
    return text + ' ' + ' '.join(tokens)


# emails.csv has columns: raw (str), label (0=legit, 1=phishing)
df = pd.read_csv('emails.csv')
df['text'] = df['raw'].apply(extract_text).apply(url_heuristics)

X_train, X_test, y_train, y_test = train_test_split(
    df['text'], df['label'], test_size=0.2, stratify=df['label'], random_state=42
)

pipeline = Pipeline([
    ('tfidf', TfidfVectorizer(
        sublinear_tf=True,
        max_features=50_000,
        ngram_range=(1, 2),
        min_df=2,
    )),
    ('clf', LogisticRegression(C=1.0, max_iter=1000, class_weight='balanced')),
])

pipeline.fit(X_train, y_train)
print(classification_report(y_test, pipeline.predict(X_test), target_names=['legitimate', 'phishing']))

On a balanced 10k dataset this typically yields F1 ~0.92–0.94 on the phishing class. The model is interpretable: inspect the top coefficients with pipeline.named_steps['tfidf'].get_feature_names_out() combined with pipeline.named_steps['clf'].coef_ to understand exactly what it learned.

Fine-Tuning a Language Model for Production

When false negatives carry real cost — a missed phishing email that leads to a breach — a fine-tuned small language model consistently outperforms TF-IDF by 3–5% F1. DistilBERT (66M parameters) runs comfortably on a CPU inference endpoint and trains in under 10 minutes on a GPU.

import numpy as np
from datasets import Dataset
from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer,
)
from sklearn.metrics import f1_score, accuracy_score

MODEL_NAME = 'distilbert-base-uncased'
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)


def tokenize_batch(batch):
    return tokenizer(batch['text'], truncation=True, max_length=512, padding='max_length')


train_ds = Dataset.from_dict({'text': X_train.tolist(), 'label': y_train.tolist()})
test_ds  = Dataset.from_dict({'text': X_test.tolist(),  'label': y_test.tolist()})

train_ds = train_ds.map(tokenize_batch, batched=True)
test_ds  = test_ds.map(tokenize_batch,  batched=True)

model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=2)


def compute_metrics(eval_pred):
    logits, labels = eval_pred
    preds = np.argmax(logits, axis=-1)
    return {
        'accuracy': accuracy_score(labels, preds),
        'f1': f1_score(labels, preds, average='binary'),
    }


args = TrainingArguments(
    output_dir='./phishing-clf',
    num_train_epochs=3,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=32,
    evaluation_strategy='epoch',
    save_strategy='epoch',
    load_best_model_at_end=True,
    metric_for_best_model='f1',
    learning_rate=2e-5,
    weight_decay=0.01,
    report_to='none',
)

trainer = Trainer(
    model=model,
    args=args,
    train_dataset=train_ds,
    eval_dataset=test_ds,
    compute_metrics=compute_metrics,
)

trainer.train()
trainer.save_model('./phishing-clf-final')

Three epochs on a V100 takes around 8 minutes for 30k examples. On CPU inference, expect roughly 50ms per email — acceptable for a mail gateway integration.

Serving the Classifier

Wrap the trained model in a FastAPI endpoint. Keep the preprocessing pipeline identical to what you used at training time; skipping this step is the most common source of silent accuracy regression in production.

from fastapi import FastAPI
from pydantic import BaseModel
from transformers import pipeline as hf_pipeline

app = FastAPI()
_clf = hf_pipeline(
    'text-classification',
    model='./phishing-clf-final',
    tokenizer=MODEL_NAME,
    truncation=True,
    max_length=512,
)


class EmailRequest(BaseModel):
    subject: str
    body: str


@app.post('/classify')
def classify(req: EmailRequest):
    text = url_heuristics(f'{req.subject} {req.body}')
    result = _clf(text)[0]
    return {
        'phishing': result['label'] == 'LABEL_1',
        'confidence': round(result['score'], 4),
    }

Wire this behind your mail gateway, an IMAP IDLE listener, or a milter integration and you have a classifier that updates with each monthly retrain cycle.

Production Considerations

A few things that are not obvious from the training code:

Retraining cadence. Phishing campaigns evolve continuously. Retrain monthly at minimum; weekly if analysts are feeding labeled examples from reported emails back into the pipeline.

Threshold tuning. The default 0.5 decision threshold is arbitrary. Plot the precision-recall curve for the phishing class and pick a threshold based on your organization's acceptable false-positive rate. A 0.3 threshold that catches 98% of phishing at the cost of 2% false positives may be better than a 0.5 threshold that catches 94% at 0.5%.

Header-based features. SPF/DKIM/DMARC pass/fail status, reply-to address mismatch, and routing anomalies are strong signals the body model cannot see. Add them as supplementary binary features fed into a stacked ensemble on top of the classifier's probability output.

Adversarial robustness. Attackers obfuscate text with Unicode lookalikes, zero-width characters, or image-only content. Normalize Unicode to NFKC before tokenization. For image-only attachments, run OCR before classification.

The classifier handles content analysis well, but it does not replace infrastructure controls. If you are building this as part of a broader email security program, the security hardening checklists at ayinedjimi-consultants.fr cover the SPF/DKIM/DMARC enforcement, attachment sandboxing, and link rewriting controls that the model alone cannot enforce.

The Takeaway

The TF-IDF baseline gets you to F1 ~0.93 in an afternoon and is worth shipping while you train the transformer. The fine-tuned language model pushes recall on phishing to ~0.97 and generalizes better to novel campaigns. In both cases: measure recall on the phishing class specifically, not overall accuracy; tune your decision threshold against a realistic false-positive budget; and plan for monthly retraining. A classifier trained six months ago without updates is losing ground to adversaries every day.

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

Implementing Secure Configuration Management with Vault and Go

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

Most Go services ship with secrets spread across environment variables, config files, and the occasional hardcoded constant nobody remembers is there. That's manageable until you need to rotate a database password without restarting the service, or answer the question: "who accessed the production API key last Tuesday?" HashiCorp Vault solves both problems, and integrating it cleanly with Go is less work than most tutorials imply.

Why environment variables are not enough

Environment variables are convenient, not secure. They appear in process listings, get dumped in crash reports, and are inherited by every subprocess your service spawns. When a developer accidentally logs os.Environ() during debugging, every secret in the process goes to stdout — and probably to your centralized log platform.

More critically, env vars are static. Once set at startup, they don't rotate. If a credential leaks, your only option is to update the value and restart every affected service. In a microservices environment with dozens of instances, that's a painful and error-prone operation.

Vault offers three capabilities that env vars fundamentally can't provide:

Dynamic secrets — Vault generates short-lived credentials on demand and revokes them automatically when the lease expires
Fine-grained access policies — each service gets a token scoped to exactly the secrets it needs, nothing more
Audit logging — every read, write, and login is recorded with timestamps and identities

Setting up Vault with AppRole authentication

For local development, Vault's dev mode is the fastest path:

vault server -dev -dev-root-token-id="dev-root-token"
export VAULT_ADDR='http://127.0.0.1:8200'
export VAULT_TOKEN='dev-root-token'

# Store a test secret in KV v2
vault kv put secret/myapp/db password="s3cr3t" user="appuser"
vault kv put secret/myapp/api key="sk-prod-abc123"

For production, avoid static Vault tokens entirely. Use AppRole: your service authenticates with a role ID (static, safe to embed in deployment config) and a secret ID (short-lived, injected at runtime). The server exchanges those for a scoped token.

# Enable AppRole and create a role
vault auth enable approle
vault policy write myapp-policy - <<EOF
path "secret/data/myapp/*" {
  capabilities = ["read"]
}
EOF

vault write auth/approle/role/myapp \\
    token_ttl=1h \\
    token_max_ttl=4h \\
    policies=myapp-policy

# Fetch the role ID — static, store it in CI config
vault read -field=role_id auth/approle/role/myapp/role-id

# Generate a secret ID — rotate this on each deploy
vault write -field=secret_id -f auth/approle/role/myapp/secret-id

The separation matters: the role ID is like a username and can live in your Kubernetes deployment manifest. The secret ID is like a one-time password and should be injected by your CI system at pod startup, then discarded.

Reading secrets in Go with the Vault SDK

The github.com/hashicorp/vault/api package handles authentication and HTTP calls. Below is a minimal client that logs in via AppRole and fetches secrets from KV v2:

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "sync"

    vault "github.com/hashicorp/vault/api"
    auth "github.com/hashicorp/vault/api/auth/approle"
)

type VaultClient struct {
    client *vault.Client
    mu     sync.RWMutex
    cache  map[string]map[string]interface{}
}

func NewVaultClient(addr, roleID, secretID string) (*VaultClient, *vault.Secret, error) {
    config := vault.DefaultConfig()
    config.Address = addr

    client, err := vault.NewClient(config)
    if err != nil {
        return nil, nil, fmt.Errorf("vault client init: %w", err)
    }

    appRoleAuth, err := auth.NewAppRoleAuth(
        roleID,
        &auth.SecretID{FromString: secretID},
    )
    if err != nil {
        return nil, nil, fmt.Errorf("approle auth init: %w", err)
    }

    authInfo, err := client.Auth().Login(context.Background(), appRoleAuth)
    if err != nil {
        return nil, nil, fmt.Errorf("vault login: %w", err)
    }
    if authInfo == nil {
        return nil, nil, fmt.Errorf("no auth info returned from vault")
    }

    vc := &VaultClient{
        client: client,
        cache:  make(map[string]map[string]interface{}),
    }
    return vc, authInfo, nil
}

func (v *VaultClient) GetSecret(ctx context.Context, path string) (map[string]interface{}, error) {
    v.mu.RLock()
    if cached, ok := v.cache[path]; ok {
        v.mu.RUnlock()
        return cached, nil
    }
    v.mu.RUnlock()

    secret, err := v.client.KVv2("secret").Get(ctx, path)
    if err != nil {
        return nil, fmt.Errorf("get secret %s: %w", path, err)
    }
    if secret == nil || secret.Data == nil {
        return nil, fmt.Errorf("secret %s not found or empty", path)
    }

    v.mu.Lock()
    v.cache[path] = secret.Data
    v.mu.Unlock()

    return secret.Data, nil
}

func main() {
    vc, authInfo, err := NewVaultClient(
        os.Getenv("VAULT_ADDR"),
        os.Getenv("VAULT_ROLE_ID"),
        os.Getenv("VAULT_SECRET_ID"),
    )
    if err != nil {
        log.Fatalf("vault setup: %v", err)
    }

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go renewToken(ctx, vc.client, authInfo)

    data, err := vc.GetSecret(ctx, "myapp/db")
    if err != nil {
        log.Fatalf("fetch secret: %v", err)
    }
    fmt.Printf("connecting as user: %s\n", data["user"])
}

The cache layer prevents Vault from becoming a hot path on every request — secrets are read once and held in memory. For dynamic secrets with short TTLs you'll want to invalidate the cache before the lease expires.

Handling token renewal without restarts

Vault tokens expire. If your service runs longer than the token's max TTL without renewal, calls start returning 403s. The SDK's LifetimeWatcher handles this cleanly:

func renewToken(ctx context.Context, client *vault.Client, authInfo *vault.Secret) {
    watcher, err := client.NewLifetimeWatcher(&vault.LifetimeWatcherInput{
        Secret:    authInfo,
        Increment: 3600,
    })
    if err != nil {
        log.Printf("token watcher init failed: %v", err)
        return
    }

    go watcher.Start()
    defer watcher.Stop()

    for {
        select {
        case err := <-watcher.DoneCh():
            if err != nil {
                log.Printf("token renewal failed, vault access lost: %v", err)
            }
            return
        case renewal := <-watcher.RenewCh():
            log.Printf("vault token renewed, TTL: %ds",
                renewal.Secret.Auth.LeaseDuration)
        case <-ctx.Done():
            return
        }
    }
}

When DoneCh() fires, the token cannot be renewed and your service must re-authenticate. Wire the renewal goroutine to your application's health check so Kubernetes liveness probes can detect and recover from a Vault outage before it causes user-visible failures.

Structuring configuration in a production Go service

Keep Vault calls out of your hot path. Read all required secrets at startup into a typed struct, then use a read-write mutex for safe concurrent access and in-place hot-reloading:

type AppConfig struct {
    mu     sync.RWMutex
    DBUser string
    DBPass string
    APIKey string
}

func LoadConfig(ctx context.Context, vc *VaultClient) (*AppConfig, error) {
    dbData, err := vc.GetSecret(ctx, "myapp/db")
    if err != nil {
        return nil, fmt.Errorf("load db config: %w", err)
    }
    apiData, err := vc.GetSecret(ctx, "myapp/api")
    if err != nil {
        return nil, fmt.Errorf("load api config: %w", err)
    }
    return &AppConfig{
        DBUser: dbData["user"].(string),
        DBPass: dbData["password"].(string),
        APIKey: apiData["key"].(string),
    }, nil
}

func (c *AppConfig) Credentials() (string, string) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.DBUser, c.DBPass
}

func (c *AppConfig) Reload(ctx context.Context, vc *VaultClient) error {
    fresh, err := LoadConfig(ctx, vc)
    if err != nil {
        return err
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    c.DBUser = fresh.DBUser
    c.DBPass = fresh.DBPass
    c.APIKey = fresh.APIKey
    return nil
}

This separates "when to fetch" from "how to use", and makes testing straightforward — in unit tests, populate the struct directly without touching Vault at all.

For a complete checklist of Vault hardening steps — including policy templates, audit backend configuration, and AppRole bootstrap procedures — see our free security hardening checklists in PDF and Excel format.

The takeaway

Environment variables are a footgun for secrets at any scale beyond a side project. Vault gives you short-lived credentials, immutable audit trails, and automatic revocation — the three things that matter during post-incident forensics or an access review.

The Go SDK handles AppRole authentication and token renewal cleanly once you understand the flow. The pattern above — authenticate at startup, cache secrets in a typed struct, renew the token in a background goroutine — covers most production Go services without meaningful added complexity.

One thing most teams skip: write a runbook for Vault unavailability. A service that fails hard at boot when Vault is down, then retries from a local cache if connectivity drops mid-run, behaves predictably under failure. A service that silently falls back to stale secrets or skips auth checks is a different problem entirely — the kind that surfaces in breach post-mortems, not in health checks.

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

Building a Kubernetes Security Audit Tool with Go

Ayi NEDJIMI — Tue, 18 Aug 2026 10:05:28 +0000

Most Kubernetes clusters ship with insecure defaults that nobody touches until something goes wrong. A misconfigured Pod security context, a container running as root, missing resource limits — these are the first things an attacker looks for after gaining access. Writing a Go tool that connects directly to the Kubernetes API lets you catch these issues automatically, before your next audit or incident.

What We're Auditing

A typical cluster has dozens of potential misconfigurations. For this tool, we focus on the controls most frequently violated in the wild:

Containers running as root (UID 0)
Privileged containers (privileged: true)
Missing resource limits (CPU and memory)
Pods with hostNetwork: true or hostPID: true
Missing readOnlyRootFilesystem
allowPrivilegeEscalation set to true (which is the Kubernetes default if unset)

This maps closely to the CIS Kubernetes Benchmark. If you want a complete audit checklist with all CIS controls, we publish one at ayinedjimi-consultants.fr/checklists — PDF and Excel, free.

Setting Up the Go Client

We'll use client-go, the official Kubernetes Go client. Add the dependencies:

go get k8s.io/client-go@latest k8s.io/api@latest k8s.io/apimachinery@latest

The client authenticates via in-cluster config when running inside a Pod, or falls back to a local kubeconfig. We handle both cases with a single function:

package main

import (
    "fmt"
    "os"
    "path/filepath"

    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "k8s.io/client-go/tools/clientcmd"
)

func buildClient() (*kubernetes.Clientset, error) {
    // Try in-cluster config first (running inside a Pod)
    cfg, err := rest.InClusterConfig()
    if err != nil {
        // Fall back to local kubeconfig
        home, _ := os.UserHomeDir()
        kubeconfigPath := filepath.Join(home, ".kube", "config")
        cfg, err = clientcmd.BuildConfigFromFlags("", kubeconfigPath)
        if err != nil {
            return nil, fmt.Errorf("failed to build config: %w", err)
        }
    }
    return kubernetes.NewForConfig(cfg)
}

This pattern works whether you run the auditor locally, in a CI job, or deploy it as a CronJob inside the cluster.

Core Audit Logic

We iterate over all pods across all namespaces and check each container's security context. The Finding struct is the central data type:

package main

import (
    "context"
    "fmt"

    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
)

type Finding struct {
    Namespace string `json:"namespace"`
    Pod       string `json:"pod"`
    Container string `json:"container,omitempty"`
    Check     string `json:"check"`
    Severity  string `json:"severity"`
}

func auditPods(ctx context.Context, client *kubernetes.Clientset) ([]Finding, error) {
    var findings []Finding

    pods, err := client.CoreV1().Pods("").List(ctx, metav1.ListOptions{})
    if err != nil {
        return nil, fmt.Errorf("listing pods: %w", err)
    }

    for _, pod := range pods.Items {
        if pod.Spec.HostNetwork {
            findings = append(findings, Finding{
                Namespace: pod.Namespace, Pod: pod.Name,
                Check: "hostNetwork=true", Severity: "HIGH",
            })
        }
        if pod.Spec.HostPID {
            findings = append(findings, Finding{
                Namespace: pod.Namespace, Pod: pod.Name,
                Check: "hostPID=true", Severity: "HIGH",
            })
        }
        for _, c := range pod.Spec.Containers {
            findings = append(findings, checkContainer(pod, c)...)
        }
    }
    return findings, nil
}

func checkContainer(pod corev1.Pod, c corev1.Container) []Finding {
    var findings []Finding

    loc := func(check, sev string) Finding {
        return Finding{
            Namespace: pod.Namespace, Pod: pod.Name,
            Container: c.Name, Check: check, Severity: sev,
        }
    }

    sc := c.SecurityContext
    if sc == nil {
        return append(findings, loc("missing SecurityContext", "MEDIUM"))
    }

    if sc.Privileged != nil && *sc.Privileged {
        findings = append(findings, loc("privileged=true", "CRITICAL"))
    }

    if sc.RunAsNonRoot == nil || !*sc.RunAsNonRoot {
        if sc.RunAsUser == nil || *sc.RunAsUser == 0 {
            findings = append(findings, loc("container may run as root", "HIGH"))
        }
    }

    if sc.ReadOnlyRootFilesystem == nil || !*sc.ReadOnlyRootFilesystem {
        findings = append(findings, loc("readOnlyRootFilesystem not enforced", "LOW"))
    }

    // allowPrivilegeEscalation defaults to true in Kubernetes if unset
    if sc.AllowPrivilegeEscalation == nil || *sc.AllowPrivilegeEscalation {
        findings = append(findings, loc("allowPrivilegeEscalation=true", "MEDIUM"))
    }

    if c.Resources.Limits == nil {
        findings = append(findings, loc("no resource limits", "MEDIUM"))
    }

    return findings
}

A few things worth noting. We treat a missing SecurityContext as a single MEDIUM finding and return early — no point emitting five sub-findings when the root cause is one missing block. AllowPrivilegeEscalation defaults to true in Kubernetes if unset, so we flag the nil case explicitly.

Reporting and Exit Codes

For a first version, JSON output piped into your SIEM or a Slack webhook is sufficient:

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
)

func main() {
    client, err := buildClient()
    if err != nil {
        log.Fatalf("init client: %v", err)
    }

    ctx := context.Background()
    findings, err := auditPods(ctx, client)
    if err != nil {
        log.Fatalf("audit failed: %v", err)
    }

    if len(findings) == 0 {
        fmt.Println("No findings.")
        os.Exit(0)
    }

    enc := json.NewEncoder(os.Stdout)
    enc.SetIndent("", "  ")
    for _, f := range findings {
        if err := enc.Encode(f); err != nil {
            log.Fatalf("encode: %v", err)
        }
    }

    // Exit non-zero on CRITICAL findings — useful as a CI gate
    for _, f := range findings {
        if f.Severity == "CRITICAL" {
            os.Exit(2)
        }
    }
}

The exit code pattern matters. Exit 0 means clean, exit 2 means at least one CRITICAL finding. This lets you wire the tool into a CI job or a pre-deployment check: if exit code is 2, block the pipeline.

To run the auditor inside the cluster on a schedule:

apiVersion: batch/v1
kind: CronJob
metadata:
  name: k8s-auditor
  namespace: security
spec:
  schedule: "0 3 * * 1"   # Every Monday at 03:00
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: k8s-auditor
          restartPolicy: OnFailure
          containers:
          - name: auditor
            image: your-registry/k8s-auditor:latest

The service account needs only get and list on pods across all namespaces — nothing more. Read-only access means the auditor itself cannot be weaponized if compromised.

Extending the Auditor

Once the pod checks are solid, adding new rules is straightforward — each is a function that takes a *kubernetes.Clientset and returns []Finding.

Network policies: Check whether each namespace has at least one NetworkPolicy. A namespace without one is completely open to lateral movement from any other pod in the cluster.

RBAC: List ClusterRoleBindings and flag any that grant cluster-admin to a service account. One API call surfaces over-privileged accounts that nobody remembers creating.

Image hygiene: Flag containers using the latest tag or pulling from unverified registries. Kubernetes doesn't enforce this by default, and latest makes deployments non-deterministic under pressure.

Secrets exposure: Detect secrets mounted as environment variables instead of volumes. Environment variables are visible in process listings and often end up in application logs.

The Takeaway

The Kubernetes API exposes everything you need to audit your cluster's security posture without installing a third-party agent. Go's client-go library handles authentication and serialization, and the compiled binary is a single static artifact you can drop into any environment or container image.

Start with the pod security checks above. Wire the exit code into your CI pipeline. Add RBAC checks in the next iteration. An auditor you run every week beats a comprehensive one that's too complex to maintain. Security tooling that actually gets used is the only kind that matters.

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