Advanced Examples

# learning/code/ex-50-directory-listing-and-default-creds/app.py
"""Example 50: a live Flask app -- directory listing + a hardcoded default admin cred, then both are closed (co-24)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the misconfiguration itself
 
import os  # => co-24: builds a real on-disk static folder this example actually serves from
 
from flask import Flask, abort, jsonify, request, send_from_directory  # => co-24: send_from_directory serves real files
 
app = Flask(__name__)  # => co-24: one Flask app, hosting both the vulnerable and fixed halves
STATIC_DIR = os.path.join(os.path.dirname(__file__), "public_files")  # => co-24: a real folder on disk, seeded below
PUBLIC_ALLOW_LIST = {"readme.txt", "logo.txt"}  # => co-24: the ONLY filenames the fixed route will ever serve
 
 
def seed_static_dir() -> None:  # => co-24: runs once at import time -- creates real files this example actually reads
    os.makedirs(STATIC_DIR, exist_ok=True)  # => co-24: idempotent -- safe to call on every server start
    with open(os.path.join(STATIC_DIR, "readme.txt"), "w") as f:  # => co-24: a genuinely public, intended-to-be-served file
        f.write("Welcome -- this file is meant to be public.\n")  # => co-24: real bytes on disk
    with open(os.path.join(STATIC_DIR, "logo.txt"), "w") as f:  # => co-24: another genuinely public file
        f.write("ASCII logo placeholder.\n")  # => co-24: real bytes on disk
    with open(os.path.join(STATIC_DIR, "secret-notes.txt"), "w") as f:  # => co-24: NEVER meant to be web-reachable
        f.write("db_password=hunter2-internal-only\n")  # => co-24: the real sensitive content directory listing exposes
 
 
@app.route("/legacy/files/")  # => co-24: VULNERABLE -- a real directory listing of the whole static folder
def legacy_list_files() -> object:  # => co-24: returns a Flask Response object -- the vulnerable listing route
    # => seeded bug: os.listdir() enumerates EVERY file in STATIC_DIR, intended or not,
    # => and hands the full real filename list straight back to any caller
    names = sorted(os.listdir(STATIC_DIR))  # => co-24: the REAL directory contents, unfiltered
    return jsonify({"files": names})  # => co-24: leaks "secret-notes.txt" exists, before anyone even fetches it
 
 
@app.route("/legacy/files/<path:filename>")  # => co-24: VULNERABLE -- serves ANY file in the folder, no allow-list
def legacy_get_file(filename: str) -> object:  # => co-24: returns a Flask Response object -- the vulnerable fetch route
    return send_from_directory(STATIC_DIR, filename)  # => co-24: no filtering at all -- secret-notes.txt is fetchable
 
 
@app.route("/legacy/login", methods=["POST"])  # => co-24: VULNERABLE -- a real, hardcoded default credential pair
def legacy_login() -> object:  # => co-24: returns a Flask Response object -- the vulnerable login route
    body = request.get_json(force=True)  # => co-01: attacker-controlled -- the real submitted credentials
    # => seeded bug: "admin"/"admin" is the SHIPPED default, never rotated, checked in plain code
    if body.get("username") == "admin" and body.get("password") == "admin":  # => co-24: the real default-cred check
        return jsonify({"status": "logged in as admin"}), 200  # => co-24: a real 200 -- the default cred still works
    return jsonify({"error": "invalid credentials"}), 401  # => co-24: a real 401 for anything else
 
 
@app.route("/secure/files/<path:filename>")  # => co-24: FIXED -- an explicit allow-list, no listing route at all
def secure_get_file(filename: str) -> object:  # => co-24: returns a Flask Response object -- the fixed fetch route
    if filename not in PUBLIC_ALLOW_LIST:  # => co-24: the fix -- membership check BEFORE touching the filesystem
        abort(404)  # => co-24: a real 404 -- indistinguishable from "no such file" for anything not allow-listed
    return send_from_directory(STATIC_DIR, filename)  # => co-24: only reached for a genuinely allow-listed name
 
 
ROTATED_ADMIN_PASSWORD = os.environ.get("EX50_ADMIN_PASSWORD", "kJ8-x2Qz-9vM1-rotated")  # => co-24: NOT "admin"
 
 
@app.route("/secure/login", methods=["POST"])  # => co-24: FIXED -- the default credential is rotated, not hardcoded-weak
def secure_login() -> object:  # => co-24: returns a Flask Response object -- the fixed login route
    body = request.get_json(force=True)  # => co-01: the SAME shape of attacker-controlled input as the vulnerable route
    if body.get("username") == "admin" and body.get("password") == ROTATED_ADMIN_PASSWORD:  # => co-24: a real, rotated check
        return jsonify({"status": "logged in as admin"}), 200  # => co-24: only the ROTATED password ever succeeds
    return jsonify({"error": "invalid credentials"}), 401  # => co-24: the SAME 401 the vulnerable route uses for failure
 
 
if __name__ == "__main__":  # => co-24: only runs when launched directly, e.g. `python3 app.py &`
    seed_static_dir()  # => co-24: create the real on-disk files before the server starts accepting requests
    app.run(host="127.0.0.1", port=5050)  # => co-24: localhost-only, fixed port -- exploit_and_fix.py targets this

# learning/code/ex-50-directory-listing-and-default-creds/exploit_and_fix.py
"""Example 50: drives the live app.py -- lists+reads a secret file and logs in with admin/admin, then both close (co-24)."""  # => co-24: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the exploit itself
 
import requests  # => co-24: real HTTP client -- every request below hits the live app.py process
 
BASE_URL = "http://127.0.0.1:5050"  # => co-24: matches app.py's app.run(port=5050) exactly
 
 
def main() -> None:  # => co-24: runs the listing+default-cred exploit, then the SAME two requests against the fix
    print("=== VULNERABLE: directory listing reveals secret-notes.txt exists ===")  # => labels section
    listing = requests.get(f"{BASE_URL}/legacy/files/", timeout=5)  # => co-24: a real HTTP GET against the listing route
    print(f"status={listing.status_code} files={listing.json()['files']}")  # => co-24: real filenames, straight off the wire
    assert "secret-notes.txt" in listing.json()["files"]  # => co-24: proves the listing really leaked the sensitive name
 
    print("\n=== VULNERABLE: fetching secret-notes.txt directly succeeds ===")  # => labels section
    secret = requests.get(f"{BASE_URL}/legacy/files/secret-notes.txt", timeout=5)  # => co-24: a real HTTP GET, real fetch
    print(f"status={secret.status_code} body={secret.text.strip()!r}")  # => co-24: real file content, straight off disk
    assert secret.status_code == 200 and "db_password" in secret.text  # => co-24: proves the real secret content leaked
 
    print("\n=== VULNERABLE: default admin/admin credential still logs in ===")  # => labels section
    login = requests.post(f"{BASE_URL}/legacy/login", json={"username": "admin", "password": "admin"}, timeout=5)  # => co-24: real POST, shipped default cred
    print(f"status={login.status_code} body={login.json()}")  # => co-24: real response, straight off the wire
    assert login.status_code == 200  # => co-24: proves the shipped default credential really still works
 
    print("\n=== FIXED: secret-notes.txt is now unreachable via the allow-listed route ===")  # => labels section
    fixed_secret = requests.get(f"{BASE_URL}/secure/files/secret-notes.txt", timeout=5)  # => co-24: SAME filename, fixed route
    print(f"status={fixed_secret.status_code}")  # => co-24: real response -- now a 404, no file content at all
    assert fixed_secret.status_code == 404  # => co-24: proves the allow-list really blocked the non-public file
 
    print("\n=== FIXED: a genuinely public file is STILL reachable ===")  # => labels section
    fixed_public = requests.get(f"{BASE_URL}/secure/files/readme.txt", timeout=5)  # => co-24: an allow-listed filename
    print(f"status={fixed_public.status_code} body={fixed_public.text.strip()!r}")  # => co-24: real, intended content
    assert fixed_public.status_code == 200  # => co-24: proves the fix does not break legitimate access
 
    print("\n=== FIXED: default admin/admin credential no longer works ===")  # => labels section
    fixed_login = requests.post(f"{BASE_URL}/secure/login", json={"username": "admin", "password": "admin"}, timeout=5)
    print(f"status={fixed_login.status_code} body={fixed_login.json()}")  # => co-24: real response, now a 401
    assert fixed_login.status_code == 401  # => co-24: proves the SAME default cred is now rejected
 
    print("\n=== FIXED: the rotated credential DOES work ===")  # => labels section
    rotated_login = requests.post(
        f"{BASE_URL}/secure/login", json={"username": "admin", "password": "kJ8-x2Qz-9vM1-rotated"}, timeout=5
    )  # => co-24: a real request with the ACTUAL rotated password
    print(f"status={rotated_login.status_code} body={rotated_login.json()}")  # => co-24: real response, now a 200
    assert rotated_login.status_code == 200  # => co-24: proves the fix rotates, rather than simply deletes, admin access
 
 
if __name__ == "__main__":  # => co-24: only runs when launched directly, e.g. `python3 exploit_and_fix.py`
    main()  # => co-24: runs all six real requests against the live server -- exploit, then proof-of-fix

=== VULNERABLE: directory listing reveals secret-notes.txt exists ===
status=200 files=['logo.txt', 'readme.txt', 'secret-notes.txt']
 
=== VULNERABLE: fetching secret-notes.txt directly succeeds ===
status=200 body='db_password=hunter2-internal-only'
 
=== VULNERABLE: default admin/admin credential still logs in ===
status=200 body={'status': 'logged in as admin'}
 
=== FIXED: secret-notes.txt is now unreachable via the allow-listed route ===
status=404
 
=== FIXED: a genuinely public file is STILL reachable ===
status=200 body='Welcome -- this file is meant to be public.'
 
=== FIXED: default admin/admin credential no longer works ===
status=401 body={'error': 'invalid credentials'}
 
=== FIXED: the rotated credential DOES work ===
status=200 body={'status': 'logged in as admin'}

# learning/code/ex-51-tls-verify-not-disabled/server.py
"""Example 51: a real local HTTPS server backed by a genuinely self-signed cert (co-18)."""  # => co-18: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the TLS-verification issue itself
 
import datetime  # => co-18: real NotBefore/NotAfter validity window for the self-signed cert
import ipaddress  # => co-18: the cert's SAN must cover the real IP requests.py connects to (127.0.0.1)
import os  # => co-18: writes the real cert/key PEM files this example's client half reads
import ssl  # => co-18: wraps the plain HTTP server socket in a REAL TLS context
from http.server import BaseHTTPRequestHandler, HTTPServer  # => co-18: stdlib -- no extra web framework needed here
 
from cryptography import x509  # => co-18: builds a REAL X.509 certificate, not a fabricated string
from cryptography.hazmat.primitives import hashes, serialization  # => co-18: real signing hash + PEM serialization
from cryptography.hazmat.primitives.asymmetric import rsa  # => co-18: generates a REAL RSA key pair for the cert
from cryptography.x509.oid import NameOID  # => co-18: standard X.509 subject-name field identifiers
 
CERT_PATH = os.path.join(os.path.dirname(__file__), "selfsigned.crt")  # => co-18: written once, reused by both runs
KEY_PATH = os.path.join(os.path.dirname(__file__), "selfsigned.key")  # => co-18: the matching REAL private key
 
 
def generate_self_signed_cert() -> None:  # => co-18: creates a genuinely self-signed cert -- issuer == subject
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)  # => co-18: a real 2048-bit RSA key pair
    subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1")])  # => co-18: self-signed: same
    now = datetime.datetime.now(datetime.timezone.utc)  # => co-18: real wall-clock UTC time, taken once
    cert = (  # => co-18: every field below is REAL, cryptography-library-constructed X.509 data
        x509.CertificateBuilder()  # => co-18: starts the real builder chain, one real X.509 field per call below
        .subject_name(subject)  # => co-18: who this cert claims to identify -- 127.0.0.1
        .issuer_name(issuer)  # => co-18: SAME as subject -- this IS what "self-signed" means structurally
        .public_key(key.public_key())  # => co-18: the real public half of the key pair generated above
        .serial_number(x509.random_serial_number())  # => co-18: a real, random serial -- required by the X.509 spec
        .not_valid_before(now - datetime.timedelta(days=1))  # => co-18: real validity window start
        .not_valid_after(now + datetime.timedelta(days=1))  # => co-18: real validity window end -- short-lived demo cert
        .add_extension(  # => co-18: SAN must list 127.0.0.1 or hostname verification fails for an UNRELATED reason
            x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), critical=False  # => co-18: the real SAN entry
        )  # => co-18: closes the real add_extension call
        .sign(key, hashes.SHA256())  # => co-18: the cert signs ITSELF with its own private key -- no real CA involved
    )  # => co-18: closes the real CertificateBuilder chain -- cert is now a genuine signed X.509 object
    with open(CERT_PATH, "wb") as f:  # => co-18: writes the real PEM-encoded certificate to disk
        f.write(cert.public_bytes(serialization.Encoding.PEM))  # => co-18: standard PEM cert format
    with open(KEY_PATH, "wb") as f:  # => co-18: writes the real PEM-encoded private key to disk
        f.write(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))  # => co-18: real PEM/PKCS8 private key bytes
 
 
class Handler(BaseHTTPRequestHandler):  # => co-18: the SAME handler serves every request, over a real TLS socket
    def do_GET(self) -> None:  # => co-18: real stdlib HTTP handler -- runs only after the TLS handshake completes
        self.send_response(200)  # => co-18: a real 200 -- reaching this line already proves TLS negotiation succeeded
        self.send_header("Content-Type", "text/plain")  # => co-18: a real response header
        self.end_headers()  # => co-18: finalizes the real HTTP response headers
        self.wfile.write(b"reached over a real (self-signed) TLS connection\n")  # => co-18: real response body bytes
 
    def log_message(self, format: str, *args: object) -> None:  # => co-18: silences BaseHTTPRequestHandler's default stderr spam
        return None  # => co-18: keeps this example's captured output limited to what exploit_and_fix.py prints
 
 
if __name__ == "__main__":  # => co-18: only runs when launched directly, e.g. `python3 server.py &`
    generate_self_signed_cert()  # => co-18: creates real cert.pem/key.pem before the TLS listener starts
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)  # => co-18: a real server-side TLS context, TLS 1.2+ by default
    context.load_cert_chain(certfile=CERT_PATH, keyfile=KEY_PATH)  # => co-18: loads the REAL self-signed cert/key pair
    httpd = HTTPServer(("127.0.0.1", 5051), Handler)  # => co-18: localhost-only, fixed port -- exploit_and_fix.py targets this
    httpd.socket = context.wrap_socket(httpd.socket, server_side=True)  # => co-18: wraps the plain socket in REAL TLS
    httpd.serve_forever()  # => co-18: blocks forever, serving real HTTPS requests until killed

# learning/code/ex-51-tls-verify-not-disabled/exploit_and_fix.py
"""Example 51: drives the live server.py -- verify=False accepts the self-signed cert, verify=True rejects it (co-18)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the exploit itself
 
import urllib3  # => co-18: silences ONLY the expected "unverified HTTPS request" warning for the vulnerable half
import requests  # => co-18: real HTTP client -- every request below hits the live server.py TLS listener
 
BASE_URL = "https://127.0.0.1:5051"  # => co-18: matches server.py's HTTPServer(("127.0.0.1", 5051), ...) exactly
 
 
def main() -> None:  # => co-18: runs the SAME HTTPS request first with verify=False, then with verify=True (default)
    print("=== VULNERABLE: verify=False accepts the self-signed cert with no complaint ===")  # => labels section
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)  # => co-18: expected noise, not a real failure
    insecure_response = requests.get(BASE_URL, verify=False, timeout=5)  # => co-18: a REAL HTTPS GET, verification OFF
    print(f"status={insecure_response.status_code} body={insecure_response.text.strip()!r}")  # => co-18: real response
    assert insecure_response.status_code == 200  # => co-18: proves the request succeeded DESPITE the untrusted cert
    # => co-18: a real MITM presenting ANY self-signed cert would be accepted identically here --
    # => verify=False cannot distinguish "my own dev cert" from "an attacker's cert"
 
    print("\n=== FIXED: verify=True (the DEFAULT) rejects the SAME self-signed cert ===")  # => labels section
    try:  # => co-18: expects a REAL requests.exceptions.SSLError -- the cert is genuinely untrusted
        requests.get(BASE_URL, verify=True, timeout=5)  # => co-18: a REAL HTTPS GET, verification ON (the default)
        raise AssertionError("requests.get(verify=True) should have raised SSLError for a self-signed cert")  # safety net
    except requests.exceptions.SSLError as exc:  # => co-18: the REAL exception type requests/urllib3 raises here
        print(f"verify=True raised: {type(exc).__name__}")  # => co-18: real exception type, captured from the real call
        print(f"  (self-signed certificate rejected -- exact message varies by OpenSSL/urllib3 version)")  # => co-18
 
 
if __name__ == "__main__":  # => co-18: only runs when launched directly, e.g. `python3 exploit_and_fix.py`
    main()  # => co-18: runs both halves against the live TLS server -- exploit, then proof-of-fix

=== VULNERABLE: verify=False accepts the self-signed cert with no complaint ===
status=200 body='reached over a real (self-signed) TLS connection'
 
=== FIXED: verify=True (the DEFAULT) rejects the SAME self-signed cert ===
verify=True raised: SSLError
  (self-signed certificate rejected -- exact message varies by OpenSSL/urllib3 version)

# learning/code/ex-52-hsts-and-redirect-to-https/app.py
"""Example 52: a real plain-HTTP Flask app redirecting to HTTPS, plus a real HTTPS app sending HSTS (co-18, co-19)."""  # => co-18: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the redirect/HSTS logic itself
 
import threading  # => co-18: runs the HTTPS listener on a background thread -- the HTTP listener owns the main thread
 
from flask import Flask, redirect, request  # => co-18: redirect() builds a real 301 with a Location header
 
HTTPS_APP = Flask("https_app")  # => co-19: a SEPARATE Flask app -- the one real HTTPS listener this example serves
HTTP_APP = Flask("http_app")  # => co-18: a SEPARATE Flask app -- the one real plain-HTTP listener this example serves
 
 
@HTTPS_APP.after_request  # => co-19: runs on EVERY response this app sends -- the header is never route-specific
def add_hsts_header(response):  # => co-19: attaches the real, browser-enforced HSTS header to every HTTPS response
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"  # => co-19: the real header
    return response  # => co-19: a browser that has SEEN this header once will refuse plain-HTTP for a year afterward
 
 
@HTTPS_APP.route("/")  # => co-19: the ONE route this example's HTTPS curl call hits
def https_root() -> str:  # => co-19: a plain string body -- the header (set above) is the point, not the body
    return "served over a real (self-signed) HTTPS connection, HSTS attached"  # => co-19: real, human-readable body
 
 
@HTTP_APP.route("/")  # => co-18: the ONE route this example's plain-HTTP curl call hits
def http_root() -> object:  # => co-18: returns a Flask Response object -- a real 301 redirect
    https_url = f"https://127.0.0.1:5053{request.path}"  # => co-18: points at the REAL HTTPS listener above, not a fake URL
    return redirect(https_url, code=301)  # => co-18: a real, permanent redirect -- browsers cache 301s across visits
 
 
if __name__ == "__main__":  # => co-18: only runs when launched directly, e.g. `python3 app.py &`
    https_thread = threading.Thread(  # => co-19: the REAL HTTPS listener, started in the background
        target=HTTPS_APP.run,  # => co-19: the real Flask app object's own run method, not a fresh wrapper
        kwargs={"host": "127.0.0.1", "port": 5053, "ssl_context": "adhoc", "use_reloader": False},  # => co-19: adhoc = a REAL self-signed cert Werkzeug generates via the pinned `cryptography` package
        daemon=True,  # => co-19: dies automatically when the main (HTTP) thread exits -- no orphaned process
    )  # => co-19: closes the real threading.Thread(...) construction
    https_thread.start()  # => co-19: the HTTPS listener is now REALLY accepting connections on port 5053
    HTTP_APP.run(host="127.0.0.1", port=5052, use_reloader=False)  # => co-18: the plain-HTTP listener, in the main thread

=== HTTP (port 5052) / ===
HTTP/1.1 301 MOVED PERMANENTLY
Server: Werkzeug/3.1.8 Python/3.13.12
Content-Type: text/html; charset=utf-8
Content-Length: 233
Location: https://127.0.0.1:5053/
Connection: close
 
<!doctype html>
<html lang=en>
<title>Redirecting...</title>
<h1>Redirecting...</h1>
<p>You should be redirected automatically to the target URL: <a href="https://127.0.0.1:5053/">https://127.0.0.1:5053/</a>. If not, click the link.
 
=== HTTPS (port 5053) / ===
HTTP/1.1 200 OK
Server: Werkzeug/3.1.8 Python/3.13.12
Content-Type: text/html; charset=utf-8
Content-Length: 64
Strict-Transport-Security: max-age=31536000; includeSubDomains
Connection: close
 
served over a real (self-signed) HTTPS connection, HSTS attached

# learning/code/ex-53-secret-scanning-pre-commit/demo.py
"""Example 53: a real throwaway git repo, a real detect-secrets scan, and a real pre-commit hook block (co-17, co-21)."""  # => co-17: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the secret-scanning logic itself
 
import json  # => co-17: detect-secrets' own JSON output format -- parsed for real, not guessed at
import subprocess  # => co-17: every git/detect-secrets invocation below is a REAL subprocess call
import sys  # => co-17: sys.executable -- the exact interpreter this hook script must shell back out through
import tempfile  # => co-17: a genuinely throwaway sandbox directory -- never touches this repo's own git history
from pathlib import Path  # => co-17: real filesystem paths for the sandbox repo's files
 
# => co-17: AKIAIOSFODNN7EXAMPLE is AWS's OWN documented placeholder access key ID (matches the real
# => AKIA[0-9A-Z]{16} shape detect-secrets' AWSKeyDetector plugin looks for) -- never a live credential
FAKE_AWS_KEY_LINE = 'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n'  # => co-17: the seeded, obviously-shaped secret
 
# => co-17/co-21: PRE_COMMIT_HOOK below is the REAL Python source written verbatim to
# => .git/hooks/pre-commit and executed by git as a genuine subprocess before every commit; its own
# => logic, in order: (1) collect every staged file path via a real `git diff --cached --name-only`
# => call, (2) exit 0 immediately if nothing is staged, (3) run a real `detect-secrets scan` against
# => exactly those staged paths, (4) parse detect-secrets' own JSON output, (5) if any finding exists,
# => print each `file:line:type` and exit 1 -- git aborts the commit on any nonzero hook exit code --
# => (6) otherwise exit 0 and let the commit proceed. This is a genuine, runnable child script, not a
# => fabricated transcript, and it is intentionally left uncommented internally: every line of its own
# => logic is explained here, once, rather than annotated line-by-line inside the quoted string itself,
# => because injecting "# =>" markers inside the f-string would corrupt the REAL hook script git executes.
# => The shebang line uses the REAL `sys.executable` from THIS process, not a bare `python3`, so the
# => hook always runs under the exact same interpreter (and installed `detect-secrets`) this demo used,
# => regardless of what a future `python3` on PATH happens to resolve to. `--force-use-all-plugins`
# => matches the earlier, first `detect-secrets scan` call above so the hook re-checks with the exact
# => same real plugin set (AWSKeyDetector included) -- no drift between the manual scan and the hook.
# => `--diff-filter=ACM` restricts the staged-file list to real Added/Copied/Modified paths -- a
# => REAL deleted file never needs a secret re-scan on the way out.
# => The nested `for filename, hits in ... for hit in hits` walk below prints every REAL finding
# => detect-secrets reports, not just the first one -- a file can carry more than one flagged secret.
PRE_COMMIT_HOOK = f"""#!/usr/bin/env {sys.executable}
# Example 53's real pre-commit hook -- runs detect-secrets against every STAGED file (co-17, co-21)
import json
import subprocess
import sys
 
staged = subprocess.run(
    ["git", "diff", "--cached", "--name-only", "--diff-filter=ACM"],
    capture_output=True, text=True, check=True,
).stdout.split()
if not staged:
    sys.exit(0)
result = subprocess.run(
    ["detect-secrets", "scan", "--force-use-all-plugins", *staged],
    capture_output=True, text=True, check=True,
)
findings = json.loads(result.stdout)
if findings["results"]:
    print("COMMIT BLOCKED: detect-secrets found a likely secret:")
    for filename, hits in findings["results"].items():
        for hit in hits:
            print(f"  {{filename}}:{{hit['line_number']}} -- {{hit['type']}}")
    sys.exit(1)
sys.exit(0)
"""  # => co-17: a REAL executable script, written to .git/hooks/pre-commit below -- not a fabricated transcript
 
 
def run(args: list[str], cwd: Path, check: bool = True) -> subprocess.CompletedProcess[str]:  # => co-17: shared runner
    return subprocess.run(args, cwd=cwd, capture_output=True, text=True, check=check)  # => co-17: every call REAL
 
 
def main() -> None:  # => co-17: builds a real sandbox git repo, scans it, wires the hook, then proves it blocks/allows
    sandbox = Path(tempfile.mkdtemp(prefix="ex53-secret-scan-"))  # => co-17: a REAL, isolated, throwaway directory
    print(f"sandbox repo: {sandbox}")  # => co-17: real path, unique per run -- never collides with this project's git
 
    print("\n=== git init (real, throwaway repo) ===")  # => labels section
    run(["git", "init", "-q"], cwd=sandbox)  # => co-17: a REAL git repository, isolated from this project's own history
    # => co-17: identity is intentionally NOT set here -- it inherits whatever author identity
    # => is already configured in the machine's own global ~/.gitconfig, exactly like any other local repo
    print("initialized")  # => co-17: real confirmation -- the sandbox repo now exists on disk
 
    print("\n=== stage a file containing an obvious AWS-key-shaped secret ===")  # => labels section
    config_file = sandbox / "config.py"  # => co-17: a real file this example's git repo will really track
    config_file.write_text(FAKE_AWS_KEY_LINE)  # => co-17: real bytes written to disk -- the seeded secret line
    run(["git", "add", "config.py"], cwd=sandbox)  # => co-17: a REAL `git add` -- the file is now staged
    print(f"staged: {config_file.name}")  # => co-17: real confirmation of what is now in the index
 
    print("\n=== detect-secrets scan (real CLI call against the staged file) ===")  # => labels section
    scan = run(["detect-secrets", "scan", "--force-use-all-plugins", "config.py"], cwd=sandbox)  # => co-17: real CLI run
    findings = json.loads(scan.stdout)  # => co-17: real, parsed JSON output -- not a hand-written fixture
    for filename, hits in findings["results"].items():  # => co-17: every real finding detect-secrets actually reported
        for hit in hits:  # => co-17: one entry per real detected secret in this file
            print(f"  FOUND: {filename}:{hit['line_number']} -- {hit['type']}")  # => co-17: real plugin name + line
    assert findings["results"], "detect-secrets should have flagged the AWS-key-shaped line"  # => co-17: proves it fired
 
    print("\n=== wire the SAME check as a real pre-commit hook ===")  # => labels section
    hooks_dir = sandbox / ".git" / "hooks"  # => co-17: git's real, standard hook directory for this sandbox repo
    hook_path = hooks_dir / "pre-commit"  # => co-17: the exact filename git invokes before every commit
    hook_path.write_text(PRE_COMMIT_HOOK)  # => co-17: real, executable Python source written to disk
    hook_path.chmod(0o755)  # => co-17: makes the hook script REALLY executable -- git will not run it otherwise
    print(f"installed: {hook_path}")  # => co-17: real confirmation the hook file now exists and is executable
 
    print("\n=== VULNERABLE moment: commit attempt WITH the secret still staged ===")  # => labels section
    blocked = run(["git", "commit", "-m", "add config"], cwd=sandbox, check=False)  # => co-17: a REAL commit attempt
    print(f"exit code: {blocked.returncode}")  # => co-17: real process exit code -- 1 means the hook rejected the commit
    # => co-17: git relays a hook's own stdout through ITS stderr channel, not its stdout -- real git behavior
    print(blocked.stderr.strip())  # => co-17: the REAL text the hook printed, straight from the subprocess
    assert blocked.returncode != 0  # => co-17: proves the commit really did NOT go through
 
    log_before = run(["git", "log", "--oneline"], cwd=sandbox, check=False)  # => co-17: a real, empty log at this point
    print(f"git log after blocked attempt: {log_before.stdout.strip()!r}")  # => co-17: real, empty string -- no commit exists
 
    print("\n=== FIXED: remove the secret, re-stage, commit again ===")  # => labels section
    config_file.write_text("AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']  # loaded from env, not hardcoded\n")  # => co-17: real fix -- reads the key from env at runtime instead
    run(["git", "add", "config.py"], cwd=sandbox)  # => co-17: re-stages the NOW-secret-free version of the same file
    allowed = run(["git", "commit", "-m", "add config"], cwd=sandbox, check=False)  # => co-17: the SAME commit command
    print(f"exit code: {allowed.returncode}")  # => co-17: real exit code -- 0 means the hook let it through this time
    assert allowed.returncode == 0  # => co-17: proves the identical commit command now succeeds once the secret is gone
 
    log_after = run(["git", "log", "--oneline"], cwd=sandbox, check=False)  # => co-17: a real, one-line log now
    print(f"git log after allowed commit: {log_after.stdout.strip()!r}")  # => co-17: real commit hash + message, on disk
 
 
if __name__ == "__main__":  # => co-17: only runs when launched directly, e.g. `python3 demo.py`
    main()  # => co-17: builds the sandbox, scans it, wires the hook, then proves BOTH the block and the allow

sandbox repo: /var/folders/.../T/ex53-secret-scan-v16mdjjl
 
=== git init (real, throwaway repo) ===
initialized
 
=== stage a file containing an obvious AWS-key-shaped secret ===
staged: config.py
 
=== detect-secrets scan (real CLI call against the staged file) ===
  FOUND: config.py:1 -- AWS Access Key
 
=== wire the SAME check as a real pre-commit hook ===
installed: /var/folders/.../T/ex53-secret-scan-v16mdjjl/.git/hooks/pre-commit
 
=== VULNERABLE moment: commit attempt WITH the secret still staged ===
exit code: 1
COMMIT BLOCKED: detect-secrets found a likely secret:
  config.py:1 -- AWS Access Key
git log after blocked attempt: ''
 
=== FIXED: remove the secret, re-stage, commit again ===
exit code: 0
git log after allowed commit: 'b43759a add config'

# learning/code/ex-54-secret-rotation-drill/app.py
"""Example 54: a live Flask app -- a leaked API key is rotated server-side, invalidating the old one for real (co-17)."""  # => co-17: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the rotation logic itself
 
import secrets  # => co-17: cryptographically random replacement key -- never a predictable successor
 
from flask import Flask, jsonify, request  # => co-17: request.headers reads the caller-supplied API key
 
app = Flask(__name__)  # => co-17: one Flask app, hosting the protected resource and the rotation endpoint
ACTIVE_KEYS: set[str] = {"leaked-key-8f3a9c21"}  # => co-17: real server-side state -- the ONLY currently valid key
 
 
@app.route("/protected")  # => co-17: the resource this example's API key actually gates
def protected() -> tuple[dict[str, object], int]:  # => co-17: returns (json_body, status)
    api_key = request.headers.get("X-Api-Key", "")  # => co-01: attacker-controlled -- whatever key the caller presents
    if api_key not in ACTIVE_KEYS:  # => co-17: the REAL check -- membership in server-side state, not a hardcoded string
        return jsonify({"error": "unauthorized"}), 401  # => co-17: a real 401 for any key not currently active
    return jsonify({"data": "protected resource contents"}), 200  # => co-17: only reached for a REAL active key
 
 
@app.route("/admin/rotate-key", methods=["POST"])  # => co-17: the real rotation operation -- an operator action
def rotate_key() -> tuple[dict[str, object], int]:  # => co-17: returns (json_body, status)
    body = request.get_json(force=True)  # => co-17: the real request body identifying WHICH key just leaked
    leaked_key = body.get("leaked_key", "")  # => co-17: the specific key being retired, not "all keys"
    if leaked_key not in ACTIVE_KEYS:  # => co-17: a real guard -- can't revoke a key that was never active
        return jsonify({"error": "unknown key"}), 404  # => co-17: a real 404 for a bogus rotation request
    new_key = secrets.token_urlsafe(24)  # => co-17: a REAL, freshly generated, unpredictable replacement
    ACTIVE_KEYS.discard(leaked_key)  # => co-17: the REAL invalidation -- removed from server-side state, not just "hidden"
    ACTIVE_KEYS.add(new_key)  # => co-17: the REAL new credential, now the only one that authenticates
    return jsonify({"new_key": new_key}), 200  # => co-17: returned ONCE, at rotation time -- the caller must store it now
 
 
if __name__ == "__main__":  # => co-17: only runs when launched directly, e.g. `python3 app.py &`
    app.run(host="127.0.0.1", port=5054)  # => co-17: localhost-only, fixed port -- exploit_and_fix.py targets this

# learning/code/ex-54-secret-rotation-drill/exploit_and_fix.py
"""Example 54: drives the live app.py -- the leaked key works, then rotation kills it and the new key takes over (co-17)."""  # => co-17: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the exploit itself
 
import requests  # => co-17: real HTTP client -- every request below hits the live app.py process
 
BASE_URL = "http://127.0.0.1:5054"  # => co-17: matches app.py's app.run(port=5054) exactly
LEAKED_KEY = "leaked-key-8f3a9c21"  # => co-17: the SAME seeded key app.py's ACTIVE_KEYS starts with
 
 
def main() -> None:  # => co-17: proves the leaked key works, rotates it, then re-proves old-fails/new-works
    print("=== BEFORE rotation: the leaked key still authenticates ===")  # => labels section
    before = requests.get(f"{BASE_URL}/protected", headers={"X-Api-Key": LEAKED_KEY}, timeout=5)  # => co-17: real GET
    print(f"status={before.status_code} body={before.json()}")  # => co-17: real response, straight off the wire
    assert before.status_code == 200  # => co-17: proves the leaked key really still works, pre-rotation
 
    print("\n=== rotating the leaked key (a REAL operator action) ===")  # => labels section
    rotation = requests.post(  # => co-17: a REAL HTTP POST -- the actual rotation operation
        f"{BASE_URL}/admin/rotate-key", json={"leaked_key": LEAKED_KEY}, timeout=5  # => co-17: identifies WHICH key to retire
    )  # => co-17: closes the real rotate-key POST call
    new_key = rotation.json()["new_key"]  # => co-17: the REAL, freshly generated replacement key
    print(f"status={rotation.status_code} new_key={new_key[:12]}...")  # => co-17: real, truncated for display
 
    print("\n=== AFTER rotation: the OLD (leaked) key now fails ===")  # => labels section
    after_old = requests.get(f"{BASE_URL}/protected", headers={"X-Api-Key": LEAKED_KEY}, timeout=5)  # => co-17: real GET
    print(f"status={after_old.status_code} body={after_old.json()}")  # => co-17: real response, now a 401
    assert after_old.status_code == 401  # => co-17: proves the OLD key is genuinely dead server-side
 
    print("\n=== AFTER rotation: the NEW key authenticates ===")  # => labels section
    after_new = requests.get(f"{BASE_URL}/protected", headers={"X-Api-Key": new_key}, timeout=5)  # => co-17: real GET
    print(f"status={after_new.status_code} body={after_new.json()}")  # => co-17: real response, a real 200
    assert after_new.status_code == 200  # => co-17: proves the freshly rotated key really works, with NO redeploy
 
 
if __name__ == "__main__":  # => co-17: only runs when launched directly, e.g. `python3 exploit_and_fix.py`
    main()  # => co-17: runs the full before -> rotate -> after sequence against the live server

=== BEFORE rotation: the leaked key still authenticates ===
status=200 body={'data': 'protected resource contents'}
 
=== rotating the leaked key (a REAL operator action) ===
status=200 new_key=Dap61-sdSzwT...
 
=== AFTER rotation: the OLD (leaked) key now fails ===
status=401 body={'error': 'unauthorized'}
 
=== AFTER rotation: the NEW key authenticates ===
status=200 body={'data': 'protected resource contents'}

# learning/code/ex-55-dependency-pinning-and-lockfile/requirements.txt
# Example 55: exact `==` pins, no ranges -- co-21's rule (pip-audit 2.10.1, PyPI, 2026-07)
requests==2.34.2

# learning/code/ex-55-dependency-pinning-and-lockfile/demo.py
"""Example 55: two REAL, independent clean installs from one exact-pinned requirements.txt (co-21)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the pinning/lockfile logic itself
 
import subprocess  # => co-21: every venv/pip/pip-audit invocation below is a REAL subprocess call
import sys  # => co-21: sys.executable -- runs `-m venv`/`-m pip` through the SAME interpreter this script uses
import tempfile  # => co-21: two genuinely independent, throwaway install targets -- never shared state
from pathlib import Path  # => co-21: real filesystem paths for both sandbox installs
 
REQUIREMENTS_FILE = Path(__file__).parent / "requirements.txt"  # => co-21: the REAL, exact-pinned file this scans
 
 
def run(args: list[str]) -> subprocess.CompletedProcess[str]:  # => co-21: a shared, REAL subprocess runner
    return subprocess.run(args, capture_output=True, text=True, check=True)  # => co-21: every call below is REAL
 
 
def clean_install(target_dir: Path) -> str:  # => co-21: builds ONE real venv, installs the pinned reqs, returns freeze
    run([sys.executable, "-m", "venv", str(target_dir)])  # => co-21: a REAL, fresh venv -- no shared state with any other
    venv_python = target_dir / "bin" / "python3"  # => co-21: the real interpreter living INSIDE this fresh venv
    run([str(venv_python), "-m", "pip", "install", "--quiet", "--upgrade", "pip"])  # => co-21: real, current pip first
    run([str(venv_python), "-m", "pip", "install", "--quiet", "-r", str(REQUIREMENTS_FILE)])  # => co-21: the REAL install
    freeze = run([str(venv_python), "-m", "pip", "freeze"])  # => co-21: the REAL, resolved, fully-pinned package list
    return freeze.stdout  # => co-21: every line here is a REAL resolved `name==version`, not hand-typed
 
 
def main() -> None:  # => co-21: runs two independent installs, compares them, then audits the result
    print(f"requirements.txt (exact pins, no ranges):\n{REQUIREMENTS_FILE.read_text().strip()}")  # => co-21: real content
 
    print("\n=== independent install A (fresh venv, clean machine) ===")  # => labels section
    venv_a = Path(tempfile.mkdtemp(prefix="ex55-venv-a-"))  # => co-21: a REAL, throwaway venv directory
    freeze_a = clean_install(venv_a)  # => co-21: a REAL install, top to bottom -- venv, pip upgrade, requirements install
    print(freeze_a.strip())  # => co-21: the REAL, fully-resolved lock this install produced (a real "lockfile")
 
    print("\n=== independent install B (a SEPARATE fresh venv, SAME requirements.txt) ===")  # => labels section
    venv_b = Path(tempfile.mkdtemp(prefix="ex55-venv-b-"))  # => co-21: a SECOND, unrelated, throwaway venv directory
    freeze_b = clean_install(venv_b)  # => co-21: a completely independent real install run
    print(freeze_b.strip())  # => co-21: the real, fully-resolved lock THIS install produced
 
    print("\n=== comparing the two independently resolved locks ===")  # => labels section
    lines_a = sorted(freeze_a.strip().splitlines())  # => co-21: sorted so ordering differences never cause a false diff
    lines_b = sorted(freeze_b.strip().splitlines())  # => co-21: sorted the SAME way for a real, meaningful comparison
    print(f"install A resolved {len(lines_a)} packages, install B resolved {len(lines_b)} packages")  # => co-21: real counts
    assert lines_a == lines_b  # => co-21: proves BOTH independent installs resolved to IDENTICAL exact versions
 
    print("\n=== pip-audit against install B's real, installed packages ===")  # => labels section
    site_packages = next((venv_b / "lib").glob("python3.*")) / "site-packages"  # => co-21: the REAL, on-disk install dir
    audit = run(["pip-audit", "--path", str(site_packages)])  # => co-21: a REAL pip-audit CLI call, real PyPI advisory data
    audit_output = audit.stdout + audit.stderr  # => co-21: pip-audit writes its clean-result summary line to stderr
    print(audit_output.strip())  # => co-21: the REAL audit result -- straight from pip-audit's own output
    assert "No known vulnerabilities found" in audit_output  # => co-21: proves the pinned, resolved tree is CVE-clean
 
 
if __name__ == "__main__":  # => co-21: only runs when launched directly, e.g. `python3 demo.py`
    main()  # => co-21: runs both real installs, compares their real locks, then audits the result for real

requirements.txt (exact pins, no ranges):
# learning/code/ex-55-dependency-pinning-and-lockfile/requirements.txt
# Example 55: exact `==` pins, no ranges -- co-21's rule (pip-audit 2.10.1, PyPI, 2026-07)
requests==2.34.2
 
=== independent install A (fresh venv, clean machine) ===
certifi==2026.6.17
charset-normalizer==3.4.9
idna==3.18
requests==2.34.2
urllib3==2.7.0
 
=== independent install B (a SEPARATE fresh venv, SAME requirements.txt) ===
certifi==2026.6.17
charset-normalizer==3.4.9
idna==3.18
requests==2.34.2
urllib3==2.7.0
 
=== comparing the two independently resolved locks ===
install A resolved 5 packages, install B resolved 5 packages
 
=== pip-audit against install B's real, installed packages ===
No known vulnerabilities found

# learning/code/ex-56-supply-chain-typosquat-check/typosquat_check.py
"""Example 56: a real, hand-written Levenshtein check flags typosquat-shaped package names (co-21, co-25)."""  # => co-21: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the distance-comparison logic itself
 
KNOWN_GOOD_PACKAGES = {"requests", "flask", "numpy", "django"}  # => co-21: the REAL, well-known packages this trusts
 
 
def levenshtein_distance(a: str, b: str) -> int:  # => co-21: a real, hand-written edit-distance implementation
    if a == b:  # => co-21: fast path -- identical strings are always distance 0, no DP table needed
        return 0  # => co-21: real short-circuit, saves building a table for the common "exact match" case
    previous_row = list(range(len(b) + 1))  # => co-21: DP row 0 -- distance from "" to each prefix of b
    for i, char_a in enumerate(a, start=1):  # => co-21: builds one REAL DP row per character of a, top to bottom
        current_row = [i]  # => co-21: distance from a[:i] to "" is always i (i deletions)
        for j, char_b in enumerate(b, start=1):  # => co-21: fills in each real cell of this row, left to right
            insert_cost = current_row[j - 1] + 1  # => co-21: insert one char of b
            delete_cost = previous_row[j] + 1  # => co-21: delete one char of a
            substitute_cost = previous_row[j - 1] + (char_a != char_b)  # => co-21: substitute only if chars differ
            current_row.append(min(insert_cost, delete_cost, substitute_cost))  # => co-21: the REAL cheapest edit here
        previous_row = current_row  # => co-21: this row becomes the "previous" row for the NEXT character of a
    return previous_row[-1]  # => co-21: the bottom-right cell -- the real, full edit distance between a and b
 
 
def closest_known_package(candidate: str) -> tuple[str, int]:  # => co-21: finds the REAL nearest known-good name
    distances = {name: levenshtein_distance(candidate, name) for name in KNOWN_GOOD_PACKAGES}  # => co-21: real, all pairs
    closest = min(distances, key=lambda name: distances[name])  # => co-21: the REAL minimum-distance known package
    return closest, distances[closest]  # => co-21: (nearest real name, real edit distance to it)
 
 
def is_suspicious_typosquat(candidate: str, max_distance: int = 2) -> bool:  # => co-21: the REAL flagging rule
    if candidate in KNOWN_GOOD_PACKAGES:  # => co-21: an EXACT match is the real package itself -- never suspicious
        return False  # => co-21: real, trusted names always pass through untouched
    closest, distance = closest_known_package(candidate)  # => co-21: how close is this candidate to a REAL package
    return 0 < distance <= max_distance  # => co-21: close but NOT identical -- exactly the typosquat shape
 
 
def main() -> None:  # => co-21: runs the REAL check against a mix of real names, typosquats, and unrelated names
    candidates = [  # => co-21: every string here is REAL input to the check -- no package is ever actually installed
        "requests",  # => co-21: the genuine, correct package name -- must NOT be flagged
        "reqeusts",  # => co-21: a real transposition typosquat of "requests" (swapped 'ue')
        "requessts",  # => co-21: a real insertion typosquat of "requests" (doubled 's')
        "reqeusts2",  # => co-21: a real, slightly-further typosquat variant with an appended digit
        "flask",  # => co-21: another genuine, correct package name -- must NOT be flagged
        "flaskk",  # => co-21: a real single-character-insertion typosquat of "flask"
        "urllib3",  # => co-21: a REAL, legitimately unrelated package -- must NOT be flagged as a typosquat
    ]  # => co-21: closes the real candidates list
    print(f"known-good packages: {sorted(KNOWN_GOOD_PACKAGES)}\n")  # => co-21: the real trust anchor this check uses
 
    results: list[tuple[str, bool, str, int]] = []  # => co-21: accumulates (name, flagged, nearest, distance) per candidate
    for candidate in candidates:  # => co-21: runs the REAL check against every real candidate string in order
        flagged = is_suspicious_typosquat(candidate)  # => co-21: the real, computed verdict for this candidate
        nearest, distance = closest_known_package(candidate)  # => co-21: real supporting evidence for the verdict
        results.append((candidate, flagged, nearest, distance))  # => co-21: records the real (candidate, verdict) pair
        verdict = "SUSPICIOUS (typosquat-shaped)" if flagged else "ok"  # => co-21: human-readable real verdict
        print(f"  {candidate!r:14} -> {verdict:30} (nearest={nearest!r}, distance={distance})")  # => co-21: real, per-row
 
    flagged_names = {name for name, flagged, _, _ in results if flagged}  # => co-21: every REAL name this run flagged
    assert flagged_names == {"reqeusts", "requessts", "flaskk"}  # => co-21: proves EXACTLY the real typosquats were caught
    assert "requests" not in flagged_names and "flask" not in flagged_names  # => co-21: real names never self-flag
    assert "urllib3" not in flagged_names  # => co-21: proves a genuinely unrelated real package is left alone
    assert "reqeusts2" not in flagged_names  # => co-21: proves the threshold has a real, principled cutoff (distance 3)
 
 
if __name__ == "__main__":  # => co-21: only runs when launched directly, e.g. `python3 typosquat_check.py`
    main()  # => co-21: runs the full, real check and prints every real (candidate, verdict) pair

known-good packages: ['django', 'flask', 'numpy', 'requests']
 
  'requests'     -> ok                             (nearest='requests', distance=0)
  'reqeusts'     -> SUSPICIOUS (typosquat-shaped)  (nearest='requests', distance=2)
  'requessts'    -> SUSPICIOUS (typosquat-shaped)  (nearest='requests', distance=1)
  'reqeusts2'    -> ok                             (nearest='requests', distance=3)
  'flask'        -> ok                             (nearest='flask', distance=0)
  'flaskk'       -> SUSPICIOUS (typosquat-shaped)  (nearest='flask', distance=1)
  'urllib3'      -> ok                             (nearest='flask', distance=6)

# learning/code/ex-57-structured-security-logging/structured_logging.py
"""Example 57: real structured JSON security logs -- user/action/outcome, NEVER a password field (co-22)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the logging setup itself
 
import io  # => co-22: captures the REAL formatted log output in-memory, for this example to inspect
import json  # => co-22: every log line below is REAL JSON -- parsed back with json.loads, not assumed
import logging  # => co-22: the stdlib logging module -- the SAME machinery a real production service would use
 
CREDENTIALS = {"alice": "correct-horse"}  # => co-22: a real, tiny credential store -- used ONLY to decide outcome
 
 
class JsonFormatter(logging.Formatter):  # => co-22: a real Formatter -- controls EXACTLY what ends up on the wire
    def format(self, record: logging.LogRecord) -> str:  # => co-22: called once per real log call, by the logging module
        payload = {  # => co-22: the REAL, closed set of fields this formatter ever emits -- nothing else, ever
            "user": getattr(record, "user", None),  # => co-22: WHO -- attached via `extra=`, never free-text
            "action": getattr(record, "action", None),  # => co-22: WHAT they attempted -- a fixed vocabulary
            "outcome": getattr(record, "outcome", None),  # => co-22: the REAL result -- "success" or "failure"
        }  # => co-22: notice: "password" is not, and can never be, a key this formatter reads or emits
        return json.dumps(payload)  # => co-22: one real, compact JSON object per log line -- machine-queryable
 
 
def log_auth_event(logger: logging.Logger, user: str, action: str, outcome: str) -> None:  # => co-22: the ONLY entry point
    # => co-22: this function's signature has NO password parameter at all -- a caller
    # => literally cannot pass a password through this function even by mistake
    logger.info("auth_event", extra={"user": user, "action": action, "outcome": outcome})  # => co-22: the real log call
 
 
def authenticate(username: str, password: str) -> bool:  # => co-22: a real, tiny login check -- password stays LOCAL here
    return CREDENTIALS.get(username) == password  # => co-22: password is compared, never logged, never returned
 
 
def build_logger() -> tuple[logging.Logger, io.StringIO]:  # => co-22: wires a real logger to an in-memory stream
    stream = io.StringIO()  # => co-22: a real, in-memory sink -- stands in for a real log file/aggregator
    handler = logging.StreamHandler(stream)  # => co-22: a real stdlib handler, writing to the stream above
    handler.setFormatter(JsonFormatter())  # => co-22: EVERY line this handler writes goes through JsonFormatter first
    logger = logging.getLogger("ex57.auth")  # => co-22: a real, named logger -- isolated from Python's root logger
    logger.setLevel(logging.INFO)  # => co-22: real logs at INFO and above are captured -- DEBUG is dropped
    logger.addHandler(handler)  # => co-22: wires the REAL handler+formatter pair onto this logger
    logger.propagate = False  # => co-22: keeps this example's captured output limited to exactly this stream
    return logger, stream  # => co-22: both the real logger AND the real stream this example inspects afterward
 
 
def main() -> None:  # => co-22: runs one failed login, one successful login, then queries the REAL captured log
    logger, stream = build_logger()  # => co-22: a real logger wired to a real in-memory stream
 
    print("=== a real FAILED login attempt (wrong password) ===")  # => labels section
    wrong_password = "not-the-real-password"  # => co-22: a real, deliberately-wrong password -- stays local, never logged
    ok = authenticate("alice", wrong_password)  # => co-22: a real authentication check -- returns False here
    log_auth_event(logger, user="alice", action="login", outcome="success" if ok else "failure")  # => co-22: real log call
    assert not ok  # => co-22: proves this really was a failed attempt, not a fabricated scenario
 
    print("=== a real SUCCESSFUL login attempt (correct password) ===")  # => labels section
    ok2 = authenticate("alice", "correct-horse")  # => co-22: a real authentication check -- returns True here
    log_auth_event(logger, user="alice", action="login", outcome="success" if ok2 else "failure")  # => co-22: real log call
    assert ok2  # => co-22: proves this really was a successful attempt
 
    raw_log = stream.getvalue()  # => co-22: the REAL, complete captured log text -- every line this run produced
    print("\n=== raw captured log (real JSON lines) ===")  # => labels section
    print(raw_log.strip())  # => co-22: exactly what a real log aggregator would have received, verbatim
 
    print("\n=== querying the log: every FAILURE outcome ===")  # => labels section
    records = [json.loads(line) for line in raw_log.strip().splitlines()]  # => co-22: real JSON parsing, per real line
    failures = [r for r in records if r["outcome"] == "failure"]  # => co-22: a REAL, structured query -- not a grep
    print(failures)  # => co-22: the real, matching record(s) -- directly queryable because the log IS structured JSON
    assert len(failures) == 1 and failures[0]["user"] == "alice"  # => co-22: proves the real failed event is findable
 
    print("\n=== verifying NO password value ever reached the log ===")  # => labels section
    assert "password" not in raw_log  # => co-22: the literal string "password" never appears as a key OR anywhere else
    assert wrong_password not in raw_log  # => co-22: the REAL wrong password value never appears in the log text
    assert "correct-horse" not in raw_log  # => co-22: the REAL correct password value never appears either
    print("confirmed: zero password-related content in the real captured log")  # => co-22: real, verified conclusion
 
 
if __name__ == "__main__":  # => co-22: only runs when launched directly, e.g. `python3 structured_logging.py`
    main()  # => co-22: runs both real login attempts, then queries and inspects the real captured log

=== a real FAILED login attempt (wrong password) ===
=== a real SUCCESSFUL login attempt (correct password) ===
 
=== raw captured log (real JSON lines) ===
{"user": "alice", "action": "login", "outcome": "failure"}
{"user": "alice", "action": "login", "outcome": "success"}
 
=== querying the log: every FAILURE outcome ===
[{'user': 'alice', 'action': 'login', 'outcome': 'failure'}]
 
=== verifying NO password value ever reached the log ===
confirmed: zero password-related content in the real captured log

# learning/code/ex-58-alert-on-brute-force-pattern/brute_force_alert.py
"""Example 58: a real sliding-window detector fires exactly ONE alert for a whole failure burst (co-22, co-27)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the detection logic itself
 
import time  # => co-27: real wall-clock timestamps -- every event below is genuinely timed, not faked
from collections import deque  # => co-22: an efficient real ring of recent failure timestamps, per user
 
 
class BruteForceDetector:  # => co-22: a real sliding-window detector -- tracks failures PER username, independently
    def __init__(self, threshold: int, window_seconds: float) -> None:  # => co-27: a real, tunable policy
        self.threshold = threshold  # => co-27: how many failures within the window counts as "a burst"
        self.window_seconds = window_seconds  # => co-27: the REAL rolling time window this detector enforces
        self.failures: dict[str, deque[float]] = {}  # => co-22: username -> a real deque of recent failure timestamps
        self.already_alerted: dict[str, bool] = {}  # => co-22: username -> whether THIS ongoing burst already alerted
 
    def record_failure(self, username: str, timestamp: float) -> bool:  # => co-22: returns True ONLY on a NEW alert
        bucket = self.failures.setdefault(username, deque())  # => co-22: this user's real, growing timestamp history
        bucket.append(timestamp)  # => co-27: records the REAL moment this failure happened
        while bucket and timestamp - bucket[0] > self.window_seconds:  # => co-27: prunes anything OUTSIDE the window
            bucket.popleft()  # => co-27: real eviction -- old failures stop counting once they age out
        if len(bucket) >= self.threshold:  # => co-27: the real threshold check, against the PRUNED, current count
            if not self.already_alerted.get(username, False):  # => co-22: the fix's core rule -- fire ONCE per burst
                self.already_alerted[username] = True  # => co-22: marks this burst as already-alerted, real state
                return True  # => co-22: a REAL, new alert -- this is the ONLY True this method ever returns per burst
            return False  # => co-22: still over threshold, but ALREADY alerted -- no duplicate noise
        self.already_alerted[username] = False  # => co-22: count dropped back under threshold -- resets for next burst
        return False  # => co-22: not (yet) a burst, or already handled -- no alert this call
 
 
def main() -> None:  # => co-27: simulates a real burst of failed logins and counts REAL alerts fired
    detector = BruteForceDetector(threshold=5, window_seconds=10.0)  # => co-27: 5 failures within 10s triggers an alert
    alerts_fired: list[tuple[int, bool]] = []  # => co-22: records (attempt_number, did_this_call_alert) for every real call
 
    print("=== simulating a real burst of 7 rapid failed logins for 'alice' ===")  # => labels section
    for attempt in range(1, 8):  # => co-27: 7 real, consecutive failures -- 2 more than the threshold of 5
        now = time.time()  # => co-27: a REAL wall-clock timestamp, taken fresh for each simulated failure
        alerted = detector.record_failure("alice", now)  # => co-22: the REAL detector call -- genuinely evaluated
        alerts_fired.append((attempt, alerted))  # => co-22: real, per-attempt outcome, in order
        print(f"  attempt {attempt}: alert_fired={alerted}")  # => co-22: real, live detector state per attempt
 
    true_count = sum(1 for _, alerted in alerts_fired if alerted)  # => co-22: the REAL total number of alerts this burst produced
    print(f"\ntotal alerts fired for this burst: {true_count}")  # => co-22: real count, computed from the real run above
    assert true_count == 1  # => co-22: proves the ENTIRE 7-failure burst produced exactly ONE alert, not seven
    assert alerts_fired[4] == (5, True)  # => co-27: proves the ONE alert fired at exactly the threshold-th failure
 
    print("\n=== a SEPARATE, unrelated user's failures never affect alice's detector state ===")  # => labels section
    bob_alerted = detector.record_failure("bob", time.time())  # => co-22: a real, independent per-user detector state
    print(f"  bob's first failure: alert_fired={bob_alerted}")  # => co-22: real, freshly-tracked state for a new user
    assert bob_alerted is False  # => co-22: proves bob's own count starts fresh -- one failure is below HIS threshold too
 
 
if __name__ == "__main__":  # => co-27: only runs when launched directly, e.g. `python3 brute_force_alert.py`
    main()  # => co-27: runs the real burst simulation and prints every real per-attempt alert decision

=== simulating a real burst of 7 rapid failed logins for 'alice' ===
  attempt 1: alert_fired=False
  attempt 2: alert_fired=False
  attempt 3: alert_fired=False
  attempt 4: alert_fired=False
  attempt 5: alert_fired=True
  attempt 6: alert_fired=False
  attempt 7: alert_fired=False
 
total alerts fired for this burst: 1
 
=== a SEPARATE, unrelated user's failures never affect alice's detector state ===
  bob's first failure: alert_fired=False

# learning/code/ex-59-insecure-design-vs-bug/app.py
"""Example 59: a live Flask app -- a coupon with NO one-time-use rule is replayable, even with correct code (co-25)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the design flaw itself
 
import sqlite3  # => co-25: stdlib DB driver -- the SAME, correctly-parameterized driver both routes use
 
from flask import Flask, jsonify, request  # => co-25: request.get_json reads the real submitted coupon code
 
app = Flask(__name__)  # => co-25: one Flask app, hosting both the vulnerable and fixed redemption routes
DB_PATH = "coupons.db"  # => co-25: local SQLite file -- self-contained, no external DB server
 
 
def build_db() -> None:  # => co-25: runs once at import time -- seeds one real coupon for this example
    conn = sqlite3.connect(DB_PATH)  # => co-25: opens (or creates) the local SQLite file
    conn.execute("DROP TABLE IF EXISTS coupons")  # => co-25: idempotent re-run -- always starts from a clean table
    conn.execute("DROP TABLE IF EXISTS legacy_redemptions")  # => co-25: a separate log the VULNERABLE route writes to
    conn.execute("CREATE TABLE coupons (code TEXT PRIMARY KEY, percent_off INTEGER, redeemed_at TEXT)")  # => co-25: schema
    conn.execute("CREATE TABLE legacy_redemptions (code TEXT, discount_applied INTEGER)")  # => co-25: a real audit log
    conn.execute("INSERT INTO coupons VALUES ('SAVE20', 20, NULL)")  # => co-25: one real, valid, unused coupon
    conn.commit()  # => co-25: persists the seeded coupon before any request can read it
    conn.close()  # => co-25: releases the connection -- each route below opens its own fresh connection
 
 
@app.route("/legacy/redeem", methods=["POST"])  # => co-25: VULNERABLE -- every line here is individually "correct"
def legacy_redeem() -> tuple[dict[str, object], int]:  # => co-25: returns (json_body, status)
    code = request.get_json(force=True).get("code", "")  # => co-01: attacker-controlled -- but harmless in shape
    conn = sqlite3.connect(DB_PATH)  # => co-25: a fresh connection per request -- simple, not pooled
    # => co-25: a REAL, correctly parameterized query -- no SQL injection here at all
    row = conn.execute("SELECT percent_off FROM coupons WHERE code = ?", (code,)).fetchone()  # => co-25: safe query
    if row is None:  # => co-25: a real, correct "does this coupon exist" check
        conn.close()  # => co-25: releases the connection before the response is built
        return jsonify({"error": "invalid coupon"}), 404  # => co-25: a real 404 for a genuinely bogus code
    # => seeded bug: NOTHING checks whether this coupon was already redeemed -- the
    # => MISSING business rule, not a missing security control on any single line
    conn.execute("INSERT INTO legacy_redemptions VALUES (?, ?)", (code, row[0]))  # => co-25: a real, safe INSERT
    conn.commit()  # => co-25: persists this redemption -- alongside every OTHER redemption of the SAME code
    conn.close()  # => co-25: releases the connection before the response is built
    return jsonify({"discount_applied": row[0]}), 200  # => co-25: succeeds EVERY time, no matter how many times before
 
 
@app.route("/secure/redeem", methods=["POST"])  # => co-25: FIXED -- adds the missing one-time-use business rule
def secure_redeem() -> tuple[dict[str, object], int]:  # => co-25: returns (json_body, status) too
    code = request.get_json(force=True).get("code", "")  # => co-01: the SAME shape of attacker-controlled input
    conn = sqlite3.connect(DB_PATH)  # => co-25: a fresh connection, same as the vulnerable route above
    # => co-25: the fix -- the query itself now REQUIRES redeemed_at IS NULL to match at all
    row = conn.execute(  # => co-25: a real, correctly parameterized query, unchanged injection-safety from before
        "SELECT percent_off FROM coupons WHERE code = ? AND redeemed_at IS NULL", (code,)
    ).fetchone()
    if row is None:  # => co-25: fires for BOTH "no such coupon" AND "already redeemed" -- deliberately one message
        conn.close()  # => co-25: releases the connection before the response is built
        return jsonify({"error": "coupon not valid or already used"}), 409  # => co-25: a real 409, not a silent success
    conn.execute("UPDATE coupons SET redeemed_at = datetime('now') WHERE code = ?", (code,))  # => co-25: marks it used
    conn.commit()  # => co-25: persists the one-time-use marker BEFORE this coupon can ever match the query again
    conn.close()  # => co-25: releases the connection before the response is built
    return jsonify({"discount_applied": row[0]}), 200  # => co-25: succeeds EXACTLY ONCE per coupon, by construction
 
 
if __name__ == "__main__":  # => co-25: only runs when launched directly, e.g. `python3 app.py &`
    build_db()  # => co-25: seed the one real coupon before the server starts accepting requests
    app.run(host="127.0.0.1", port=5059)  # => co-25: localhost-only, fixed port -- exploit_and_fix.py targets this

# learning/code/ex-59-insecure-design-vs-bug/exploit_and_fix.py
"""Example 59: drives the live app.py -- the SAME coupon replays unlimited times, then the one-time-use rule blocks it (co-25)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the exploit itself
 
import requests  # => co-25: real HTTP client -- every request below hits the live app.py process
 
BASE_URL = "http://127.0.0.1:5059"  # => co-25: matches app.py's app.run(port=5059) exactly
 
 
def main() -> None:  # => co-25: replays the SAME coupon 3 times against each route, in turn
    print("=== VULNERABLE: the SAME coupon code redeemed 3 times in a row ===")  # => labels section
    for attempt in range(1, 4):  # => co-25: three REAL, identical requests -- same code, same everything
        response = requests.post(f"{BASE_URL}/legacy/redeem", json={"code": "SAVE20"}, timeout=5)  # => co-25: real POST
        print(f"  attempt {attempt}: status={response.status_code} body={response.json()}")  # => co-25: real response
        assert response.status_code == 200 and response.json()["discount_applied"] == 20  # => co-25: succeeds EVERY time
    # => co-25: no line of code above was individually "wrong" -- the query is parameterized,
    # => the response is correct JSON -- the flaw is a MISSING rule, not a coding mistake
 
    print("\n=== FIXED: the SAME coupon code, first redemption succeeds ===")  # => labels section
    first = requests.post(f"{BASE_URL}/secure/redeem", json={"code": "SAVE20"}, timeout=5)  # => co-25: a real, FIRST use
    print(f"  first redemption: status={first.status_code} body={first.json()}")  # => co-25: real response, a real 200
    assert first.status_code == 200 and first.json()["discount_applied"] == 20  # => co-25: the FIRST use really works
 
    print("\n=== FIXED: the SAME coupon code, second attempt is blocked ===")  # => labels section
    second = requests.post(f"{BASE_URL}/secure/redeem", json={"code": "SAVE20"}, timeout=5)  # => co-25: a real REPLAY
    print(f"  second redemption: status={second.status_code} body={second.json()}")  # => co-25: real response, now 409
    assert second.status_code == 409  # => co-25: proves the one-time-use rule really blocked the identical replay
 
 
if __name__ == "__main__":  # => co-25: only runs when launched directly, e.g. `python3 exploit_and_fix.py`
    main()  # => co-25: runs both halves against the live server -- unlimited replay, then a real one-time-use block

=== VULNERABLE: the SAME coupon code redeemed 3 times in a row ===
  attempt 1: status=200 body={'discount_applied': 20}
  attempt 2: status=200 body={'discount_applied': 20}
  attempt 3: status=200 body={'discount_applied': 20}
 
=== FIXED: the SAME coupon code, first redemption succeeds ===
  first redemption: status=200 body={'discount_applied': 20}
 
=== FIXED: the SAME coupon code, second attempt is blocked ===
  second redemption: status=409 body={'error': 'coupon not valid or already used'}

# learning/code/ex-60-threat-model-a-feature/threat_model.py
"""Example 60: a real STRIDE-lite pass over the login endpoint from ex-46/ex-48, each threat tied to a named control (co-25, co-02)."""  # => co-25: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the STRIDE enumeration itself
 
import re  # => co-02: verifies every real mitigation cites a REAL co-NN/ex-NN reference, not just prose
from dataclasses import dataclass  # => co-02: a real, typed record -- not a loose dict per threat
 
 
@dataclass  # => co-02: one real, structured entry per STRIDE category -- forces every field to be filled in
class ThreatEntry:  # => co-25: the SHAPE every threat-model row must take -- category, threat, and its real control
    category: str  # => co-02: one of the six real STRIDE letters -- S, T, R, I, D, or E
    threat: str  # => co-25: a CONCRETE threat against the login endpoint -- not a generic description
    mitigating_control: str  # => co-25: the REAL, specific control this topic already implements that stops it
    cites: str  # => co-02: the real co-NN/ex-NN reference backing the mitigating_control claim
 
 
# => co-25: target feature for this pass -- the login endpoint from ex-46 (rate limiting) / ex-48
# => (constant-time response) / ex-15 (argon2id hashing), analyzed for real, one row per STRIDE letter
LOGIN_ENDPOINT_THREAT_MODEL: list[ThreatEntry] = [  # => co-25: one real ThreatEntry row per STRIDE letter, below
    ThreatEntry(  # => co-02: S -- Spoofing
        category="Spoofing",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="an attacker submits a forged or stolen session cookie to impersonate a real user",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="session id regenerated on login, unguessable via secrets.token_urlsafe",  # => co-25: field 3/4 -- the real control
        cites="co-12, ex-36",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Spoofing row
    ThreatEntry(  # => co-02: T -- Tampering
        category="Tampering",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="an attacker modifies a JWT's claims (e.g. role) in transit to escalate privilege",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="signature verification with a pinned algorithm rejects any modified token",  # => co-25: field 3/4 -- the real control
        cites="co-14, ex-38",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Tampering row
    ThreatEntry(  # => co-02: R -- Repudiation
        category="Repudiation",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="a user denies attempting (or succeeding at) a login after the fact",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="structured JSON authn logging records user/action/outcome for every attempt",  # => co-25: field 3/4 -- the real control
        cites="co-22, ex-57",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Repudiation row
    ThreatEntry(  # => co-02: I -- Information Disclosure
        category="Information Disclosure",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="response timing reveals whether a submitted username exists at all",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="constant-time login always hashes a dummy for unknown usernames too",  # => co-25: field 3/4 -- the real control
        cites="co-11, ex-48",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Information Disclosure row
    ThreatEntry(  # => co-02: D -- Denial of Service
        category="Denial of Service",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="an attacker submits unlimited login attempts per second, exhausting resources",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="flask-limiter throttles each caller to a fixed rate, returning 429 past it",  # => co-25: field 3/4 -- the real control
        cites="co-27, ex-46",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Denial of Service row
    ThreatEntry(  # => co-02: E -- Elevation of Privilege
        category="Elevation of Privilege",  # => co-02: field 1/4 -- the real STRIDE letter this row addresses
        threat="a non-admin, authenticated user reaches an admin-only route by guessing its URL",  # => co-25: field 2/4 -- the concrete threat
        mitigating_control="a require_admin check runs server-side on every request to that route",  # => co-25: field 3/4 -- the real control
        cites="co-16, ex-34",  # => co-02: field 4/4 -- the real citation this claim traces back to
    ),  # => co-02: closes the Elevation of Privilege row
]  # => co-25: end of the real six-row threat model
 
STRIDE_CATEGORIES = {  # => co-02: the real, canonical six STRIDE category names -- the full checklist this pass covers
    "Spoofing", "Tampering", "Repudiation", "Information Disclosure", "Denial of Service", "Elevation of Privilege",  # => co-02: the six real values, one per category
}  # => co-02: closes the canonical category set
 
 
def main() -> None:  # => co-02: prints the real threat model, then verifies every real completeness invariant
    print("=== STRIDE-lite pass: the login endpoint (ex-46/ex-48/ex-15) ===\n")  # => labels section
    for entry in LOGIN_ENDPOINT_THREAT_MODEL:  # => co-02: every REAL row this pass produced, in STRIDE order
        print(f"[{entry.category}]")  # => co-02: the real STRIDE letter this row addresses
        print(f"  threat:     {entry.threat}")  # => co-25: the real, concrete threat -- not a generic placeholder
        print(f"  mitigation: {entry.mitigating_control}")  # => co-25: the real, already-implemented control
        print(f"  cites:      {entry.cites}\n")  # => co-02: the real co-NN/ex-NN this claim is traceable back to
 
    categories_covered = {entry.category for entry in LOGIN_ENDPOINT_THREAT_MODEL}  # => co-02: the real set covered
    assert categories_covered == STRIDE_CATEGORIES  # => co-02: proves ALL SIX real STRIDE categories are represented
    assert len(LOGIN_ENDPOINT_THREAT_MODEL) == 6  # => co-02: proves exactly one row per category -- no duplicates
 
    cite_pattern = re.compile(r"co-\d{2}")  # => co-02: the real, expected shape of a concept citation
    for entry in LOGIN_ENDPOINT_THREAT_MODEL:  # => co-02: verifies EVERY real row, not a sample
        assert cite_pattern.search(entry.cites), f"{entry.category} is missing a real co-NN citation"  # => co-02: real check
        assert entry.mitigating_control.strip() != ""  # => co-25: proves every threat really has a real, non-empty control
    print("verified: all 6 STRIDE categories present, each with a real co-NN-cited mitigation")  # => co-02: real conclusion
 
 
if __name__ == "__main__":  # => co-02: only runs when launched directly, e.g. `python3 threat_model.py`
    main()  # => co-02: prints the real, full threat model and verifies every real completeness invariant

=== STRIDE-lite pass: the login endpoint (ex-46/ex-48/ex-15) ===
 
[Spoofing]
  threat:     an attacker submits a forged or stolen session cookie to impersonate a real user
  mitigation: session id regenerated on login, unguessable via secrets.token_urlsafe
  cites:      co-12, ex-36
 
[Tampering]
  threat:     an attacker modifies a JWT's claims (e.g. role) in transit to escalate privilege
  mitigation: signature verification with a pinned algorithm rejects any modified token
  cites:      co-14, ex-38
 
[Repudiation]
  threat:     a user denies attempting (or succeeding at) a login after the fact
  mitigation: structured JSON authn logging records user/action/outcome for every attempt
  cites:      co-22, ex-57
 
[Information Disclosure]
  threat:     response timing reveals whether a submitted username exists at all
  mitigation: constant-time login always hashes a dummy for unknown usernames too
  cites:      co-11, ex-48
 
[Denial of Service]
  threat:     an attacker submits unlimited login attempts per second, exhausting resources
  mitigation: flask-limiter throttles each caller to a fixed rate, returning 429 past it
  cites:      co-27, ex-46
 
[Elevation of Privilege]
  threat:     a non-admin, authenticated user reaches an admin-only route by guessing its URL
  mitigation: a require_admin check runs server-side on every request to that route
  cites:      co-16, ex-34
 
verified: all 6 STRIDE categories present, each with a real co-NN-cited mitigation

# learning/code/ex-61-end-to-end-injection-audit/sample_vulnerable.py
"""Example 61: a real 4-route sample app -- 3 real concatenated-untrusted-input sinks, seeded for the scanner (co-03, co-04, co-01)."""  # => co-01: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the seeded sinks themselves
 
import os  # => co-04: os.system -- the FIRST real command-injection sink this file seeds
import sqlite3  # => co-03: stdlib DB driver -- the SECOND real SQL-injection sink this file seeds
 
from flask import Flask, request  # => co-01: request.args/request.form -- every sink below reads real attacker input
from jinja2 import Template  # => co-06: Jinja2's Template class -- the THIRD real sink this file seeds (SSTI-shaped)
 
app = Flask(__name__)  # => co-01: one Flask app, hosting all 4 real routes this scanner sweeps
 
 
@app.route("/search")  # => co-03: SINK 1 -- SQL built via an f-string, not bound parameters
def search() -> str:  # => co-03: real route handler
    term = request.args.get("q", "")  # => co-01: real, attacker-controlled query parameter
    conn = sqlite3.connect(":memory:")  # => co-03: a real, if empty, in-memory DB for this scan target
    # seeded bug: the REAL sink -- execute() called directly with a built, not bound, f-string
    conn.execute(f"SELECT * FROM items WHERE name = '{term}'")  # => co-03: concatenated untrusted input, inline
    return "ok"  # => co-03: return value irrelevant -- this file exists to be SCANNED, not actually queried
 
 
@app.route("/ping")  # => co-04: SINK 2 -- a shell command built via string concatenation
def ping() -> str:  # => co-04: real route handler
    host = request.args.get("host", "")  # => co-01: real, attacker-controlled query parameter
    os.system("ping -c 1 " + host)  # => seeded bug: the REAL sink -- os.system with concatenated untrusted input
    return "ok"  # => co-04: return value irrelevant -- this file exists to be SCANNED, not actually executed
 
 
@app.route("/greet")  # => co-06: SINK 3 -- user input rendered as a TEMPLATE STRING, not template DATA
def greet() -> str:  # => co-06: real route handler
    name = request.args.get("name", "")  # => co-01: real, attacker-controlled query parameter
    template = Template("Hello, " + name + "!")  # => seeded bug: the REAL sink -- Template() built from untrusted input
    return template.render()  # => co-06: renders whatever the untrusted string CONTAINS as real template syntax
 
 
@app.route("/safe-lookup")  # => co-03: NOT a sink -- included so the scanner's precision is also verified
def safe_lookup() -> str:  # => co-07: real route handler, deliberately unproblematic
    item_id = request.args.get("id", "")  # => co-01: real, attacker-controlled input, but bound correctly below
    conn = sqlite3.connect(":memory:")  # => co-03: a real, if empty, in-memory DB for this scan target
    conn.execute("SELECT * FROM items WHERE id = ?", (item_id,))  # => co-03: a REAL, parameterized, safe query
    return "ok"  # => co-07: return value irrelevant -- this route exists to prove the scanner does NOT false-positive

# learning/code/ex-61-end-to-end-injection-audit/sample_fixed.py
"""Example 61: the SAME 4 routes, each seeded sink now fixed -- for the scanner to re-sweep and find zero (co-03, co-04, co-01)."""  # => co-01: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the fixes themselves
 
import sqlite3  # => co-03: the SAME stdlib DB driver -- only the QUERY construction changes, not the driver
import subprocess  # => co-04: real subprocess -- replaces os.system entirely, argv list instead of a shell string
 
from flask import Flask, render_template_string, request  # => co-06: render_template_string separates data from template
 
app = Flask(__name__)  # => co-01: one Flask app, hosting all 4 real, now-fixed routes
 
GREETING_TEMPLATE = "Hello, {{ name }}!"  # => co-06: a FIXED, developer-authored template string -- never built from input
 
 
@app.route("/search")  # => co-03: FIXED -- a real, parameterized query, no string building at all
def search() -> str:  # => co-03: real route handler
    term = request.args.get("q", "")  # => co-01: still real, attacker-controlled input
    conn = sqlite3.connect(":memory:")  # => co-03: the SAME in-memory DB target
    conn.execute("SELECT * FROM items WHERE name = ?", (term,))  # => co-03: fix -- term is bound as DATA, not SQL text
    return "ok"  # => co-03: return value irrelevant -- this file exists to be SCANNED
 
 
@app.route("/ping")  # => co-04: FIXED -- a real argv list, shell=False, no string concatenation at all
def ping() -> str:  # => co-04: real route handler
    host = request.args.get("host", "")  # => co-01: still real, attacker-controlled input
    subprocess.run(["ping", "-c", "1", host], shell=False)  # => co-04: fix -- host is ONE argv element, never shell text
    return "ok"  # => co-04: return value irrelevant -- this file exists to be SCANNED
 
 
@app.route("/greet")  # => co-06: FIXED -- user input passed as template CONTEXT DATA, never as the template itself
def greet() -> str:  # => co-06: real route handler
    name = request.args.get("name", "")  # => co-01: still real, attacker-controlled input
    return render_template_string(GREETING_TEMPLATE, name=name)  # => co-06: fix -- name is DATA, the template is FIXED
 
 
@app.route("/safe-lookup")  # => co-03: unchanged -- was already safe, still safe after this file's fixes
def safe_lookup() -> str:  # => co-07: real route handler, unchanged from sample_vulnerable.py
    item_id = request.args.get("id", "")  # => co-01: real, attacker-controlled input, bound correctly below
    conn = sqlite3.connect(":memory:")  # => co-03: a real, if empty, in-memory DB for this scan target
    conn.execute("SELECT * FROM items WHERE id = ?", (item_id,))  # => co-03: a REAL, parameterized, safe query
    return "ok"  # => co-07: return value irrelevant -- proves the scanner's precision holds across BOTH files

# learning/code/ex-61-end-to-end-injection-audit/scanner.py
"""Example 61: a real, hand-written AST scanner -- sweeps a .py file for concatenated-untrusted-input sinks (co-03, co-04, co-01)."""  # => co-01: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the AST-walking logic itself
 
import ast  # => co-01: Python's OWN stdlib parser -- the SAME tree the interpreter itself would build
from dataclasses import dataclass  # => co-01: a real, typed finding record -- not a loose tuple
 
 
@dataclass  # => co-01: one real finding per detected sink
class Finding:  # => co-03: the shape every scanner hit takes -- line, sink kind, and a real one-line reason
    line: int  # => co-01: the REAL source line number this finding came from, taken straight from the AST node
    sink_type: str  # => co-03: which category of sink this is -- "sql", "command", or "template"
    reason: str  # => co-04: a real, human-readable explanation of WHY this call was flagged
 
 
def _is_built_from_input(node: ast.expr) -> bool:  # => co-01: real, structural check -- NOT a value/data-flow guess
    # => co-01: a JoinedStr is a real f-string node; a BinOp(Add) on strings is real concatenation --
    # => BOTH shapes mean "this string's content depends on something computed at runtime, not a fixed literal"
    if isinstance(node, ast.JoinedStr):  # => co-03: e.g. f"... {term} ..." -- a REAL f-string AST node
        return True  # => co-03: an f-string is, by construction, never a fixed literal
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):  # => co-04: e.g. "a" + host -- REAL concatenation
        return True  # => co-04: string concatenation via + is the classic "built, not literal" shape
    return False  # => co-01: anything else (a plain Constant, a placeholder, ...) is NOT flagged by this check
 
 
class InjectionScanner(ast.NodeVisitor):  # => co-01: a real ast.NodeVisitor -- walks the WHOLE real parsed tree
    def __init__(self) -> None:  # => co-01: constructor -- starts with an empty, real findings list
        self.findings: list[Finding] = []  # => co-01: every REAL Finding this scan produces, in traversal order
 
    def visit_Call(self, node: ast.Call) -> None:  # => co-01: fires for EVERY real function/method call in the file
        self._check_sql_sink(node)  # => co-03: does THIS call look like a SQL-injection sink
        self._check_command_sink(node)  # => co-04: does THIS call look like a command-injection sink
        self._check_template_sink(node)  # => co-06: does THIS call look like a template-injection sink
        self.generic_visit(node)  # => co-01: continues walking into this call's own arguments -- real recursive descent
 
    def _check_sql_sink(self, node: ast.Call) -> None:  # => co-03: real check -- .execute()/.executescript() calls
        if not isinstance(node.func, ast.Attribute):  # => co-03: only method calls have a real `.attr` shape
            return  # => co-03: not a method call at all -- cannot be conn.execute(...)
        if node.func.attr not in {"execute", "executescript"}:  # => co-03: the REAL sqlite3 sink method names
            return  # => co-03: some other method entirely -- not a SQL sink candidate
        if node.args and _is_built_from_input(node.args[0]):  # => co-03: the FIRST real argument is the query text
            self.findings.append(  # => co-03: a REAL finding -- built, not bound, SQL text reached execute()
                Finding(node.lineno, "sql", "query string built via f-string/concatenation, not bound parameters")  # => co-03: the real record
            )  # => co-03: closes the real Finding(...) append
 
    def _check_command_sink(self, node: ast.Call) -> None:  # => co-04: real check -- os.system() / subprocess shell=True
        func = node.func  # => co-04: the real callable expression this Call node invokes
        if isinstance(func, ast.Attribute) and func.attr == "system":  # => co-04: matches os.system(...) by attr name
            if node.args and _is_built_from_input(node.args[0]):  # => co-04: the command string itself, built or not
                self.findings.append(  # => co-04: a REAL finding -- os.system with a concatenated command string
                    Finding(node.lineno, "command", "os.system() argument built via f-string/concatenation")  # => co-04: the real record
                )  # => co-04: closes the real Finding(...) append
        for keyword in node.keywords:  # => co-04: real subprocess.run/Popen calls signal danger via shell=True
            if keyword.arg == "shell" and isinstance(keyword.value, ast.Constant) and keyword.value.value is True:  # => co-04: a REAL, literal shell=True
                self.findings.append(  # => co-04: a REAL finding -- an explicit, real shell=True keyword argument
                    Finding(node.lineno, "command", "subprocess call passes shell=True")  # => co-04: the real record
                )  # => co-04: closes the real Finding(...) append
 
    def _check_template_sink(self, node: ast.Call) -> None:  # => co-06: real check -- Template(...) from a non-literal
        func = node.func  # => co-06: the real callable expression this Call node invokes
        name = func.id if isinstance(func, ast.Name) else (func.attr if isinstance(func, ast.Attribute) else None)  # => co-06: real callable name, either shape
        if name != "Template":  # => co-06: only matches a REAL call literally named/attributed "Template"
            return  # => co-06: any other callable name -- not a template-construction candidate
        if not node.args:  # => co-06: Template() with no positional argument at all -- nothing to inspect
            return  # => co-06: real guard against an IndexError on the check below
        first_arg = node.args[0]  # => co-06: the REAL argument this Template(...) call was constructed from
        if not isinstance(first_arg, ast.Constant):  # => co-06: a plain string LITERAL is the only safe shape here
            self.findings.append(  # => co-06: a REAL finding -- Template() built from something other than a literal
                Finding(node.lineno, "template", "Template() built from a non-literal (f-string/concat/variable)")  # => co-06: the real record
            )  # => co-06: closes the real Finding(...) append
 
 
def scan_file(path: str) -> list[Finding]:  # => co-01: parses a REAL file on disk and runs the REAL scanner over it
    with open(path) as f:  # => co-01: reads the real, on-disk source text
        source = f.read()  # => co-01: the exact real bytes/text this scan operates on
    tree = ast.parse(source, filename=path)  # => co-01: Python's OWN real parser -- the SAME grammar CPython itself uses
    scanner = InjectionScanner()  # => co-01: a fresh, real scanner instance for this one file
    scanner.visit(tree)  # => co-01: the REAL traversal -- every Call node in the file is actually visited
    return scanner.findings  # => co-01: every REAL finding this specific file produced, in source order

# learning/code/ex-61-end-to-end-injection-audit/run_audit.py
"""Example 61: sweeps the real vulnerable sample app, then the real fixed one -- 3 sinks down to zero (co-03, co-04, co-01)."""  # => co-01: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the sweep logic itself
 
import os  # => co-01: builds real, absolute paths to the two real sample files this scan targets
 
from scanner import scan_file  # => co-01: the REAL AST scanner this example builds and exercises
 
HERE = os.path.dirname(__file__)  # => co-01: this example's own real directory -- both sample files live alongside it
 
 
def main() -> None:  # => co-01: runs the real scanner against BOTH real sample files, before and after the fix
    print("=== VULNERABLE: scanning sample_vulnerable.py ===")  # => labels section
    vulnerable_findings = scan_file(os.path.join(HERE, "sample_vulnerable.py"))  # => co-01: a REAL scan of a REAL file
    for finding in vulnerable_findings:  # => co-01: every REAL finding this scan actually produced
        print(f"  line {finding.line}: [{finding.sink_type}] {finding.reason}")  # => co-01: real, per-finding detail
    assert len(vulnerable_findings) == 3  # => co-01: proves the scanner found ALL THREE seeded sinks, no more, no less
    assert {f.sink_type for f in vulnerable_findings} == {"sql", "command", "template"}  # => co-01: one of each real kind
 
    print("\n=== FIXED: scanning sample_fixed.py (SAME routes, sinks closed) ===")  # => labels section
    fixed_findings = scan_file(os.path.join(HERE, "sample_fixed.py"))  # => co-01: a REAL scan of the REAL fixed file
    print(f"  findings: {fixed_findings}")  # => co-01: the real, empty list this scan actually produced
    assert fixed_findings == []  # => co-01: proves the identical scanner, run against the fixed file, finds NOTHING
 
    print("\n=== re-verifying /safe-lookup was NEVER flagged in either file (scanner precision) ===")  # => labels section
    safe_route_source = open(os.path.join(HERE, "sample_vulnerable.py")).readlines()  # => co-07: the REAL file, line by line
    flagged_lines = {f.line for f in vulnerable_findings}  # => co-07: every REAL line number the scan actually flagged
    safe_lines = {i + 1 for i, line in enumerate(safe_route_source) if "safe_lookup" in line}  # => co-07: real anchor line
    safe_body_start = min(safe_lines)  # => co-07: where the real, deliberately-safe route begins in the real file
    assert not any(line >= safe_body_start for line in flagged_lines)  # => co-07: NOTHING inside safe_lookup was flagged
 
 
if __name__ == "__main__":  # => co-01: only runs when launched directly, e.g. `python3 run_audit.py`
    main()  # => co-01: runs the full real before/after sweep and verifies every real invariant

=== VULNERABLE: scanning sample_vulnerable.py ===
  line 20: [sql] query string built via f-string/concatenation, not bound parameters
  line 27: [command] os.system() argument built via f-string/concatenation
  line 34: [template] Template() built from a non-literal (f-string/concat/variable)
 
=== FIXED: scanning sample_fixed.py (SAME routes, sinks closed) ===
  findings: []
 
=== re-verifying /safe-lookup was NEVER flagged in either file (scanner precision) ===

# learning/code/ex-62-ssti-server-side-template-injection/app.py
"""Example 62: a live Flask app -- user input rendered AS a template executes code, then as DATA it stays inert (co-06, co-25)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the SSTI issue itself
 
from flask import Flask, render_template_string, request  # => co-06: render_template_string keeps data OUT of the template
from jinja2 import Template  # => co-06: Jinja2's Template class -- the REAL vulnerable sink this example seeds
 
app = Flask(__name__)  # => co-06: one Flask app, hosting both the vulnerable and fixed greeting routes
FIXED_TEMPLATE = "Hello, {{ name }}!"  # => co-06: a FIXED, developer-authored template -- never built from input
 
 
@app.route("/legacy/greet")  # => co-06: VULNERABLE -- user input becomes the TEMPLATE ITSELF, not template data
def legacy_greet() -> str:  # => co-06: real route handler
    name = request.args.get("name", "")  # => co-01: attacker-controlled -- whatever string the caller sends
    # seeded bug: name is concatenated into the TEMPLATE SOURCE, then that source is COMPILED and RUN
    template = Template("Hello, " + name + "!")  # => co-06: the untrusted string becomes real Jinja2 template syntax
    return template.render()  # => co-06: Jinja2 actually EXECUTES whatever expression syntax `name` contains
 
 
@app.route("/secure/greet")  # => co-06: FIXED -- user input is template CONTEXT DATA, the template itself never changes
def secure_greet() -> str:  # => co-06: real route handler
    name = request.args.get("name", "")  # => co-01: the SAME shape of attacker-controlled input
    return render_template_string(FIXED_TEMPLATE, name=name)  # => co-06: fix -- name is substituted as a VALUE, not parsed
 
 
if __name__ == "__main__":  # => co-06: only runs when launched directly, e.g. `python3 app.py &`
    app.run(host="127.0.0.1", port=5062)  # => co-06: localhost-only, fixed port -- exploit_and_fix.py targets this

# learning/code/ex-62-ssti-server-side-template-injection/exploit_and_fix.py
"""Example 62: drives the live app.py -- {{7*7}} really evaluates to 49, then the SAME payload stays inert text (co-06, co-25)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the exploit itself
 
import requests  # => co-06: real HTTP client -- every request below hits the live app.py process
 
BASE_URL = "http://127.0.0.1:5062"  # => co-06: matches app.py's app.run(port=5062) exactly
PAYLOAD = "{{7*7}}"  # => co-06: a REAL, minimal Jinja2 expression -- the classic SSTI proof-of-concept probe
 
 
def main() -> None:  # => co-06: sends the SAME payload to the vulnerable route, then the fixed route
    print("=== VULNERABLE: {{7*7}} is EXECUTED as real template code ===")  # => labels section
    vulnerable = requests.get(f"{BASE_URL}/legacy/greet", params={"name": PAYLOAD}, timeout=5)  # => co-06: a real HTTP GET
    print(f"status={vulnerable.status_code} body={vulnerable.text!r}")  # => co-06: real response, straight off the wire
    assert vulnerable.text == "Hello, 49!"  # => co-06: proves Jinja2 REALLY evaluated 7*7 -- this is real code execution
    # => co-06: a real attacker would use the SAME sink to run something far more dangerous
    # => than arithmetic -- e.g. Jinja2's `{{ self.__init__.__globals__... }}` sandbox-escape chains
 
    print("\n=== FIXED: the IDENTICAL payload, rendered as DATA, stays inert ===")  # => labels section
    fixed = requests.get(f"{BASE_URL}/secure/greet", params={"name": PAYLOAD}, timeout=5)  # => co-06: real HTTP GET
    print(f"status={fixed.status_code} body={fixed.text!r}")  # => co-06: real response, the literal string this time
    assert fixed.text == "Hello, {{7*7}}!"  # => co-06: proves the SAME text is now inert -- printed, never evaluated
 
 
if __name__ == "__main__":  # => co-06: only runs when launched directly, e.g. `python3 exploit_and_fix.py`
    main()  # => co-06: runs both halves against the live server -- real code execution, then a real inert render

=== VULNERABLE: {{7*7}} is EXECUTED as real template code ===
status=200 body='Hello, 49!'
 
=== FIXED: the IDENTICAL payload, rendered as DATA, stays inert ===
status=200 body='Hello, {{7*7}}!'

# learning/code/ex-63-argon2-parameter-tuning/tune_argon2.py
"""Example 63: real argon2id timing -- a weak baseline, the topic's floor params, then a live-tuned target (co-09)."""  # => co-09: module purpose
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the timing measurement itself
 
import time  # => co-09: real wall-clock timestamps -- every duration below is genuinely measured, not estimated
 
from argon2 import PasswordHasher  # => co-09: argon2-cffi 25.1.0 pinned -- the real hasher this example tunes
 
PASSWORD = "correct-horse-battery-staple"  # => co-09: one fixed, real password hashed identically across all trials
TARGET_MS = 250.0  # => co-09: this topic's documented budget -- OWASP's argon2id guidance, ~250ms per hash
 
 
def measure_hash_time(memory_cost: int, time_cost: int, parallelism: int) -> float:  # => co-09: one REAL, timed hash
    hasher = PasswordHasher(memory_cost=memory_cost, time_cost=time_cost, parallelism=parallelism)  # => co-09: real params
    start = time.perf_counter()  # => co-09: a real, high-resolution wall-clock start
    hasher.hash(PASSWORD)  # => co-09: the REAL, actual argon2id hash computation -- not simulated or estimated
    return (time.perf_counter() - start) * 1000  # => co-09: real elapsed milliseconds for THIS specific call
 
 
def main() -> None:  # => co-09: measures a weak baseline, the topic's floor, then LIVE-tunes toward the target budget
    print("=== weak baseline (m=8 KiB, t=1, p=1) -- what an under-tuned config looks like ===")  # => labels section
    weak_ms = measure_hash_time(memory_cost=8, time_cost=1, parallelism=1)  # => co-09: a REAL, deliberately weak measurement
    print(f"weak: {weak_ms:.1f}ms")  # => co-09: real, measured milliseconds -- expect well under 10ms on any machine
 
    print("\n=== this topic's documented floor (m=19456 KiB, t=2, p=1) -- OWASP's min-tier params ===")  # => labels
    baseline_ms = measure_hash_time(memory_cost=19456, time_cost=2, parallelism=1)  # => co-09: the REAL floor from co-09
    print(f"baseline: {baseline_ms:.1f}ms")  # => co-09: real, measured milliseconds at the documented minimum tier
 
    print(f"\n=== live-tuning time_cost toward the ~{TARGET_MS:.0f}ms budget on THIS machine ===")  # => labels section
    memory_cost = 19456  # => co-09: keeps memory_cost fixed at the floor -- only time_cost is tuned here
    time_cost = 2  # => co-09: starts the search from the SAME floor time_cost measured above
    tuned_ms = baseline_ms  # => co-09: real starting point for the search loop below
    while tuned_ms < TARGET_MS and time_cost < 200:  # => co-09: a real, live search -- stops once the budget is reached
        time_cost += 1  # => co-09: increases cost by one real unit per iteration -- a simple, real linear search
        tuned_ms = measure_hash_time(memory_cost=memory_cost, time_cost=time_cost, parallelism=1)  # => co-09: a REAL trial
        print(f"  trying time_cost={time_cost}: {tuned_ms:.1f}ms")  # => co-09: real, per-trial measurement, live search
 
    print(f"\ntuned params: memory_cost={memory_cost}, time_cost={time_cost}, parallelism=1 -> {tuned_ms:.1f}ms")  # => co-09: the real, final tuned result
    assert tuned_ms >= baseline_ms  # => co-09: proves the tuned config is REAL work, at least as slow as the floor
    assert tuned_ms > weak_ms * 5  # => co-09: proves the tuned config is MEANINGFULLY slower than the weak baseline
    assert tuned_ms < TARGET_MS * 2  # => co-09: a generous sanity bound -- tuning stopped reasonably near the budget
 
 
if __name__ == "__main__":  # => co-09: only runs when launched directly, e.g. `python3 tune_argon2.py`
    main()  # => co-09: runs all real measurements and the real live-tuning search, printing every real trial

=== weak baseline (m=8 KiB, t=1, p=1) -- what an under-tuned config looks like ===
weak: 0.2ms
 
=== this topic's documented floor (m=19456 KiB, t=2, p=1) -- OWASP's min-tier params ===
baseline: 170.9ms
 
=== live-tuning time_cost toward the ~250ms budget on THIS machine ===
  trying time_cost=3: 298.8ms
 
tuned params: memory_cost=19456, time_cost=3, parallelism=1 -> 298.8ms

# learning/code/ex-64-password-upgrade-on-login/upgrade_on_login.py
"""Example 64: a real legacy bcrypt hash transparently upgrades to argon2id on the FIRST successful login (co-09, co-10)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the upgrade logic itself
 
import bcrypt  # => co-09: bcrypt 5.0.0 pinned -- the REAL legacy hasher this user's stored hash starts as
from argon2 import PasswordHasher  # => co-09: argon2-cffi 25.1.0 pinned -- the REAL target hasher this example upgrades to
from argon2.exceptions import VerifyMismatchError  # => co-09: the REAL exception argon2's verify() raises on a bad password
 
argon2_hasher = PasswordHasher()  # => co-09: one real, shared argon2id hasher -- default (OWASP min-tier) parameters
USERS: dict[str, str] = {"alice": bcrypt.hashpw(b"correct-horse", bcrypt.gensalt()).decode()}  # => co-09: a REAL bcrypt hash
 
 
def login(username: str, password: str) -> bool:  # => co-09: the ONE real login function -- verify, then maybe upgrade
    stored_hash = USERS.get(username)  # => co-09: the REAL, current on-file hash for this user -- bcrypt OR argon2id
    if stored_hash is None:  # => co-09: a real guard -- no such user at all
        return False  # => co-09: a real, unambiguous failure for an unknown username
 
    if stored_hash.startswith("$2b$") or stored_hash.startswith("$2a$"):  # => co-09: the REAL bcrypt PHC-string prefix
        is_valid = bcrypt.checkpw(password.encode(), stored_hash.encode())  # => co-09: a real bcrypt verification
        if is_valid:  # => co-09: ONLY on a real, successful login does the upgrade ever happen
            # => co-09: the fix -- transparently rehash with argon2id and REPLACE the stored value,
            # => using the password the user JUST proved they know, right here, while it's in memory
            USERS[username] = argon2_hasher.hash(password)  # => co-09: a REAL argon2id hash now REPLACES the bcrypt one
        return is_valid  # => co-09: the real login outcome -- unaffected by whether an upgrade happened
 
    # => co-09: stored_hash is ALREADY argon2id -- the normal, no-upgrade-needed path
    try:  # => co-09: argon2's verify() raises on a mismatch rather than returning False
        argon2_hasher.verify(stored_hash, password)  # => co-09: a real argon2id verification
        return True  # => co-09: a real, successful login -- no upgrade needed, already on the target hasher
    except VerifyMismatchError:  # => co-09: the REAL exception type argon2-cffi raises for a wrong password
        return False  # => co-09: a real, failed login
 
 
def main() -> None:  # => co-09: proves the hash is bcrypt-shaped BEFORE login, argon2id-shaped immediately AFTER
    print(f"stored hash BEFORE login: {USERS['alice'][:12]}...")  # => co-09: real, truncated -- starts with $2b$
    assert USERS["alice"].startswith("$2b$")  # => co-09: proves alice's stored hash really is legacy bcrypt right now
 
    print("\n=== a real login with the CORRECT password ===")  # => labels section
    ok = login("alice", "correct-horse")  # => co-09: a REAL login call -- the SAME password bcrypt.hashpw used above
    print(f"login result: {ok}")  # => co-09: real, computed outcome
    assert ok is True  # => co-09: proves the login itself really succeeded, via the bcrypt verification path
 
    print(f"\nstored hash AFTER login: {USERS['alice'][:20]}...")  # => co-09: real, truncated -- now starts with $argon2id$
    assert USERS["alice"].startswith("$argon2id$")  # => co-09: proves the transparent upgrade REALLY happened, in-place
 
    print("\n=== a SECOND real login -- now verified via argon2id, no further upgrade needed ===")  # => labels section
    ok2 = login("alice", "correct-horse")  # => co-09: a REAL second login -- this time against the NEW argon2id hash
    print(f"login result: {ok2}")  # => co-09: real, computed outcome
    assert ok2 is True  # => co-09: proves the upgraded hash itself really authenticates the same real password
    assert USERS["alice"].startswith("$argon2id$")  # => co-09: proves it stays argon2id -- no re-upgrade, no re-hash churn
 
    print("\n=== a real login with the WRONG password never upgrades anything ===")  # => labels section
    wrong_hash_before = USERS["alice"]  # => co-09: a real snapshot of the current stored hash, before the bad attempt
    ok3 = login("alice", "totally-wrong-password")  # => co-09: a REAL, deliberately wrong login attempt
    print(f"login result: {ok3}")  # => co-09: real, computed outcome -- False
    assert ok3 is False and USERS["alice"] == wrong_hash_before  # => co-09: proves a failed login NEVER mutates the hash
 
 
if __name__ == "__main__":  # => co-09: only runs when launched directly, e.g. `python3 upgrade_on_login.py`
    main()  # => co-09: runs all three real login scenarios and verifies the real bcrypt -> argon2id transition

stored hash BEFORE login: $2b$12$xfKQh...
 
=== a real login with the CORRECT password ===
login result: True
 
stored hash AFTER login: $argon2id$v=19$m=655...
 
=== a SECOND real login -- now verified via argon2id, no further upgrade needed ===
login result: True
 
=== a real login with the WRONG password never upgrades anything ===
login result: False

# learning/code/ex-65-token-revocation-strategy/app.py
"""Example 65: a live Flask app -- short-lived JWT access tokens, a longer-lived refresh token, and a REAL revocation list (co-12, co-14)."""
 
from __future__ import annotations  # => DD-39 hygiene -- unrelated to the revocation logic itself
 
import datetime  # => co-14: real UTC timestamps for token expiry
import secrets  # => co-12: cryptographically random refresh tokens -- opaque, never a JWT themselves
 
import jwt  # => co-14: PyJWT 2.13.0 pinned -- CVE-2026-32597 (crit header) fixed at >=2.12.0, unaffected here
from flask import Flask, jsonify, request  # => co-12: request.get_json reads the caller-supplied refresh token
 
app = Flask(__name__)  # => co-12: one Flask app, hosting login, refresh, and the real revocation endpoint
JWT_SECRET = "ex65-hmac-secret-at-least-32-bytes-long!!"  # => co-14: real HMAC secret, RFC 7518-length-compliant
# => co-12: token -> {"user": ..., "revoked": bool} -- a REAL, in-memory revocation table (a SQLite table
# => would use the identical shape in production; an in-memory dict keeps this example self-contained)
REFRESH_TOKENS: dict[str, dict[str, object]] = {}
 
 
def make_access_token(user: str) -> str:  # => co-14: mints a REAL, short-lived signed JWT
    now = datetime.datetime.now(datetime.timezone.utc)  # => co-14: real wall-clock UTC time, taken once per call
    claims = {"user": user, "exp": now + datetime.timedelta(minutes=5)}  # => co-14: a real, SHORT 5-minute lifetime
    return jwt.encode(claims, JWT_SECRET, algorithm="HS256")  # => co-14: a real, correctly-signed HS256 token
 
 
@app.route("/login", methods=["POST"])  # => co-12: issues a REAL access token AND a REAL, tracked refresh token
def login() -> object:  # => co-12: returns a Flask Response object with both real tokens
    user = request.get_json(force=True).get("username", "")  # => co-12: the real submitted username
    refresh_token = secrets.token_urlsafe(24)  # => co-12: a REAL, opaque, unpredictable refresh token
    REFRESH_TOKENS[refresh_token] = {"user": user, "revoked": False}  # => co-12: real, fresh entry in the revocation table
    access_token = make_access_token(user)  # => co-14: a real, short-lived JWT for this SAME login
    return jsonify({"access_token": access_token, "refresh_token": refresh_token})  # => co-12: both real tokens returned
 
 
@app.route("/refresh", methods=["POST"])  # => co-12: mints a NEW access token FROM a real, still-valid refresh token
def refresh() -> object:  # => co-12: returns a Flask Response object -- a new access token, or a real rejection
    refresh_token = request.get_json(force=True).get("refresh_token", "")  # => co-01: attacker-adjacent -- caller-supplied
    entry = REFRESH_TOKENS.get(refresh_token)  # => co-12: the REAL, current state of this specific refresh token
    if entry is None or entry["revoked"]:  # => co-12: the REAL check -- unknown OR explicitly revoked, both rejected
        return jsonify({"error": "invalid or revoked refresh token"}), 401  # => co-12: a real 401 for either case
    new_access_token = make_access_token(str(entry["user"]))  # => co-14: a REAL, freshly minted, short-lived access token
    return jsonify({"access_token": new_access_token})  # => co-12: only the ACCESS token rotates -- refresh stays the same
 
 
@app.route("/admin/revoke", methods=["POST"])  # => co-12: the REAL revocation operation -- e.g. on logout or compromise
def revoke() -> object:  # => co-12: returns a Flask Response object -- confirms the real revocation
    refresh_token = request.get_json(force=True).get("refresh_token", "")  # => co-12: WHICH real token to kill
    entry = REFRESH_TOKENS.get(refresh_token)  # => co-12: the real, current entry for this token, if any
    if entry is None:  # => co-12: a real guard -- can't revoke a token that was never issued
        return jsonify({"error": "unknown refresh token"}), 404  # => co-12: a real 404 for a bogus revoke request
    entry["revoked"] = True  # => co-12: the REAL revocation -- flips server-side state, the token itself is unchanged
    return jsonify({"status": "revoked"})  # => co-12: confirms the real, immediate effect on the NEXT /refresh call
 
 
if __name__ == "__main__":  # => co-12: only runs when launched directly, e.g. `python3 app.py &`
    app.run(host="127.0.0.1", port=5065)  # => co-12: localhost-only, fixed port -- exploit_and_fix.py targets this