Initial commit

Author: script3r

.gitignore

@@ -0,0 +1,156 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+build/
+
+# PyBuilder
+.pybuilder/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#   Therefore, this is a recommendation and not a requirement.
+# Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially true for binary packages to ensure reproducibility, and is more commonly
+#   ignored for libraries.
+# poetry.lock
+
+# pdm
+.pdm-build/
+.pdm-python
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype type checker
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# Ruff
+.ruff_cache/
+
+# PyCharm
+.idea/
+
+# VS Code
+.vscode/
+
+# MacOS
+.DS_Store
+
+# Windows
+Thumbs.db
+
+# Local tooling
+.pytest_cache/

README.md

@@ -0,0 +1,71 @@
+# SQLModel Encrypted Fields
+
+Encrypt SQLModel fields with Tink AEAD in a few lines, and keep ciphertext out of sight.
+
+## Install
+
+```bash
+pip install sqlmodel-encrypted-fields
+```
+
+## Quickstart
+
+```python
+from sqlalchemy import Column
+from sqlmodel import Field, SQLModel
+
+from sqlmodel_encrypted_fields import (
+    configure_keysets,
+    EncryptedString,
+    DeterministicEncryptedString,
+)
+
+configure_keysets(
+    {
+        "default": {"path": "/path/to/aead_keyset.json", "cleartext": True},
+        "searchable": {"path": "/path/to/daead_keyset.json", "cleartext": True},
+    }
+)
+
+
+class Customer(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    email: str = Field(sa_column=Column(EncryptedString()))
+    email_lookup: str = Field(sa_column=Column(DeterministicEncryptedString(keyset="searchable")))
+```
+
+## Notes
+
+Regular AEAD fields change ciphertext on every write. Use deterministic fields only when you need equality lookups.
+
+## Supported Fields
+
+- `EncryptedType` (custom serializer/deserializer)
+- `EncryptedString`
+- `EncryptedJSON`
+- `EncryptedBytes`
+- `DeterministicEncryptedType` (custom serializer/deserializer)
+- `DeterministicEncryptedString`
+- `DeterministicEncryptedJSON`
+- `DeterministicEncryptedBytes`
+
+## Custom Serialization
+
+```python
+from datetime import date
+
+from sqlmodel_encrypted_fields import EncryptedType
+
+
+def serialize_date(value: date) -> str:
+    return value.isoformat()
+
+
+def deserialize_date(value: str) -> date:
+    return date.fromisoformat(value)
+
+
+class User(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    birthday: date = Field(sa_column=EncryptedType(serializer=serialize_date, deserializer=deserialize_date))
+```

example_app/__init__.py

@@ -0,0 +1 @@
+"""Minimal FastAPI + SQLModel example app."""

example_app/database.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from sqlmodel import Session, SQLModel, create_engine
+
+engine = create_engine("sqlite:///./example_app.db", echo=False)
+
+
+def init_db() -> None:
+    SQLModel.metadata.create_all(engine)
+
+
+def get_session() -> Session:
+    return Session(engine)

example_app/main.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+from fastapi import Depends, FastAPI, HTTPException
+from sqlmodel import Session, select
+
+from example_app.database import get_session, init_db
+from example_app.models import Customer
+from sqlmodel_encrypted_fields import configure_keysets
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    configure_keysets(
+        {
+            "default": {"path": "./tests/fixtures/aead_keyset.json", "cleartext": True},
+            "deterministic": {"path": "./tests/fixtures/daead_keyset.json", "cleartext": True},
+        }
+    )
+    init_db()
+    yield
+
+
+app = FastAPI(title="SQLModel Encrypted Fields Example", lifespan=lifespan)
+
+
+@app.post("/customers", response_model=Customer)
+def create_customer(customer: Customer, session: Session = Depends(get_session)) -> Customer:
+    session.add(customer)
+    session.commit()
+    session.refresh(customer)
+    return customer
+
+
+@app.get("/customers/{customer_id}", response_model=Customer)
+def get_customer(customer_id: int, session: Session = Depends(get_session)) -> Customer:
+    customer = session.get(Customer, customer_id)
+    if customer is None:
+        raise HTTPException(status_code=404, detail="Customer not found")
+    return customer
+
+
+@app.get("/customers/by-email/{email}", response_model=Customer)
+def get_customer_by_email(email: str, session: Session = Depends(get_session)) -> Customer:
+    statement = select(Customer).where(Customer.email_lookup == email)
+    customer = session.exec(statement).first()
+    if customer is None:
+        raise HTTPException(status_code=404, detail="Customer not found")
+    return customer

example_app/models.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from sqlalchemy import Column
+from sqlmodel import Field, SQLModel
+
+from sqlmodel_encrypted_fields import DeterministicEncryptedString, EncryptedString
+
+
+class Customer(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    email: str = Field(sa_column=Column(EncryptedString()))
+    email_lookup: str = Field(sa_column=Column(DeterministicEncryptedString(keyset="deterministic")))

example_app/tests/test_integration.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session, SQLModel, create_engine, select
+
+from example_app.database import get_session
+from example_app.main import app
+from example_app.models import Customer
+from sqlmodel_encrypted_fields import configure_keysets
+
+
+def _project_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _configure_keysets() -> None:
+    root = _project_root()
+    configure_keysets(
+        {
+            "default": {"path": str(root / "tests" / "fixtures" / "aead_keyset.json"), "cleartext": True},
+            "deterministic": {"path": str(root / "tests" / "fixtures" / "daead_keyset.json"), "cleartext": True},
+        }
+    )
+
+
+def _test_engine(tmp_path: Path):
+    return create_engine(f"sqlite:///{tmp_path / 'test.db'}", echo=False)
+
+
+def test_customer_create_and_lookup(tmp_path: Path) -> None:
+    _configure_keysets()
+
+    engine = _test_engine(tmp_path)
+    SQLModel.metadata.create_all(engine)
+
+    def _session_override():
+        with Session(engine) as session:
+            yield session
+
+    app.dependency_overrides[get_session] = _session_override
+    client = TestClient(app)
+
+    payload = {"email": "alice@example.com", "email_lookup": "alice@example.com"}
+    response = client.post("/customers", json=payload)
+    assert response.status_code == 200
+    customer_id = response.json()["id"]
+
+    response = client.get(f"/customers/{customer_id}")
+    assert response.status_code == 200
+    assert response.json()["email"] == payload["email"]
+
+    response = client.get(f"/customers/by-email/{payload['email']}")
+    assert response.status_code == 200
+    assert response.json()["id"] == customer_id
+
+    with engine.connect() as connection:
+        raw = connection.exec_driver_sql(
+            "select email from customer where id = ?",
+            (customer_id,),
+        ).fetchone()[0]
+        assert isinstance(raw, (bytes, memoryview))
+        raw_bytes = raw.tobytes() if isinstance(raw, memoryview) else raw
+        assert payload["email"].encode("utf-8") not in raw_bytes
+
+    with Session(engine) as session:
+        statement = select(Customer).where(Customer.email_lookup == payload["email"])
+        customer = session.exec(statement).first()
+        assert customer is not None
+
+    app.dependency_overrides.clear()