diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/conftest.py b/documentdb_tests/compatibility/tests/system/security/encryption/conftest.py new file mode 100644 index 000000000..0dab3e11d --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/security/encryption/conftest.py @@ -0,0 +1,9 @@ +"""Shared fixtures for Queryable Encryption tests in this directory.""" + +from documentdb_tests.compatibility.tests.system.security.encryption.utils.qe_collections import ( + qe_collection, + qe_collection_multi, + qe_collection_nested, +) + +__all__ = ["qe_collection", "qe_collection_multi", "qe_collection_nested"] diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_crud.py b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_crud.py new file mode 100644 index 000000000..101700da6 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_crud.py @@ -0,0 +1,135 @@ +"""Tests for CRUD operations against Queryable Encryption collections. + +Verifies insert, update, and delete operations with encryption feature. Any +plaintext value at an encrypted path is rejected, including null and regardless +of bsonType. Absent fields are unaffected. +""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.compatibility.tests.core.utils.command_test_case import ( + CommandContext, + CommandTestCase, +) +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.framework.error_codes import DOCUMENT_VALIDATION_FAILURE_ERROR +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params + +pytestmark = pytest.mark.requires(queryable_encryption=True) + + +# Property [Missing Field Acceptance]: insert succeeds when an encrypted field is absent. +@pytest.mark.insert +def test_encryption_insert_missing_field_succeeds(qe_collection): + """Test insert succeeds when the encrypted field is entirely absent.""" + result = execute_command( + qe_collection, {"insert": qe_collection.name, "documents": [{"_id": 1}]} + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 1}, + msg="insert should accept a document missing the encrypted field.", + ) + + +# Property [Plaintext Rejection]: insert rejects null and any plaintext value at an +# encrypted path, regardless of bsonType. +INSERT_PLAINTEXT_REJECTION_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "null_value", + command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": 1, "ssn": None}]}, + error_code=DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject an explicit null at an encrypted path.", + ), + CommandTestCase( + "plaintext_matching_bsontype", + command=lambda ctx: { + "insert": ctx.collection, + "documents": [{"_id": 2, "ssn": "123-45-6789"}], + }, + error_code=DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject a plaintext value even when it matches the declared bsonType.", + ), + CommandTestCase( + "plaintext_wrong_bsontype", + command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": 3, "ssn": 123}]}, + error_code=DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject a plaintext value of the wrong bsonType.", + ), +] + + +@pytest.mark.insert +@pytest.mark.parametrize("test", pytest_params(INSERT_PLAINTEXT_REJECTION_TESTS)) +def test_encryption_insert_rejects_plaintext(qe_collection, test: CommandTestCase): + """Test insert rejects plaintext values at an encrypted path.""" + ctx = CommandContext.from_collection(qe_collection) + result = execute_command(qe_collection, test.build_command(ctx)) + assertFailureCode(result, test.error_code, msg=test.msg) + + +# Property [Multi-Field Independence]: each declared encrypted field is validated +# independently of the others. +@pytest.mark.insert +def test_encryption_insert_multiple_fields_all_absent(qe_collection_multi): + """Test insert succeeds when every declared encrypted field is absent.""" + result = execute_command( + qe_collection_multi, + {"insert": qe_collection_multi.name, "documents": [{"_id": 1, "name": "a"}]}, + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 1}, + msg="insert should accept a document missing every encrypted field.", + ) + + +@pytest.mark.insert +def test_encryption_insert_one_of_multiple_fields_plaintext_rejected(qe_collection_multi): + """Test insert rejects a plaintext value on one of several encrypted fields.""" + result = execute_command( + qe_collection_multi, + {"insert": qe_collection_multi.name, "documents": [{"_id": 1, "dob": "2000-01-01"}]}, + ) + assertFailureCode( + result, + DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject a plaintext value on any one of several encrypted fields.", + ) + + +# Property [Update Enforcement]: update applies the same validation as insert. +@pytest.mark.update +def test_encryption_update_rejects_plaintext(qe_collection): + """Test update rejects setting an encrypted field to a plaintext value.""" + qe_collection.insert_one({"_id": 1}) + result = execute_command( + qe_collection, + { + "update": qe_collection.name, + "updates": [{"q": {"_id": 1}, "u": {"$set": {"ssn": "123-45-6789"}}}], + }, + ) + assertFailureCode( + result, + DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="update should reject a plaintext value written to an encrypted path.", + ) + + +# Property [Delete Unaffected]: delete is unaffected by the collection's encryption schema. +@pytest.mark.delete +def test_encryption_delete_succeeds(qe_collection): + """Test delete succeeds for a document with an absent encrypted field.""" + qe_collection.insert_one({"_id": 1}) + result = execute_command( + qe_collection, {"delete": qe_collection.name, "deletes": [{"q": {"_id": 1}, "limit": 1}]} + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 1}, + msg="delete should succeed on a Queryable Encryption collection.", + ) diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_edge_cases.py b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_edge_cases.py new file mode 100644 index 000000000..25fe20575 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_edge_cases.py @@ -0,0 +1,112 @@ +"""Tests for encryption edge cases: nested paths, value size, and multiplicity. + +Verifies that no value shape other than absence can be written to an encrypted +path without a client FLE driver, regardless of nesting, size, or document +count. +""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.framework.error_codes import DOCUMENT_VALIDATION_FAILURE_ERROR +from documentdb_tests.framework.executor import execute_command + +pytestmark = pytest.mark.requires(queryable_encryption=True) + + +# Property [Nested Path Independence]: a nested encrypted path is unaffected by +# insert as long as it is absent, regardless of whether its parent object is present. +@pytest.mark.insert +def test_encryption_insert_nested_path_missing_parent_succeeds(qe_collection_nested): + """Test insert succeeds when the parent of a nested encrypted path is absent.""" + result = execute_command( + qe_collection_nested, {"insert": qe_collection_nested.name, "documents": [{"_id": 1}]} + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 1}, + msg="insert should succeed when the parent object of a nested encrypted path is absent.", + ) + + +@pytest.mark.insert +def test_encryption_insert_nested_path_parent_present_child_absent_succeeds(qe_collection_nested): + """Test insert succeeds when the parent object is present but the encrypted child is absent.""" + result = execute_command( + qe_collection_nested, + {"insert": qe_collection_nested.name, "documents": [{"_id": 1, "address": {"city": "x"}}]}, + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 1}, + msg="insert should succeed when a nested encrypted field is absent" + " even if its parent object is present.", + ) + + +@pytest.mark.insert +def test_encryption_insert_nested_path_plaintext_rejected(qe_collection_nested): + """Test insert rejects a plaintext value at a nested encrypted path.""" + result = execute_command( + qe_collection_nested, + { + "insert": qe_collection_nested.name, + "documents": [{"_id": 1, "address": {"ssn": "123-45-6789"}}], + }, + ) + assertFailureCode( + result, + DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject a plaintext value at a nested encrypted path.", + ) + + +# Property [Validation Ignores Size]: a plaintext value at an encrypted path is +# rejected regardless of its size. +@pytest.mark.insert +def test_encryption_insert_large_plaintext_value_rejected(qe_collection): + """Test insert rejects a large plaintext value at an encrypted path.""" + result = execute_command( + qe_collection, + {"insert": qe_collection.name, "documents": [{"_id": 1, "ssn": "x" * 4_000}]}, + ) + assertFailureCode( + result, + DOCUMENT_VALIDATION_FAILURE_ERROR, + msg="insert should reject an oversized plaintext value at an encrypted path" + " just as it would a small one.", + ) + + +# Property [Absence Has No Multiplicity Constraint]: several documents may each omit +# the same encrypted field in one insert; equality does not impose uniqueness on absence. +@pytest.mark.insert +def test_encryption_insert_multiple_documents_all_missing_field(qe_collection): + """Test a batch insert succeeds when every document omits the same encrypted field.""" + result = execute_command( + qe_collection, + {"insert": qe_collection.name, "documents": [{"_id": 1}, {"_id": 2}]}, + ) + assertSuccessPartial( + result, + {"ok": 1.0, "n": 2}, + msg="a batch insert should succeed when every document omits the same encrypted field.", + ) + + +# Property [Explain Compatibility]: explain runs normally against a +# Queryable Encryption collection. +@pytest.mark.find +def test_encryption_explain_find_on_encrypted_collection(qe_collection): + """Test explain succeeds for a find against a Queryable Encryption collection.""" + result = execute_command( + qe_collection, + {"explain": {"find": qe_collection.name, "filter": {"ssn": "x"}}}, + ) + assertSuccessPartial( + result, + {"ok": 1.0}, + msg="explain should succeed for a find against a Queryable Encryption collection.", + ) diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_query.py b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_query.py new file mode 100644 index 000000000..0e8bc9bfb --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_query.py @@ -0,0 +1,92 @@ +"""Tests for query and aggregation behavior against Queryable Encryption collections. + +Verifies that a raw filter against an encrypted path, including a shape a real +client would never send such as a range comparison on an equality-only field, is +evaluated as an ordinary filter and never raises an encryption-specific error. +This repo has no client-side FLE driver, so query rewriting never happens. +""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_command + +pytestmark = pytest.mark.requires(queryable_encryption=True) + + +@pytest.fixture() +def qe_collection_seeded(qe_collection): + """A qe_collection populated with two documents that both omit the encrypted field.""" + qe_collection.insert_many([{"_id": 1, "name": "a"}, {"_id": 2, "name": "b"}]) + return qe_collection + + +@pytest.mark.find +def test_encryption_find_equality_filter_returns_empty(qe_collection_seeded): + """Test an equality filter on an encrypted field returns no matches, not an error.""" + result = execute_command( + qe_collection_seeded, {"find": qe_collection_seeded.name, "filter": {"ssn": "123-45-6789"}} + ) + assertResult( + result, + expected=[], + msg="an equality filter on an encrypted field should return no matches" + " when no document has that field set.", + ) + + +@pytest.mark.find +def test_encryption_find_range_operator_on_equality_field_no_error(qe_collection_seeded): + """Test a range operator on an equality-only encrypted field executes without error.""" + result = execute_command( + qe_collection_seeded, {"find": qe_collection_seeded.name, "filter": {"ssn": {"$gt": "a"}}} + ) + assertResult( + result, + expected=[], + msg="a range filter on an equality-only encrypted field should execute as an" + " ordinary filter rather than raise a query-type error, since raw commands" + " bypass client-side query rewriting.", + ) + + +@pytest.mark.find +def test_encryption_find_exists_false_matches_absent_field(qe_collection_seeded): + """Test $exists:false on an encrypted field matches documents where it is absent.""" + result = execute_command( + qe_collection_seeded, + { + "find": qe_collection_seeded.name, + "filter": {"ssn": {"$exists": False}}, + "sort": {"_id": 1}, + }, + ) + assertResult( + result, + expected=[{"_id": 1, "name": "a"}, {"_id": 2, "name": "b"}], + msg="$exists:false on an encrypted field should match documents where the field is absent.", + ) + + +@pytest.mark.aggregate +def test_encryption_aggregate_match_expr_on_encrypted_field(qe_collection_seeded): + """Test $match+$expr referencing an encrypted field in aggregation.""" + result = execute_command( + qe_collection_seeded, + { + "aggregate": qe_collection_seeded.name, + "pipeline": [ + {"$match": {"$expr": {"$eq": [{"$type": "$ssn"}, "missing"]}}}, + {"$project": {"_id": 1}}, + {"$sort": {"_id": 1}}, + ], + "cursor": {}, + }, + ) + assertResult( + result, + expected=[{"_id": 1}, {"_id": 2}], + msg="$match+$expr referencing an encrypted field should evaluate normally in aggregation.", + ) diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/utils/__init__.py b/documentdb_tests/compatibility/tests/system/security/encryption/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/utils/qe_collections.py b/documentdb_tests/compatibility/tests/system/security/encryption/utils/qe_collections.py new file mode 100644 index 000000000..c9ad3b1a3 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/security/encryption/utils/qe_collections.py @@ -0,0 +1,64 @@ +"""Shared Queryable Encryption collection fixtures for tests in this directory. + +Creating a QE collection needs a per-field keyId, and each test file here needs +a differently-shaped one. The fixtures here are the single source of truth; +this directory's conftest re-exports them so every test file picks them up. +""" + +from __future__ import annotations + +from typing import Any +from uuid import uuid4 + +import pytest +from bson import Binary +from pymongo.collection import Collection + + +def _create_qe_collection( + collection: Collection, name: str, fields: list[dict[str, Any]] +) -> Collection: + """Create and return a Queryable Encryption collection with the given fields.""" + db = collection.database + resolved_fields = [{**field, "keyId": Binary(uuid4().bytes, 4)} for field in fields] + db.command("create", name, encryptedFields={"fields": resolved_fields}) + return db[name] + + +@pytest.fixture() +def qe_collection(collection): + """A Queryable Encryption collection with ssn as encrypted field.""" + qe = _create_qe_collection( + collection, + f"{collection.name}_qe", + [{"path": "ssn", "bsonType": "string", "queries": {"queryType": "equality"}}], + ) + yield qe + qe.database.drop_collection(qe.name) + + +@pytest.fixture() +def qe_collection_multi(collection): + """A Queryable Encryption collection with two independent encrypted fields.""" + qe = _create_qe_collection( + collection, + f"{collection.name}_qe_multi", + [ + {"path": "ssn", "bsonType": "string"}, + {"path": "dob", "bsonType": "string"}, + ], + ) + yield qe + qe.database.drop_collection(qe.name) + + +@pytest.fixture() +def qe_collection_nested(collection): + """A Queryable Encryption collection with address.ssn as a nested encrypted path.""" + qe = _create_qe_collection( + collection, + f"{collection.name}_qe_nested", + [{"path": "address.ssn", "bsonType": "string"}], + ) + yield qe + qe.database.drop_collection(qe.name)