diff --git a/nutkit/frontend/driver.py b/nutkit/frontend/driver.py index 095cdd779..108c27ba7 100644 --- a/nutkit/frontend/driver.py +++ b/nutkit/frontend/driver.py @@ -24,7 +24,8 @@ def __init__(self, backend, uri, auth_token, user_agent=None, notifications_disabled_categories=None, telemetry_disabled=None, client_certificate=None, - disable_auto_commit_retries=None): + disable_auto_commit_retries=None, + property_encryption_profiles=None): self._backend = backend self._resolver_fn = resolver_fn self._domain_name_resolver_fn = domain_name_resolver_fn @@ -54,6 +55,13 @@ def __init__(self, backend, uri, auth_token, user_agent=None, else: client_certificate_provider_id_ = client_certificate.id + property_encryption_profiles_ = None + if property_encryption_profiles is not None: + property_encryption_profiles_ = [ + self._encryption_profile_wire(p) + for p in property_encryption_profiles + ] + req = protocol.NewDriver( uri, self._auth_token, auth_token_manager_id, userAgent=user_agent, resolverRegistered=resolver_fn is not None, @@ -71,6 +79,7 @@ def __init__(self, backend, uri, auth_token, user_agent=None, client_certificate=client_certificate_, client_certificate_provider_id=client_certificate_provider_id_, disable_auto_commit_retries=disable_auto_commit_retries, + property_encryption_profiles=property_encryption_profiles_, ) res = backend.send_and_receive(req) if not isinstance(res, protocol.Driver): @@ -78,6 +87,16 @@ def __init__(self, backend, uri, auth_token, user_agent=None, self._driver = res self._closed = False + @staticmethod + def _encryption_profile_wire(profile): + if isinstance(profile, str): + return {"name": profile} + wire = {"name": profile["name"]} + kek = profile.get("kek") + if kek is not None: + wire["kek"] = kek.hex(" ") + return wire + def receive(self, timeout=None, hooks=None, *, allow_resolution): while True: res = self._backend.receive(timeout=timeout, hooks=hooks) @@ -188,6 +207,48 @@ def is_encrypted(self): raise Exception(f"Should be DriverIsEncrypted but was {res}") return res.encrypted + def encrypt_to_bytes(self, value, *, aad=None, profile_name=None, + key_alias=None, key_id=None, iv=None): + iv_ = iv.hex(" ") if iv is not None else None + req = protocol.EncryptToBytes( + self._driver.id, value, aad=aad, profile_name=profile_name, + key_alias=key_alias, key_id=key_id, iv=iv_ + ) + res = self.send_and_receive(req, allow_resolution=False) + if not isinstance(res, protocol.EncryptedValue): + raise Exception(f"Should be EncryptedValue but was: {res}") + return res.encrypted_bytes + + def decrypt(self, value, *, aad=None, use_persisted_aad=False): + req = protocol.Decrypt( + self._driver.id, value.hex(" "), aad=aad, + use_persisted_aad=use_persisted_aad + ) + res = self.send_and_receive(req, allow_resolution=False) + if not isinstance(res, protocol.DecryptedValue): + raise Exception(f"Should be DecryptedValue but was: {res}") + return res.decrypted_value + + def create_encapsulated_key(self, alias, *, profile_name=None): + req = protocol.CreateEncapsulatedKey( + self._driver.id, alias, profile_name=profile_name + ) + res = self.send_and_receive(req, allow_resolution=False) + if not isinstance(res, protocol.EncapsulatedKey): + raise Exception(f"Should be EncapsulatedKey but was: {res}") + return res + + def import_encapsulated_key(self, alias, encapsulation, metadata, *, + profile_name=None): + req = protocol.ImportEncapsulatedKey( + self._driver.id, alias, encapsulation.hex(" "), metadata, + profile_name=profile_name + ) + res = self.send_and_receive(req, allow_resolution=False) + if not isinstance(res, protocol.EncapsulatedKey): + raise Exception(f"Should be EncapsulatedKey but was: {res}") + return res + def close(self): req = protocol.DriverClose(self._driver.id) res = self.send_and_receive(req, allow_resolution=False) diff --git a/nutkit/protocol/feature.py b/nutkit/protocol/feature.py index f836a3a17..504aea581 100644 --- a/nutkit/protocol/feature.py +++ b/nutkit/protocol/feature.py @@ -44,6 +44,10 @@ class Feature(Enum): API_DRIVER_SUPPORTS_SESSION_AUTH = "Feature:API:Driver.SupportsSessionAuth" # The driver supports connection liveness check. API_LIVENESS_CHECK = "Feature:API:Liveness.Check" + # The driver offers a public API for client-side property encryption: + # encrypting and decrypting individual property values, and managing + # encapsulated data encryption keys. + API_PROPERTY_ENCRYPTION = "Feature:API:PropertyEncryption" # The driver offers a method for the result to return all records as a list # or array. This method should exhaust the result. API_RESULT_LIST = "Feature:API:Result.List" diff --git a/nutkit/protocol/requests.py b/nutkit/protocol/requests.py index 7583f7148..c7990a03a 100644 --- a/nutkit/protocol/requests.py +++ b/nutkit/protocol/requests.py @@ -79,6 +79,7 @@ def __init__( telemetry_disabled=None, client_certificate=None, client_certificate_provider_id=None, disable_auto_commit_retries=None, + property_encryption_profiles=None, ): # Neo4j URI to connect to self.uri = uri @@ -110,6 +111,8 @@ def __init__( self.telemetryDisabled = telemetry_disabled if disable_auto_commit_retries is not None: self.disableAutoCommitRetries = disable_auto_commit_retries + if property_encryption_profiles is not None: + self.propertyEncryptionProfiles = property_encryption_profiles # (bool) whether to enable or disable encryption # field missing in message: use driver default (should be False) if encrypted is not None: @@ -891,3 +894,111 @@ class FakeTimeUninstall: """ pass + + +class EncryptToBytes: + """ + Request to encrypt a value using client-side property encryption. + + The backend should respond with an EncryptedValue or an Error response. + + :param driver_id: The id of the driver to encrypt with. + :param value: The property value to encrypt. + :param aad: The additional authenticated data (AAD) to bind to the + ciphertext, or None for no AAD. + :param profile_name: The name of the encryption profile to use, or None + to use the sole configured profile. + :param key_alias: The alias of the data encryption key to encrypt with. + Mutually exclusive with key_id; exactly one of the two must be set. + :param key_id: The repository-assigned id of the data encryption key to + encrypt with. Mutually exclusive with key_alias; exactly one of the + two must be set. + :param iv: The exact 12-byte IV the driver must use for this + encrypt call, or None to draw a random one. The backend raises if + the IV is not exactly 12 bytes or the operation doesn't consume it. + Used to assert byte-exact ciphertext in the deterministic + encryption tests. + """ + + def __init__(self, driver_id, value, aad=None, profile_name=None, + key_alias=None, key_id=None, iv=None): + self.driverId = driver_id + self.value = value + self.aad = aad + self.profileName = profile_name + self.keyAlias = key_alias + self.keyId = key_id + self.iv = iv + + +class Decrypt: + """ + Request to decrypt a value using client-side property encryption. + + The backend should respond with a DecryptedValue or an Error response. + + :param driver_id: The id of the driver to decrypt with. + :param value: The encrypted value to decrypt, as returned by + EncryptToBytes. + :param aad: The additional authenticated data (AAD) to reproduce, or + None to use the AAD persisted alongside the encrypted value. + Mutually exclusive with use_persisted_aad; exactly one of the two + must be set. + :param use_persisted_aad: Whether to use the AAD persisted alongside the + encrypted value. Mutually exclusive with aad; exactly one of the two + must be set. + """ + + def __init__(self, driver_id, value, aad=None, use_persisted_aad=False): + self.driverId = driver_id + self.value = value + self.aad = aad + self.usePersistedAad = use_persisted_aad + + +class CreateEncapsulatedKey: + """ + Request to create a new encapsulated data encryption key. + + The backend should respond with an EncapsulatedKey or an Error response. + + :param driver_id: The id of the driver to create the key with. + :param alias: The alias to bind to the new key. + :param profile_name: The name of the encryption profile to create the + key for, or None to use the sole configured profile. + """ + + def __init__(self, driver_id, alias, profile_name=None): + self.driverId = driver_id + self.alias = alias + self.profileName = profile_name + + +class ImportEncapsulatedKey: + """ + Request to register a pre-existing encapsulated data encryption key. + + Unlike CreateEncapsulatedKey, this does not generate a new key via the + profile's KeyEncapsulationService; it seeds the profile's + EncapsulatedKeyRepository directly with an encapsulation obtained + elsewhere (e.g. a fixture, or a prior CreateEncapsulatedKey response). + + The backend should respond with an EncapsulatedKey or an Error response. + + :param driver_id: The id of the driver to import the key into. + :param alias: The alias to bind to the imported key. + :param encapsulation: The encapsulated (wrapped) data encryption key + bytes. + :param metadata: The key encapsulation service's metadata for the + encapsulation. + :param profile_name: The name of the encryption profile to import the + key into, or None to use the sole configured profile. + """ + + def __init__(self, driver_id, alias, encapsulation, metadata, + profile_name=None): + self.driverId = driver_id + self.alias = alias + self.encapsulation = encapsulation + self.metadata = metadata + self.profileName = profile_name diff --git a/nutkit/protocol/responses.py b/nutkit/protocol/responses.py index 95a9ecd3c..bf187900d 100644 --- a/nutkit/protocol/responses.py +++ b/nutkit/protocol/responses.py @@ -343,6 +343,42 @@ def __init__(self, id): self.id = id +class EncryptedValue: + """ + The result of encrypting a value with client-side property encryption. + + Sent in response to an EncryptToBytes request. + """ + + def __init__(self, encryptedBytes): + self.encrypted_bytes = bytes.fromhex(encryptedBytes) + + +class DecryptedValue: + """ + The result of decrypting a value with client-side property encryption. + + Sent in response to a Decrypt request. + """ + + def __init__(self, decryptedValue): + self.decrypted_value = decryptedValue + + +class EncapsulatedKey: + """ + An encapsulated data encryption key. + + Sent in response to a CreateEncapsulatedKey request. + """ + + def __init__(self, id, alias, encapsulatedBytes, metadata): + self.id = id + self.alias = alias + self.encapsulated_bytes = bytes.fromhex(encapsulatedBytes) + self.metadata = metadata + + class Result: """Represents a result instance on the backend.""" diff --git a/tests/stub/property_encryption/__init__.py b/tests/stub/property_encryption/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/stub/property_encryption/decrypt_interop_fixtures.py b/tests/stub/property_encryption/decrypt_interop_fixtures.py new file mode 100644 index 000000000..26d50e600 --- /dev/null +++ b/tests/stub/property_encryption/decrypt_interop_fixtures.py @@ -0,0 +1,78 @@ +""" +Shared cross-driver fixture list for the decrypt-interop check. + +Each entry was produced by one driver team's own implementation, using +generate_decrypt_interop_fixture.py against their own backend. Every driver's +test suite decrypts every entry here, proving it can read values encrypted by +every other driver. + +Add your own entry by running that script and pasting its output below. +""" + +from dataclasses import dataclass + +import nutkit.protocol as types + +# don't change this or the interop decryption test will break +INTEROP_PROFILE_NAME = "interop" + + +@dataclass(frozen=True) +class DecryptInteropFixture: + driver: str + kek: bytes + encapsulation: bytes + metadata: dict + encrypted: bytes + value: object + + +DECRYPT_INTEROP_TEST_CASES = [ + DecryptInteropFixture( + driver="dotnet", + kek=bytes.fromhex( + "1974e1620d8d540ffba202d9eeb616f4" + "266a731f3eadd3637f25c71bb96ad024" + ), + encapsulation=bytes.fromhex( + "014603135847721b9f3a1b1089ec98712" + "1e03c8d703df8c35b7ca95f8893c84f6f" + "e23c661aae26ef78bc9b1b6bb06905" + ), + metadata={"iv": "nJvPChdeMDkE/FDM"}, + encrypted=bytes.fromhex( + "01b66587696e7465726f70cc24eab4839" + "fe33026669beb1413b7a3581cb32edae0" + "1e2c94e41e6e4ed741e6302df4da42208" + "6535452494e470100a5866b65795f6964" + "8130826976cc0ca3feff5e0278a951c49" + "0a11483616164cc00d0126161645f7072" + "6f746f636f6c5f6d616a6f7201d012616" + "1645f70726f746f636f6c5f6d696e6f72" + "00" + ), + value=types.CypherString("hello from dotnet!"), + ), + DecryptInteropFixture( + driver="javascript", + kek=bytes.fromhex( + "ac4828b563d2dd626d34214c3dcd8168" + "ed77c5ed9620b383e4b63665286302f9" + ), + encapsulation=bytes.fromhex( + "4320112f90e2228f4dabffdc82c904bf" + "ff33eb70b4b162354474cb6a389717e9" + "dfe3077d9255913ee6d0e30f2a9f55f0" + ), + metadata={"iv": "lL9ga7QC9WWcKXH6"}, + encrypted=bytes.fromhex( + "01b66587696e7465726f70cc28e4d50c" + "776bdf240ccb0afd823376dfda1b55ba" + "682601933fe5fc15a9276dcdd20e2cda" + "2768236a1686535452494e470100a282" + "6976cc0c343f2b095b3385d0d7adf455" + "866b65795f69648130" + ), + value=types.CypherString("hello from javascript!"), + ), +] diff --git a/tests/stub/property_encryption/deterministic_fixtures.py b/tests/stub/property_encryption/deterministic_fixtures.py new file mode 100644 index 000000000..82485cce5 --- /dev/null +++ b/tests/stub/property_encryption/deterministic_fixtures.py @@ -0,0 +1,128 @@ +# These entries were initially generated with the .NET driver. + +from dataclasses import dataclass + +import nutkit.protocol as types + +DETERMINISTIC_PROFILE_NAME = "deterministic" + +DETERMINISTIC_KEK = bytes.fromhex( + "f0de94eb5a2d4da6f17ea74b14e9e556" + "d367cb22b053e01798aa2677bfcf5761" +) +DETERMINISTIC_ENCAPSULATION = bytes.fromhex( + "9e1f562dee78c6c2d47f4378d2949774" + "c3a56339b824abaf276c4ca7fcf5a8cd" + "63976ae348104d6757b9e419bf9ea325" +) +DETERMINISTIC_KEY_METADATA = {"iv": "P02Pc7vInYIQ7k93"} + + +@dataclass(frozen=True) +class DeterministicFixture: + value: object + iv: bytes + encrypted: bytes + aad: object = None + + +DETERMINISTIC_TEST_CASES = [ + DeterministicFixture( + value=types.CypherBool(True), + iv=bytes.fromhex( + "000102030405060708090a0b" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc11877fe22670d0d3433e2a9c4dd5" + "fd17994b87424f4f4c45414e0100a282" + "6976cc0c000102030405060708090a0b" + "866b65795f69648130" + ), + ), + DeterministicFixture( + value=types.CypherInt(32768), + iv=bytes.fromhex( + "0c0d0e0f1011121314151617" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc153022e4a29e68285f6fddac8604" + "5e26b63ba5da995087494e5445474552" + "0100a2826976cc0c0c0d0e0f10111213" + "14151617866b65795f69648130" + ), + ), + DeterministicFixture( + value=types.CypherFloat(3.25), + iv=bytes.fromhex( + "18191a1b1c1d1e1f20212223" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc1959a942a76621fe2aa1f2d388ab" + "4e91010e4b39b48520328e9585464c4f" + "41540100a2826976cc0c18191a1b1c1d" + "1e1f20212223866b65795f69648130" + ), + ), + DeterministicFixture( + value=types.CypherString("hello world"), + iv=bytes.fromhex( + "2425262728292a2b2c2d2e2f" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc1c19b0e5f67ee23e78eb73899546" + "4420a6fd3626fc052501e325cee12586" + "535452494e470100a2826976cc0c2425" + "262728292a2b2c2d2e2f866b65795f69" + "648130" + ), + ), + DeterministicFixture( + value=types.CypherBytes(b"\x00\x01\x02"), + iv=bytes.fromhex( + "303132333435363738393a3b" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc1529ea2acd117b82841138917fc8" + "9cdf15cb4f0d809d8542595445530100" + "a2826976cc0c30313233343536373839" + "3a3b866b65795f69648130" + ), + ), + DeterministicFixture( + value=types.CypherList([types.CypherInt(1), types.CypherInt(2)]), + iv=bytes.fromhex( + "3c3d3e3f4041424344454647" + ), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc13b5011505031e789718bd92136f" + "766baa191036844c4953540100a28269" + "76cc0c3c3d3e3f404142434445464786" + "6b65795f69648130" + ), + ), + DeterministicFixture( + value=types.CypherString("aad-bound"), + iv=bytes.fromhex( + "48494a4b4c4d4e4f50515253" + ), + aad=types.CypherString("row-42"), + encrypted=bytes.fromhex( + "01b6658d64657465726d696e69737469" + "63cc1a3a8af0d3820a0a549d75e42e59" + "6a18ff85ee74fb51dce4bc0300865354" + "52494e470100a583616164cc0786726f" + "772d3432d0196161645f656e636f6469" + "6e675f736368656d655f6d616a6f7201" + "d0196161645f656e636f64696e675f73" + "6368656d655f6d696e6f7200826976cc" + "0c48494a4b4c4d4e4f50515253866b65" + "795f69648130" + ), + ), +] diff --git a/tests/stub/property_encryption/generate_decrypt_interop_fixture.py b/tests/stub/property_encryption/generate_decrypt_interop_fixture.py new file mode 100644 index 000000000..9c69722e3 --- /dev/null +++ b/tests/stub/property_encryption/generate_decrypt_interop_fixture.py @@ -0,0 +1,101 @@ +""" +Generates an entry for the fixture list in decrypt_interop_fixtures.py. + +Make sure the backend is running, then run the script. The script connects to +the backend and pretends to be starting a test, and creates a driver. The +connection details are nonsense because it's not actually going to try to +connect to a database, just using the encryption property. + +If your backend is passing the encryption stub tests then this script will +work, and it will print out an entry for you to paste into the FIXTURES list +in decrypt_interop_fixtures.py. The actual test in test_property_encryption.py +picks it up from there. + +Run as a module from the repo root with your backend already listening on +:9876: + + cd ~/dev/testkit # repo root + python3 -m tests.stub.property_encryption.generate_decrypt_interop_fixture + +""" + +import secrets + +import nutkit.protocol as types +from nutkit.frontend import Driver +from tests.shared import ( + get_driver_name, + new_backend, +) +from tests.stub.property_encryption.decrypt_interop_fixtures import ( + INTEROP_PROFILE_NAME, +) + + +def main(): + backend = new_backend() + try: + backend.send_and_receive(types.GetFeatures()) + backend.send_and_receive( + types.StartTest("generate_decrypt_interop_fixture") + ) + + string_to_encrypt = f"hello from {get_driver_name()}!" + kek = secrets.token_bytes(32) + value = types.CypherString(string_to_encrypt) + auth = types.AuthorizationToken( + "basic", principal="neo4j", credentials="pass" + ) + driver = Driver( + backend, "bolt://localhost:9999", auth, + property_encryption_profiles=[ + {"name": INTEROP_PROFILE_NAME, "kek": kek} + ], + ) + try: + key = driver.create_encapsulated_key( + "k", profile_name=INTEROP_PROFILE_NAME + ) + encrypted = driver.encrypt_to_bytes( + value, profile_name=INTEROP_PROFILE_NAME, key_alias="k" + ) + finally: + driver.close() + + print_fixture_literal( + driver=get_driver_name(), + kek=kek, + encapsulation=key.encapsulated_bytes, + metadata=key.metadata, + encrypted=encrypted, + value=string_to_encrypt, + ) + finally: + backend.close() + + +def print_fixture_literal(*, driver, kek, encapsulation, metadata, encrypted, + value): + metadata_literal = ", ".join( + f'"{key}": "{val}"' for key, val in metadata.items() + ) + print(" DecryptInteropFixture(") + print(f' driver="{driver}",') + print_bytes_literal("kek", kek) + print_bytes_literal("encapsulation", encapsulation) + print(f" metadata={{{metadata_literal}}},") + print_bytes_literal("encrypted", encrypted) + print(f' value=types.CypherString("{value}"),') + print(" ),") + + +def print_bytes_literal(field, value): + print(f" {field}=bytes.fromhex(") + hex_string = value.hex() + for i in range(0, len(hex_string), 32): + print(f' "{hex_string[i:i + 32]}"') + print(" ),") + + +if __name__ == "__main__": + main() diff --git a/tests/stub/property_encryption/test_property_encryption.py b/tests/stub/property_encryption/test_property_encryption.py new file mode 100644 index 000000000..cafd8d893 --- /dev/null +++ b/tests/stub/property_encryption/test_property_encryption.py @@ -0,0 +1,241 @@ +import random + +import nutkit.protocol as types +from nutkit.frontend import Driver +from tests.shared import TestkitTestCase +from tests.stub.property_encryption.decrypt_interop_fixtures import ( + DECRYPT_INTEROP_TEST_CASES, + INTEROP_PROFILE_NAME, +) +from tests.stub.property_encryption.deterministic_fixtures import ( + DETERMINISTIC_ENCAPSULATION, + DETERMINISTIC_KEK, + DETERMINISTIC_KEY_METADATA, + DETERMINISTIC_PROFILE_NAME, + DETERMINISTIC_TEST_CASES, +) +from tests.stub.shared import StubServer + + +class TestPropertyEncryption(TestkitTestCase): + required_features = (types.Feature.API_PROPERTY_ENCRYPTION,) + + def setUp(self): + super().setUp() + self._server = StubServer(9020) + self._driver = None + + def tearDown(self): + if self._driver: + self._driver.close() + return super().tearDown() + + def _new_driver(self, profiles=("default",)): + auth = types.AuthorizationToken("basic", principal="neo4j", + credentials="pass") + uri = "bolt://%s" % self._server.address + self._driver = Driver( + self._backend, uri, auth, + property_encryption_profiles=list(profiles) + ) + return self._driver + + def test_round_trips_a_single_value(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + encrypted = driver.encrypt_to_bytes( + types.CypherString("hello world"), key_alias="k1" + ) + decrypted = driver.decrypt(encrypted, use_persisted_aad=True) + + self.assertEqual(decrypted, types.CypherString("hello world")) + + def test_round_trips_many_values_in_shuffled_order(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + values = [ + types.CypherBool(True), + types.CypherBool(False), + types.CypherInt(0), + types.CypherInt(-1), + types.CypherInt(32768), + types.CypherFloat(3.25), + types.CypherString(""), + types.CypherString("a"), + types.CypherString("hello world"), + types.CypherBytes(b""), + types.CypherBytes(b"\x00\x01\x02"), + types.CypherList([types.CypherInt(1), types.CypherInt(2)]) + ] + + vectors = [ + (value, driver.encrypt_to_bytes(value, key_alias="k1")) + for value in values + ] + + order = list(range(len(vectors))) + random.shuffle(order) + + for i in order: + original, encrypted = vectors[i] + decrypted = driver.decrypt(encrypted, use_persisted_aad=True) + self.assertEqual( + decrypted, original, + f"vector {i} ({original!r}) round-tripped to {decrypted!r}" + ) + + def test_encrypting_the_same_value_twice_yields_different_ciphertext(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + value = types.CypherString("hello world") + first = driver.encrypt_to_bytes(value, key_alias="k1") + second = driver.encrypt_to_bytes(value, key_alias="k1") + + self.assertNotEqual(first, second) + + def test_fixed_iv_produces_identical_ciphertext(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + iv = bytes(range(12)) + value = types.CypherString("hello world") + first = driver.encrypt_to_bytes(value, key_alias="k1", iv=iv) + second = driver.encrypt_to_bytes(value, key_alias="k1", iv=iv) + + self.assertEqual(first, second) + + def test_decrypts_an_aad_bound_value_with_the_persisted_aad(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + encrypted = driver.encrypt_to_bytes( + types.CypherString("aad-bound"), + aad=types.CypherString("row-42"), + key_alias="k1" + ) + decrypted = driver.decrypt(encrypted, use_persisted_aad=True) + + self.assertEqual(decrypted, types.CypherString("aad-bound")) + + def test_decrypt_raises_on_wrong_aad(self): + driver = self._new_driver() + driver.create_encapsulated_key("k1") + + encrypted = driver.encrypt_to_bytes( + types.CypherString("aad-bound"), + aad=types.CypherString("row-42"), + key_alias="k1" + ) + + with self.assertRaises(types.DriverError): + driver.decrypt(encrypted, aad=types.CypherString("row-999")) + + def test_encrypt_raises_on_unknown_alias(self): + driver = self._new_driver() + + with self.assertRaises(types.DriverError): + driver.encrypt_to_bytes( + types.CypherString("hello world"), key_alias="no-such-key" + ) + + def test_encrypt_raises_when_alias_belongs_to_a_different_profile(self): + driver = self._new_driver(profiles=("p1", "p2")) + driver.create_encapsulated_key("k1", profile_name="p1") + + with self.assertRaises(types.DriverError): + driver.encrypt_to_bytes( + types.CypherString("hello world"), + profile_name="p2", key_alias="k1" + ) + + def test_encrypt_raises_on_unknown_key_id(self): + driver = self._new_driver() + + with self.assertRaises(types.DriverError): + driver.encrypt_to_bytes( + types.CypherString("hello world"), key_id="no-such-id" + ) + + def test_key_manager_raises_when_ambiguous_and_no_profile_given(self): + driver = self._new_driver(profiles=("p1", "p2")) + + with self.assertRaises(types.DriverError): + driver.create_encapsulated_key("k1") + + def test_imported_key_decrypts_with_a_fixed_kek(self): + kek = bytes(range(32)) + + driver_1 = self._new_driver( + profiles=({"name": "fx", "kek": kek},) + ) + key = driver_1.create_encapsulated_key("k1", profile_name="fx") + encrypted = driver_1.encrypt_to_bytes( + types.CypherString("hello world"), + profile_name="fx", key_alias="k1" + ) + driver_1.close() + + driver_2 = self._new_driver( + profiles=({"name": "fx", "kek": kek},) + ) + driver_2.import_encapsulated_key( + "k1", key.encapsulated_bytes, key.metadata, profile_name="fx" + ) + + decrypted = driver_2.decrypt(encrypted, use_persisted_aad=True) + + self.assertEqual(decrypted, types.CypherString("hello world")) + + def test_encrypts_to_known_bytes(self): + driver = self._new_driver( + profiles=( + { + "name": DETERMINISTIC_PROFILE_NAME, + "kek": DETERMINISTIC_KEK, + }, + ) + ) + driver.import_encapsulated_key( + "k", DETERMINISTIC_ENCAPSULATION, + DETERMINISTIC_KEY_METADATA, + profile_name=DETERMINISTIC_PROFILE_NAME + ) + + for case in DETERMINISTIC_TEST_CASES: + with self.subTest(value=case.value): + encrypted = driver.encrypt_to_bytes( + case.value, profile_name=DETERMINISTIC_PROFILE_NAME, + key_alias="k", iv=case.iv, aad=case.aad + ) + self.assertEqual(encrypted, case.encrypted) + + def test_decrypts_values_produced_by_other_drivers(self): + for case in DECRYPT_INTEROP_TEST_CASES: + with self.subTest(driver=case.driver): + driver = self._new_driver( + profiles=( + { + "name": INTEROP_PROFILE_NAME, + "kek": case.kek, + }, + ) + ) + driver.import_encapsulated_key( + "k", case.encapsulation, case.metadata, + profile_name=INTEROP_PROFILE_NAME + ) + + decrypted = driver.decrypt( + case.encrypted, use_persisted_aad=True + ) + + self.assertEqual( + decrypted, case.value, + "Could not decrypt value encrypted by driver: " + f"{case.driver}" + ) + driver.close() + self._driver = None