DEV Community

Ayi NEDJIMI
Ayi NEDJIMI

Posted on

How to Build a Canary Token System for Breach Detection

#security #python #devops #tutorial

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

What Is a Canary Token?

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

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

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

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

Setting Up the Token Server

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

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

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

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

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

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

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

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

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

Generating Tokens for Different Trap Types

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

# canary_gen.py
import uuid, json

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

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

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

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

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

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

Each trap targets a different threat:

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

Where to Plant Tokens

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

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

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

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

Handling Alert Noise Without Missing Real Hits

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

import time
from collections import defaultdict

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

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

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

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

The Takeaway

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

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

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

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

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

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

Top comments (0)