Criptografia em uso: a maciez do MongoDB para queries seguras

1
pip install pymongo
2
python -m pip install 'pymongo[encryption]'

1
echo $(head -c 96 /dev/urandom | base64 | tr -d '\n')

1
$r=[byte[]]::new(64);$g=[System.Security.Cryptography.RandomNumberGenerator]::Create();$g.GetBytes($r);[Convert]::ToBase64String($r)

1
def load_local_master_key(filename):
2
    if not os.path.exists(filename):
3
        key = "YOUR_MASTER_KEY_HERE"
4
        with open(filename, "w") as f:
5
            f.write(key)
6
    with open(filename, "r") as f:
7
        return f.read()

1
uri = "<URI>"
2
key_vault_namespace = "encryption.__keyVault"
3
encrypted_database_name = "employee_data"
4
encrypted_collection_name = "employee_salary"
5
6
local_master_key_file = "local_master_key.txt"
7
local_master_key = load_local_master_key(local_master_key_file)
8
9
kms_providers = {"local": {"key": local_master_key}}
10
encrypted_fields_map = get_encrypted_fields_map()

1
def get_encrypted_fields_map():
2
    return {
3
        "fields": [
4
            {
5
                "keyId": None,
6
                "path": "name",
7
                "bsonType": "string",
8
                "queries": [{"queryType": "equality"}],
9
            },
10
            {
11
                "keyId": None,
12
                "path": "salary",
13
                "bsonType": "int",
14
                "queries": [{"queryType": "range", "min": 0, "max": 1000000}],
15
            },
16
        ]
17
    }

1
auto_encryption_opts = AutoEncryptionOpts(
2
	kms_providers=kms_providers,
3
	key_vault_namespace=key_vault_namespace,
4
	crypt_shared_lib_path="./mongo_crypt_shared_v1-macos-arm64-enterprise-8.0.3/lib/mongo_crypt_v1.dylib",
5
	encrypted_fields_map=encrypted_fields_map,
6
)

1
client = MongoClient(uri, auto_encryption_opts=auto_encryption_opts)
2
db = client[encrypted_database_name]
3
collection = db[encrypted_collection_name]
4
collection.drop()

1
employees = [
2
	EmployeeDocument("Alice Johnson", "Software Engineer", "MongoDB", 100000, "USD", datetime.datetime(2019, 3, 5)),
3
	EmployeeDocument("Bob Smith", "Product Manager", "MongoDB", 150000, "USD", datetime.datetime(2018, 6, 15)),
4
	EmployeeDocument("Charlie Brown", "Data Analyst", "MongoDB", 200000, "USD", datetime.datetime(2020, 4, 20)),
5
	EmployeeDocument("Diana Prince", "HR Specialist", "MongoDB", 250000, "USD", datetime.datetime(2021, 12, 3)),
6
	EmployeeDocument("Evan Peters", "Marketing Coordinator", "MongoDB", 80000, "USD", datetime.datetime(2022, 10, 7)),
7
]
8
9
for employee in employees:
10
	collection.insert_one(employee.__dict__)
11
	print(f"Inserted document for {employee.name}")

1
def search_by_name(coll, name):
2
    filter = {"name": name}
3
    cursor = coll.find(filter)
4
    print(f"Documents with name '{name}':")
5
    for doc in cursor:
6
        print(json.dumps(doc, indent=4, default=json_util.default))
7
8
def search_by_salary_range(coll, min_salary, max_salary):
9
    filter = {"salary": {"$gte": min_salary, "$lte": max_salary}}
10
    cursor = coll.find(filter)
11
    print(f"Documents with salary between {min_salary} and {max_salary}:")
12
    for doc in cursor:
13
        print(json.dumps(doc, indent=4, default=json_util.default))
14
15
search_by_name(collection, "Alice Johnson")
16
search_by_salary_range(collection, 150000, 200000)

1
uri = "<URI>"
2
local_master_key_file = "local_master_key.txt"
3
local_master_key = load_local_master_key(local_master_key_file)
4
kms_providers = setup_kms_providers(local_master_key)
5
key_vault_namespace = "encryption.__keyVault"
6
7
client = connect_mongo(uri)
8
database_name = "employee_data"
9
collection_name = "employee_salary"
10
coll = client[database_name][collection_name]
11
coll.drop()
12
key_vault_coll = client["encryption"]["__keyVault"]
13
ensure_key_vault_index(key_vault_coll)

1
def ensure_key_vault_index(key_vault_coll):
2
    index_name = "keyAltNames_1"
3
    indexes = key_vault_coll.index_information()
4
    if index_name not in indexes:
5
        key_vault_coll.create_index(
6
            [("keyAltNames", 1)], unique=True, partialFilterExpression={"keyAltNames": {"$exists": True}}
7
        )
8
        print("Index created in key vault.")
9
    else:
10
        print("Index already exists in key vault.")

1
def ensure_data_key(client_encryption, key_vault_coll, key_alt_name):
2
    existing_key = key_vault_coll.find_one({"keyAltNames": key_alt_name})
3
    if existing_key:
4
        print("Data key already exists. Reusing existing key.")
5
        return existing_key["_id"]
6
7
    print("Creating new data key.")
8
    data_key_id = client_encryption.create_data_key("local", key_alt_names=[key_alt_name])
9
    return data_key_id

1
def encrypt_salary(client_encryption, data_key_id, salary):
2
    salary_in_cents = int(salary * 100)
3
    encrypted_salary = client_encryption.encrypt(
4
        value=salary_in_cents,
5
        algorithm="AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic",
6
        key_id=data_key_id
7
    )
8
    return encrypted_salary

1
def insert_employee_doc(coll, name, position, company, salary_encrypted, currency, start_date):
2
    employee_doc = {
3
        "name": name,
4
        "position": position,
5
        "company": company,
6
        "salary": salary_encrypted,
7
        "currency": currency,
8
        "startDate": start_date
9
    }
10
    coll.insert_one(employee_doc)

1
employees = [
2
	{"name": "Alice Johnson", "position": "Software Engineer", "company": "MongoDB", "salary": 50000, "currency": "USD", "start_date": datetime(2007, 2, 3)},
3
	{"name": "Bob Smith", "position": "Product Manager", "company": "MongoDB", "salary": 70000, "currency": "USD", "start_date": datetime(2009, 3, 14)},
4
]
5
for emp in employees:
6
	encrypted_salary = encrypt_salary(client_encryption, data_key_id, emp["salary"])
7
	insert_employee_doc(coll, emp["name"], emp["position"], emp["company"], encrypted_salary, emp["currency"], emp["start_date"])

1
def find_all_and_decrypt_salaries(coll, client_encryption):
2
    for doc in coll.find():
3
        encrypted_salary = doc["salary"]
4
        decrypted_salary = client_encryption.decrypt(encrypted_salary)
5
        salary_in_dollars = decrypted_salary / 100.0
6
        print(f"Employee: {doc['name']}, Salary: {salary_in_dollars} USD")

1
def find_all_without_decryption(coll):
2
    for doc in coll.find():
3
        print(f"Employee: {doc['name']}, Encrypted Salary: {doc['salary']}")