Szyfrowanie mapowania pseudonimizacji w Pythonie: cryptography i Fernet

W poprzednich częściach pokazaliśmy dwie niewygodne rzeczy: mapowanie token → oryginał leży w stanie grafu, a stan w pliku, w otwartym tekście. Pseudonimizacja, która zostawia klucz do odwrócenia obok pseudonimizowanego tekstu, niewiele daje, a dotyczy to wszystkich danych osobowych (PII, od ang. personally identifiable information), nie tylko PESEL-u. Zaszyfrujemy więc mapowanie biblioteką cryptography, a potem cały plik checkpointów, na którym opiera się human-in-the-loop w LangGraph.

Opieramy się na pseudonimizacji danych w Pythonie (funkcja restore) i na checkpointach w LangGraph i SQLite. Kod uruchomiłem na Pythonie 3.13, cryptography 50.0.2, LangGraph 1.2.18 i langgraph-checkpoint-sqlite 3.1.2. To tylko technika; nie jest to porada prawna ani gwarancja zgodności z RODO.

pip install cryptography langgraph langgraph-checkpoint-sqlite

Czym jest Fernet i jak go użyć?

Fernet to gotowy sposób szyfrowania symetrycznego (symmetric encryption), czyli takiego, w którym ten sam klucz szyfruje i odszyfrowuje, z biblioteki cryptography. Według dokumentacji biblioteki łączy szyfr AES z podpisem HMAC, więc to szyfrowanie uwierzytelnione (authenticated encryption): nie tylko ukrywa dane, ale też wykrywa ich zmianę. Nie musisz samodzielnie wybierać trybu szyfru ani wektora inicjującego, co jest najczęstszym źródłem błędów w „własnym” szyfrowaniu. Dostajesz dwie operacje: encrypt i decrypt.

from cryptography.fernet import Fernet, InvalidToken

key = Fernet.generate_key()
fernet = Fernet(key)
token = fernet.encrypt("44051401359".encode("utf-8"))
print(len(key), token[:6])
print(fernet.decrypt(token).decode("utf-8"))

try:
    Fernet(Fernet.generate_key()).decrypt(token)
except InvalidToken:
    print("zły klucz: InvalidToken")

changed = token[:-4] + b"AAAA"
try:
    fernet.decrypt(changed)
except InvalidToken:
    print("zmieniony token: InvalidToken")

Wynik:

44 b'gAAAAA'
44051401359
zły klucz: InvalidToken
zmieniony token: InvalidToken

Klucz ma 44 znaki (to 32 bajty zapisane w base64), a token zawsze zaczyna się od gAAAAA. Odszyfrowanie innym kluczem i odszyfrowanie zmienionego tokenu kończą się tym samym wyjątkiem InvalidToken. Dzięki temu z samego wyjątku nie dowiesz się, czy klucz był zły, czy dane zostały uszkodzone, i to jest zamierzone.

Jak zaszyfrować mapowanie token → oryginał?

Mapowanie to słownik, więc wystarczy zamienić go na JSON, zaszyfrować i trzymać jako bajty. Zamiast pojedynczego Fernet używamy od razu MultiFernet, bo przyda się przy wymianie klucza (osobna sekcja niżej). Z jednym kluczem działa tak samo. W pliku vault.py zaczynamy od czterech funkcji:

import json
import os

from cryptography.fernet import Fernet, MultiFernet


def make_fernet(keys: str) -> MultiFernet:
    """Klucze podajemy po przecinku, najnowszy pierwszy (do szyfrowania)."""
    return MultiFernet([Fernet(key.strip()) for key in keys.split(",")])


def fernet_from_env(name: str = "PSEUDONYM_KEYS") -> MultiFernet:
    keys = os.environ.get(name)
    if not keys:
        raise RuntimeError(f"Brak klucza: ustaw zmienną środowiskową {name}")
    return make_fernet(keys)


def seal_mapping(mapping: dict[str, str], fernet: MultiFernet) -> bytes:
    return fernet.encrypt(json.dumps(mapping).encode("utf-8"))


def open_mapping(sealed: bytes, fernet: MultiFernet) -> dict[str, str]:
    return json.loads(fernet.decrypt(sealed))

Pseudonimizowany tekst może wyjść do zewnętrznego narzędzia, a zaszyfrowane mapowanie zostaje u Ciebie. Gdy tekst wróci z tokenami, otwierasz mapowanie i przywracasz oryginały funkcją restore z artykułu o pseudonimizacji:

from cryptography.fernet import Fernet

from pseudonymize import restore
from vault import make_fernet, open_mapping, seal_mapping

fernet = make_fernet(Fernet.generate_key().decode())
mapping = {"[PESEL_1]": "44051401359", "[EMAIL_1]": "a@b.pl"}

sealed = seal_mapping(mapping, fernet)
print(type(sealed).__name__, len(sealed), b"44051401359" in sealed)

opened = open_mapping(sealed, fernet)
print(restore("Klient [PESEL_1] pisze z [EMAIL_1].", opened))

Wynik:

bytes 164 False
Klient 44051401359 pisze z a@b.pl.

Zaszyfrowane mapowanie z dwoma wpisami ma 164 bajty i nie zawiera PESEL-u. Jest dłuższe niż dane, bo token niesie ze sobą m.in. wektor inicjujący i podpis.

Skąd wziąć klucz i gdzie go trzymać?

Klucz generujesz raz, poza kodem programu:

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
export PSEUDONYM_KEYS="<wklej wygenerowany klucz>"

Funkcja fernet_from_env czyta klucz ze zmiennej środowiskowej i przerywa program z czytelnym komunikatem, gdy go brakuje. To lepsze niż cicha wartość domyślna, bo domyślny klucz byłby kluczem znanym każdemu, kto przeczyta kod.

Kilka zasad, które decydują o tym, czy szyfrowanie ma sens:

• Klucz nie trafia do repozytorium ani do pliku obok danych. Zaszyfrowane mapowanie z kluczem w tym samym katalogu to zamknięte drzwi z kluczem w zamku.
• Utrata klucza oznacza utratę danych. Nie odszyfrujesz mapowania, więc nie odtworzysz oryginałów. Zrób kopię klucza w innym miejscu niż kopia danych.
• Na produkcji zmienna środowiskowa to minimum. Lepiej pobierać klucz z menedżera sekretów (secrets manager), do którego dostęp jest osobno kontrolowany.

Jak zaszyfrować cały plik checkpointów w LangGraph?

Zaszyfrowanie samego mapowania nie wystarczy, jeśli oryginalny tekst i propozycje do zatwierdzenia leżą w stanie, a stan w checkpointach. LangGraph przewiduje na to wymienny serializer: EncryptedSerializer szyfruje bajty każdego checkpointu szyfrem, który mu podasz. Szyfr musi mieć dwie metody, encrypt i decrypt, więc opakowujemy w niego nasz MultiFernet. Dopisz to do vault.py (importy na górze pliku, klasy i funkcje na końcu):

import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver


class FernetCipher:
    """Szyfr zgodny z CipherProtocol z LangGraph."""

    def __init__(self, fernet: MultiFernet) -> None:
        self.fernet = fernet

    def encrypt(self, plaintext: bytes) -> tuple[str, bytes]:
        return "fernet", self.fernet.encrypt(plaintext)

    def decrypt(self, ciphername: str, ciphertext: bytes) -> bytes:
        if ciphername != "fernet":
            raise ValueError(f"Nieznany szyfr: {ciphername}")
        return self.fernet.decrypt(ciphertext)


@contextmanager
def encrypted_saver(path: str, fernet: MultiFernet) -> Iterator[SqliteSaver]:
    connection = sqlite3.connect(path, check_same_thread=False)
    try:
        yield SqliteSaver(connection, serde=EncryptedSerializer(FernetCipher(fernet)))
    finally:
        connection.close()

SqliteSaver nie jest menedżerem kontekstu, a from_conn_string nie przyjmuje serializera, dlatego encrypted_saver sam otwiera połączenie i zamyka je po użyciu. Porównajmy plik bez szyfrowania i z szyfrowaniem na grafie z artykułu o human-in-the-loop:

from pathlib import Path

from cryptography.fernet import Fernet, InvalidToken
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command

import review
from vault import encrypted_saver, make_fernet

TEXT = "Klient 44051401359 pisze z a@b.pl. Faktura 600123456."
start = {"text": TEXT, "findings": [], "approved": [], "pseudonymized": "", "mapping": {}}
config = {"configurable": {"thread_id": "z1"}}
fernet = make_fernet(Fernet.generate_key().decode())

for name in ("plain.db", "encrypted.db"):
    Path(name).unlink(missing_ok=True)

with SqliteSaver.from_conn_string("plain.db") as saver:
    review.build(saver).invoke(start, config)
with encrypted_saver("encrypted.db", fernet) as saver:
    review.build(saver).invoke(start, config)

for name in ("plain.db", "encrypted.db"):
    raw = Path(name).read_bytes()
    print(name, "| PESEL:", b"44051401359" in raw, "| e-mail:", b"a@b.pl" in raw, "| thread_id:", b"z1" in raw)

with encrypted_saver("encrypted.db", fernet) as saver:
    graph = review.build(saver)
    result = graph.invoke(Command(resume={"approved": [0, 1]}), config)
print(result["pseudonymized"])
print("PESEL w pliku po wznowieniu:", b"44051401359" in Path("encrypted.db").read_bytes())

other = make_fernet(Fernet.generate_key().decode())
with encrypted_saver("encrypted.db", other) as saver:
    try:
        review.build(saver).get_state(config)
    except InvalidToken:
        print("zły klucz: InvalidToken")

Wynik:

plain.db | PESEL: True | e-mail: True | thread_id: True
encrypted.db | PESEL: False | e-mail: False | thread_id: True
Klient [PESEL_1] pisze z [EMAIL_1]. Faktura 600123456.
PESEL w pliku po wznowieniu: False
zły klucz: InvalidToken

W zwykłym pliku PESEL i e-mail są widoczne, w zaszyfrowanym nie ma żadnego z nich, także po wznowieniu grafu. Graf działa tak samo: zatrzymuje się na interrupt, a po Command(resume=...) kończy pracę. Z innym kluczem nie odczytasz nawet stanu, bo get_state kończy się InvalidToken.

Co zostaje w otwartym tekście?

Szyfrowane są bajty checkpointów, nie cała baza. W moim teście thread_id nadal było widoczne w pliku, a metadane sprawdzonego wiersza (source, step, parents) to zwykły JSON bez treści stanu. Wniosek praktyczny jest taki sam jak w artykule o checkpointach: nazwa wątku nie może zawierać danych osobowych.

Co zrobić ze starymi checkpointami sprzed szyfrowania?

Włączenie szyfrowania nie szyfruje wstecz. Zaszyfrowany saver odczytuje stare, niezaszyfrowane wiersze bez błędu, a w pliku nadal leży PESEL w otwartym tekście. Sprawdza to test test_encrypted_saver_reads_old_plain_checkpoints z końca artykułu. Dlatego przy włączaniu szyfrowania na istniejącym pliku usuń stare wątki albo cały plik. Migracji wierszy (odczyt i zapis szyfrem) nie sprawdzałem, więc jej nie opisuję.

Czy szyfrować cały plik, czy tylko mapowanie?

Zacznij od pytania, co w ogóle musi leżeć w stanie. Jeśli oryginalny tekst i mapowanie nie muszą przetrwać restartu, najtańsze zabezpieczenie to ich nie zapisywać: po podmianie usuń je ze stanu, a mapowanie zapisz osobno, zaszyfrowane. Szyfrowanie całego pliku checkpointów dodaj wtedy, gdy graf musi czekać na człowieka godzinami i stan z oryginałami przez ten czas leży na dysku. Obie warstwy się uzupełniają, ale pierwsza oznacza mniej danych do ochrony.

Jak wymienić klucz szyfrowania (key rotation)?

Klucze warto wymieniać okresowo i zawsze po podejrzeniu wycieku. MultiFernet przyjmuje listę kluczy: szyfruje pierwszym, a odszyfrowuje dowolnym z listy. Dzięki temu nowy klucz dodajesz na początek, a stare dane nadal czytasz. Metoda rotate przepisuje token nowym kluczem:

old, new = Fernet.generate_key(), Fernet.generate_key()
sealed = seal_mapping({"[PESEL_1]": "44051401359"}, make_fernet(old.decode()))

both = make_fernet(f"{new.decode()},{old.decode()}")
rotated = both.rotate(sealed)

Po rotate token odczytasz samym nowym kluczem, a stary klucz już go nie odczyta. Zachowanie to pokrywa test test_rotation_keeps_old_data_readable. Rotację pokazałem tylko dla zaszyfrowanego mapowania. Dla pliku checkpointów oznaczałaby przepisanie wszystkich wierszy, czego nie sprawdzałem.

Jakie są ograniczenia takiego szyfrowania?

Szyfrowanie mapowania jest środkiem technicznym. Nie zamienia pseudonimizacji w anonimizację i samo nie zapewnia zgodności z RODO; o tym, czym pseudonimizacja jest, piszę w artykule o pseudonimizacji danych.

Jak przetestować szyfrowanie w pytest?

Testy sprawdzają zarówno szyfrowanie mapowania, jak i plik checkpointów:

import sqlite3

import pytest
from cryptography.fernet import Fernet, InvalidToken
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command

import review
from pseudonymize import restore
from vault import encrypted_saver, fernet_from_env, make_fernet, open_mapping, seal_mapping

PESEL = "44051401359"
TEXT = f"Klient {PESEL} pisze z a@b.pl. Faktura 600123456."
START = {"text": TEXT, "findings": [], "approved": [], "pseudonymized": "", "mapping": {}}
CONFIG = {"configurable": {"thread_id": "z1"}}


@pytest.fixture
def fernet():
    return make_fernet(Fernet.generate_key().decode())


def test_mapping_round_trip(fernet):
    mapping = {"[PESEL_1]": PESEL, "[EMAIL_1]": "a@b.pl"}
    assert open_mapping(seal_mapping(mapping, fernet), fernet) == mapping


def test_sealed_mapping_hides_values(fernet):
    sealed = seal_mapping({"[PESEL_1]": PESEL}, fernet)
    assert PESEL.encode() not in sealed
    assert b"PESEL_1" not in sealed


def test_wrong_key_is_rejected(fernet):
    sealed = seal_mapping({"[PESEL_1]": PESEL}, fernet)
    other = make_fernet(Fernet.generate_key().decode())
    with pytest.raises(InvalidToken):
        open_mapping(sealed, other)


def test_tampered_mapping_is_rejected(fernet):
    sealed = bytearray(seal_mapping({"[PESEL_1]": PESEL}, fernet))
    sealed[-5] ^= 1
    with pytest.raises(InvalidToken):
        open_mapping(bytes(sealed), fernet)


def test_restore_with_opened_mapping(fernet):
    sealed = seal_mapping({"[PESEL_1]": PESEL}, fernet)
    assert restore("Klient [PESEL_1].", open_mapping(sealed, fernet)) == f"Klient {PESEL}."


def test_rotation_keeps_old_data_readable():
    old, new = Fernet.generate_key(), Fernet.generate_key()
    sealed = seal_mapping({"[PESEL_1]": PESEL}, make_fernet(old.decode()))
    both = make_fernet(f"{new.decode()},{old.decode()}")
    assert open_mapping(sealed, both) == {"[PESEL_1]": PESEL}
    rotated = both.rotate(sealed)
    assert open_mapping(rotated, make_fernet(new.decode())) == {"[PESEL_1]": PESEL}
    with pytest.raises(InvalidToken):
        open_mapping(rotated, make_fernet(old.decode()))


def test_missing_env_key(monkeypatch):
    monkeypatch.delenv("PSEUDONYM_KEYS", raising=False)
    with pytest.raises(RuntimeError, match="PSEUDONYM_KEYS"):
        fernet_from_env()


def test_env_key(monkeypatch, fernet):
    key = Fernet.generate_key().decode()
    monkeypatch.setenv("PSEUDONYM_KEYS", key)
    assert open_mapping(seal_mapping({"a": "b"}, make_fernet(key)), fernet_from_env()) == {"a": "b"}


def test_checkpoint_file_has_no_plaintext(tmp_path, fernet):
    path = str(tmp_path / "enc.db")
    with encrypted_saver(path, fernet) as saver:
        review.build(saver).invoke(START, CONFIG)
    raw = (tmp_path / "enc.db").read_bytes()
    assert PESEL.encode() not in raw
    assert b"a@b.pl" not in raw


def test_encrypted_graph_resumes_with_same_key(tmp_path, fernet):
    path = str(tmp_path / "enc.db")
    with encrypted_saver(path, fernet) as saver:
        review.build(saver).invoke(START, CONFIG)
    with encrypted_saver(path, fernet) as saver:
        out = review.build(saver).invoke(Command(resume={"approved": [0, 1]}), CONFIG)
    assert out["pseudonymized"] == "Klient [PESEL_1] pisze z [EMAIL_1]. Faktura 600123456."
    assert PESEL.encode() not in (tmp_path / "enc.db").read_bytes()


def test_encrypted_graph_cannot_resume_with_wrong_key(tmp_path, fernet):
    path = str(tmp_path / "enc.db")
    with encrypted_saver(path, fernet) as saver:
        review.build(saver).invoke(START, CONFIG)
    other = make_fernet(Fernet.generate_key().decode())
    with encrypted_saver(path, other) as saver:
        with pytest.raises(InvalidToken):
            review.build(saver).get_state(CONFIG)


def test_plain_file_has_plaintext(tmp_path):
    path = str(tmp_path / "plain.db")
    with SqliteSaver.from_conn_string(path) as saver:
        review.build(saver).invoke(START, CONFIG)
    assert PESEL.encode() in (tmp_path / "plain.db").read_bytes()


def test_encrypted_saver_reads_old_plain_checkpoints(tmp_path, fernet):
    path = str(tmp_path / "old.db")
    with SqliteSaver.from_conn_string(path) as saver:
        review.build(saver).invoke(START, CONFIG)
    with encrypted_saver(path, fernet) as saver:
        state = review.build(saver).get_state(CONFIG)
    assert state.values["text"] == TEXT
    assert PESEL.encode() in (tmp_path / "old.db").read_bytes()

Test test_rotation_keeps_old_data_readable dokumentuje rotację, a test_encrypted_saver_reads_old_plain_checkpoints opisaną wyżej pułapkę: gdyby w przyszłej wersji LangGraph albo biblioteki to zachowanie się zmieniło, test to wykryje.

Co zbudowaliśmy w tym cyklu i co dalej?

Przeszliśmy całą drogę od podstaw do gotowego programu:

1. Graf w programowaniu: węzły, krawędzie, przejście po grafie.
2. Stan, węzły i krawędzie w LangGraph.
3. Wykrywanie danych wrażliwych: wyrażenia regularne i suma kontrolna PESEL.
4. Pseudonimizacja: tokeny i mapowanie.
5. Checkpointy w SQLite.
6. Human-in-the-loop: zatwierdzanie przez operatora.
7. Szyfrowanie mapowania i pliku checkpointów (ten artykuł).

Z tych elementów składa się graf, który wykrywa dane wrażliwe, pokazuje je operatorowi, podmienia tylko zatwierdzone, zapisuje stan w zaszyfrowanym pliku i da się go wznowić po przerwie. Podobny projekt realizujemy w mentoringu AI Developer: budujesz go krok po kroku z mentorem, który ocenia Twój kod. Jeśli chcesz zrobić taki projekt razem z nami, zobacz ścieżki rozwoju w devmentor.pl i skontaktuj się z nami.