How to Encrypt PII in Python and Keep It Searchable With Blind Indexes

Almost every application stores personal data such as email addresses and Social Security numbers (SSNs). If those values sit in a database column as plain text, anyone who copies the database file, a backup or a SQL dump can read every customer’s details. Encrypting the columns fixes that, and it breaks something your support team does all day: looking up a customer by email address.

In this tutorial you will build a small Python and SQLite lab that solves both problems. You will encrypt email addresses, SSNs and last names with AES-256-GCM, find a customer again by email or SSN using a blind index (a keyed hash stored next to the ciphertext), and measure what each design leaks, because every searchable encryption scheme leaks something and you should know exactly what. By the end you will have a 131-line module, ten passing tests and a clear rule for how many bits of index each column deserves. Plan on about an hour.

The idea came from freeCodeCamp’s concept article How to Encrypt PII in Data Pipelines While Keeping It Searchable, which compares randomized encryption, deterministic encryption, HMAC and tokenization in diagrams but contains no runnable code. This tutorial turns those ideas into code you can run, measures the leakage that article warns about, and checks the design against the documentation of two libraries built on the same technique: CipherSweet and the AWS Database Encryption SDK.

What you will build, and who you are defending against

The lab keeps customer data in one SQLite table. Email and SSN are searchable: each has an encrypted column (suffix _ct, for ciphertext) and an integer blind-index column (suffix _bi). Last name is encrypted but deliberately not searchable. Two columns, plan and signup month, stay in plain text because they are not personal data. A lookup always follows the same four moves: normalize the search term, compute its blind index, fetch the few rows that share that index value, then decrypt those rows and keep only the real matches.

Three attackers appear in the steps. The thief copies the database file but has no keys, for example through a stolen backup or a SQL injection that reads tables. The writer can change cells in the database but has no keys, for example a bad migration or a curious administrator. The key thief holds the database and the blind-index key but not the encryption key. Attackers who run code inside your application or obtain every key are out of scope, because no database-level scheme can help once they have the keys.

Six ideas in plain language

You will meet each of these again in a concrete step, so a short definition is enough for now.

Randomized encryption means the same plaintext encrypts to a different ciphertext every time. AES-GCM does this with a nonce, a number used once, which you generate fresh for every value. Deterministic encryption is the opposite: the same plaintext under the same key always produces the same ciphertext, which makes equality search easy and leaks which rows hold equal values.

An HMAC is a keyed hash: a function that turns any input and a secret key into a fixed-size fingerprint that nobody can compute without the key. HKDF derives many independent keys from one master key by mixing in a different context string for each. Associated data is extra information, such as a row number, that the cipher authenticates without encrypting, so changing it makes decryption fail. A blind index is a truncated HMAC of the normalized plaintext, stored next to the ciphertext and used as a lookup key.

Prerequisites and setup

You need Python 3 with pip, a terminal and basic SQL. No cryptography background is assumed. I ran every step on Windows 11 with the versions below and did not test other versions. SQLite comes with Python, and the only packages you install are cryptography and pytest.

One honest warning before you start. The cryptography project labels the module that holds AES-GCM and AES-SIV a “Hazardous Materials” module and says you should only use it if you are “100% absolutely sure that you know what you’re doing”. This lab uses it to teach the design. For production data, prefer a vetted high-level library, and see the last section for options.

Create a folder for the lab, open a terminal in it and run these commands (on macOS or Linux, activate the environment with source labvenv/bin/activate instead). Keep every file from this tutorial in that one folder, and run each command from it with the environment active.

python -m venv labvenv
labvenv\Scripts\activate
python -m pip install cryptography pytest
python -c "import sys, sqlite3, cryptography, pytest; print(sys.version.split()[0], cryptography.__version__, sqlite3.sqlite_version, pytest.__version__)"

The last command prints the versions of Python, cryptography, SQLite and pytest. This is what I got:

3.13.14 50.0.2 3.50.4 9.1.1

In the lab every script creates a fresh random master key with os.urandom(32), so each run starts clean. In production the master key lives in a key management service or hardware security module, never in source code and never in the same place as the data.

Step 1: Create the lab data and see what a stolen file reveals

Save the first file as data.py. It generates 20,000 fake customers with a seeded random generator so your numbers match this article. The surname column is deliberately skewed, so that “Smith” is far more common than “Jimenez”, because real name columns are skewed too and that skew is what makes the attacks in steps 3 and 6 work. The order of the surname list doubles as the public frequency table an attacker would use. Everything here is generated; nobody’s real data is involved, and the SSN-shaped values are random.

# data.py
"""Synthetic customers for the lab. Everything is generated; nobody's real data is involved."""
import random

# 100 common US surnames, most common first. The ORDER is the public frequency table an attacker would use.
SURNAMES = [
    "smith", "johnson", "williams", "brown", "jones", "garcia", "miller", "davis", "rodriguez", "martinez",
    "hernandez", "lopez", "gonzalez", "wilson", "anderson", "thomas", "taylor", "moore", "jackson", "martin",
    "lee", "perez", "thompson", "white", "harris", "sanchez", "clark", "ramirez", "lewis", "robinson",
    "walker", "young", "allen", "king", "wright", "scott", "torres", "nguyen", "hill", "flores",
    "green", "adams", "nelson", "baker", "hall", "rivera", "campbell", "mitchell", "carter", "roberts",
    "gomez", "phillips", "evans", "turner", "diaz", "parker", "cruz", "edwards", "collins", "reyes",
    "stewart", "morris", "morales", "murphy", "cook", "rogers", "gutierrez", "ortiz", "morgan", "cooper",
    "peterson", "bailey", "reed", "kelly", "howard", "ramos", "kim", "cox", "ward", "richardson",
    "watson", "brooks", "chavez", "wood", "james", "bennett", "gray", "mendoza", "ruiz", "hughes",
    "price", "alvarez", "castillo", "sanders", "patel", "myers", "long", "ross", "foster", "jimenez",
]
PUBLIC_RANKING = [name.title() for name in SURNAMES]

FIRST_NAMES = [
    "james", "mary", "john", "patricia", "robert", "jennifer", "michael", "linda", "william", "elizabeth",
    "david", "barbara", "richard", "susan", "joseph", "jessica", "thomas", "sarah", "charles", "karen",
    "chris", "nancy", "daniel", "lisa", "matthew", "betty", "anthony", "margaret", "mark", "sandra",
    "donald", "ashley", "steven", "kimberly", "paul", "emily", "andrew", "donna", "joshua", "michelle",
]


def zipf(n, s=1.0):
    weights = [1.0 / rank**s for rank in range(1, n + 1)]
    total = sum(weights)
    return [w / total for w in weights]


def make_customers(n=20_000, seed=7):
    rng = random.Random(seed)
    weights = zipf(len(SURNAMES))
    used_ssn = set()
    rows = []
    for i in range(1, n + 1):
        first = rng.choice(FIRST_NAMES)
        last = rng.choices(SURNAMES, weights)[0]
        while True:
            ssn = f"9{rng.randrange(100):02d}-{rng.randrange(100):02d}-{rng.randrange(10_000):04d}"
            if ssn not in used_ssn:
                used_ssn.add(ssn)
                break
        rows.append({
            "id": i,
            "email": f"{first}.{last}{i}@example.test",
            "ssn": ssn,
            "last_name": last.title(),
            "plan": rng.choices(["free", "pro", "team"], [70, 22, 8])[0],
            "signup_month": f"2026-{rng.randrange(1, 10):02d}",
        })
    return rows

Now save step01_dataset.py. It builds an ordinary, unencrypted SQLite file, shows how many distinct values each column holds, then plays the thief: it reads the raw bytes of the database file and searches them.

# step01_dataset.py
import os
import re
import sqlite3
from collections import Counter

from data import make_customers

rows = make_customers()

if os.path.exists("plain.db"):
    os.remove("plain.db")
conn = sqlite3.connect("plain.db")
conn.execute(
    "CREATE TABLE customers (id INTEGER PRIMARY KEY, email TEXT, ssn TEXT, last_name TEXT, plan TEXT, signup_month TEXT)"
)
conn.executemany("INSERT INTO customers VALUES (:id, :email, :ssn, :last_name, :plan, :signup_month)", rows)
conn.commit()

print("rows:", conn.execute("SELECT COUNT(*) FROM customers").fetchone()[0])
for row in conn.execute("SELECT * FROM customers LIMIT 3"):
    print(row)

print()
for column in ("email", "ssn", "last_name", "plan", "signup_month"):
    distinct = conn.execute(f"SELECT COUNT(DISTINCT {column}) FROM customers").fetchone()[0]
    print(f"{column:<13} distinct values: {distinct:,}")

print("\nmost common last names:", Counter(r["last_name"] for r in rows).most_common(3))

raw = open("plain.db", "rb").read()
found = re.findall(rb"[a-z]+\.[a-z]+\d+@example\.test", raw)
print(f"\nsomeone copies plain.db ({len(raw):,} bytes) and reads {len(set(found)):,} distinct email addresses straight out of it")
print(f"matches in the raw bytes: {len(found):,} ({len(found) - len(set(found))} addresses are stored twice)")

python step01_dataset.py

rows: 20000
(1, '[email protected]', '950-83-0791', 'Kim', 'free', '2026-09')
(2, '[email protected]', '907-64-3517', 'Brown', 'free', '2026-07')
(3, '[email protected]', '911-70-6955', 'Smith', 'free', '2026-02')

email         distinct values: 20,000
ssn           distinct values: 20,000
last_name     distinct values: 100
plan          distinct values: 3
signup_month  distinct values: 9

most common last names: [('Smith', 3923), ('Johnson', 2000), ('Williams', 1265)]

someone copies plain.db (1,433,600 bytes) and reads 20,000 distinct email addresses straight out of it
matches in the raw bytes: 20,018 (18 addresses are stored twice)

Look at the distinct counts. Cardinality is the number of distinct values in a column. Email and SSN have 20,000 distinct values, one per customer, so they identify people. Last name has only 100, plan has 3 and signup month has 9, so they describe groups. Keep that difference in mind, because it decides which columns can safely be searchable.

Then look at the last two lines. The thief needed no password and no SQL, only the file, and read 20,000 distinct email addresses straight out of the bytes. The 18 addresses stored twice are a curiosity worth a sentence: I checked, and the second copies sit on pages 2 and 3 of the file, where page 2 is the table’s root page, which fits SQLite leaving old row bytes behind when it moved the first page’s rows to a new page. Whatever the cause, nothing here needed a key.

Step 2: Encrypt with AES-GCM and meet the search problem

AES-GCM is an authenticated encryption mode. Besides hiding the data it produces a 16-byte tag, and decryption fails loudly if anyone changed a single bit of the ciphertext. You give it a key, a nonce and the data. Our stored value will be the 12-byte nonce followed by the ciphertext and tag, so each value carries everything but the key.

Save step02_randomized.py. It encrypts every email with a fresh random nonce, shows three encryptions of the same address, tries the obvious search, then tries the brute-force alternative of decrypting everything.

# step02_randomized.py
import os
import sqlite3
import statistics
import time

from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from data import make_customers

aes = AESGCM(AESGCM.generate_key(bit_length=256))   # one key, for this first experiment only


def seal(text):
    nonce = os.urandom(12)                           # a fresh random nonce for every value
    return nonce + aes.encrypt(nonce, text.encode(), None)


def unseal(blob):
    return aes.decrypt(blob[:12], blob[12:], None).decode()


rows = make_customers()
if os.path.exists("randomized.db"):
    os.remove("randomized.db")
conn = sqlite3.connect("randomized.db")
conn.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, email_ct BLOB)")
conn.executemany("INSERT INTO customers VALUES (?, ?)", [(r["id"], seal(r["email"])) for r in rows])
conn.commit()

target = rows[4241]["email"]
print("the same email encrypted three times:")
for _ in range(3):
    print("  ", seal(target).hex()[:48], "...")

hits = conn.execute("SELECT id FROM customers WHERE email_ct = ?", (seal(target),)).fetchall()
print("\nWHERE email_ct = <encrypt(target)> finds", len(hits), "rows")


def decrypt_everything(needle):
    return [i for i, blob in conn.execute("SELECT id, email_ct FROM customers") if unseal(blob) == needle]


times = []
for _ in range(5):
    start = time.perf_counter()
    found = decrypt_everything(target)
    times.append(time.perf_counter() - start)
print(f"decrypt-everything search finds {found} in {statistics.median(times) * 1000:.0f} ms (median of 5, {len(rows):,} rows)")

raw = open("randomized.db", "rb").read()
print(f"\nsomeone copies randomized.db: '@example.test' appears {raw.count(b'@example.test')} times")

python step02_randomized.py

the same email encrypted three times:
   f97eafb9e5339f416d5260ca24bb56deb1d74d1f633c70f1 ...
   fae419bf56a2d32a44e176bbefef04e595028133f89b2dc1 ...
   ddcee65913ff0194b313668b23f4d5d364bda0fda4d36fe7 ...

WHERE email_ct = <encrypt(target)> finds 0 rows
decrypt-everything search finds [4242] in 16 ms (median of 5, 20,000 rows)

someone copies randomized.db: '@example.test' appears 0 times

Your hex digits will differ from mine, and that is the point. The same email produced three unrelated ciphertexts, so the query WHERE email_ct = ? found 0 rows however carefully you encrypted the search term. That randomness is a feature. It stops anyone looking at the column from seeing which customers share a value, and the stolen-file check at the bottom shows that the address text is gone from the file.

The fallback is to decrypt every row and compare, and at 20,000 rows it is quick: 16 ms in this run. Two things make it a poor foundation anyway. The work grows in proportion to the table, because every lookup decrypts every row. And every lookup puts every customer’s email address into the memory of whichever service runs the query, which defeats much of the reason you encrypted the column.

Common mistake: encrypting the search term again and comparing ciphertexts. With a random nonce that can never match, as the 0 above shows.

Step 3: See why deterministic encryption is not the shortcut it looks like

The idea most people have next is to remove the randomness so equal values encrypt equally. There are two ways to do that, one dangerous and one legitimate, and the lab tries both. Then it plays the thief against the legitimate one.

The fixed-nonce trap

One way is to pin the nonce to a constant. The cryptography documentation says “NEVER REUSE A NONCE with a key” and explains why: “Reuse of a nonce with a given key compromises the security of any message with that nonce and key pair.” The script shows the simplest consequence. GCM turns the key and nonce into a keystream that is XORed with the plaintext, so two messages encrypted under the same nonce share the keystream, and XORing the two ciphertexts cancels it out.

A deterministic mode built on purpose: AES-SIV

The legitimate way is a construction designed to be deterministic. AES-SIV is defined in RFC 5297, whose abstract says: “Depending on how it is used, SIV achieves either the goal of deterministic authenticated encryption or the goal of nonce-based, misuse-resistant authenticated encryption.” The cryptography library takes a list of associated-data strings, and the script passes the column name so that the same value in two columns encrypts differently (I checked: the same surname with a different column name produced different bytes).

What a thief can still do: count

A deterministic ciphertext is a perfect label for a value. The thief cannot decrypt it, but can count how often each label occurs. The script runs a simple rank matching attack: the most common ciphertext is probably the most common surname, the second most common is probably the second most common surname, and so on down the public frequency table.

# step03_deterministic.py
from collections import Counter

from cryptography.hazmat.primitives.ciphers.aead import AESGCM, AESSIV

from data import PUBLIC_RANKING, make_customers

# 1. The tempting shortcut: pin the nonce so AES-GCM becomes deterministic.
key = AESGCM.generate_key(bit_length=256)
fixed = bytes(12)                                   # twelve zero bytes, used for every value
c1 = AESGCM(key).encrypt(fixed, b"smith", None)
c2 = AESGCM(key).encrypt(fixed, b"jones", None)
print("same plaintext, same ciphertext:", AESGCM(key).encrypt(fixed, b"smith", None) == c1)


def xor(a, b):
    return bytes(x ^ y for x, y in zip(a, b))


leak = xor(c1[:5], c2[:5])                          # the first 5 bytes are the encrypted text, the tag follows
print("c1 xor c2           :", leak.hex())
print("'smith' xor 'jones' :", xor(b"smith", b"jones").hex())
print("attacker who knows 'smith' recovers:", xor(leak, b"smith"))

# 2. A construction built to be deterministic: AES-SIV (RFC 5297).
siv = AESSIV(AESSIV.generate_key(bit_length=512))


def det(value):
    return siv.encrypt(value.encode(), [b"customers.last_name"])


print("\nAES-SIV, Smith twice equal:", det("Smith") == det("Smith"), "| Smith vs Jones equal:", det("Smith") == det("Jones"))
print("AES-SIV decrypts back to :", siv.decrypt(det("Smith"), [b"customers.last_name"]).decode())

# 3. What a thief with only the file can still do: frequency analysis.
rows = make_customers()
cts = [det(r["last_name"]) for r in rows]                       # one opaque value per row
by_frequency = [token for token, _ in Counter(cts).most_common()]   # most common token first
guess = dict(zip(by_frequency, PUBLIC_RANKING))                 # rank 1 token -> most common surname, and so on
truth = {ct: r["last_name"] for ct, r in zip(cts, rows)}        # known to us, never to the attacker

print(f"\ndistinct ciphertexts the thief sees: {len(set(cts))}")
print("rank  true name   attacker's guess")
for rank, token in enumerate(by_frequency[:8], 1):
    verdict = "right" if guess[token] == truth[token] else "wrong"
    print(f"{rank:>4}  {truth[token]:<10}  {guess[token]:<10}  {verdict}")

right = sum(guess[ct] == r["last_name"] for ct, r in zip(cts, rows))
print(f"\nrows whose last name the thief names correctly: {right:,} of {len(rows):,} ({right / len(rows):.1%})")

python step03_deterministic.py

same plaintext, same ciphertext: True
c1 xor c2           : 190207111b
'smith' xor 'jones' : 190207111b
attacker who knows 'smith' recovers: b'jones'

AES-SIV, Smith twice equal: True | Smith vs Jones equal: False
AES-SIV decrypts back to : Smith

distinct ciphertexts the thief sees: 100
rank  true name   attacker's guess
   1  Smith       Smith       right
   2  Johnson     Johnson     right
   3  Williams    Williams    right
   4  Brown       Brown       right
   5  Jones       Jones       right
   6  Garcia      Garcia      right
   7  Miller      Miller      right
   8  Davis       Davis       right

rows whose last name the thief names correctly: 13,634 of 20,000 (68.2%)

Read the output in three parts. First, with the pinned nonce the two ciphertexts XOR to exactly the same bytes as the two plaintexts, so an attacker who knows one message recovers the other: given “smith” the script recovers “jones”. Second, AES-SIV is deterministic and decrypts correctly, so equality search works. Third, the thief with only the file and a public frequency table named the correct last name for 13,634 of the 20,000 rows, 68.2 percent, and got the top eight surnames completely right.

Be careful about what that number means. My surname column is a synthetic, steeply skewed list of 100 names that is the whole universe, so the percentage belongs to this data, not to yours. Real columns are usually flatter. The mechanism is what transfers: the thief needed no key and no queries. freeCodeCamp’s article puts it as “deterministic encryption leaks equality patterns” and names the columns where that bites hardest: state codes, boolean values, gender categories and other small categorical fields. The academic literature agrees. Kamara and Moataz note in the abstract of a 2016 paper that property-preserving schemes such as deterministic encryption “have recently been shown to reveal a lot of information in certain settings”, citing Naveed and colleagues at CCS 2015.

AES-SIV is not broken. The leak is its feature: equal in, equal out. That is why the next step moves the equality test out of the encrypted value and into a separate column that only reveals what you decide it should.

Step 4: Build a blind index with HMAC and HKDF

The design has two columns per searchable field. The value stays encrypted with random AES-GCM, so the stored ciphertext reveals nothing. Next to it sits the blind index, a keyed hash of the normalized value. It is “blind” because without the key the number says nothing about the value. The same email always produces the same index, so WHERE email_bi = ? finds the row, while the ciphertext stays randomized.

CipherSweet’s design notes define it in three lines: “A deterministic one-way hash of the plaintext”, “Truncated to a specified number of bits” and “Treated as a Bloom filter for database lookups”. The truncation and the Bloom filter idea are the subject of step 6. A 2017 article on the Paragon Initiative Enterprises blog, Building Searchable Encrypted Databases with PHP and SQL by Scott Arciszewski, describes the lookup in one sentence: “Your application code will need to perform the decryption for each candidate row and then only serve the actual matches.” Our find function does exactly that.

You will build piivault.py in pieces across steps 4, 5 and 8. Every piece is printed in order, so you can type or paste them one after another into a single file.

Derive one key per column, then seal each value

Using one key for everything is a trap. The key that encrypts should not be the key that hashes, and email and SSN should not share keys, or a compromise of one column opens the others. HKDF solves this by deriving independent keys from one master key. RFC 5869 explains the purpose of its info input: “Its main objective is to bind the derived key material to application- and context-specific information.” The same RFC says “HKDF is defined to operate with and without random salt” and stresses that salt adds strength where you have one. Our input is already a random key, so the code passes None for the salt.

# piivault.py
"""Field-level encryption with blind indexes (lab code for Python 3.13 and cryptography 50.x)."""
import hmac
import os
import unicodedata

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

FORMAT = 1                        # first byte of every sealed blob
BITS = {"email": 16, "ssn": 16}   # blind-index length per searchable column (step 6 explains the choice)


class KeyRing:
    """Master keys by id. Each column gets its own encryption key and its own blind-index key."""

    def __init__(self, masters, current):
        self.masters = dict(masters)
        self.current = current
        self._cache = {}

    def active_ids(self):
        return sorted(self.masters)

    def derive(self, key_id, purpose, field):
        slot = (key_id, purpose, field)
        if slot not in self._cache:
            info = f"piivault/{FORMAT}/{purpose}/{field}".encode()
            self._cache[slot] = HKDF(hashes.SHA256(), 32, None, info).derive(self.masters[key_id])
        return self._cache[slot]


def seal(ring, field, row_id, text):
    """AES-256-GCM with a random nonce. The field name and row id are authenticated, not encrypted."""
    nonce = os.urandom(12)
    aad = f"{field}|{row_id}".encode()
    ct = AESGCM(ring.derive(ring.current, "enc", field)).encrypt(nonce, text.encode(), aad)
    return bytes([FORMAT, ring.current]) + nonce + ct


def unseal(ring, field, row_id, blob):
    if blob[0] != FORMAT:
        raise ValueError(f"unknown blob format {blob[0]}")
    key = ring.derive(blob[1], "enc", field)          # KeyError if that key id was retired
    aad = f"{field}|{row_id}".encode()
    return AESGCM(key).decrypt(blob[2:14], blob[14:], aad).decode()

Take it in order. KeyRing holds master keys by number and derives a 32-byte key for each purpose and field, so the key that encrypts emails, the key that encrypts SSNs and the key that hashes emails are three unrelated values. seal picks a random 12-byte nonce, builds the associated data from the field name and row id (step 7 explains why), encrypts with AES-256-GCM and returns a blob made of a format byte, a key id byte, the nonce and the ciphertext with its tag. Those two header bytes are what make key rotation possible in step 8. unseal reads the key id from the blob, so one table can hold rows sealed under several keys, and an unknown key id raises KeyError, which is the behavior you want from a retired key.

Compute the blind index

# piivault.py (continued)
def norm_text(value):
    return unicodedata.normalize("NFKC", value).strip().casefold()


def norm_digits(value):
    return "".join(ch for ch in unicodedata.normalize("NFKC", value) if ch in "0123456789")


COLUMNS = {"email": norm_text, "ssn": norm_digits}    # the only searchable columns, with their normalizers


def blind_index(ring, key_id, field, normalized, bits):
    """First `bits` bits of HMAC-SHA256(per-column index key, normalized value) as an integer."""
    assert 1 <= bits <= 63
    mac = hmac.digest(ring.derive(key_id, "bidx", field), normalized.encode(), "sha256")
    return int.from_bytes(mac[:8], "big") >> (64 - bits)

Every value is normalized before it is hashed; step 5 shows why. blind_index computes HMAC-SHA256 under the per-column index key, takes the first 64 bits of the result and shifts right to keep only the top bits bits. The result is a small integer that SQLite can index compactly. The assertion keeps bits at 63 or below because SQLite integers are signed 64-bit. COLUMNS lists the only searchable columns together with their normalizers, which both documents the design and keeps the SQL below safe from injection through column names.

Create the table and the search function

# piivault.py (continued)
def create_table(conn):
    conn.executescript("""
        DROP TABLE IF EXISTS customers;
        CREATE TABLE customers (
            id INTEGER PRIMARY KEY,
            key_id INTEGER NOT NULL,
            email_ct BLOB NOT NULL, email_bi INTEGER NOT NULL,
            ssn_ct BLOB NOT NULL, ssn_bi INTEGER NOT NULL,
            last_name_ct BLOB NOT NULL,
            plan TEXT NOT NULL, signup_month TEXT NOT NULL);
        CREATE INDEX customers_email ON customers (key_id, email_bi);
        CREATE INDEX customers_ssn ON customers (key_id, ssn_bi);
    """)


def protect_row(ring, row, bits=BITS):
    k, rid = ring.current, row["id"]
    return (
        rid, k,
        seal(ring, "customers.email", rid, row["email"]),
        blind_index(ring, k, "customers.email", norm_text(row["email"]), bits["email"]),
        seal(ring, "customers.ssn", rid, row["ssn"]),
        blind_index(ring, k, "customers.ssn", norm_digits(row["ssn"]), bits["ssn"]),
        seal(ring, "customers.last_name", rid, row["last_name"]),
        row["plan"], row["signup_month"],
    )


def find(conn, ring, column, value):
    """Ids of rows whose decrypted column equals value, after normalization."""
    norm = COLUMNS[column]          # KeyError: there is no blind index for this column, by design
    field = f"customers.{column}"
    wanted = norm(value)
    probes = [(k, blind_index(ring, k, field, wanted, BITS[column])) for k in ring.active_ids()]
    where = " OR ".join(f"(key_id = ? AND {column}_bi = ?)" for _ in probes)
    candidates = conn.execute(
        f"SELECT id, {column}_ct FROM customers WHERE {where}", [x for probe in probes for x in probe]
    ).fetchall()
    return [rid for rid, blob in candidates if norm(unseal(ring, field, rid, blob)) == wanted]

The table stores a key id per row, an encrypted column and an index column for email and SSN, an encrypted column for last name, and the two plain columns. The two SQLite indexes on (key_id, email_bi) and (key_id, ssn_bi) make lookups fast. protect_row seals three values and computes two indexes, in the same order as the table columns.

Now read find slowly, because it is the whole idea in eight lines. It looks up the normalizer, which raises KeyError for any column without a blind index. It normalizes the search term. It computes one probe per active key id, which is what lets search keep working during a key rotation. It fetches the candidate rows with an indexed query. Then it decrypts each candidate, normalizes the plaintext and keeps only exact matches. That last line is the one people forget: a blind index narrows the search, but only the decrypted comparison proves a match.

Run it

Save step04_build_index.py. It encrypts and indexes all 20,000 rows, searches by email and SSN, tries to search the unindexed column, compares the lookup time with the decrypt-everything scan from step 2, shows the SQLite query plan, and then plays the thief against the new file.

# step04_build_index.py
import os
import sqlite3
import statistics
import time

import piivault as pv
from data import make_customers

ring = pv.KeyRing({1: os.urandom(32)}, current=1)
rows = make_customers()

if os.path.exists("vault.db"):
    os.remove("vault.db")
conn = sqlite3.connect("vault.db")
pv.create_table(conn)
start = time.perf_counter()
conn.executemany("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)", [pv.protect_row(ring, r) for r in rows])
conn.commit()
print(f"encrypted and indexed {len(rows):,} rows in {time.perf_counter() - start:.2f} s")

target = rows[4241]
print("email search   :", pv.find(conn, ring, "email", target["email"]))
print("ssn search     :", pv.find(conn, ring, "ssn", target["ssn"]))
try:
    pv.find(conn, ring, "last_name", "Smith")
except KeyError as exc:
    print("last_name search: KeyError", exc, "(no blind index for this column, by design)")


def median_ms(fn, repeat):
    times = []
    for _ in range(repeat):
        start = time.perf_counter()
        fn()
        times.append(time.perf_counter() - start)
    return statistics.median(times) * 1000


def scan():
    return [i for i, blob in conn.execute("SELECT id, email_ct FROM customers")
            if pv.unseal(ring, "customers.email", i, blob) == target["email"]]


indexed = median_ms(lambda: pv.find(conn, ring, "email", target["email"]), 200)
scanned = median_ms(scan, 5)
print(f"\nblind-index lookup: {indexed:.2f} ms | decrypt-everything scan: {scanned:.0f} ms | ratio {scanned / indexed:,.0f}x")

plan = conn.execute("EXPLAIN QUERY PLAN SELECT id, email_ct FROM customers WHERE key_id = 1 AND email_bi = 5").fetchall()
print("query plan:", plan[0][3])

raw = open("vault.db", "rb").read()
print("\nsomeone copies vault.db:")
print("  '@example.test' appears", raw.count(b"@example.test"), "times, 'Smith' appears", raw.count(b"Smith"), "times")
print("  plan and signup_month stay readable on purpose: '2026-0' appears", f"{raw.count(b'2026-0'):,}", "times")
lengths = conn.execute(
    "SELECT length(last_name_ct) - 30 AS chars, COUNT(*) FROM customers GROUP BY chars ORDER BY 2 DESC LIMIT 4"
).fetchall()
print("  last-name lengths readable from blob sizes (blob bytes minus 30 of overhead):", lengths)

python step04_build_index.py

encrypted and indexed 20,000 rows in 0.25 s
email search   : [4242]
ssn search     : [4242]
last_name search: KeyError 'last_name' (no blind index for this column, by design)

blind-index lookup: 0.04 ms | decrypt-everything scan: 39 ms | ratio 959x
query plan: SEARCH customers USING INDEX customers_email (key_id=? AND email_bi=?)

someone copies vault.db:
  '@example.test' appears 0 times, 'Smith' appears 0 times
  plan and signup_month stay readable on purpose: '2026-0' appears 20,000 times
  last-name lengths readable from blob sizes (blob bytes minus 30 of overhead): [(5, 8429), (6, 3778), (7, 3004), (8, 2846)]

Go through the output line by line. Encrypting and indexing 20,000 rows took a quarter of a second. Searching by email and by SSN both returned [4242], the id of the customer the script picked. Searching the last-name column raised KeyError on purpose: not every column deserves an index, and step 6 shows why last name is the clearest example. The indexed lookup took 0.04 ms against 39 ms for decrypting everything, 959 times faster on my machine, and the query plan confirms SQLite used customers_email. If you want to read plans like that one fluently, my tutorial on reading SQLite EXPLAIN QUERY PLAN covers it.

The stolen-file check at the bottom is the payoff. The address text and “Smith” appear zero times in the file. Two things remain visible, and you should know them. The plan and signup month columns stay readable on purpose. And the size of each encrypted blob reveals the length of the value, because AES-GCM does not hide length: subtracting the 30 bytes of overhead (2 header bytes, a 12-byte nonce and a 16-byte tag) shows that 8,429 customers have a five-letter last name. If length is sensitive for a field, pad the value to a fixed size before encrypting.

Step 5: Normalize before you hash

A blind index compares bytes, so “[email protected]” and “[email protected]” are different values to it. Users type emails in capitals, paste them with a trailing space, or arrive from a form that uses full-width letters. If you hash the raw text, those searches silently find nothing. Save step05_normalize.py, which searches for the same stored address in five spellings, once hashing the text as typed and once normalizing first.

# step05_normalize.py
import os
import sqlite3

import piivault as pv
from data import make_customers

ring = pv.KeyRing({1: os.urandom(32)}, current=1)
rows = make_customers(2000)
conn = sqlite3.connect(":memory:")
pv.create_table(conn)
conn.executemany("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)", [pv.protect_row(ring, r) for r in rows])

stored = rows[41]["email"]
fullwidth = stored.replace("e", chr(0xFF45)).replace("a", chr(0xFF41))   # Latin letters in their full-width forms


def hash_as_typed(text):
    """What you get if you hash the query exactly the way the user typed it."""
    bi = pv.blind_index(ring, 1, "customers.email", text, pv.BITS["email"])
    hits = conn.execute("SELECT id, email_ct FROM customers WHERE key_id = 1 AND email_bi = ?", (bi,)).fetchall()
    return [i for i, blob in hits if pv.unseal(ring, "customers.email", i, blob) == text]


queries = [
    ("as stored", stored),
    ("UPPER CASE", stored.upper()),
    ("spaces around", f"  {stored} "),
    ("full-width letters", fullwidth),
    ("one dot removed", stored.replace(".", "", 1)),
]
print("stored email:", stored, "\n")
print(f"{'query':<20}{'hash as typed':<16}normalize first")
for label, query in queries:
    print(f"{label:<20}{str(hash_as_typed(query)):<16}{pv.find(conn, ring, 'email', query)}")

print("\nssn spellings that reach the same index value:")
for spelling in ("900-12-3456", "900123456", " 900 12 3456 "):
    print(f"  {spelling!r:<18} -> {pv.norm_digits(spelling)}")

python step05_normalize.py

stored email: [email protected] 

query               hash as typed   normalize first
as stored           [42]            [42]
UPPER CASE          []              [42]
spaces around       []              [42]
full-width letters  []              [42]
one dot removed     []              []

ssn spellings that reach the same index value:
  '900-12-3456'      -> 900123456
  '900123456'        -> 900123456
  ' 900 12 3456 '    -> 900123456

Hashing the text as typed found the customer only when the spelling matched byte for byte. Normalizing first found them in all four equivalent spellings. norm_text applies Unicode NFKC normalization (which turns full-width letters into ordinary ones), strips spaces at both ends and case-folds, which is a more thorough form of lowercasing. norm_digits keeps only the digits, which is why the three SSN spellings at the bottom reach the same index value.

The last row of the table is a deliberate non-match. Dropping a dot could merge two different people’s addresses, so the normalizer leaves dots alone. Normalize only what you know to be equivalent, and treat the rules as part of your stored data: if you change a normalizer later, every existing index value must be recomputed, with a job like the rotation job in step 8.

Step 6: Decide how many bits of index to keep

HMAC-SHA256 produces 256 bits and the module keeps 16. Truncating means many different values share each index value. Those collisions cost you something and buy you something. The cost is that a lookup returns the true match plus some strangers, which the decrypted comparison throws away. The benefit is that the index tells a thief less.

Both libraries describe this trade-off. The Paragon Initiative Enterprises article suggests you “Truncate your blind indexes to e.g. 16, 32, or 64 bits, and treat them as a Bloom filter”. The AWS Database Encryption SDK calls the same thing a beacon and says that its collisions, “also known as false positives”, “limit an unauthorized user’s ability to infer distinguishing information about the underlying plaintext”. It states the trade-off directly: “Shorter beacon lengths and additional partitions increase collisions and reduce frequency leakage, while longer beacon lengths and fewer partitions improve query precision.”

Rather than take that on trust, measure it. Save step06_truncation.py. For the last-name column it plays the rank-matching thief against indexes of different lengths, repeating the experiment with 20 different index keys because which names happen to collide depends on the key. The script derives those keys from a seeded generator so that your numbers match mine; never generate real keys that way. For the email column it measures the cost side: how many rows a lookup must decrypt.

# step06_truncation.py
import random
from collections import Counter

import piivault as pv
from data import PUBLIC_RANKING, make_customers

rows = make_customers()
N = len(rows)
truths = [r["last_name"] for r in rows]
TRIALS = 20


def keyring(seed):
    """Lab only: repeatable keys so your numbers match the article. Real keys come from a KMS."""
    return pv.KeyRing({1: random.Random(seed).randbytes(32)}, current=1)


def recovery(index_values):
    """Rank matching: the biggest bucket is called the most common surname, the next one the next surname..."""
    order = [bucket for bucket, _ in Counter(index_values).most_common()]
    guess = dict(zip(order, PUBLIC_RANKING))
    return sum(guess.get(b) == t for b, t in zip(index_values, truths)) / N


def rows_per_lookup(index_values):
    """Average number of rows a lookup has to decrypt: the size of the bucket the searched row falls in."""
    sizes = Counter(index_values)
    return sum(sizes[b] for b in index_values) / N


baseline = Counter(truths).most_common(1)[0][1] / N
print(f"last_name: 100 distinct values, Zipf-skewed, {N:,} rows, {TRIALS} different index keys")
print(f"baseline: naming the most common surname for every row, without looking at the index, is right {baseline:.1%} of the time\n")

CHOICES = (62, 16, 8, 6, 5, 4, 3)
results = {bits: [] for bits in CHOICES}
for seed in range(TRIALS):
    ring = keyring(seed)
    full = [pv.blind_index(ring, 1, "customers.last_name", pv.norm_text(t), 62) for t in truths]
    for bits in CHOICES:
        idx = [value >> (62 - bits) for value in full]       # fewer bits = the top bits of the same value
        results[bits].append((len(set(idx)), rows_per_lookup(idx), recovery(idx)))

print(f"{'bits':>5} {'buckets':>8} {'rows per lookup':>16}   thief names the right surname: mean (lowest to highest)")
for bits, trials in results.items():
    buckets = sum(t[0] for t in trials) / TRIALS
    per_lookup = sum(t[1] for t in trials) / TRIALS
    hits = [t[2] for t in trials]
    print(f"{bits:>5} {buckets:>8.1f} {per_lookup:>16,.0f}   {sum(hits) / TRIALS:.1%} ({min(hits):.1%} to {max(hits):.1%})")

ring = keyring(0)
print(f"\nemail: {N:,} unique values")
print(f"{'bits':>5} {'rows per lookup':>16} {'formula 1 + (N-1)/2^bits':>26}")
for bits in (32, 24, 20, 16, 12, 8):
    idx = [pv.blind_index(ring, 1, "customers.email", pv.norm_text(r["email"]), bits) for r in rows]
    print(f"{bits:>5} {rows_per_lookup(idx):>16.2f} {1 + (N - 1) / 2**bits:>26.2f}")

python step06_truncation.py

last_name: 100 distinct values, Zipf-skewed, 20,000 rows, 20 different index keys
baseline: naming the most common surname for every row, without looking at the index, is right 19.6% of the time

 bits  buckets  rows per lookup   thief names the right surname: mean (lowest to highest)
   62    100.0            1,254   68.2% (68.2% to 68.2%)
   16     99.8            1,254   68.0% (64.1% to 68.3%)
    8     82.7            1,333   50.8% (25.9% to 58.6%)
    6     50.2            1,620   40.6% (21.3% to 54.2%)
    5     30.5            2,027   34.3% (19.6% to 49.1%)
    4     15.9            2,536   32.6% (19.6% to 45.5%)
    3      8.0            3,696   27.5% (19.6% to 38.6%)

email: 20,000 unique values
 bits  rows per lookup   formula 1 + (N-1)/2^bits
   32             1.00                       1.00
   24             1.00                       1.00
   20             1.02                       1.02
   16             1.31                       1.31
   12             5.87                       5.88
    8            79.04                      79.12

Reading the last-name table

Start with the baseline: naming “Smith” for every row, without looking at any index, is right 19.6 percent of the time, so nothing the thief does can be called a win below that. With the longest index in the experiment (62 bits, effectively untruncated) the thief names 68.2 percent of the rows correctly, exactly the deterministic-encryption result, because no two names collide. At 16 bits almost nothing changes. At 8 bits the average drops to 50.8 percent, but across the 20 keys it ranged from 25.9 to 58.6 percent, so truncation is a probabilistic defense that depends on which names share a bucket with the big ones. At 4 bits the average is 32.6 percent and some keys land on the 19.6 percent floor.

Now read the cost column. A lookup on last name returns 1,254 rows on average even with no truncation, because a surname really is shared by thousands of customers, and the number only grows as you truncate. An index on a repeated-value column buys almost nothing and leaks the most. The right decision is the one the lab made: do not index it. If you need to search such a column, use few bits and accept that every lookup is expensive, or switch to a different technique, as the AWS documentation does with partitions for “hot” values.

Reading the email table

Email is the opposite case. Every value is unique, so a thief sees no frequencies to count, and truncation exists only to control cost and to blur brute-force guesses (step 9). The measured rows per lookup match the formula 1 plus (N minus 1) divided by 2 to the power of the number of bits, where N is the number of rows. At 20,000 rows, 16 bits means a lookup decrypts 1.31 rows on average, 12 bits means 5.87 and 8 bits means 79.04. That is why the module uses 16 bits for both email and SSN.

Use the formula as your rule. Pick the smallest number of bits that keeps the average candidates per lookup in single digits at the table size you expect, then recheck when the table grows. Computed from the formula rather than measured, a table of 2,000,000 rows at 16 bits would decrypt about 31.5 rows per lookup and need roughly 24 bits to get back to about 1.1. One more caveat: if a unique-looking value ever repeats across rows, for example an events table that stores the email on every login, treat that column like last name, because frequency analysis applies to it again. Changing the bit length later means recomputing every index for that column with a job like the rotation job in step 8, so choose with some growth in mind.

Step 7: Bind every ciphertext to its row and column

Encryption alone does not stop the writer. Someone who can edit the database, through SQL injection or a careless migration, does not need to decrypt anything: they can copy the encrypted SSN cell from one customer into another. If the cells are not tied to their position, the application decrypts the copied value happily and shows customer 1 with customer 2’s SSN.

AES-GCM’s associated data fixes this. The cryptography library describes it as additional data that is authenticated with the key but does not need to be encrypted. The module passes the field name and row id as associated data, and the tag only validates if the same values are supplied when decrypting. Save step07_bind_aad.py. It first seals values with no associated data and swaps two cells, then does the same swap against the vault.

# step07_bind_aad.py
import os
import sqlite3

from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

import piivault as pv
from data import make_customers

rows = make_customers(3)

# 1. Sealed with no associated data: nothing ties a value to its row.
aes = AESGCM(AESGCM.generate_key(bit_length=256))


def naive_seal(text):
    nonce = os.urandom(12)
    return nonce + aes.encrypt(nonce, text.encode(), None)


naive = sqlite3.connect(":memory:")
naive.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, last_name TEXT, ssn_ct BLOB)")
naive.executemany("INSERT INTO customers VALUES (?,?,?)", [(r["id"], r["last_name"], naive_seal(r["ssn"])) for r in rows])

# an attacker who can write to the database (SQL injection, a bad migration, a curious DBA) swaps two cells
a, b = [blob for (blob,) in naive.execute("SELECT ssn_ct FROM customers WHERE id IN (1, 2) ORDER BY id")]
naive.execute("UPDATE customers SET ssn_ct = ? WHERE id = 1", (b,))
naive.execute("UPDATE customers SET ssn_ct = ? WHERE id = 2", (a,))

name, blob = naive.execute("SELECT last_name, ssn_ct FROM customers WHERE id = 1").fetchone()
print(f"no associated data : customer 1 is {name}; the app decrypts ssn {aes.decrypt(blob[:12], blob[12:], None).decode()}")
print(f"                     (that is customer 2's real ssn: {rows[1]['ssn']}), and no error was raised")

# 2. The vault binds every value to "<table.column>|<row id>".
ring = pv.KeyRing({1: os.urandom(32)}, current=1)
vault = sqlite3.connect(":memory:")
pv.create_table(vault)
vault.executemany("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)", [pv.protect_row(ring, r) for r in rows])
blob1, blob2 = [blob for (blob,) in vault.execute("SELECT ssn_ct FROM customers WHERE id IN (1, 2) ORDER BY id")]
(email1,) = vault.execute("SELECT email_ct FROM customers WHERE id = 1").fetchone()

attempts = [
    ("row 2's ssn placed in row 1", "customers.ssn", 1, blob2),
    ("row 1's email placed in its ssn cell", "customers.ssn", 1, email1),
    ("row 1's own ssn", "customers.ssn", 1, blob1),
]
print()
for label, field, rid, blob in attempts:
    try:
        print(f"with associated data: {label:<38} -> {pv.unseal(ring, field, rid, blob)}")
    except InvalidTag:
        print(f"with associated data: {label:<38} -> InvalidTag (rejected)")

python step07_bind_aad.py

no associated data : customer 1 is Kim; the app decrypts ssn 907-64-3517
                     (that is customer 2's real ssn: 907-64-3517), and no error was raised

with associated data: row 2's ssn placed in row 1            -> InvalidTag (rejected)
with associated data: row 1's email placed in its ssn cell   -> InvalidTag (rejected)
with associated data: row 1's own ssn                        -> 950-83-0791

The first line is the problem. The application read customer 1 (Kim), decrypted the cell and showed customer 2’s real SSN with no error. The second block is the fix: moving row 2’s SSN into row 1, and moving row 1’s email into its own SSN cell, both raise InvalidTag, while row 1’s own SSN still decrypts. Two defenses work together here. The associated data differs, and the cross-column attempt also uses a different derived key, because each field has its own key. If you ever merge tables or change primary keys, you must re-seal the affected values, because the row id is part of what the tag proves.

Step 8: Rotate keys without losing search

Keys must be replaceable: an employee leaves, a key is exposed, or policy demands rotation. The blob’s key id byte and the table’s key_id column let old and new keys coexist. The plan is simple. Add a new master key and make it current, so new writes use it. Keep the old key in the ring so old rows stay readable, and let find probe every active key id. Migrate rows in batches. Only when no row depends on the old key, retire it.

Add the two helper functions to the end of piivault.py. rotate_batch decrypts rows still on an older key and re-seals them under the current one, recomputing both blind indexes, because an HMAC under a new key cannot be derived from the old index value.

# piivault.py (continued)
def rows_left(conn, key_id):
    return conn.execute("SELECT COUNT(*) FROM customers WHERE key_id = ?", (key_id,)).fetchone()[0]


def rotate_batch(conn, ring, limit=500):
    """Re-seal up to `limit` rows still on an older key under the current key. Returns how many moved."""
    old = conn.execute(
        "SELECT id, email_ct, ssn_ct, last_name_ct FROM customers WHERE key_id != ? LIMIT ?", (ring.current, limit)
    ).fetchall()
    for rid, email_ct, ssn_ct, last_ct in old:
        row = {
            "id": rid,
            "email": unseal(ring, "customers.email", rid, email_ct),
            "ssn": unseal(ring, "customers.ssn", rid, ssn_ct),
            "last_name": unseal(ring, "customers.last_name", rid, last_ct),
        }
        new = protect_row(ring, dict(row, plan="", signup_month=""))
        conn.execute(
            "UPDATE customers SET key_id=?, email_ct=?, email_bi=?, ssn_ct=?, ssn_bi=?, last_name_ct=? WHERE id=?",
            (new[1], new[2], new[3], new[4], new[5], new[6], rid),
        )
    conn.commit()
    return len(old)

Save step08_rotate.py. It also demonstrates the mistake to avoid: retiring the old key while rows still depend on it.

# step08_rotate.py
import os
import sqlite3

import piivault as pv
from data import make_customers

rows = make_customers(5000)
key1, key2 = os.urandom(32), os.urandom(32)

conn = sqlite3.connect(":memory:")
pv.create_table(conn)
ring = pv.KeyRing({1: key1}, current=1)
conn.executemany("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)", [pv.protect_row(ring, r) for r in rows])


def per_key():
    return dict(conn.execute("SELECT key_id, COUNT(*) FROM customers GROUP BY key_id ORDER BY key_id"))


# Day 1: key 2 becomes current. Old rows stay readable and searchable because the ring still holds key 1.
ring = pv.KeyRing({1: key1, 2: key2}, current=2)
old_row, new_row = rows[10], {**rows[11], "id": 5001, "email": "[email protected]"}
conn.execute("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)", pv.protect_row(ring, new_row))
conn.commit()
print("rows per key id after adding key 2:", per_key())
print("search finds a key-1 row :", pv.find(conn, ring, "email", old_row["email"]))
print("search finds a key-2 row :", pv.find(conn, ring, "email", new_row["email"]))

# The mistake to avoid: retiring key 1 while rows still depend on it.
too_early = pv.KeyRing({2: key2}, current=2)
print("\nretired too early:")
print("  search for the key-1 row ->", pv.find(conn, too_early, "email", old_row["email"]), "(a silent miss, no error)")
try:
    pv.unseal(too_early, "customers.email", 11, conn.execute("SELECT email_ct FROM customers WHERE id = 11").fetchone()[0])
except KeyError as exc:
    print("  decrypting that row     -> KeyError", exc)

# Migrate in batches, then check the counter before retiring anything.
print("\nmigration:")
while moved := pv.rotate_batch(conn, ring, limit=1500):
    print(f"  moved {moved:,} rows, rows still on key 1: {pv.rows_left(conn, 1):,}")
print("rows per key id after migration:", per_key())

retired = pv.KeyRing({2: key2}, current=2)
print("\nkey 1 retired:")
print("  search finds the old row  :", pv.find(conn, retired, "email", old_row["email"]))
print("  ssn search, a random row  :", pv.find(conn, retired, "ssn", rows[2500]["ssn"]) == [rows[2500]["id"]])

python step08_rotate.py

rows per key id after adding key 2: {1: 5000, 2: 1}
search finds a key-1 row : [11]
search finds a key-2 row : [5001]

retired too early:
  search for the key-1 row -> [] (a silent miss, no error)
  decrypting that row     -> KeyError 1

migration:
  moved 1,500 rows, rows still on key 1: 3,500
  moved 1,500 rows, rows still on key 1: 2,000
  moved 1,500 rows, rows still on key 1: 500
  moved 500 rows, rows still on key 1: 0
rows per key id after migration: {2: 5001}

key 1 retired:
  search finds the old row  : [11]
  ssn search, a random row  : True

After adding key 2, the table holds 5,000 rows on key 1 and the one new row on key 2, and searches find rows under both. The “retired too early” block is the one to remember. Searching for the key 1 row returned an empty list with no error, because find only probes key ids that are in the ring, while decrypting the same row raised KeyError: 1. A silent miss is worse than a crash, so check the counter, rows_left(conn, 1), and retire the key only when it reads zero. The migration moved all 5,000 rows in four batches, and after retiring key 1 every search still worked. Backups taken before the migration are still sealed under the old key, so archive that key for as long as you keep those backups.

Step 9: Know the limits of the design

A blind index is a fast hash by design, and that matters in the key thief scenario. If the index key leaks together with the database, the thief can hash every possible value and compare. SSNs have only nine digits, so the whole space is small. The script below simulates exactly that for the 214 customers whose SSN starts with 900, and then measures what a slow index costs.

# step09_small_domain.py
import hashlib
import hmac
import random
import time
from collections import Counter

import piivault as pv
from data import make_customers

ring = pv.KeyRing({1: random.Random(9).randbytes(32)}, current=1)   # lab only: a repeatable key
rows = make_customers()
key = ring.derive(1, "bidx", "customers.ssn")      # imagine this key leaked together with the database

targets = [r for r in rows if r["ssn"].startswith("900")]
print(f"customers whose ssn starts with 900: {len(targets)}")


def index_of(digits, bits):
    mac = hmac.digest(key, digits.encode(), "sha256")
    return int.from_bytes(mac[:8], "big") >> (64 - bits)


# 1. Full-length index: try every ssn from 900000000 to 900999999.
wanted = {index_of(pv.norm_digits(r["ssn"]), 62) for r in targets}
start = time.perf_counter()
found = sum(index_of(f"900{n:06d}", 62) in wanted for n in range(1_000_000))
elapsed = time.perf_counter() - start
print(f"62-bit index : recovered {found} of {len(targets)} ssns in {elapsed:.1f} s ({1_000_000 / elapsed:,.0f} guesses per second)")

# 2. Truncated to 16 bits: the same attack, but each target index matches many candidates.
buckets = Counter(index_of(f"900{n:06d}", 16) for n in range(1_000_000))
matches = [buckets[index_of(pv.norm_digits(r["ssn"]), 16)] for r in targets]
average = sum(matches) / len(matches)
print(f"16-bit index : each target index matches {average:.1f} candidate ssns on average, so a guess is right 1 time in {average:.0f}")


# 3. A slow index: PBKDF2 per guess, the derived key used as the salt.
def slow_index(digits, bits, iterations):
    mac = hashlib.pbkdf2_hmac("sha256", digits.encode(), key, iterations, dklen=8)
    return int.from_bytes(mac, "big") >> (64 - bits)


def human(seconds):
    for unit, size in (("days", 86400), ("hours", 3600), ("min", 60)):
        if seconds >= size:
            return f"{seconds / size:.1f} {unit}"
    return f"{seconds:.1f} s"


print()
print(f"{'index':<20}{'ms per guess':>13}{'10^6 guesses':>15}{'10^8 guesses':>15}")
per_guess = {"HMAC-SHA256": elapsed / 1_000_000}
for iterations in (10_000, 100_000):
    start = time.perf_counter()
    for n in range(20):
        slow_index(f"900{n:06d}", 62, iterations)
    per_guess[f"PBKDF2 x {iterations:,}"] = (time.perf_counter() - start) / 20
for name, seconds in per_guess.items():
    print(f"{name:<20}{seconds * 1000:>13.4f}{human(seconds * 1e6):>15}{human(seconds * 1e8):>15}")
print("(the last two columns multiply the measured per-guess time out for one CPU core; they are not separate runs)")

python step09_small_domain.py

customers whose ssn starts with 900: 214
62-bit index : recovered 214 of 214 ssns in 1.6 s (617,680 guesses per second)
16-bit index : each target index matches 16.1 candidate ssns on average, so a guess is right 1 time in 16

index                ms per guess   10^6 guesses   10^8 guesses
HMAC-SHA256                0.0016          1.6 s        2.7 min
PBKDF2 x 10,000            2.4765       41.3 min       2.9 days
PBKDF2 x 100,000          27.3149      7.6 hours      31.6 days
(the last two columns multiply the measured per-guess time out for one CPU core; they are not separate runs)

What the numbers say

With the full-length index, the thief tried all 1,000,000 values from 900000000 to 900999999 in 1.6 seconds, about 620,000 guesses per second in plain Python on one core, and recovered all 214 SSNs. That is the cost of a fast hash on a small keyspace. The same attack against the 16-bit index still finds the real SSN, but every index value now matches about 16 candidate SSNs in that slice, so a guess is right about one time in 16. Scaling that up is simple arithmetic rather than a measurement: the full 9XX space of 100,000,000 values spread over 65,536 index values is about 1,526 candidates per index value, which turns recovery into guesswork.

The table shows the second defense, a slow index that runs PBKDF2 with the derived key as salt. CipherSweet offers the same idea and states its purpose plainly: “The purpose of a slow blind index is to make attacks more expensive (e.g. for slightly smaller keyspaces).” At 100,000 iterations each guess costs 27 ms, so the 1,000,000-guess slice takes about 7.6 hours on one core and the whole 100,000,000-value space about 31.6 days, while each legitimate lookup pays the same 27 ms once. Pick the iteration count from the latency you can afford, and remember this is defense in depth for small keyspaces, not a substitute for protecting the key. My tutorial on hashing passwords with bcrypt and Argon2 covers how to choose slow-hash settings.

Protect the key, and know what you cannot search

The OWASP Cryptographic Storage Cheat Sheet gives the rule for where keys live: “encryption keys should be stored in a separate location from encrypted data”. In production that means a key management service or hardware security module, and for backing up the master key itself you can split it between people, as in my Shamir’s Secret Sharing tutorial.

Finally, set expectations about queries. CipherSweet says its design “does not aim to provide full-text searching or allow the database to order ciphertexts”. In practice that rules out LIKE searches, range queries and ORDER BY on the encrypted columns. For other lookups, CipherSweet’s notes describe “Functional indexes (by applying domain-specific transformations to the plaintext before it encounters the final hash/KDF function)”. One example is a second blind index over only the last four digits of the SSN, with few bits. Each extra index is another place where equality leaks, so add them one at a time and measure.

Step 10: Lock it in with tests

Save test_piivault.py. Each test pins down one property from the steps above: a fresh nonce on every seal, the row and column binding, tamper detection, key and column separation of the index, the normalizers, messy search input, the unindexed column refusing to be searched, a truncated index that returns extra candidates but still gets filtered to the exact match, a complete rotation followed by retiring the old key, and a retired key failing closed.

# test_piivault.py
import os
import sqlite3

import pytest
from cryptography.exceptions import InvalidTag

import piivault as pv
from data import make_customers

INSERT = "INSERT INTO customers VALUES (?,?,?,?,?,?,?,?,?)"


@pytest.fixture
def ring():
    return pv.KeyRing({1: os.urandom(32)}, current=1)


@pytest.fixture
def db(ring):
    rows = make_customers(300)
    conn = sqlite3.connect(":memory:")
    pv.create_table(conn)
    conn.executemany(INSERT, [pv.protect_row(ring, r) for r in rows])
    return conn, rows


def test_roundtrip_with_a_fresh_nonce_every_time(ring):
    first = pv.seal(ring, "customers.email", 1, "[email protected]")
    second = pv.seal(ring, "customers.email", 1, "[email protected]")
    assert first != second
    assert pv.unseal(ring, "customers.email", 1, first) == "[email protected]"


def test_a_sealed_value_only_opens_in_its_own_row_and_column(ring):
    blob = pv.seal(ring, "customers.ssn", 1, "900-12-3456")
    with pytest.raises(InvalidTag):
        pv.unseal(ring, "customers.ssn", 2, blob)
    with pytest.raises(InvalidTag):
        pv.unseal(ring, "customers.email", 1, blob)


def test_tampering_is_detected(ring):
    blob = bytearray(pv.seal(ring, "customers.ssn", 1, "900-12-3456"))
    blob[-1] ^= 1
    with pytest.raises(InvalidTag):
        pv.unseal(ring, "customers.ssn", 1, bytes(blob))


def test_blind_index_is_stable_and_separated_by_column_and_key(ring):
    value = pv.blind_index(ring, 1, "customers.email", "x", 32)
    assert value == pv.blind_index(ring, 1, "customers.email", "x", 32)
    assert value != pv.blind_index(ring, 1, "customers.ssn", "x", 32)
    other = pv.KeyRing({1: os.urandom(32)}, current=1)
    assert value != pv.blind_index(other, 1, "customers.email", "x", 32)


def test_normalizers():
    assert pv.norm_text("  [email protected] ") == "[email protected]"
    assert pv.norm_text(chr(0xFF21) + chr(0xFF4C)) == "al"
    assert pv.norm_digits("900-12-3456") == pv.norm_digits(" 900 12 3456 ") == "900123456"


def test_find_survives_messy_input(db, ring):
    conn, rows = db
    row = rows[17]
    assert pv.find(conn, ring, "email", f"  {row['email'].upper()} ") == [row["id"]]
    assert pv.find(conn, ring, "ssn", row["ssn"].replace("-", "")) == [row["id"]]
    assert pv.find(conn, ring, "email", "[email protected]") == []


def test_a_column_without_a_blind_index_cannot_be_searched(db, ring):
    conn, _ = db
    with pytest.raises(KeyError):
        pv.find(conn, ring, "last_name", "Smith")


def test_truncated_index_returns_candidates_and_find_filters_them(ring, monkeypatch):
    monkeypatch.setitem(pv.BITS, "email", 3)
    rows = make_customers(300)
    conn = sqlite3.connect(":memory:")
    pv.create_table(conn)
    conn.executemany(INSERT, [pv.protect_row(ring, r) for r in rows])
    target = rows[5]
    bucket = pv.blind_index(ring, 1, "customers.email", pv.norm_text(target["email"]), 3)
    candidates = conn.execute("SELECT COUNT(*) FROM customers WHERE key_id = 1 AND email_bi = ?", (bucket,)).fetchone()[0]
    assert candidates > 5
    assert pv.find(conn, ring, "email", target["email"]) == [target["id"]]


def test_rotation_moves_every_row_and_the_old_key_can_then_retire(db, ring):
    conn, rows = db
    both = pv.KeyRing({1: ring.masters[1], 2: os.urandom(32)}, current=2)
    while pv.rotate_batch(conn, both, limit=100):
        pass
    assert pv.rows_left(conn, 1) == 0
    only_new = pv.KeyRing({2: both.masters[2]}, current=2)
    for row in (rows[0], rows[150], rows[299]):
        assert pv.find(conn, only_new, "email", row["email"]) == [row["id"]]
        assert pv.find(conn, only_new, "ssn", row["ssn"]) == [row["id"]]


def test_a_retired_key_fails_closed_when_rows_still_use_it(db, ring):
    conn, _ = db
    blob = conn.execute("SELECT email_ct FROM customers WHERE id = 1").fetchone()[0]
    retired = pv.KeyRing({2: os.urandom(32)}, current=2)
    with pytest.raises(KeyError):
        pv.unseal(retired, "customers.email", 1, blob)

python -m pytest -q test_piivault.py

..........                                                               [100%]
10 passed in 0.06s

Ten tests pass in under a second. The most valuable are the truncation test, which proves the final comparison is what makes a lookup exact, and the rotation tests, which protect you from the silent miss in step 8.

Check the whole lab end to end

Delete the .db files and run the scripts again in order. You are done when all of these hold. The plain database gives up 20,000 addresses to a raw byte search, while randomized.db and vault.db give up none. Searching vault.db by email and by SSN returns the same customer id, and searching last name raises KeyError. Swapping encrypted cells raises InvalidTag. After the rotation, every row sits on key 2, searches still work with key 1 removed from the ring, and rows_left reads zero. The tests report 10 passed.

Here is the whole comparison in one place, using only what you measured.

Common mistakes and how to spot them

Reusing a nonce. Symptom: equal plaintexts give equal ciphertexts, or the XOR of two ciphertexts equals the XOR of the plaintexts. Cause: a constant or repeated nonce. Fix: os.urandom(12) for every value, as in seal.

One key for everything. Symptom: the same email gives the same index in two columns, so the columns can be linked. Fix: derive per-column keys with HKDF and distinct info strings.

Skipping normalization. Symptom: support says a customer exists but search finds nothing. Fix: normalize on both sides, and version the rules.

Indexing a repeated-value column. Symptom: lookups return thousands of rows, and a thief can count buckets. Fix: leave it unindexed, or use very few bits and accept the cost.

Trusting the index. Symptom: a false positive returned to a user. Fix: always decrypt and compare, as find does.

Retiring a key early. Symptom: searches return empty lists, or KeyError on read. Fix: check rows_left and your backups first.

Keys next to data, or plaintext in logs. Symptom: a single leaked artifact reveals everything. Fix: keep keys in a key management service, and never log decrypted values or search terms.

Where to go next

Move the master key into a key management service so the application never holds it for longer than needed; freeCodeCamp’s article compares keeping encryption local with calling the service for each value, which is a useful decision to make deliberately. For production, use a vetted library instead of this lab module: CipherSweet implements the blind-index design and publishes an informal security analysis, and the AWS Database Encryption SDK applies the same idea with beacons and partitions. If a few values are very common, read the AWS partition guidance before you index them. And keep checking your assumptions with measurements, as you did here: run the truncation experiment on the real distribution of your column (the number of rows per value, not the values themselves) before you choose bit lengths.

Sources and further reading

freeCodeCamp, How to Encrypt PII in Data Pipelines While Keeping It Searchable, the concept article that inspired this lab. CipherSweet, security design notes, and Paragon Initiative Enterprises, Building Searchable Encrypted Databases with PHP and SQL. AWS, Choosing a beacon length and partitions. The cryptography project, Authenticated encryption. IETF, RFC 5869 (HKDF) and RFC 5297 (AES-SIV). OWASP, Cryptographic Storage Cheat Sheet. Seny Kamara and Tarik Moataz, SQL on Structurally-Encrypted Databases.