add release docs, add docs, finish tests
This commit is contained in:
163
docs/fields/encryption.md
Normal file
163
docs/fields/encryption.md
Normal file
@ -0,0 +1,163 @@
|
|||||||
|
# Encryption
|
||||||
|
|
||||||
|
`ormar` provides you with a way to encrypt a field in the database only.
|
||||||
|
Provided encryption backends allow for both one-way encryption (`HASH` backend) as
|
||||||
|
well as both-way encryption/decryption (`FERNET` backend).
|
||||||
|
|
||||||
|
!!!warning
|
||||||
|
Note that in order for encryption to work you need to install optional `cryptography` package.
|
||||||
|
|
||||||
|
You can do it manually `pip install cryptography` or with ormar by `pip install ormar[crypto]`
|
||||||
|
|
||||||
|
!!!warning
|
||||||
|
Note that adding `encrypt_backend` changes the database column type to `TEXT`,
|
||||||
|
which needs to be reflected in db either by migration (`alembic`) or manual change
|
||||||
|
|
||||||
|
## Defining a field encryption
|
||||||
|
|
||||||
|
To encrypt a field you need to pass at minimum `encrypt_secret` and `encrypt_backend` parameters.
|
||||||
|
|
||||||
|
```python hl_lines="7-8"
|
||||||
|
class Filter(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "filters"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=100,
|
||||||
|
encrypt_secret="secret123",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.FERNET)
|
||||||
|
```
|
||||||
|
|
||||||
|
!!!warning
|
||||||
|
You can encrypt all `Field` types apart from `primary_key` column and relation
|
||||||
|
columns (`ForeignKey` and `ManyToMany`). Check backends details for more information.
|
||||||
|
|
||||||
|
## Available backends
|
||||||
|
|
||||||
|
### HASH
|
||||||
|
|
||||||
|
HASH is a one-way hash (like for password), never decrypted on retrieval
|
||||||
|
|
||||||
|
To set it up pass appropriate backend value.
|
||||||
|
|
||||||
|
```python
|
||||||
|
... # rest of model definition
|
||||||
|
password: str = ormar.String(max_length=128,
|
||||||
|
encrypt_secret="secret123",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.HASH)
|
||||||
|
```
|
||||||
|
|
||||||
|
Note that since this backend never decrypt the stored value it's only applicable for
|
||||||
|
`String` fields. Used hash is a `sha512` hash, so the field length has to be >=128.
|
||||||
|
|
||||||
|
!!!warning
|
||||||
|
Note that in `HASH` backend you can filter by full value but filters like `contain` will not work as comparison is make on encrypted values
|
||||||
|
|
||||||
|
!!!note
|
||||||
|
Note that provided `encrypt_secret` is first hashed itself and used as salt, so in order to
|
||||||
|
compare to stored string you need to recreate this steps.
|
||||||
|
|
||||||
|
```python
|
||||||
|
class Hash(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "hashes"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=128,
|
||||||
|
encrypt_secret="udxc32",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.HASH)
|
||||||
|
|
||||||
|
|
||||||
|
await Hash(name='test1').save()
|
||||||
|
|
||||||
|
# note the steps to recreate the stored value
|
||||||
|
# you can use also cryptography package instead of hashlib
|
||||||
|
secret = hashlib.sha256("udxc32".encode()).digest()
|
||||||
|
secret = base64.urlsafe_b64encode(secret)
|
||||||
|
hashed_test1 = hashlib.sha512(secret + 'test1'.encode()).hexdigest()
|
||||||
|
|
||||||
|
# full value comparison works
|
||||||
|
hash1 = await Hash.objects.get(name='test1')
|
||||||
|
assert hash1.name == hashed_test1
|
||||||
|
|
||||||
|
# but partial comparison does not (hashed strings are compared)
|
||||||
|
with pytest.raises(NoMatch):
|
||||||
|
await Filter.objects.get(name__icontains='test')
|
||||||
|
```
|
||||||
|
|
||||||
|
### FERNET
|
||||||
|
|
||||||
|
FERNET is a two-way encrypt/decrypt backend
|
||||||
|
|
||||||
|
To set it up pass appropriate backend value.
|
||||||
|
|
||||||
|
```python
|
||||||
|
... # rest of model definition
|
||||||
|
year: int = ormar.Integer(encrypt_secret="secret123",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.FERNET)
|
||||||
|
```
|
||||||
|
|
||||||
|
Value is encrypted on way to database end decrypted on way out. Can be used on all types,
|
||||||
|
as the returned value is parsed to corresponding python type.
|
||||||
|
|
||||||
|
!!!warning
|
||||||
|
Note that in `FERNET` backend you loose `filtering` possibility altogether as part of the encrypted value is a timestamp
|
||||||
|
|
||||||
|
```python
|
||||||
|
class Filter(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "filters"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=100,
|
||||||
|
encrypt_secret="asd123",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.FERNET)
|
||||||
|
|
||||||
|
await Filter(name='test1').save()
|
||||||
|
await Filter(name='test1').save()
|
||||||
|
|
||||||
|
# values are properly encrypted and later decrypted
|
||||||
|
filters = await Filter.objects.all()
|
||||||
|
assert filters[0].name == filters[1].name == 'test1'
|
||||||
|
|
||||||
|
# but you cannot filter at all since part of the fernet hash is a timestamp
|
||||||
|
# which means that even if you encrypt the same string 2 times it will be different
|
||||||
|
with pytest.raises(NoMatch):
|
||||||
|
await Filter.objects.get(name='test1')
|
||||||
|
```
|
||||||
|
|
||||||
|
## Custom Backends
|
||||||
|
|
||||||
|
If you wish to support other type of encryption (i.e. AES) you can provide your own `EncryptionBackend`.
|
||||||
|
|
||||||
|
To setup a backend all you need to do is subclass `ormar.fields.EncryptBackend` class and provide required backend.
|
||||||
|
|
||||||
|
Sample dummy backend (that does nothing) can look like following:
|
||||||
|
|
||||||
|
```python
|
||||||
|
class DummyBackend(ormar.fields.EncryptBackend):
|
||||||
|
def _initialize_backend(self, secret_key: bytes) -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
|
def encrypt(self, value: Any) -> str:
|
||||||
|
return value
|
||||||
|
|
||||||
|
def decrypt(self, value: Any) -> str:
|
||||||
|
return value
|
||||||
|
```
|
||||||
|
|
||||||
|
To use this backend set `encrypt_backend` to `CUSTOM` and provide your backend as
|
||||||
|
argument by `encrypt_custom_backend`.
|
||||||
|
|
||||||
|
```python
|
||||||
|
class Filter(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "filters"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=100,
|
||||||
|
encrypt_secret="secret123",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.CUSTOM,
|
||||||
|
encrypt_custom_backend=DummyBackend
|
||||||
|
)
|
||||||
|
```
|
||||||
@ -1,3 +1,21 @@
|
|||||||
|
# 0.9.8
|
||||||
|
|
||||||
|
## Features
|
||||||
|
* Add possibility to encrypt the selected field(s) in the database
|
||||||
|
* As minimum you need to provide `encrypt_secret` and `encrypt_backend`
|
||||||
|
* `encrypt_backend` can be one of the `ormar.EncryptBackends` enum (`NONE, FERNET, HASH, CUSTOM`) - default: `NONE`
|
||||||
|
* When custom backend is selected you need to provide your backend class that subclasses `ormar.fields.EncryptBackend`
|
||||||
|
* You cannot encrypt `primary_key` column and relation columns (FK and M2M).
|
||||||
|
* Provided are 2 backends: HASH and FERNET
|
||||||
|
* HASH is a one-way hash (like for password), never decrypted on retrieval
|
||||||
|
* FERNET is a two-way encrypt/decrypt backend
|
||||||
|
* Note that in FERNET backend you loose `filtering` possibility altogether as part of the encrypted value is a timestamp.
|
||||||
|
* Note that in HASH backend you can filter by full value but filters like `contain` will not work as comparison is make on encrypted values
|
||||||
|
* Note that adding `encrypt_backend` changes the database column type to `TEXT`, which needs to be reflected in db either by migration or manual change
|
||||||
|
|
||||||
|
## Fixes
|
||||||
|
* (Advanced/ Internal) Restore custom sqlalchemy types (by `types.TypeDecorator` subclass) functionality that ceased to working so `process_result_value` was never called
|
||||||
|
|
||||||
# 0.9.7
|
# 0.9.7
|
||||||
|
|
||||||
## Features
|
## Features
|
||||||
|
|||||||
@ -12,6 +12,7 @@ nav:
|
|||||||
- Fields:
|
- Fields:
|
||||||
- Common parameters: fields/common-parameters.md
|
- Common parameters: fields/common-parameters.md
|
||||||
- Fields types: fields/field-types.md
|
- Fields types: fields/field-types.md
|
||||||
|
- Fields encryption: fields/encryption.md
|
||||||
- Relations:
|
- Relations:
|
||||||
- relations/index.md
|
- relations/index.md
|
||||||
- relations/postponed-annotations.md
|
- relations/postponed-annotations.md
|
||||||
|
|||||||
@ -31,7 +31,6 @@ class EncryptBackend(abc.ABC):
|
|||||||
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
|
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
|
||||||
digest.update(key)
|
digest.update(key)
|
||||||
engine_key = digest.finalize()
|
engine_key = digest.finalize()
|
||||||
|
|
||||||
self._initialize_backend(engine_key)
|
self._initialize_backend(engine_key)
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
@ -165,7 +164,8 @@ class EncryptedString(types.TypeDecorator):
|
|||||||
if encoder:
|
if encoder:
|
||||||
value = encoder(value) # type: ignore
|
value = encoder(value) # type: ignore
|
||||||
|
|
||||||
return self.backend.encrypt(value)
|
encrypted_value = self.backend.encrypt(value)
|
||||||
|
return encrypted_value
|
||||||
|
|
||||||
def process_result_value(self, value: Any, dialect: DefaultDialect) -> Any:
|
def process_result_value(self, value: Any, dialect: DefaultDialect) -> Any:
|
||||||
if value is None:
|
if value is None:
|
||||||
|
|||||||
3
setup.py
3
setup.py
@ -62,7 +62,8 @@ setup(
|
|||||||
"postgresql": ["asyncpg", "psycopg2"],
|
"postgresql": ["asyncpg", "psycopg2"],
|
||||||
"mysql": ["aiomysql", "pymysql"],
|
"mysql": ["aiomysql", "pymysql"],
|
||||||
"sqlite": ["aiosqlite"],
|
"sqlite": ["aiosqlite"],
|
||||||
"orjson": ["orjson"]
|
"orjson": ["orjson"],
|
||||||
|
"crypto": ["cryptography"]
|
||||||
},
|
},
|
||||||
classifiers=[
|
classifiers=[
|
||||||
"Development Status :: 4 - Beta",
|
"Development Status :: 4 - Beta",
|
||||||
|
|||||||
@ -1,5 +1,7 @@
|
|||||||
# type: ignore
|
# type: ignore
|
||||||
|
import base64
|
||||||
import decimal
|
import decimal
|
||||||
|
import hashlib
|
||||||
import uuid
|
import uuid
|
||||||
import datetime
|
import datetime
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@ -9,7 +11,7 @@ import pytest
|
|||||||
import sqlalchemy
|
import sqlalchemy
|
||||||
|
|
||||||
import ormar
|
import ormar
|
||||||
from ormar import ModelDefinitionError
|
from ormar import ModelDefinitionError, NoMatch
|
||||||
from ormar.fields.sqlalchemy_encrypted import EncryptedString
|
from ormar.fields.sqlalchemy_encrypted import EncryptedString
|
||||||
from tests.settings import DATABASE_URL
|
from tests.settings import DATABASE_URL
|
||||||
|
|
||||||
@ -75,6 +77,26 @@ class Author(ormar.Model):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class Filter(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "filters"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=100, **default_fernet)
|
||||||
|
|
||||||
|
|
||||||
|
class Hash(ormar.Model):
|
||||||
|
class Meta(BaseMeta):
|
||||||
|
tablename = "hashes"
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(
|
||||||
|
max_length=128,
|
||||||
|
encrypt_secret="udxc32",
|
||||||
|
encrypt_backend=ormar.EncryptBackends.HASH,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True, scope="module")
|
@pytest.fixture(autouse=True, scope="module")
|
||||||
def create_test_database():
|
def create_test_database():
|
||||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||||
@ -184,3 +206,33 @@ async def test_save_and_retrieve():
|
|||||||
assert author.test_decimal == 3.5
|
assert author.test_decimal == 3.5
|
||||||
assert author.test_decimal2 == 5.5
|
assert author.test_decimal2 == 5.5
|
||||||
assert author.custom_backend == "test12"
|
assert author.custom_backend == "test12"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_fernet_filters_nomatch():
|
||||||
|
async with database:
|
||||||
|
await Filter(name="test1").save()
|
||||||
|
await Filter(name="test1").save()
|
||||||
|
|
||||||
|
filters = await Filter.objects.all()
|
||||||
|
assert filters[0].name == filters[1].name == "test1"
|
||||||
|
|
||||||
|
with pytest.raises(NoMatch):
|
||||||
|
await Filter.objects.get(name="test1")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_hash_filters_works():
|
||||||
|
async with database:
|
||||||
|
await Hash(name="test1").save()
|
||||||
|
await Hash(name="test2").save()
|
||||||
|
|
||||||
|
secret = hashlib.sha256("udxc32".encode()).digest()
|
||||||
|
secret = base64.urlsafe_b64encode(secret)
|
||||||
|
hashed_test1 = hashlib.sha512(secret + "test1".encode()).hexdigest()
|
||||||
|
|
||||||
|
hash1 = await Hash.objects.get(name="test1")
|
||||||
|
assert hash1.name == hashed_test1
|
||||||
|
|
||||||
|
with pytest.raises(NoMatch):
|
||||||
|
await Filter.objects.get(name__icontains="test")
|
||||||
|
|||||||
Reference in New Issue
Block a user