Merge pull request #1527 from serengil/feat-task-1511-encrypt-embeddings-feature

encrypt module added

Author: serengil

README.md

@@ -324,13 +324,10 @@ from lightphe import LightPHE
 # build an additively homomorphic cryptosystem (e.g. Paillier) on-prem
 cs = LightPHE(algorithm_name = "Paillier", precision = 19)
 
-# define plain vectors for source and target
-alpha = DeepFace.represent("img1.jpg")[0]["embedding"]
+# define encrypted and plain vectors
+encrypted_alpha = DeepFace.represent("source.jpg", cryptosystem=cs)[0]["encrypted_embedding"]
 beta = DeepFace.represent("target.jpg")[0]["embedding"]
 
-# encrypt source embedding on-prem - private key not required
-encrypted_alpha = cs.encrypt(alpha)
-
 # dot product of encrypted & plain embedding in cloud - private key not required
 encrypted_cosine_similarity = encrypted_alpha @ beta
 

deepface/DeepFace.py

@@ -13,6 +13,7 @@
 import numpy as np
 import pandas as pd
 import tensorflow as tf
+from lightphe import LightPHE
 
 # package dependencies
 from deepface.commons import package_utils, folder_utils
@@ -398,6 +399,7 @@ def represent(
     max_faces: Optional[int] = None,
     l2_normalize: bool = False,
     minmax_normalize: bool = False,
+    cryptosystem: Optional[LightPHE] = None,
 ) -> Union[List[Dict[str, Any]], List[List[Dict[str, Any]]]]:
     """
     Represent facial images as multi-dimensional vector embeddings.
@@ -441,6 +443,20 @@ def represent(
         minmax_normalize (bool): Flag to enable min-max normalization of the output embeddings
             to the range [0, 1].
 
+        cryptosystem (LightPHE): An instance of a partially homomorphic encryption system
+            to encrypt the output embeddings. If provided, the embeddings will be encrypted
+            using the specified cryptosystem. Then, you will be able to perform homomorphic
+            operations on the encrypted embeddings without decrypting them first.
+            Check out these papers to find out more:
+
+            [1] Serengil, S. I. & Ozpinar, A. (2025). "LightPHE: Integrating Partially
+            Homomorphic Encryption into Python with Extensive Cloud Environment Evaluations",
+            https://arxiv.org/abs/2408.05219.
+
+            [2] Serengil, S. I. & Ozpinar, A. (2025). "Encrypted Vector Similarity Computations
+            Using Partially Homomorphic Encryption: Applications and Performance Analysis",
+            https://arxiv.org/abs/2503.05850.
+
     Returns:
         results (List[Dict[str, Any]] or List[Dict[str, Any]]): A list of dictionaries.
             Result type becomes List of List of Dict if batch input passed.
@@ -457,6 +473,9 @@ def represent(
 
         - face_confidence (float): Confidence score of face detection. If `detector_backend` is set
             to 'skip', the confidence will be 0 and is nonsensical.
+
+        - encrypted_embedding (List[Any]): Encrypted multidimensional vector representing
+            facial features. This field is included only if a `cryptosystem` is provided.
     """
     return representation.represent(
         img_path=img_path,
@@ -470,6 +489,7 @@ def represent(
         max_faces=max_faces,
         l2_normalize=l2_normalize,
         minmax_normalize=minmax_normalize,
+        cryptosystem=cryptosystem,
     )
 
 

deepface/commons/embed_utils.py

@@ -0,0 +1,13 @@
+from typing import List, Union
+
+
+def is_flat_embedding(x: Union[List[float], List[List[float]]]) -> bool:
+    """
+    Check if the embeddings represent a single flat list of floats
+        rather than a list of list of float.
+    Args:
+        x (List[float] or List[List[float]]): Embeddings to check.
+    Returns:
+        bool: True if x is a flat list of floats, False otherwise.
+    """
+    return isinstance(x, list) and all(isinstance(i, (int, float)) for i in x)

deepface/modules/encryption.py

@@ -0,0 +1,74 @@
+# built-in dependencies
+from typing import List, Union, Optional, cast
+
+# third-party dependencies
+from lightphe import LightPHE
+from lightphe.models.Tensor import EncryptedTensor
+import numpy as np
+
+# project dependencies
+from deepface.commons.embed_utils import is_flat_embedding
+from deepface.commons.logger import Logger
+
+logger = Logger()
+
+
+# pylint: disable=no-else-return
+def encrypt_embeddings(
+    embeddings: Union[List[float], List[List[float]]], cryptosystem: Optional[LightPHE] = None
+) -> Union[EncryptedTensor, List[EncryptedTensor], None]:
+    """
+    Encrypt embeddings using a provided cryptosystem.
+    Args:
+        embeddings (List[float] or List[List[float]]): Embeddings to encrypt.
+        cryptosystem (LightPHE): Cryptosystem to use for encryption.
+    Returns:
+        EncryptedTensor or List[EncryptedTensor] or None: Encrypted embeddings or None
+            if no cryptosystem is provided.
+    """
+    if cryptosystem is None:
+        return None
+
+    if is_flat_embedding(embeddings):
+        embedding = cast(List[float], embeddings)  # let type checker know
+        encrypted_embedding = encrypt_embedding(embedding, cryptosystem)
+        return encrypted_embedding
+    else:
+        encrypted_embeddings: List[EncryptedTensor] = []
+        for embedding in embeddings:
+            embedding = cast(List[float], embedding)  # let type checker know
+            encrypted_embedding = encrypt_embedding(embedding, cryptosystem)
+            encrypted_embeddings.append(encrypted_embedding)
+
+        if all(item is None for item in encrypted_embeddings):
+            return None
+
+        return encrypted_embeddings
+
+
+def encrypt_embedding(embeddings: List[float], cryptosystem: LightPHE) -> Optional[EncryptedTensor]:
+    """
+    Encrypt an embedding using a provided cryptosystem.
+    Args:
+        embeddings (List[float]): Embedding to encrypt.
+        cryptosystem (LightPHE): Cryptosystem to use for encryption.
+    Returns:
+        EncryptedTensor or None: Encrypted embedding or None if encryption is skipped.
+    """
+    if any(x < 0 for x in embeddings):
+        logger.warn(
+            "Skipping encryption because it contains negative values."
+            "Consider to set minmax_normalize=True in DeepFace.represent method."
+        )
+        return None
+
+    norm = np.linalg.norm(embeddings)
+    if not np.isclose(norm, 1.0):
+        logger.warn(
+            "Skipping encryption because given embedding is not l_2 normalized."
+            "Consider to set l2_normalize=True in DeepFace.represent method."
+        )
+        return None
+
+    encrypted_embeddings = cryptosystem.encrypt(embeddings, silent=True)
+    return encrypted_embeddings

deepface/modules/normalization.py

@@ -6,18 +6,7 @@
 
 # project dependencies
 from deepface.config.minmax import get_minmax_values
-
-
-def is_flat_embedding(x: Union[List[float], List[List[float]]]) -> bool:
-    """
-    Check if the embeddings represent a single flat list of floats
-        rather than a list of list of float.
-    Args:
-        x (List[float] or List[List[float]]): Embeddings to check.
-    Returns:
-        bool: True if x is a flat list of floats, False otherwise.
-    """
-    return isinstance(x, list) and all(isinstance(i, (int, float)) for i in x)
+from deepface.commons.embed_utils import is_flat_embedding
 
 
 def normalize_embedding_minmax(

deepface/modules/representation.py

@@ -4,12 +4,17 @@
 
 # 3rd party dependencies
 import numpy as np
+from lightphe import LightPHE
 
 # project dependencies
 from deepface.commons import image_utils
 from deepface.modules import modeling, detection, preprocessing
 from deepface.models.FacialRecognition import FacialRecognition
 from deepface.modules.normalization import normalize_embedding_l2, normalize_embedding_minmax
+from deepface.modules.encryption import encrypt_embeddings
+from deepface.commons.logger import Logger
+
+logger = Logger()
 
 
 def represent(
@@ -24,6 +29,7 @@ def represent(
     max_faces: Optional[int] = None,
     l2_normalize: bool = False,
     minmax_normalize: bool = False,
+    cryptosystem: Optional[LightPHE] = None,
 ) -> Union[List[Dict[str, Any]], List[List[Dict[str, Any]]]]:
     """
     Represent facial images as multi-dimensional vector embeddings.
@@ -63,6 +69,10 @@ def represent(
         minmax_normalize (bool): Flag to enable min-max normalization of the output embeddings
             to the range [0, 1].
 
+        cryptosystem (LightPHE): An instance of a partially homomorphic encryption system
+            to encrypt the output embeddings. If provided, the embeddings will be encrypted
+            using the specified cryptosystem.
+
     Returns:
         results (List[Dict[str, Any]] or List[Dict[str, Any]]): A list of dictionaries.
             Result type becomes List of List of Dict if batch input passed.
@@ -77,6 +87,8 @@ def represent(
             the full image area and is nonsensical.
         - face_confidence (float): Confidence score of face detection. If `detector_backend` is set
             to 'skip', the confidence will be 0 and is nonsensical.
+        - encrypted_embedding (List[Any]): Encrypted multidimensional vector representing
+            facial features. This field is included only if a `cryptosystem` is provided.
     """
     resp_objs = []
 
@@ -178,15 +190,20 @@ def represent(
     if l2_normalize:
         embeddings = normalize_embedding_l2(embeddings)
 
+    encrypted_embeddings = encrypt_embeddings(embeddings, cryptosystem)
+
     resp_objs_dict = defaultdict(list)
     for idy, batch_index in enumerate(batch_indexes):
-        resp_objs_dict[batch_index].append(
-            {
-                "embedding": embeddings if len(batch_images) == 1 else embeddings[idy],
-                "facial_area": batch_regions[idy],
-                "face_confidence": batch_confidences[idy],
-            }
-        )
+        resp_obj = {
+            "embedding": embeddings if len(batch_images) == 1 else embeddings[idy],
+            "facial_area": batch_regions[idy],
+            "face_confidence": batch_confidences[idy],
+        }
+        if cryptosystem is not None and encrypted_embeddings is not None:
+            resp_obj["encrypted_embedding"] = (
+                encrypted_embeddings if len(batch_images) == 1 else encrypted_embeddings[idy]
+            )
+        resp_objs_dict[batch_index].append(resp_obj)
 
     resp_objs = [resp_objs_dict[idx] for idx in range(len(images))]
 

requirements.txt

@@ -13,3 +13,4 @@ mtcnn>=0.1.0
 retina-face>=0.0.14
 fire>=0.4.0
 gunicorn>=20.1.0
+lightphe>=0.0.15