TRENDING
An empty toll plaza entry with several lanes under a steel canopy, green arrow signals above the booths and a toll fare board beside the road
November 6, 2026
How to Spread Requests Across Servers in Python With the Power of Two Choices and Avoid the Herd Effect of Least-Loaded Routing
Cross-shaped arrow slit set in a recessed brick arch in a lichen-spotted castle wall at White Castle, Monmouthshire
November 6, 2026
Microsoft’s PDD Kit Turns Code Review Into Intent Review, but Verus Accepts a Login That Answers Every Attempt With a 500 Under the Contract the Post Prints
Several dried pufferfish with their spines raised hanging in a stack at a seaside market stall
November 6, 2026
OpenSSH Patches a Password Bypass on musl Systems Such as Alpine, Where a 257 to 1,024 Byte Password Logs In to Password-Disabled Accounts, and musl’s Maintainer Says the Bug Is Not in musl
Black silhouette of a wooden hourglass on a ledge against a grey cloudy sky with blurred bare branches
November 6, 2026
How to Stop Hung HTTP Calls in Python With Timeouts and Deadlines, and Catch the Slow-Drip Response That Slips Under requests, httpx, and urllib
Woodland path closed off with red and white barrier tape, traffic cones and a warning triangle
November 6, 2026
OpenAI’s “A Thousand Fewer Paths to Root” Turns $1.2 Million of API Credit Into 948 Candidate Patches, but Mainline Linux Shows 24 OpenAI-Linked Commits Since Mid-September and the Bounty Math Assumes Payouts the Rules Do Not Offer
06 Nov 2026
SXZ.io SXZ.io
  • Home
Search the Site
Popular Searches:
Technology Amazon AI
Recent Posts
A chrome ticket punch lying on a bright blue surface, the tool that makes a ticket good for one use only
How to Secure a WebSocket Endpoint in FastAPI and Stop Cross-Site Hijacking, Leaked Tokens, Floods, and Slow Readers
November 6, 2026
Yellow stenciled words Mind the gap painted on a station platform edge beside the railway tracks
OpenAI’s Astra Report Turns Silent Safety-Task Underperformance Into a Matched-Pair Gap, but the Ethical-Doubt Result’s Bar Runs From a 4-Point Gap to a 3.6-Point Reversal, and the Launch Card Quoted AISI as Seeing Very Few Refusals
November 6, 2026
Black and white photo of a speckled stoneware honey pot with a lid and a wooden dipper standing in its opening, against a dark background
OpenAI Releases GPT-6 Cyber, Rated Critical, to Approved Customers: the Post’s 99.93% Scope Score Is Measured Against the Older Cyber Model, the System Card Puts Its Base Model at 0% Honeypot Exploitation, and One Outside Benchmark Has It Solving Fewer Challenges
November 6, 2026
SXZ.io SXZ.io
  • Home

Categories

Articles 285 Posts
News 289 Posts
Learning Hub 259 Posts
Home/Learning Hub/How to Keep Secrets Out of Python Logs and Stop Forged Log Lines With a Safe Formatter and Canary Tests
Learning Hub

How to Keep Secrets Out of Python Logs and Stop Forged Log Lines With a Safe Formatter and Canary Tests

Build a small Python logging module that hides secrets wherever they leak (messages, arguments, tracebacks, library URLs and uvicorn's access log) and keeps attacker-supplied newlines from forging...

November 2, 2026 42 Min Read
30

Every log line you write is a copy. It lands in a file, a terminal scrollback, a log-shipping agent, a search index and often a ticket, and it stays there long after the request that produced it. Teams guard the original secret carefully and then print it into the copy by accident: a password in a debug message, an API key in a URL that a library helpfully logs, a token inside an exception message. Logs tend to be copied to more places than secrets are, so they are a common way for a secret to escape, and our report on exposed Claude Code logs that tied an AI pentest agent to data leaks at Korean financial firms is one recent example of what a log can reveal once someone else can read it.

Table Of Content

  • What you will build, and the words you need first
  • Prerequisites
  • Step 1: Set up the lab and watch five leaks happen
  • Step 2: Make the secret impossible to print
  • Step 3: Build a redactor for the leaks you cannot prevent
  • Check what the redactor costs
  • Make sure the redactor cannot be turned against you
  • What the rules still miss
  • Step 4: Decide where the redactor lives
  • Three places that look right
  • The formatter, line by line
  • Step 5: Stop attackers from forging log lines
  • Neutralize what a log reader would misread
  • JSON lines: one event, one line
  • Step 6: Catch what libraries write
  • Step 7: Fix the loggers uvicorn creates for itself
  • Step 8: Lock it in with canary tests
  • Two traps in pytest’s caplog
  • Break it on purpose: a mutation check
  • Step 9: Confirm everything end to end
  • Common mistakes, in one place
  • Where to go next
  • Sources

In this tutorial you will find those leaks, remove them at the source, and add a safety net that hides whatever slips through. The same small module also closes the opposite hole: an attacker who puts a newline in a username so that your log file contains a line nobody on your team wrote. You will measure every claim with fake secrets, so you never need to paste a real key into a test. If you want the debugging side of logging, our tutorial on correlation IDs and structured logging in Python covers it; this one is about making logs safe to keep.

What you will build, and the words you need first

A log record is the object Python creates each time you call log.info(...). It holds the message, its arguments, the level, the logger’s name and any exception. A logger is the named object you call. A handler sends a record somewhere, such as the console or a file. A formatter turns a record into the text that the handler writes. A filter can change or drop a record on its way.

Two problems matter here, and both have official names. CWE-532 is short: “The product writes sensitive information to a log file.” CWE-117, also called log forging, describes a product that “constructs a log message from external input, but it does not neutralize or incorrectly neutralizes special elements when the message is written to a log file.” To redact text is to replace a secret with a marker such as [REDACTED]. A canary is a fake secret that contains a unique word, so that one search finds every place it leaked.

By the end you will have:

  • a Secret wrapper that cannot be printed by accident;
  • a redactor that hides secrets by pattern and by exact value, with a timing check showing that hostile input cannot make it slow;
  • a text formatter and a JSON formatter that hide secrets in messages, arguments, tracebacks and structured fields, and keep each event on one line;
  • a live uvicorn server whose own access log and error log pass through the same protection;
  • 40 tests built on canaries, and eight deliberate bugs that those tests catch.

What this does not do: the best defence is still never to log a secret at all, and Steps 2 and 6 are about that. No pattern list finds every secret, so treat the redactor as a safety net and test it. And nothing here covers how long logs are kept, who may read them, or personal data such as email addresses; the OWASP Logging Cheat Sheet covers those.

Prerequisites

  • Python 3.11 or newer. Every command below ran on Python 3.13.14 on Windows 11. I read Requires-Python in each pinned package’s metadata: Starlette 1.10.0 (which FastAPI installs) needs 3.11 or newer, and the other pinned packages need 3.10 or newer (httpx accepts 3.8 and newer).
  • A bash shell. On Windows I used Git Bash. Linux and macOS need nothing extra.
  • Packages: FastAPI 0.162.0, Uvicorn 0.54.0, httpx 0.28.1, requests 2.34.2, Pydantic 2.14.2 and pytest 9.2.0, pinned in Step 1.
  • Background: you can read a Python function and a dictionary, and you know what an HTTP request, a header and a URL query string are. No Docker, no cloud account and no real credentials are needed.

Step 1: Set up the lab and watch five leaks happen

Create a folder and a virtual environment (a private set of installed packages for this project), then install the pinned packages:

$ mkdir safe-logging && cd safe-logging
$ python -m venv .venv
$ source .venv/bin/activate
$ pip install fastapi==0.162.0 uvicorn==0.54.0 httpx==0.28.1 requests==2.34.2 pydantic==2.14.2 pytest==9.2.0

On Windows in Git Bash the activation line is source .venv/Scripts/activate. All the files in this tutorial go in this one folder. The first line of every file is a comment with its name, so you can tell which code block belongs to which file.

Start with the fake secrets. Each contains the word CANARY and has the shape of a real credential: a Stripe-style key, a GitHub-style token, an AWS-style access key ID, a JWT (a signed login token made of three dot-separated parts) and a password. The file also holds four helpers that you will use all through the tutorial. leaked counts canaries in a text. closed_port finds a local port where nothing is listening. url_error produces the exception that requests raises when a server is down. make_logger builds a logger that writes into a string buffer, so we can inspect exactly what a handler wrote.

# labkit.py
"""Fake secrets for the lab. Every one contains the word CANARY, so one search finds every leak."""
import functools
import io
import logging
import socket

PASSWORD = "correct-horse-CANARY-battery"
API_KEY = "sk_live_CANARY51HqX7aB9cD2eF"
TOKEN = "ghp_CANARY" + "a1B2c3D4e5F6g7H8i9J0k1L2m3N4o5"
AWS_KEY = "AKIA" + "CANARY0123456789"
JWT = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhbGljZSJ9.CANARYsignature-0123"


def leaked(text):
    """How many canaries are visible in `text`."""
    return text.count("CANARY")


def closed_port():
    """A local port where nothing is listening, so a connection is refused at once."""
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


@functools.cache
def url_error():
    """The exception requests raises when a server is down. Its message repeats the URL, query string and all."""
    import requests

    try:
        requests.get(f"http://127.0.0.1:{closed_port()}/report?api_key={API_KEY}", timeout=0.5)
    except requests.RequestException as error:
        return error


def make_logger(name, formatter, level=logging.DEBUG):
    """A logger that writes to a string buffer through `formatter`. Returns (logger, handler, buffer)."""
    buffer = io.StringIO()
    handler = logging.StreamHandler(buffer)
    handler.setFormatter(formatter)
    logger = logging.getLogger(name)
    logger.handlers[:] = [handler]
    logger.filters[:] = []
    logger.setLevel(level)
    logger.propagate = False
    return logger, handler, buffer

Now write five ordinary-looking pieces of application code, each with one log call. They are the kind of code you will find in almost any project: a login function, a charge function, a handler that logs its payload, a token refresh that fails, and a report download that cannot reach its server. The script sends the log output into a buffer and prints every line that contains a canary.

# step01_leaks.py
"""Five ordinary-looking log calls, five leaks."""
import io
import logging
import re

import requests

from labkit import API_KEY, PASSWORD, TOKEN, closed_port, leaked

buffer = io.StringIO()
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s", stream=buffer)
log = logging.getLogger("shop")


def login(username, password):
    log.info(f"login attempt user={username} password={password}")


def charge():
    log.info("charging with key %s", API_KEY)


def handle(payload):
    log.info("payload: %s", payload)


def refresh():
    try:
        raise ValueError(f"refresh rejected for token {TOKEN}")
    except ValueError:
        log.exception("refresh failed")


def fetch_report():
    try:
        requests.get(f"http://127.0.0.1:{closed_port()}/report?api_key={API_KEY}", timeout=0.5)
    except requests.RequestException:
        log.exception("report download failed")


login("alice", PASSWORD)
charge()
handle({"user": "alice", "password": PASSWORD})
refresh()
fetch_report()

text = buffer.getvalue()
for line in text.splitlines():
    if "CANARY" in line:
        print(re.sub(r"port=\d+", "port=PORT", line.split(" (Caused by")[0]))
print(f"{leaked(text)} canary strings are visible in the log output")

Run it with python step01_leaks.py:

INFO shop login attempt user=alice password=correct-horse-CANARY-battery
INFO shop charging with key sk_live_CANARY51HqX7aB9cD2eF
INFO shop payload: {'user': 'alice', 'password': 'correct-horse-CANARY-battery'}
ValueError: refresh rejected for token ghp_CANARYa1B2c3D4e5F6g7H8i9J0k1L2m3N4o5
urllib3.exceptions.MaxRetryError: HTTPConnectionPool(host='127.0.0.1', port=PORT): Max retries exceeded with url: /report?api_key=sk_live_CANARY51HqX7aB9cD2eF
requests.exceptions.ConnectTimeout: HTTPConnectionPool(host='127.0.0.1', port=PORT): Max retries exceeded with url: /report?api_key=sk_live_CANARY51HqX7aB9cD2eF
6 canary strings are visible in the log output

Each of the five leaked, and each leaked in a different way:

  • The f-string (login) builds the whole message before the logging module sees it. By the time logging is involved, the password is already part of the text.
  • The argument (charge): the message and its arguments travel separately and are merged only when a formatter produces the text. Anything that inspects just the message misses the secret.
  • The whole dictionary (handle) hides the password inside a larger object, so the call site looks innocent.
  • The exception message (refresh): log.exception adds the traceback, including the text of the exception, and that text is often built from the values the code was working with.
  • The library exception (fetch_report): requests repeats the full URL, query string included, in its error. Two lines leaked because urllib3 raised first and requests wrapped its error.

Check yourself: you should see six canary strings. I ran this only on Windows, where the connection to the closed local port ended in ConnectTimeout. On other systems the exception class may differ, but the “Max retries exceeded with url” message repeats the URL either way.

Step 2: Make the secret impossible to print

The most reliable fix is the boring one: do not let the secret travel through your program as an ordinary string. A plain str is happy to be printed, formatted, logged, put in a dictionary and dumped to JSON. A wrapper object can refuse all of that, and hand the value out only when you ask for it by name, at the one place that needs it.

# secret.py
"""A wrapper that refuses to show its value, so a careless log line cannot leak it."""


class Secret:
    """Holds a value that must not reach a log line, a repr or a JSON dump by accident."""

    __slots__ = ("_value",)

    def __init__(self, value):
        self._value = value

    def reveal(self):
        """The only way to get the real value out. Call it at the last moment, where the secret is used."""
        return self._value

    def __repr__(self):
        return "Secret('[REDACTED]')"

    def __str__(self):
        return "[REDACTED]"

__slots__ keeps the object small and stops accidental extra attributes. __str__ and __repr__ are what print, f-strings, %s, %r and the default logging formatting call, so both return a marker. reveal() is the only way out, and the name is deliberately loud: a reviewer who sees it knows a secret is about to be used.

Try it in a settings object, and compare it with SecretStr from Pydantic, which does the same job in a library you may already use:

# step02_secret.py
"""Make the secret impossible to print by accident."""
import json
import logging
import sys
from dataclasses import dataclass

from pydantic import BaseModel, SecretStr

from labkit import API_KEY
from secret import Secret


@dataclass
class Settings:
    database: str
    api_key: Secret


settings = Settings("db.internal/shop", Secret(API_KEY))
print("repr of the settings object:", repr(settings))
print("in an f-string:", f"{settings.api_key}")
print("with percent formatting:", "%s" % settings.api_key)
print("repr():", repr(settings.api_key))

try:
    json.dumps({"api_key": settings.api_key})
except TypeError as error:
    print("json.dumps fails loudly:", error)
print("json.dumps with default=str:", json.dumps({"api_key": settings.api_key}, default=str))

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s", stream=sys.stdout)
logging.getLogger("shop").info("calling the API with key %s", settings.api_key)
print("the real value is still there for the call:", len(settings.api_key.reveal()), "characters")

print("\n=== The same idea in Pydantic")


class Config(BaseModel):
    api_key: SecretStr


config = Config(api_key=API_KEY)
print("repr:", repr(config))
print("str():", str(config.api_key))
print("model_dump_json():", config.model_dump_json())
print("model_dump():", config.model_dump())
print("get_secret_value() returns the real value:", config.api_key.get_secret_value() == API_KEY)

Run python step02_secret.py:

repr of the settings object: Settings(database='db.internal/shop', api_key=Secret('[REDACTED]'))
in an f-string: [REDACTED]
with percent formatting: [REDACTED]
repr(): Secret('[REDACTED]')
json.dumps fails loudly: Object of type Secret is not JSON serializable
json.dumps with default=str: {"api_key": "[REDACTED]"}
INFO shop calling the API with key [REDACTED]
the real value is still there for the call: 28 characters

=== The same idea in Pydantic
repr: Config(api_key=SecretStr('**********'))
str(): **********
model_dump_json(): {"api_key":"**********"}
model_dump(): {'api_key': SecretStr('**********')}
get_secret_value() returns the real value: True

Read the output from the top. A dataclass prints every field in its repr, so a settings object is a classic leak; here the key shows as Secret('[REDACTED]'). The f-string, the percent-style formatting and the logging call all print [REDACTED]. json.dumps raises a TypeError, which is what you want: a program that crashes in a test is better than one that quietly writes a key to a file. Passing default=str turns the secret into the marker instead. Pydantic behaves the same way, with two details worth knowing: model_dump() keeps the SecretStr object (so it is still masked when printed), and model_dump_json() writes **********.

Common mistakes. Calling reveal() early and keeping the result in a variable (key = settings.api_key.reveal()) puts the plain string back into your code, where an f-string can leak it again. Call it inline, at the point of use, for example when you build a header. And a wrapper only protects values that start life wrapped. A password in a request body, a token inside a URL and the message of a library exception are plain strings, which is what the next step is for.

One more habit costs nothing: log the fields you choose, not the object you were given. log.info("login for user %s", user_id) cannot leak a password, while log.info("request %s", request_body) can. OWASP lists what should usually not be recorded directly in logs but instead “removed, masked, sanitized, hashed, or encrypted”: “Access tokens”, “Authentication passwords”, “Database connection strings” and “Encryption keys and other primary secrets” are on that list.

Step 3: Build a redactor for the leaks you cannot prevent

Some leaks start as plain strings that come from outside your code, so a safety net has to recognise secrets in text. There are two ways to do that, and the redactor uses both. A pattern rule recognises a secret by its shape: a value after password=, a string that starts with AKIA followed by sixteen capital letters or digits (a shape that AWS access key IDs commonly have), a JWT. A registered value recognises a secret because you told it the exact value at startup, so it is found even when it has no recognisable shape. Create safelog.py. This first part is the redactor; later steps add the formatters to the same file.

# safelog.py
"""Keep secrets out of log output and stop attackers from forging log lines."""
import copy
import json
import logging
import re
from urllib.parse import quote

REDACTED = "[REDACTED]"

# Names that mark a value as sensitive. The word may sit inside a longer name (client_secret, X-Api-Key).
SENSITIVE = r"(?:password|passwd|pwd|secret|token|api[_-]?key|access[_-]?key|credential)"
SENSITIVE_NAME = re.compile(SENSITIVE + "|authorization|cookie", re.IGNORECASE)

# A value after key=value, key: value, "key": "value" or 'key': 'value'. A quoted value may contain spaces.
VALUE = r"(?P<value>\"[^\"]*\"?|'[^']*'?|[^\s&,;\"'}]+)"


def _keep_key(match):
    """Keep what explains the value (the key, the scheme) and hide only the value."""
    value = match.group("value")
    quote_mark = value[0] if value[0] in "\"'" else ""
    return match.group("key") + quote_mark + REDACTED + quote_mark


RULES = [
    # Authorization: Bearer abc.def   /   Authorization: Basic dXNlcjpwYXNz
    (re.compile(r"(?P<key>authorization\s{0,4}[:=]\s{0,4}(?:bearer|basic|token)\s{1,4})" + VALUE, re.IGNORECASE), _keep_key),
    # password=hunter2   {"api_key": "abc def"}   client_secret: abc
    (re.compile(r"(?P<key>" + SENSITIVE + r"[\w.-]{0,40}[\"']?\s{0,4}[:=]\s{0,4})" + VALUE, re.IGNORECASE), _keep_key),
    # postgresql://user:password@host/db
    (re.compile(r"(?P<key>://[^/\s:@]{1,64}:)(?P<value>[^/\s@]{1,128})(?=@)"), _keep_key),
    # Secrets whose shape is recognisable on its own
    (re.compile(r"\bAKIA[0-9A-Z]{16}\b"), REDACTED),
    (re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"), REDACTED),
    (re.compile(r"\b[sr]k_(?:live|test)_[A-Za-z0-9]{16,}\b"), REDACTED),
    (re.compile(r"(?<![A-Za-z0-9_-])eyJ[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]{5,}"), REDACTED),
]


class Redactor:
    """Hides secrets in text: the exact values you register, then pattern rules for shapes you know."""

    def __init__(self, rules=RULES):
        self.rules = list(rules)
        self._known = []

    def register(self, value):
        """Remember a secret in every shape it is likely to be printed in."""
        if len(value) < 8:
            raise ValueError("a secret shorter than 8 characters would match ordinary words")
        shapes = {value, quote(value, safe=""), json.dumps(value)[1:-1], repr(value)[1:-1]}
        self._known = sorted(set(self._known) | shapes, key=len, reverse=True)

    def scrub(self, text):
        for known in self._known:
            text = text.replace(known, REDACTED)
        for pattern, replacement in self.rules:
            text = pattern.sub(replacement, text)
        return text

    def scrub_obj(self, obj):
        """Scrub a structure for JSON logs: a sensitive key hides its whole value."""
        if isinstance(obj, str):
            return self.scrub(obj)
        if isinstance(obj, dict):
            return {
                str(key): REDACTED if SENSITIVE_NAME.search(str(key)) else self.scrub_obj(value)
                for key, value in obj.items()
            }
        if isinstance(obj, (list, tuple, set)):
            return [self.scrub_obj(item) for item in obj]
        if obj is None or isinstance(obj, (bool, int, float)):
            return obj
        return self.scrub(str(obj))


DEFAULT = Redactor()

Walk through it from the top.

  • SENSITIVE lists the words that mark a value as secret. The word may sit inside a longer name, so client_secret and X-Api-Key match as well.
  • VALUE describes what comes after the key. A quoted value may contain spaces, so it is matched up to the closing quote (or to the end of the text if the closing quote is missing, which over-redacts instead of leaking). An unquoted value stops at a space, &, a comma, a semicolon, a quote or }, so api_key=abc&page=2 hides only abc.
  • _keep_key keeps the part that explains the value (the key, the Bearer scheme, the user name in a connection string) and replaces only the value, so a redacted line is still useful for debugging.
  • The first three RULES handle authorization headers, key=value pairs in any shape (query strings, JSON, a Python dictionary printed with %s) and passwords inside connection URLs. The last four recognise a secret by its shape alone: an AWS key ID, a GitHub token, a Stripe key and a JWT.
  • Redactor.register stores a secret in four shapes: as it is, percent-encoded (how it looks in a URL), JSON-escaped, and as repr shows it. It refuses anything shorter than eight characters, because a three-letter secret would match inside ordinary words and corrupt every log line.
  • scrub replaces registered values first and then applies the rules. scrub_obj does the same for a dictionary or list, which the JSON formatter in Step 5 needs: a key whose name looks sensitive hides its whole value, whatever the value looks like.
  • DEFAULT is one shared redactor. A logging configuration written as a dictionary can only hand simple values to a formatter, so the shared state lives in a module-level object that your startup code feeds with DEFAULT.register(...).

Before the redactor goes anywhere near the logging module, try it on its own. step03_redactor.py runs four experiments: the rules, registered values, ordinary lines that the rules wrongly change, and cost.

# step03_redactor.py
"""Try the redactor on its own before it goes anywhere near the logging module."""
import json
import re
import time
from urllib.parse import quote

from labkit import API_KEY, AWS_KEY, JWT, PASSWORD, TOKEN
from safelog import Redactor

redactor = Redactor()

print("=== Pattern rules")
samples = [
    f"Authorization: Bearer {JWT}",
    f"GET /export?user=bob&api_key={API_KEY}&page=2",
    f'{{"user": "alice", "password": "two words {PASSWORD}"}}',
    f"connecting to postgresql://shop:{PASSWORD}@db.internal:5432/shop",
    f"deploy key {AWS_KEY} rotated",
    f"cloned the repository with {TOKEN}",
    "retry 3 of 5 for order 100234",
]
for sample in samples:
    print("in :", sample)
    print("out:", redactor.scrub(sample))

print("\n=== A registered value is hidden in every shape it is printed in")
odd = 'pa"ss w0rd&CANARY=1'
known = Redactor()
known.register(odd)
for shape in (odd, quote(odd, safe=""), json.dumps(odd)[1:-1], repr(odd)[1:-1]):
    print(known.scrub(f"saw {shape} in the request"))
try:
    known.register("short")
except ValueError as error:
    print("register('short'):", error)

print("\n=== Over-redaction: ordinary lines that the rules also touch")
for line in [
    "Invalid password: too short",
    "Invalid token: expired",
    "token bucket refilled: 5 tokens",
    "password reset email sent to user 42",
    "secret santa draw finished",
    "api_key missing in request",
]:
    out = redactor.scrub(line)
    print("changed  " if out != line else "unchanged", line, "->" if out != line else "", out if out != line else "")

print("\n=== Cost on ordinary lines")
line = "INFO shop order 100234 shipped to customer 5521 via carrier ACME, tracking 1Z999AA10123456784"
count = 100_000
start = time.perf_counter()
for _ in range(count):
    redactor.scrub(line)
print(f"{(time.perf_counter() - start) / count * 1e6:.1f} microseconds per line over {count:,} lines")

print("\n=== Hostile input: can an attacker make the redactor slow?")
hostile = {
    "1,000,000 letters": "a" * 1_000_000,
    "1,000,000 spaces": " " * 1_000_000,
    "100,000 x password=": "password=" * 100_000,
    "100,000 x password=quote": 'password="' * 100_000,
    "100,000 x Authorization: Bearer": "Authorization: Bearer " * 100_000,
    "250,000 x eyJ-": "eyJ-" * 250_000,
}
for name, text in hostile.items():
    start = time.perf_counter()
    redactor.scrub(text)
    print(f"{name:34} {time.perf_counter() - start:5.2f} s")

print("\n=== The same attack on two rules written the obvious way")
naive_jwt = re.compile(r"\beyJ[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]{5,}")
naive_key = re.compile(r"([\w.-]*password[\w.-]*\s*[:=]\s*)(\S+)", re.IGNORECASE)
for name, rule, unit, sizes in (
    ("\\beyJ... rule", naive_jwt, "eyJ-", (16_000, 32_000, 64_000, 128_000)),
    ("[\\w.-]*password rule", naive_key, "a", (4_000, 8_000, 16_000, 32_000)),
):
    for size in sizes:
        text = unit * (size // len(unit))
        start = time.perf_counter()
        rule.sub("X", text)
        print(f"{name:22} {len(text):8,} characters {time.perf_counter() - start:6.2f} s")

Run python step03_redactor.py. The first part shows the rules at work:

=== Pattern rules
in : Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhbGljZSJ9.CANARYsignature-0123
out: Authorization: Bearer [REDACTED]
in : GET /export?user=bob&api_key=sk_live_CANARY51HqX7aB9cD2eF&page=2
out: GET /export?user=bob&api_key=[REDACTED]&page=2
in : {"user": "alice", "password": "two words correct-horse-CANARY-battery"}
out: {"user": "alice", "password": "[REDACTED]"}
in : connecting to postgresql://shop:[email protected]:5432/shop
out: connecting to postgresql://shop:[REDACTED]@db.internal:5432/shop
in : deploy key AKIACANARY0123456789 rotated
out: deploy key [REDACTED] rotated
in : cloned the repository with ghp_CANARYa1B2c3D4e5F6g7H8i9J0k1L2m3N4o5
out: cloned the repository with [REDACTED]
in : retry 3 of 5 for order 100234
out: retry 3 of 5 for order 100234

=== A registered value is hidden in every shape it is printed in
saw [REDACTED] in the request
saw [REDACTED] in the request
saw [REDACTED] in the request
saw [REDACTED] in the request
register('short'): a secret shorter than 8 characters would match ordinary words

=== Over-redaction: ordinary lines that the rules also touch
changed   Invalid password: too short -> Invalid password: [REDACTED] short
changed   Invalid token: expired -> Invalid token: [REDACTED]
unchanged token bucket refilled: 5 tokens  
unchanged password reset email sent to user 42  
unchanged secret santa draw finished  
unchanged api_key missing in request  

The seven samples show the three jobs. The bearer token, the query-string key, the JSON password (with a space inside it) and the connection-string password keep their surroundings and lose only the secret. The AWS and GitHub values are found by shape alone. The last sample holds no secret and is untouched. A registered value is hidden in all four of its shapes, and register('short') is refused.

The over-redaction lines are the price of pattern matching. “Invalid password: too short” becomes Invalid password: [REDACTED] short, because the rule cannot tell that the word after the colon is ordinary English. That is an acceptable failure for a safety net (too much hidden, nothing leaked), but it is worth knowing about so that a log line that has lost a word does not surprise you.

Check what the redactor costs

The redactor runs on the logging path of every request, so measure it. The second part of the output times an ordinary line and then feeds the redactor hostile input:

=== Cost on ordinary lines
6.5 microseconds per line over 100,000 lines

=== Hostile input: can an attacker make the redactor slow?
1,000,000 letters                   0.06 s
1,000,000 spaces                    0.07 s
100,000 x password=                 0.01 s
100,000 x password=quote            0.05 s
100,000 x Authorization: Bearer     0.16 s
250,000 x eyJ-                      0.06 s

On this machine scrubbing an ordinary line takes about 6 microseconds, so 100,000 lines cost well under a second. Your numbers will differ. Every hostile input finished in at most about a sixth of a second, including a million characters in one run.

Make sure the redactor cannot be turned against you

The redactor reads text that attackers can influence, so a slow pattern would be a denial-of-service hole. This is called ReDoS (regular expression denial of service), and our ReDoS tutorial covers it in depth. The short version: a regular expression that can start at many positions and re-scan the same text each time costs time proportional to the square of the input, so doubling the input roughly quadruples the time. The last part of the output times two rules written the obvious way:

=== The same attack on two rules written the obvious way
\beyJ... rule            16,000 characters   0.03 s
\beyJ... rule            32,000 characters   0.12 s
\beyJ... rule            64,000 characters   0.49 s
\beyJ... rule           128,000 characters   2.29 s
[\w.-]*password rule      4,000 characters   0.07 s
[\w.-]*password rule      8,000 characters   0.31 s
[\w.-]*password rule     16,000 characters   1.08 s
[\w.-]*password rule     32,000 characters   3.36 s

Both obvious versions follow the quadratic curve. The key rule, [\w.-]*password..., lets the engine start at every position and re-scan the whole run of word characters each time. The JWT rule written with \b does the same on text such as eyJ-eyJ-eyJ-, where \b finds a word boundary in front of every eyJ and each attempt scans to the end of the run. At 128,000 characters the JWT version already needs more than two seconds. By extrapolating that curve (I did not run it), a million characters would take on the order of minutes, while the final rule handled a million characters in 0.06 seconds. The final rules avoid both problems: the key rule starts at the keyword itself and allows at most 40 more word characters before the separator, and the JWT rule uses a lookbehind, (?<![A-Za-z0-9_-]), so that it can start only at the beginning of a run.

What the rules still miss

A random password in free text with no key name before it, a secret that a library has base64-encoded, a secret split over two lines, and anything shorter than eight characters. That is why this step is the second line of defence, and why register exists: register the secrets your own process holds, and you no longer depend on their shape at all. The shapes themselves are conventions of each provider and can change, so check a provider’s current documentation before you rely on one of these rules.

Step 4: Decide where the redactor lives

Python gives you three places to put code that edits a log line, and two of them look right and fail. To see why, follow one record. You call log.info(...) on a logger. The logger runs its own filters, then hands the record to its handlers, and then to the handlers of its parent loggers. Each handler runs its own filters and, last of all, calls its formatter to produce the text.

log.info(...) -> logger filters -> handler -> handler filters -> formatter -> text
                       |
                       +-> parent loggers' handlers (their logger filters are skipped)

Two statements in the Python documentation matter. On propagation: “Messages are passed directly to the ancestor loggers’ handlers”, and “neither the level nor filters of the ancestor loggers in question are considered”. And in the section on filters: “events which have been generated by descendant loggers will not be filtered by a logger’s filter setting, unless the filter has also been applied to those descendant loggers”. Keep those two in mind while you read the next experiment.

Three places that look right

step04_where.py builds the same logger four ways (no redaction, a filter on the logger, a filter on the handler, and the formatter) and sends seven kinds of leak through each:

# step04_where.py
"""Where should the redactor live? Try a logger filter, a handler filter and the formatter."""
import logging

from labkit import API_KEY, AWS_KEY, PASSWORD, TOKEN, leaked, make_logger, url_error
from safelog import Redactor, SafeFormatter

redactor = Redactor()
FORMAT = "%(levelname)s %(name)s %(message)s"


class LoggerFilter(logging.Filter):
    """Attempt 1: rewrite the message of every record that passes through the logger."""

    def filter(self, record):
        record.msg = redactor.scrub(str(record.msg))
        return True


class HandlerFilter(logging.Filter):
    """Attempt 2: the same job on the handler, after the arguments are merged into the message."""

    def filter(self, record):
        record.msg = redactor.scrub(record.getMessage())
        record.args = None
        return True


def exception_with_token(log):
    try:
        raise ValueError(f"refresh rejected for token {TOKEN}")
    except ValueError:
        log.exception("refresh failed")


def chained_exception(log):
    try:
        try:
            raise ValueError(f"upstream said {AWS_KEY}")
        except ValueError as cause:
            raise RuntimeError("refresh failed") from cause
    except RuntimeError:
        log.exception("refresh failed")


VECTORS = {
    "f-string message": lambda log: log.info(f"login password={PASSWORD}"),
    "positional argument": lambda log: log.info("charging with key %s", API_KEY),
    "dict as argument": lambda log: log.info("payload: %s", {"user": "alice", "password": PASSWORD}),
    "exception message": exception_with_token,
    "chained exception": chained_exception,
    "library exception with URL": lambda log: log.error("report download failed", exc_info=url_error()),
    "message from a child logger": lambda log: logging.getLogger(log.name + ".payments").info(f"login password={PASSWORD}"),
}


def plain():
    return make_logger("shop", logging.Formatter(FORMAT))


def with_logger_filter():
    log, handler, buffer = plain()
    log.addFilter(LoggerFilter())
    return log, handler, buffer


def with_handler_filter():
    log, handler, buffer = plain()
    handler.addFilter(HandlerFilter())
    return log, handler, buffer


def with_formatter():
    return make_logger("shop", SafeFormatter(FORMAT, redactor=redactor))


SETUPS = {"nothing": plain, "logger filter": with_logger_filter, "handler filter": with_handler_filter, "formatter": with_formatter}

print(f"{'where the redactor lives:':30}" + "".join(f"{name:>16}" for name in SETUPS))
totals = dict.fromkeys(SETUPS, 0)
for vector_name, vector in VECTORS.items():
    row = f"{vector_name:30}"
    for setup_name, setup in SETUPS.items():
        log, _, buffer = setup()
        vector(log)
        leaks = leaked(buffer.getvalue())
        totals[setup_name] += leaks > 0
        row += f"{'LEAK' if leaks else 'ok':>16}"
    print(row)
print(f"{'vectors that leak:':30}" + "".join(f"{totals[name]:>16}" for name in SETUPS))

Run python step04_where.py:

where the redactor lives:              nothing   logger filter  handler filter       formatter
f-string message                          LEAK              ok              ok              ok
positional argument                       LEAK            LEAK              ok              ok
dict as argument                          LEAK            LEAK              ok              ok
exception message                         LEAK            LEAK            LEAK              ok
chained exception                         LEAK            LEAK            LEAK              ok
library exception with URL                LEAK            LEAK            LEAK              ok
message from a child logger               LEAK            LEAK              ok              ok
vectors that leak:                           7               6               3               0

Read the columns one at a time.

  • Logger filter. It fixed one of seven. It rewrites record.msg, which is only the message template, so a secret passed as an argument is not in it yet (positional argument, dict as argument). A traceback is not in msg at all. And it never sees the record from the child logger shop.payments, exactly as the documentation said.
  • Handler filter. It fixed four. A handler sees every record that reaches it, children included, and getMessage() merges the arguments, so the argument, dictionary and child-logger leaks are gone. But the three tracebacks still leak. At this point the record holds only exc_info, a reference to the exception; the text of the traceback is produced later, by the formatter.
  • Formatter. It fixed all seven, because the formatter is the last stop. It builds the final text, with the merged message and the traceback, and nothing can be added after it.

The formatter, line by line

Add the rest of the module. neutralize and the indentation helper belong to Step 5, so skim them for now; the redaction work happens in four lines of format and in formatException.

# safelog.py (continued)
def neutralize(text, keep=""):
    """Escape every character that str.isprintable() rejects (newlines, escape codes, line separators)."""
    if text.isprintable():
        return text
    return "".join(
        ch if ch.isprintable() or ch in keep else ch.encode("unicode_escape").decode("ascii")
        for ch in text
    )


class SafeFormatter(logging.Formatter):
    """Text lines that hide secrets and keep each event on one line. Tracebacks are indented under it."""

    def __init__(self, fmt=None, datefmt=None, redactor=None):
        super().__init__(fmt, datefmt)
        self.redactor = DEFAULT if redactor is None else redactor

    def format(self, record):
        record = copy.copy(record)  # the same record object is handed to every handler
        record.msg = neutralize(self.redactor.scrub(record.getMessage()))
        record.args = None
        if record.exc_info:
            record.exc_text = None  # never reuse a traceback another handler cached without redaction
        return super().format(record)

    def formatException(self, exc_info):
        return self._indent(super().formatException(exc_info))

    def formatStack(self, stack_info):
        return self._indent(super().formatStack(stack_info))

    def _indent(self, text):
        text = neutralize(self.redactor.scrub(text), keep="\n")
        return "\n".join("    " + line for line in text.split("\n"))
  • copy.copy(record): one record object is handed to every handler on the logger, so editing it in place would change what the next handler sees. Python 3.12 added a cleaner route for filters: “You can now return a LogRecord instance from filters to replace the log record rather than modifying it in place. This allows filters attached to a Handler to modify the log record before it is emitted, without having side effects on other handlers.” A formatter has no such return value, so it works on a copy.
  • record.getMessage() merges the arguments into the message, which is what makes %s arguments visible to the redactor. Setting record.args = None stops the base class from merging them a second time.
  • record.exc_text = None: logging.Formatter.format caches the traceback text on the record. The CPython source says why: “Cache the traceback text to avoid converting it multiple times (it’s constant anyway)”. If another handler formatted the record first, without redaction, this formatter would reuse the cached text and print it with the secret inside. The copy forgets the cache, so formatException runs again here.
  • formatException and formatStack pass their text through the redactor as well, and then through _indent.

Common mistake: attaching the redactor to a logger because that is where you wrote your log.info calls. A logger filter is the right tool for adding context to your own records. For hiding secrets, the place that every record must pass through is the formatter of every handler.

Step 5: Stop attackers from forging log lines

Redaction protects the secrets in your logs. This step protects their integrity. Anyone who can influence a logged value, such as a username typed into a login form or a path in a URL, can put a newline in it. The logging module writes the newline exactly as given, so the log file now contains a second line that looks like an event your application logged. OWASP describes it this way: “Log Injection occurs when an application includes untrusted data in an application log message”, for example when an attacker can cause “an additional log entry that looks like it came from a completely different user”.

Neutralize what a log reader would misread

The fix is to escape characters that mean something to whoever reads the log. neutralize uses one simple rule: escape every character that Python considers not printable. The documentation defines printable as “the character is suitable for repr() to use in its output”, and says that otherwise “repr() on built-in types will hex-escape the character”. So newline, carriage return, tab, the ESC character that starts terminal escape sequences and the Unicode line and paragraph separators are all escaped, while letters from any language stay as they are. The early return when text.isprintable() keeps the common case fast. Run step05_injection.py; it reads a log the way a simple parser would, starting a new event on every line that begins with a level name.

# step05_injection.py
"""A message that carries a newline can write a log line that nobody logged."""
import json
import logging
import re

from labkit import make_logger
from safelog import SafeFormatter, SafeJsonFormatter

FORMAT = "%(levelname)s %(name)s %(message)s"
EVENT = re.compile(r"^(INFO|WARNING|ERROR) ")
FORGED = "alice\nINFO shop login ok user=admin"
SEPARATOR = chr(0x2028)  # LINE SEPARATOR


def events(text):
    """What a log reader sees: a new event starts on every line that begins with a level name."""
    return [line for line in text.splitlines() if EVENT.match(line)]


def emit(formatter, message, *args):
    log, _, buffer = make_logger("shop", formatter)
    log.warning(message, *args)
    return buffer.getvalue()


plain = logging.Formatter(FORMAT)
safe = SafeFormatter(FORMAT)

print("=== A newline inside a username")
for name, formatter in (("plain Formatter", plain), ("SafeFormatter", safe)):
    out = emit(formatter, "login failed for user %s", FORGED)
    print(f"{name}:\n{out}events a reader finds: {len(events(out))}")

print("\n=== An escape code inside a username (shown with ascii() so your terminal stays calm)")
for name, formatter in (("plain Formatter", plain), ("SafeFormatter", safe)):
    print(f"{name}: {ascii(emit(formatter, 'login failed for user %s', 'bob' + chr(27) + '[2K' + chr(27) + '[1A'))}")

print("\n=== A line separator that is not a newline")
print("split on newline  :", ("a" + SEPARATOR + "b").split("\n"))
print("str.splitlines()  :", ("a" + SEPARATOR + "b").splitlines())
for name, formatter in (("plain Formatter", plain), ("SafeFormatter", safe)):
    out = emit(formatter, "login failed for user %s", "carol" + SEPARATOR + "INFO shop login ok user=admin")
    print(f"{name}: events found by splitlines() = {len(events(out))}; output = {ascii(out)}")

print("\n=== A newline inside an exception message")
for name, formatter in (("plain Formatter", plain), ("SafeFormatter", safe)):
    log, _, buffer = make_logger("shop", formatter)
    try:
        raise ValueError("bad input\nINFO shop login ok user=admin")
    except ValueError:
        log.exception("request failed")
    print(f"{name}: events found = {len(events(buffer.getvalue()))}")
    for line in buffer.getvalue().splitlines():
        if "File " not in line and "Traceback" not in line:
            print("   |", line)

print("\n=== JSON lines")
out = emit(SafeJsonFormatter(), "login failed for user %s", FORGED)
print(out, end="")
print("lines:", len(out.splitlines()), "| message round-trips:", json.loads(out)["message"] == "login failed for user " + FORGED)

print("\n=== Ordinary non-ASCII text is left alone")
print(ascii(emit(safe, "greeting from %s", "Zoë in 東京 ✓")))

Run python step05_injection.py. First, a username that contains a newline and a fake “login ok” line:

=== A newline inside a username
plain Formatter:
WARNING shop login failed for user alice
INFO shop login ok user=admin
events a reader finds: 2
SafeFormatter:
WARNING shop login failed for user alice\nINFO shop login ok user=admin
events a reader finds: 1

With the plain Formatter the reader finds two events, and the second says that admin logged in. With SafeFormatter there is one event, and the newline shows up as the two characters backslash and n.

The next two experiments use less obvious characters:

=== An escape code inside a username (shown with ascii() so your terminal stays calm)
plain Formatter: 'WARNING shop login failed for user bob\x1b[2K\x1b[1A\n'
SafeFormatter: 'WARNING shop login failed for user bob\\x1b[2K\\x1b[1A\n'

=== A line separator that is not a newline
split on newline  : ['a\u2028b']
str.splitlines()  : ['a', 'b']
plain Formatter: events found by splitlines() = 2; output = 'WARNING shop login failed for user carol\u2028INFO shop login ok user=admin\n'
SafeFormatter: events found by splitlines() = 1; output = 'WARNING shop login failed for user carol\\u2028INFO shop login ok user=admin\n'

The first puts a terminal escape sequence in the username. Terminals that understand ANSI escape codes treat ESC [ 2 K as Erase in Line with the whole line cleared and ESC [ 1 A as Cursor Up by one line, so a person who reads the log with cat or tail could see a line overwritten on screen. The script prints with ascii() so that your own terminal stays calm. The plain formatter wrote the raw ESC character (\x1b), and the safe formatter wrote it escaped, which ascii() displays with a doubled backslash.

The second is the sneakiest. str.splitlines() splits on more than newlines. The documentation lists the boundaries, among them \x1c, \x1d, \x1e, \x85 (Next Line), \u2028 (Line Separator) and \u2029 (Paragraph Separator), and calls them “a superset of universal newlines”. A tool that splits a log with splitlines() sees a new line where grep, which splits on newlines only, sees none. In the output, split("\n") keeps the text together while splitlines() cuts it, and the plain formatter lets the separator through, so that reader finds a forged event. SafeFormatter turns the separator into the six characters \u2028. This is the same family of invisible-character tricks that our Unicode tutorial covers for source code and prompts.

The last injection case is an exception message:

=== A newline inside an exception message
plain Formatter: events found = 2
   | ERROR shop request failed
   |     raise ValueError("bad input\nINFO shop login ok user=admin")
   | ValueError: bad input
   | INFO shop login ok user=admin
SafeFormatter: events found = 1
   | ERROR shop request failed
   |         raise ValueError("bad input\nINFO shop login ok user=admin")
   |     ValueError: bad input
   |     INFO shop login ok user=admin

This is the case that text logs cannot fully solve. A traceback is multi-line by nature, so its newlines cannot be escaped. SafeFormatter indents every line of the traceback by four spaces instead. The forged INFO shop login ok user=admin can no longer start at the beginning of a line, and the reader finds one event instead of two. A person, or a plain grep for those words, can still find the forged text inside the indented block. If you need a hard guarantee that one event is one line, use JSON.

JSON lines: one event, one line

OWASP’s advice is blunt: “Do not use unstructured text log formats”, because “unstructured text log formats are susceptible to log entry injection through Carriage Return (CR) or Line Feed (LF) characters”. Its alternative is “use structured log formats, such as JSON”, and it adds “Limit the size of the user input value used to create the log message”. This lab does not cap lengths, so add that when you adopt the module. Here is the JSON formatter:

# safelog.py (continued)
_STANDARD = set(vars(logging.makeLogRecord({}))) | {"message", "asctime"}


class SafeJsonFormatter(logging.Formatter):
    """One JSON object per line. json.dumps escapes newlines, so a message cannot start a second line."""

    def __init__(self, redactor=None):
        super().__init__()
        self.redactor = DEFAULT if redactor is None else redactor

    def format(self, record):
        event = {
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        event.update({key: value for key, value in vars(record).items() if key not in _STANDARD})
        if record.exc_info:
            event["exception"] = self.formatException(record.exc_info)
        return json.dumps(self.redactor.scrub_obj(event), default=str)

_STANDARD is the set of attribute names that every record has. Anything else the caller passes with extra={...} is a custom field and goes into the event. The formatter builds a dictionary and passes the whole thing through scrub_obj, so a field called api_key is hidden by its name and every string is scrubbed by pattern. json.dumps then escapes every control character and, by default, every non-ASCII character, so the output cannot contain a raw newline. default=str stops an extra that is not JSON-serializable from crashing the logging call. The last two experiments in the script exercise it:

=== JSON lines
{"level": "WARNING", "logger": "shop", "message": "login failed for user alice\nINFO shop login ok user=admin"}
lines: 1 | message round-trips: True

=== Ordinary non-ASCII text is left alone
'WARNING shop greeting from Zo\xeb in \u6771\u4eac \u2713\n'

The JSON event is one line, and it round-trips: parsing it gives back the original message, newline included, so nothing is lost. The last experiment confirms that ordinary non-ASCII text (the “ë” in Zoë, Japanese, a check mark) passes through the text formatter untouched. A single backslash in that line comes from ascii(), which shows the characters as escapes for display only; if the formatter itself had escaped them, you would see doubled backslashes, as in the escape-code experiment.

Step 6: Catch what libraries write

Your own log calls are only part of the picture. Libraries write log lines too, and a library that logs a URL logs your query string. The httpx logging guide shows its INFO line, HTTP Request: GET https://www.example.com "HTTP/1.1 200 OK", and the httpx source passes the request URL to that call. urllib3, the engine under requests, logs the request line at DEBUG level with a format string that includes the URL. step06_libraries.py starts a tiny local web server, calls it with requests and httpx, and captures what the root logger receives in four setups:

# step06_libraries.py
"""Libraries write log lines too, and their lines carry URLs."""
import http.server
import io
import logging
import threading

import httpx
import requests

from labkit import API_KEY, leaked
from safelog import SafeFormatter


class Quiet(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Length", "2")
        self.end_headers()
        self.wfile.write(b"ok")

    def log_message(self, *args):
        pass


server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Quiet)
threading.Thread(target=server.serve_forever, daemon=True).start()
PORT = str(server.server_address[1])
FORMAT = "%(name)s %(levelname)s %(message)s"


def call_both(url, headers=None):
    requests.get(url, headers=headers, timeout=5)
    httpx.get(url, headers=headers, timeout=5)


def capture(formatter, setup=None, url=None, headers=None):
    buffer = io.StringIO()
    handler = logging.StreamHandler(buffer)
    handler.setFormatter(formatter)
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(logging.DEBUG)
    for name in ("httpx", "urllib3", "httpcore"):
        logging.getLogger(name).setLevel(logging.NOTSET)
    if setup:
        setup()
    call_both(url or f"http://127.0.0.1:{PORT}/export?api_key={API_KEY}", headers)
    return buffer.getvalue()


def show(title, text):
    print(f"=== {title}")
    for line in text.splitlines():
        if "export" in line:
            print(line.replace(PORT, "PORT"))
    print(f"canary strings in the output: {leaked(text)}\n")


def quiet_libraries():
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)


show("root logger at DEBUG, plain formatter", capture(logging.Formatter(FORMAT)))
show("same, with SafeFormatter on the root handler", capture(SafeFormatter(FORMAT)))
show("same, with the two libraries set to WARNING", capture(logging.Formatter(FORMAT), setup=quiet_libraries))
show(
    "key sent in a header instead of the URL",
    capture(logging.Formatter(FORMAT), url=f"http://127.0.0.1:{PORT}/export", headers={"X-API-Key": API_KEY}),
)

Run python step06_libraries.py:

=== root logger at DEBUG, plain formatter
urllib3.connectionpool DEBUG http://127.0.0.1:PORT "GET /export?api_key=sk_live_CANARY51HqX7aB9cD2eF HTTP/1.1" 200 2
httpx INFO HTTP Request: GET http://127.0.0.1:PORT/export?api_key=sk_live_CANARY51HqX7aB9cD2eF "HTTP/1.1 200 OK"
canary strings in the output: 2

=== same, with SafeFormatter on the root handler
urllib3.connectionpool DEBUG http://127.0.0.1:PORT "GET /export?api_key=[REDACTED] HTTP/1.1" 200 2
httpx INFO HTTP Request: GET http://127.0.0.1:PORT/export?api_key=[REDACTED] "HTTP/1.1 200 OK"
canary strings in the output: 0

=== same, with the two libraries set to WARNING
canary strings in the output: 0

=== key sent in a header instead of the URL
urllib3.connectionpool DEBUG http://127.0.0.1:PORT "GET /export HTTP/1.1" 200 2
httpx INFO HTTP Request: GET http://127.0.0.1:PORT/export "HTTP/1.1 200 OK"
canary strings in the output: 0

Four results, in order:

  • Plain formatter, root at DEBUG. Both libraries print the key. They do it through different loggers (urllib3.connectionpool and httpx), which is why a fix that targets one library misses the other. Two canary strings.
  • SafeFormatter on the root handler. Both lines now say api_key=[REDACTED]. This works because both libraries leave propagate at its default of true and add no handler that prints (urllib3 attaches only a do-nothing NullHandler), so their records reach the root logger’s handler. Step 7 shows a logger that does not behave this way.
  • Both libraries set to WARNING. The lines disappear. That is a good production default, at the cost of losing the request log.
  • Key in a header. The log lines are still there, and they hold no secret: GET /export. This is the real fix. URLs are written down in many places along the way (proxy and web-server access logs, and as you just saw, client libraries), while headers usually are not, so a secret in a header has far fewer chances to land in a log.

Common mistake: turning on logging.basicConfig(level=logging.DEBUG) to chase one bug. It switches on DEBUG output for every library in the process, urllib3 included. Set the level on your own logger instead, for example logging.getLogger("shop").setLevel(logging.DEBUG).

Step 7: Fix the loggers uvicorn creates for itself

uvicorn, the server that runs your FastAPI app, writes logs of its own: an access log with one line per request, which includes the path and the query string, and an error log with a traceback when your code crashes. Look at how it wires them. Its default configuration, the dictionary uvicorn.config.LOGGING_CONFIG, gives the uvicorn and uvicorn.access loggers handlers of their own and sets propagate to False on both. The documentation for that setting says: “If this evaluates to false, logging messages are not passed to the handlers of ancestor loggers.” So a redacting formatter on the root logger never sees uvicorn’s lines.

Here is a tiny shop API with three habits that cause most leaks: /export takes its key in the query string, /login logs the username that the client sent, and /boom crashes with a message that contains a token. At startup it tells the redactor about the two secrets this process was started with.

# app.py
"""A tiny shop API with the three habits that cause most log leaks."""
import logging
import os

from fastapi import FastAPI
from pydantic import BaseModel

import safelog

log = logging.getLogger("shop")
app = FastAPI()

# Teach the redactor the exact secrets this process was started with.
for name in ("SHOP_API_KEY", "SHOP_TOKEN"):
    if os.environ.get(name):
        safelog.DEFAULT.register(os.environ[name])


class Login(BaseModel):
    username: str
    password: str


@app.post("/login")
def login(body: Login):
    log.warning("login failed for user %s", body.username)
    return {"ok": False}


@app.get("/export")
def export(api_key: str = ""):
    log.info("export requested")
    return {"rows": 3}


@app.get("/boom")
def boom():
    raise RuntimeError(f"upstream rejected token {os.environ['SHOP_TOKEN']}")

Next, logconfig.py sets logging up in one of three modes, and serve.py starts uvicorn with the result. default is what many projects have: a plain root handler plus uvicorn’s own configuration. root_only puts SafeFormatter on the root handler and leaves uvicorn alone. safe hands uvicorn a complete dictConfig in which every logger writes through SafeFormatter. In a dictConfig formatter entry, the "()" key names a factory to call and the other keys become its keyword arguments, so "fmt" reaches SafeFormatter.__init__.

# logconfig.py
"""Three ways to set up logging for the shop, so we can compare what uvicorn writes."""
import logging
import sys

from uvicorn.config import LOGGING_CONFIG

import safelog

FORMAT = "%(levelname)s %(name)s %(message)s"


def for_uvicorn(mode):
    """Set logging up in one of three ways and return the log_config to hand to uvicorn."""
    if mode == "safe":
        return {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {"safe": {"()": "safelog.SafeFormatter", "fmt": FORMAT}},
            "handlers": {
                "console": {"class": "logging.StreamHandler", "formatter": "safe", "stream": "ext://sys.stdout"},
            },
            "root": {"level": "INFO", "handlers": ["console"]},
            "loggers": {
                "uvicorn": {"handlers": [], "propagate": True},
                "uvicorn.access": {"handlers": [], "propagate": True, "level": "INFO"},
            },
        }
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(safelog.SafeFormatter(FORMAT) if mode == "root_only" else logging.Formatter(FORMAT))
    logging.basicConfig(level=logging.INFO, handlers=[handler])
    return LOGGING_CONFIG  # uvicorn's own handlers, which do not pass records up to the root logger
# serve.py
"""Start the shop under uvicorn: python serve.py MODE PORT, where MODE is default, root_only or safe."""
import sys

import uvicorn

import logconfig

mode, port = sys.argv[1], int(sys.argv[2])
uvicorn.run("app:app", host="127.0.0.1", port=port, log_config=logconfig.for_uvicorn(mode))

The driver script starts the server three times, sends the same three requests each time (an export with the key in the query string, a login whose username contains a forged line, and a crash), stops the server and prints the lines of interest. It writes the server’s output to a file instead of a pipe. My first version used a pipe that nothing was reading, and the third request hung until it timed out, most likely because the pipe filled up and blocked the server.

# step07_live_server.py
"""Run the shop under uvicorn three ways, send three requests, and read what the server wrote."""
import os
import re
import socket
import subprocess
import sys
import time

import requests

from labkit import API_KEY, PASSWORD, TOKEN, leaked

FORGED = "alice\nINFO shop login ok user=admin"
MODES = {
    "default": "uvicorn's default log config, plain root handler",
    "root_only": "SafeFormatter on the root handler only",
    "safe": "one dictConfig that gives every logger the SafeFormatter",
}


def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def wait_for(port):
    for _ in range(100):
        with socket.socket() as sock:
            sock.settimeout(0.2)
            if sock.connect_ex(("127.0.0.1", port)) == 0:
                return
        time.sleep(0.1)
    raise RuntimeError("server did not start")


def run_mode(mode):
    port = free_port()
    env = dict(os.environ, SHOP_API_KEY=API_KEY, SHOP_TOKEN=TOKEN, PYTHONUTF8="1")
    log_path = f"server_{mode}.log"
    # Write to a file, not a pipe: nobody reads a pipe while we send requests, and a full pipe blocks the server.
    with open(log_path, "w", encoding="utf-8") as sink:
        proc = subprocess.Popen([sys.executable, "serve.py", mode, str(port)], stdout=sink, stderr=subprocess.STDOUT, env=env)
        try:
            wait_for(port)
            base = f"http://127.0.0.1:{port}"
            requests.get(f"{base}/export", params={"api_key": API_KEY}, timeout=5)
            requests.post(f"{base}/login", json={"username": FORGED, "password": PASSWORD}, timeout=5)
            requests.get(f"{base}/boom", timeout=5)
            time.sleep(0.3)
        finally:
            if os.name == "nt":
                subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True)
            else:
                proc.terminate()
            proc.wait(timeout=15)
    with open(log_path, encoding="utf-8") as source:
        return source.read()


for mode, description in MODES.items():
    output = run_mode(mode)
    print(f"=== {mode}: {description}")
    for line in output.splitlines():
        if "CANARY" in line or "login" in line or "REDACTED" in line:
            print("  ", re.sub(r"127\.0\.0\.1:\d+", "127.0.0.1:PORT", line)[:150])
    forged = any(line.startswith("INFO shop login ok") for line in output.splitlines())
    print(f"   canary strings: {leaked(output)} | forged log line present: {forged}\n")

Run python step07_live_server.py. It prints only the lines that matter and counts canaries in the whole output:

=== default: uvicorn's default log config, plain root handler
   INFO:     127.0.0.1:PORT - "GET /export?api_key=sk_live_CANARY51HqX7aB9cD2eF HTTP/1.1" 200 OK
   WARNING shop login failed for user alice
   INFO shop login ok user=admin
   INFO:     127.0.0.1:PORT - "POST /login HTTP/1.1" 200 OK
   RuntimeError: upstream rejected token ghp_CANARYa1B2c3D4e5F6g7H8i9J0k1L2m3N4o5
   canary strings: 2 | forged log line present: True

=== root_only: SafeFormatter on the root handler only
   INFO:     127.0.0.1:PORT - "GET /export?api_key=sk_live_CANARY51HqX7aB9cD2eF HTTP/1.1" 200 OK
   WARNING shop login failed for user alice\nINFO shop login ok user=admin
   INFO:     127.0.0.1:PORT - "POST /login HTTP/1.1" 200 OK
   RuntimeError: upstream rejected token ghp_CANARYa1B2c3D4e5F6g7H8i9J0k1L2m3N4o5
   canary strings: 2 | forged log line present: False

=== safe: one dictConfig that gives every logger the SafeFormatter
   INFO uvicorn.access 127.0.0.1:PORT - "GET /export?api_key=[REDACTED] HTTP/1.1" 200
   WARNING shop login failed for user alice\nINFO shop login ok user=admin
   INFO uvicorn.access 127.0.0.1:PORT - "POST /login HTTP/1.1" 200
       RuntimeError: upstream rejected token [REDACTED]
   canary strings: 0 | forged log line present: False

Three results:

  • default: the access log prints the key from the query string, the traceback from /boom prints the token, and the forged INFO shop login ok user=admin appears as a line of its own. Two canaries and one forged line.
  • root_only: the application’s own logger is fixed (the forged line has become the characters \n), but uvicorn’s loggers still print both canaries. The traceback goes through uvicorn.error, which passes it up to the uvicorn logger and its own handler. This is the trap: the fix looks complete if you only read your own logger’s lines.
  • safe: the access log shows api_key=[REDACTED], the traceback shows [REDACTED] indented under its event, and the forged line is neutralized. Zero canaries.

If you do not need the access log, uvicorn can drop it: the settings page says --no-access-log will “Disable access log only, without changing log level.” For a configuration file, the same page lists --log-config as accepting dictConfig() formats in .json and .yaml. I ran the Python form here, a dictionary passed as log_config to uvicorn.run. Moving the key out of the URL, as in Step 6, beats both options.

Step 8: Lock it in with canary tests

A fix that nobody tests decays: someone adds a logger, a library or a log call, and a secret slips back in. Canary tests turn the promise that no secret reaches the logs into something a build can check. The idea is the one from Step 1: log something that contains a canary, capture what the handler wrote, and assert that the word CANARY is not in it. The tests below send the seven vectors from Step 4, plus an authorization header, through both formatters.

# test_safelog.py
import io
import json
import logging
import time
from urllib.parse import quote

import pytest

from labkit import API_KEY, AWS_KEY, JWT, PASSWORD, TOKEN, leaked, make_logger, url_error
from safelog import REDACTED, Redactor, SafeFormatter, SafeJsonFormatter, neutralize
from secret import Secret

FORMAT = "%(levelname)s %(name)s %(message)s"
SEPARATOR, NEXT_LINE = chr(0x2028), chr(0x85)


def exception_with_token(log):
    try:
        raise ValueError(f"refresh rejected for token {TOKEN}")
    except ValueError:
        log.exception("refresh failed")


def chained_exception(log):
    try:
        try:
            raise ValueError(f"upstream said {AWS_KEY}")
        except ValueError as cause:
            raise RuntimeError("refresh failed") from cause
    except RuntimeError:
        log.exception("refresh failed")


VECTORS = {
    "f-string": lambda log: log.info(f"login password={PASSWORD}"),
    "argument": lambda log: log.info("charging with key %s", API_KEY),
    "dict argument": lambda log: log.info("payload: %s", {"user": "alice", "password": PASSWORD}),
    "exception": exception_with_token,
    "chained exception": chained_exception,
    "library exception": lambda log: log.error("report download failed", exc_info=url_error()),
    "child logger": lambda log: logging.getLogger(log.name + ".payments").info(f"login password={PASSWORD}"),
    "bearer header": lambda log: log.info("sent Authorization: Bearer %s", JWT),
}


@pytest.fixture(params=["text", "json"])
def formatter(request):
    return SafeFormatter(FORMAT, redactor=Redactor()) if request.param == "text" else SafeJsonFormatter(redactor=Redactor())


@pytest.mark.parametrize("vector", list(VECTORS.values()), ids=list(VECTORS))
def test_no_canary_reaches_the_output(vector, formatter):
    log, _, buffer = make_logger("shop", formatter)
    vector(log)
    assert leaked(buffer.getvalue()) == 0
    assert REDACTED in buffer.getvalue()  # the secret was hidden, not lost along with the rest of the message


def test_json_extra_fields_are_scrubbed():
    log, _, buffer = make_logger("shop", SafeJsonFormatter(redactor=Redactor()))
    log.info("signed in", extra={"api_key": API_KEY, "session": JWT, "profile": {"password": PASSWORD, "name": "alice"}})
    event = json.loads(buffer.getvalue())
    assert leaked(buffer.getvalue()) == 0
    assert event["api_key"] == REDACTED and event["profile"]["name"] == "alice"

Two details. The fixture runs every test once per formatter (params=["text", "json"]). And the vector test asserts two things: that no canary appears, and that the marker [REDACTED] does. The second assertion looks redundant, but it exists because of a bug you will meet in the mutation check below: a broken formatter that throws the secret away together with the rest of the message passes a test that only looks for the canary.

# test_safelog.py (continued)
@pytest.mark.parametrize(
    "text, expected",
    [
        ("Authorization: Bearer abc.def.ghi", "Authorization: Bearer [REDACTED]"),
        ('{"password": "two words"}', '{"password": "[REDACTED]"}'),
        ("GET /x?api_key=abc123&page=2", "GET /x?api_key=[REDACTED]&page=2"),
        ("postgresql://shop:s3cret-pw@db/shop", "postgresql://shop:[REDACTED]@db/shop"),
        ("retry 3 of 5 for order 100234", "retry 3 of 5 for order 100234"),
        ("token bucket refilled: 5 tokens", "token bucket refilled: 5 tokens"),
    ],
)
def test_patterns(text, expected):
    assert Redactor().scrub(text) == expected


def test_a_registered_secret_is_hidden_in_every_shape():
    secret = 'pa"ss w0rd&CANARY=1'
    redactor = Redactor()
    redactor.register(secret)
    for shape in (secret, quote(secret, safe=""), json.dumps(secret)[1:-1], repr(secret)[1:-1]):
        assert "CANARY" not in redactor.scrub(f"saw {shape} here")


def test_a_short_secret_is_refused():
    with pytest.raises(ValueError):
        Redactor().register("short")


def test_secret_never_prints():
    secret = Secret(API_KEY)
    for shown in (str(secret), repr(secret), f"{secret}", "%s" % secret, "%r" % secret, repr({"k": secret}), repr([secret])):
        assert leaked(shown) == 0
    assert secret.reveal() == API_KEY

The pattern test is table-driven, so adding a new secret shape is one new row. The registered-secret test uses a value with a double quote and a space in it and checks all four shapes. test_secret_never_prints covers the wrapper from Step 2, including inside a list and a dictionary.

# test_safelog.py (continued)
@pytest.mark.parametrize(
    "hostile",
    ["a\nINFO shop login ok", "a\r\nINFO shop login ok", "a" + SEPARATOR + "INFO shop login ok",
     "a" + NEXT_LINE + "INFO shop login ok", "a\x1b[2K\x1b[1Ab"],
    ids=["newline", "crlf", "line-separator", "next-line", "escape-code"],
)
def test_a_text_event_stays_on_one_line(hostile):
    log, _, buffer = make_logger("shop", SafeFormatter(FORMAT, redactor=Redactor()))
    log.warning("login failed for user %s", hostile)
    out = buffer.getvalue()
    assert len(out.splitlines()) == 1  # splitlines() also splits on U+2028 and NEL
    assert out.rstrip("\n").isprintable()


@pytest.mark.parametrize("hostile", ["a\nb", "a" + SEPARATOR + "b", "a\x1b[2Kb"], ids=["newline", "line-separator", "escape-code"])
def test_a_json_event_is_one_line_and_round_trips(hostile):
    log, _, buffer = make_logger("shop", SafeJsonFormatter(redactor=Redactor()))
    log.warning("login failed for user %s", hostile)
    assert len(buffer.getvalue().splitlines()) == 1
    assert json.loads(buffer.getvalue())["message"] == "login failed for user " + hostile


def test_a_newline_in_an_exception_message_cannot_start_an_event():
    log, _, buffer = make_logger("shop", SafeFormatter(FORMAT, redactor=Redactor()))
    try:
        raise ValueError("bad\nINFO shop login ok user=admin")
    except ValueError:
        log.exception("request failed")
    lines = buffer.getvalue().splitlines()
    assert lines[0] == "ERROR shop request failed"
    assert all(line.startswith("    ") for line in lines[1:])


def test_ordinary_unicode_is_untouched():
    assert neutralize("Zoë in 東京 ✓") == "Zoë in 東京 ✓"


def test_a_traceback_cached_by_another_handler_is_not_reused():
    log, _, plain_buffer = make_logger("shop", logging.Formatter(FORMAT))
    safe_buffer = io.StringIO()
    safe_handler = logging.StreamHandler(safe_buffer)
    safe_handler.setFormatter(SafeFormatter(FORMAT, redactor=Redactor()))
    log.addHandler(safe_handler)  # runs second, after the plain handler cached the raw traceback
    exception_with_token(log)
    assert leaked(plain_buffer.getvalue()) == 1
    assert leaked(safe_buffer.getvalue()) == 0


def test_hostile_input_does_not_make_the_redactor_slow():
    redactor = Redactor()
    for text in ("a" * 1_000_000, "eyJ-" * 250_000, "password=" * 100_000, "Authorization: Bearer " * 50_000):
        start = time.perf_counter()
        redactor.scrub(text)
        assert time.perf_counter() - start < 2.0

The injection tests use splitlines() as their measure because it is the strictest reader, and they run the same hostile values through the JSON formatter, where the round trip must also hold. The cached-traceback test attaches a plain handler first and a safe handler second on one logger, and checks that the second does not reuse the traceback the first one cached. The timing test puts a two-second ceiling on each hostile input; that is deliberately generous, since the real figures are far lower, and the point is to fail loudly if a future rule turns quadratic.

Two traps in pytest’s caplog

The caplog fixture is the usual way to look at log output in pytest, and for this job it is the wrong tool. These last two tests show why:

# test_safelog.py (continued)
def test_caplog_shows_the_raw_record_not_your_formatted_output(caplog):
    log, _, buffer = make_logger("shop", SafeFormatter(FORMAT, redactor=Redactor()))
    log.propagate = True  # caplog listens on the root logger
    with caplog.at_level(logging.INFO):
        log.info(f"login password={PASSWORD}")
    assert leaked(buffer.getvalue()) == 0  # what the handler wrote is clean
    assert leaked(caplog.text) == 1  # caplog formatted the raw record with its own formatter


def test_caplog_sees_nothing_when_propagate_is_false(caplog):
    log, _, _ = make_logger("shop", SafeFormatter(FORMAT, redactor=Redactor()))
    with caplog.at_level(logging.INFO):
        log.info("hello")
    assert caplog.text == ""

caplog attaches its own handler to the root logger and formats the raw records with its own formatter, so your formatter never runs: the first test shows caplog.text holding the password while the handler’s own output is clean. And when a logger has propagate set to false, as the loggers in these tests do, caplog sees nothing at all (second test), so an assertion such as "secret" not in caplog.text would pass for the wrong reason. Capture with a handler of your own, as make_logger does, and keep these two tests as a reminder.

Run python -m pytest -q:

........................................                                                     [100%]
40 passed in 0.80s

Break it on purpose: a mutation check

Passing tests prove little if they would also pass on broken code. mutations.py copies the module, breaks it in eight specific ways, one at a time, and runs the test suite against each copy. Every broken copy should fail at least one test.

# mutations.py
"""Break safelog.py on purpose, one way at a time, and check that the tests notice."""
import pathlib
import shutil
import subprocess
import sys
import tempfile

LAB = pathlib.Path(__file__).parent
FILES = ["labkit.py", "secret.py", "safelog.py", "test_safelog.py"]

MUTANTS = {
    "tracebacks are not scrubbed": (
        r'text = neutralize(self.redactor.scrub(text), keep="\n")',
        r'text = neutralize(text, keep="\n")',
    ),
    "arguments are not merged before scrubbing": (
        "record.msg = neutralize(self.redactor.scrub(record.getMessage()))",
        "record.msg = neutralize(self.redactor.scrub(str(record.msg)))",
    ),
    "a cached traceback is reused": (
        "record.exc_text = None  # never reuse a traceback another handler cached without redaction",
        "pass",
    ),
    "registered secrets match only their raw form": (
        'shapes = {value, quote(value, safe=""), json.dumps(value)[1:-1], repr(value)[1:-1]}',
        "shapes = {value}",
    ),
    "only newline, return and escape are neutralized": (
        "ch if ch.isprintable() or ch in keep else",
        r'ch if ch not in "\n\r\x1b" or ch in keep else',
    ),
    "JSON extras are not scrubbed": (
        "return json.dumps(self.redactor.scrub_obj(event), default=str)",
        "return json.dumps(event, default=str)",
    ),
    "quoted values stop at the first space": (
        r'''\"[^\"]*\"?|'[^']*'?|''',
        "",
    ),
    "traceback lines are not indented": (
        r'return "\n".join("    " + line for line in text.split("\n"))',
        "return text",
    ),
}


def run_mutant(name, old, new):
    with tempfile.TemporaryDirectory() as folder:
        for file in FILES:
            shutil.copy(LAB / file, folder)
        target = pathlib.Path(folder) / "safelog.py"
        source = target.read_text(encoding="utf-8")
        assert source.count(old) == 1, f"{name}: the text to replace occurs {source.count(old)} times"
        target.write_text(source.replace(old, new), encoding="utf-8", newline="\n")
        result = subprocess.run(
            [sys.executable, "-m", "pytest", "-q", "-x", "-p", "no:cacheprovider", "--no-header"],
            cwd=folder, capture_output=True, text=True, encoding="utf-8", errors="replace",
        )
    failed = [line for line in result.stdout.splitlines() if line.startswith("FAILED")]
    return result.returncode != 0, (failed[0][:110] if failed else result.stdout.strip().splitlines()[-1])


caught = 0
for name, (old, new) in MUTANTS.items():
    detected, evidence = run_mutant(name, old, new)
    caught += detected
    print(f"{'caught ' if detected else 'MISSED '} {name}\n         {evidence}")
print(f"{caught} of {len(MUTANTS)} planted bugs were caught")

Run python mutations.py:

caught  tracebacks are not scrubbed
         FAILED test_safelog.py::test_no_canary_reaches_the_output[text-exception] - assert 1 == 0
caught  arguments are not merged before scrubbing
         FAILED test_safelog.py::test_no_canary_reaches_the_output[text-argument] - AssertionError: assert...
caught  a cached traceback is reused
         FAILED test_safelog.py::test_a_traceback_cached_by_another_handler_is_not_reused - assert 1 == 0
caught  registered secrets match only their raw form
         FAILED test_safelog.py::test_a_registered_secret_is_hidden_in_every_shape - AssertionError: asser...
caught  only newline, return and escape are neutralized
         FAILED test_safelog.py::test_a_text_event_stays_on_one_line[line-separator] - AssertionError: ass...
caught  JSON extras are not scrubbed
         FAILED test_safelog.py::test_no_canary_reaches_the_output[json-f-string] - assert 1 == 0
caught  quoted values stop at the first space
         FAILED test_safelog.py::test_no_canary_reaches_the_output[text-dict argument] - assert 1 == 0
caught  traceback lines are not indented
         FAILED test_safelog.py::test_a_newline_in_an_exception_message_cannot_start_an_event - assert False
8 of 8 planted bugs were caught

All eight were caught. The second mutant slipped through the first time I ran this: it dropped the arguments from the message, so the secret was lost instead of leaked, and a test that only looks for the canary accepts that. Adding the REDACTED in buffer.getvalue() assertion fixed it. The lesson generalises: test that the secret is hidden, not merely absent.

Step 9: Confirm everything end to end

Run three checks in order. First, the tests and the mutation check above. Second, the live-server comparison from Step 7. Third, a plain search of the files the servers wrote, which does not depend on any of the code you wrote. Step 7 saved each server’s full output, so count the canaries in each file:

$ grep -c CANARY server_default.log server_root_only.log server_safe.log
server_default.log:2
server_root_only.log:2
server_safe.log:0

The two setups with gaps leave two canaries each (the key in the access log and the token in the traceback), and the final setup leaves none. That is the result you want from your own project as well: run it with a canary secret in its environment, send a few requests that touch the paths you worry about, and search the output files for the canary.

To adopt this in a real project:

  1. Call DEFAULT.register(...) at startup for every secret your process loads, and again after a rotation.
  2. Configure logging once, with SafeFormatter or SafeJsonFormatter on every handler, including the handlers that libraries and servers create for themselves.
  3. Wrap secrets in Secret (or SecretStr) when you load them, and call reveal() only where they are used.
  4. Move secrets out of URLs, and set third-party loggers to WARNING in production.
  5. Keep the canary tests in your continuous integration, and add a new vector whenever a leak teaches you one.

Common mistakes, in one place

  • A filter on a logger instead of a formatter on every handler. It misses child loggers, arguments and tracebacks (Step 4).
  • Checking only your own logger. uvicorn’s access and error loggers kept leaking until they were configured too (Step 7).
  • Testing with caplog. It bypasses your formatter (Step 8).
  • Asserting only that the secret is absent. Also assert that the marker is present (Step 8).
  • Registering a short or empty secret. It would match ordinary words, so register refuses anything under eight characters (Step 3).
  • Debug level for the whole process. It turns on request logging in libraries (Step 6).
  • Trusting the regular expressions of a redactor without timing them on hostile input (Step 3).

Where to go next

  • Cap the length of logged user input, as OWASP advises. The formatter is the natural place for it.
  • Extend the rules to personal data such as email addresses and phone numbers, and decide with your privacy team what must never be stored.
  • Check where the formatter sits if you ship logs through a queue or an agent, and repeat the canary test at the far end of the pipeline.
  • For secrets that reach a repository instead of a log, see our Gitleaks pre-commit tutorial. For secrets in HTTP error bodies, see RFC 9457 problem details in FastAPI. For logs that must prove they were not edited, see our tamper-evident audit log tutorial.

Sources

  • Python documentation: logging (filters, propagate, the 3.12 change to filters) and logging.config.
  • Python documentation: str.isprintable and str.splitlines.
  • OWASP Logging Cheat Sheet, CWE-532 and CWE-117.
  • uvicorn settings, the httpx logging guide and the pytest logging guide.

Tags:

Application SecurityFastAPILoggingpytestPythonSecrets Management

Share

Stone stepping blocks in a river in Nagasaki, with the arched Megane Bridge and its mirror reflection behind them
Previous Post

MIT Technology Review’s “The Agentic Web Is Here” Turns a Partial-Credit Benchmark Score Into a 73% Success Rate, and DoorDash’s CEO Counts Outside AI Agent Orders in the Hundreds

A hand pressing a chrome doorbell button on a stucco wall
Next Post

Wordfence Reproduces the Brotli Flaw as Remote Code Execution Through WordPress, but the Public Exploit Was Written for One Docker Image and Posted a Day After CISA Called the Bug Exploited

No Comment! Be the first one.

Leave a Reply Cancel reply

Your email address will not be published. Required fields are marked *

Latest
06 Nov
How to Secure a WebSocket Endpoint in FastAPI and Stop Cross-Site Hijacking, Leaked Tokens, Floods, and Slow Readers
06 Nov
OpenAI’s Astra Report Turns Silent Safety-Task Underperformance Into a Matched-Pair Gap, but the Ethical-Doubt Result’s Bar Runs From a 4-Point Gap to a 3.6-Point Reversal, and the Launch Card Quoted AISI as Seeing Very Few Refusals
Trending
November 6, 2026
How to Secure a WebSocket Endpoint in FastAPI and Stop Cross-Site Hijacking, Leaked Tokens, Floods, and Slow Readers
November 6, 2026
OpenAI’s Astra Report Turns Silent Safety-Task Underperformance Into a Matched-Pair Gap, but the Ethical-Doubt Result’s Bar Runs From a 4-Point Gap to a 3.6-Point Reversal, and the Launch Card Quoted AISI as Seeing Very Few Refusals
November 6, 2026
OpenAI Releases GPT-6 Cyber, Rated Critical, to Approved Customers: the Post’s 99.93% Scope Score Is Measured Against the Older Cyber Model, the System Card Puts Its Base Model at 0% Honeypot Exploitation, and One Outside Benchmark Has It Solving Fewer Challenges
November 5, 2026
How to Spread Requests Across Servers in Python With the Power of Two Choices and Avoid the Herd Effect of Least-Loaded Routing
November 5, 2026
Microsoft’s PDD Kit Turns Code Review Into Intent Review, but Verus Accepts a Login That Answers Every Attempt With a 500 Under the Contract the Post Prints
November 5, 2026
OpenSSH Patches a Password Bypass on musl Systems Such as Alpine, Where a 257 to 1,024 Byte Password Logs In to Password-Disabled Accounts, and musl’s Maintainer Says the Bug Is Not in musl

Related Posts

A laptop wrapped in a chain and padlock, illustrating least-privilege controls for AI agents.
Learning Hub

How to Secure Tool-Using AI Agents Before They Touch Production

June 8, 2026
Colorful sticky notes arranged on an office wall, symbolizing governance checklists and planning.
Learning Hub

AI Governance for Agentic Apps: A Practical Checklist for Builders

June 8, 2026
A technician connects green fiber optic cables at a data center, representing a private production inference endpoint.
Learning Hub

How to Deploy a Fine-Tuned LLM Behind a Private Production Inference Endpoint

June 8, 2026
Narrow aisle behind black supercomputer racks in a data center
Learning Hub

Kubernetes SELinux Volume Labeling: What Cluster Operators Should Audit Before v1.37

June 8, 2026
SXZ.io SXZ.io
  • [email protected]

Categories

Articles
Learning Hub
News

All Rights Reserved by SXZ.io ©2026