CSFLE で消去する権限の実装

# flaskapp/db_queries.py
@aws_credential_handler
def create_key(userId):
   data_key_id = \app.mongodb_encryption_client.create_data_key
      (kms_provider, master_key, key_alt_names=[userId])
   return data_key_id

# flaskapp/user.py
def save(self):
   dek_id = db_queries.create_key(self.username)
   result = app.mongodb[db_name].user.insert_one(
      {
         "username": self.username,
         "password_hash": self.password_hash,
         "dek_id": dek_id,
         "createdAt": datetime.now(),
      }
   )
   if result:
      self.id = result.inserted_id
      return True
   else:
      return False

{
   "name": "shoe size",
   "value": "10",
   "username": "tom"
}

# flaskapp/db_queries.py
# Fields to encrypt, and the algorithm to encrypt them with
ENCRYPTED_FIELDS = {
   # Deterministic encryption for username, because we need to search on it
   "username": Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic,
   # Random encryption for value, as we don't need to search on it
   "value": Algorithm.AEAD_AES_256_CBC_HMAC_SHA_512_Random,
}

# flaskapp/db_queries.py
def insert_data(document):
   document["username"] = current_user.username
   # Loop over the field names (and associated algorithm) we want to encrypt
   for field, algo in ENCRYPTED_FIELDS.items():
      # if the field exists in the document, encrypt it
      if document.get(field):
         document[field] = encrypt_field(document[field], algo)
   # Insert document (now with encrypted fields) to the data collection
   app.data_collection.insert_one(document)

# flaskapp/db_queries.py
# Encrypt a single field with the given algorithm
@aws_credential_handler
def encrypt_field(field, algorithm):
   try:
      field = app.mongodb_encryption_client.encrypt(
         field,
         algorithm,
         key_alt_name=current_user.username,
      )
      return field
   except pymongo.errors.EncryptionError as ex:
      # Catch this error in case the DEK doesn't exist. Log a warning and
      # re-raise the exception
      if "not all keys requested were satisfied" in ex._message:
         app.logger.warn(
            f"Encryption failed: could not find data encryption key for user: {current_user.username}"
         )
      raise ex

# flaskapp/db_queries.py
def fetch_all_data_unencrypted(decrypt=False):
   results = list(app.data_collection.find())
   if decrypt:
      for field in ENCRYPTED_FIELDS.keys():
         for result in results:
            if result.get(field):
               result[field], result["encryption_succeeded"] = decrypt_field(result[field])
   return results

# flaskapp/db_queries.py
# Try to decrypt a field, returning a tuple of (value, status). This will be
either (decrypted_value, True), or (raw_cipher_text, False) if we
couldn't decrypt
def decrypt_field(field):
   try:
      # We don't need to pass the DEK or algorithm to decrypt a field
      field = app.mongodb_encryption_client.decrypt(field)
      return field, True
   # Catch this error in case the DEK doesn't exist.
   except pymongo.errors.EncryptionError as ex:
      if "not all keys requested were satisfied" in ex._message:
         app.logger.warn(
            "Decryption failed: could not find data encryption key to decrypt the record."
         )
         # If we can't decrypt due to missing DEK, return the "raw" value.
         return field, False
      raise ex