Merge pull request #16 from serengil/feat-task-1212-homomorphic-suppo…

…rt-for-tensors

Feat task 1212 homomorphic support for tensors

README.md

@@ -28,6 +28,8 @@ Even though fully homomorphic encryption (FHE) has become available in recent ti
 - 📏 Generating smaller ciphertexts
 - 🧠 Well-suited for memory-constrained environments
 - ⚖️ Strikes a favorable balance for practical use cases
+- 🔑 Supporting encryption and decryption of vectors
+- 🗝️ Performing homomorphic addition, homomorphic element-wise multiplication and scalar multiplication on encrypted vectors
 
 # Installation [![PyPI](https://img.shields.io/pypi/v/lightphe.svg)](https://pypi.org/project/lightphe/)
 
@@ -157,24 +159,30 @@ with pytest.raises(ValueError, match="Paillier is not homomorphic with respect t
 
 However, if you tried to multiply ciphertexts with RSA, or xor ciphertexts with Goldwasser-Micali, these will be succeeded because those cryptosystems support those homomorphic operations.
 
-# Encrypt & Decrypt Tensors
+# Working with vectors
 
-You can encrypt the output tensors of machine learning models with LightPHE.
+You can encrypt the output vectors of machine learning models with LightPHE. These encrypted tensors come with homomorphic operation support.
 
 ```python
+# build an additively homomorphic cryptosystem
 cs = LightPHE(algorithm_name="Paillier")
 
-# define plain tensor
-tensor = [1.005, 2.005, 3.005, -4.005, 5.005]
+# define plain tensors
+t1 = [1.005, 2.05, -3.5, 4]
+t2 = [5, 6.2, 7.002, 8.02]
 
-# encrypt tensor
-encrypted_tensors = cs.encrypt(tensor)
+# encrypt tensors
+c1 = cs.encrypt(t1)
+c2 = cs.encrypt(t2)
 
-# decrypt tensor
-decrypted_tensors = cs.decrypt(encrypted_tensors)
+# perform homomorphic addition
+c3 = c1 + c2
+
+# decrypt the addition tensor
+t3 = cs.decrypt(c3)
 
-for i, decrypted_tensor in enumerate(decrypted_tensors):
-  assert tensor[i] == decrypted_tensor
+for i, tensor in enumerate(t3):
+ assert abs((t1[i] + t2[i]) - restored_tensor) < 0.5
 ```
 
 # Contributing

lightphe/__init__.py

@@ -4,7 +4,7 @@
 from lightphe.models.Homomorphic import Homomorphic
 from lightphe.models.Ciphertext import Ciphertext
 from lightphe.models.Algorithm import Algorithm
-from lightphe.models.Tensor import EncryptedTensor, EncryptedTensors
+from lightphe.models.Tensor import Fraction, EncryptedTensor
 from lightphe.cryptosystems.RSA import RSA
 from lightphe.cryptosystems.ElGamal import ElGamal
 from lightphe.cryptosystems.Paillier import Paillier
@@ -29,6 +29,7 @@ def __init__(
  keys: Optional[dict] = None,
  key_file: Optional[str] = None,
  key_size: Optional[int] = None,
+ precision: int = 5,
  ):
  """
  Build LightPHE class
@@ -39,8 +40,10 @@ def __init__(
  keys (dict): optional private-public key pair
  key_file (str): if keys are exported, you can load them into cryptosystem
  key_size (int): key size in bits
+ precision (int): precision for homomorphic operations on tensors
  """
  self.algorithm_name = algorithm_name
+ self.precision = precision
 
  if key_file is not None:
  keys = self.restore_keys(target_file=key_file)
@@ -105,7 +108,7 @@ def __build_cryptosystem(
  raise ValueError(f"unimplemented algorithm - {algorithm_name}")
  return cs
 
- def encrypt(self, plaintext: Union[int, float, list]) -> Union[Ciphertext, EncryptedTensors]:
+ def encrypt(self, plaintext: Union[int, float, list]) -> Union[Ciphertext, EncryptedTensor]:
  """
  Encrypt a plaintext with a built cryptosystem
  Args:
@@ -126,7 +129,7 @@ def encrypt(self, plaintext: Union[int, float, list]) -> Union[Ciphertext, Encry
  return Ciphertext(algorithm_name=self.algorithm_name, keys=self.cs.keys, value=ciphertext)
 
  def decrypt(
- self, ciphertext: Union[Ciphertext, EncryptedTensors]
+ self, ciphertext: Union[Ciphertext, EncryptedTensor]
  ) -> Union[int, List[int], List[float]]:
  """
  Decrypt a ciphertext with a buit cryptosystem
@@ -138,51 +141,62 @@ def decrypt(
  if self.cs.keys.get("private_key") is None:
  raise ValueError("You must have private key to perform decryption")
 
- if isinstance(ciphertext, EncryptedTensors):
+ if isinstance(ciphertext, EncryptedTensor):
  # then this is encrypted tensor
  return self.__decrypt_tensors(encrypted_tensor=ciphertext)
 
  return self.cs.decrypt(ciphertext=ciphertext.value)
 
- def __encrypt_tensors(self, tensor: list) -> EncryptedTensors:
+ def __encrypt_tensors(self, tensor: list) -> EncryptedTensor:
  """
  Encrypt a given tensor
  Args:
  tensor (list of int or float)
  Returns
  encrypted tensor (list of encrypted tensor object)
  """
- encrypted_tensor: List[EncryptedTensor] = []
+ encrypted_tensor: List[Fraction] = []
  for m in tensor:
- sign = 1 if m >= 0 else -1
- # get rid of sign anyway
- m = m * sign
- sign_encrypted = self.cs.encrypt(plaintext=sign)
  if isinstance(m, int):
- dividend_encrypted = self.cs.encrypt(plaintext=m)
- divisor_encrypted = self.cs.encrypt(plaintext=1)
- c = EncryptedTensor(
+ dividend_encrypted = self.cs.encrypt(
+ plaintext=(m % self.cs.plaintext_modulo) * pow(10, self.precision)
+ )
+ abs_dividend_encrypted = self.cs.encrypt(
+ plaintext=(abs(m) % self.cs.plaintext_modulo) * pow(10, self.precision)
+ )
+ divisor_encrypted = self.cs.encrypt(plaintext=pow(10, self.precision))
+ c = Fraction(
  dividend=dividend_encrypted,
  divisor=divisor_encrypted,
- sign=sign_encrypted,
+ abs_dividend=abs_dividend_encrypted,
+ sign=1 if m >= 0 else -1,
  )
  elif isinstance(m, float):
- dividend, divisor = phe_utils.fractionize(value=m, modulo=self.cs.plaintext_modulo)
+ dividend, divisor = phe_utils.fractionize(
+ value=(m % self.cs.plaintext_modulo),
+ modulo=self.cs.plaintext_modulo,
+ precision=self.precision,
+ )
+ abs_dividend, _ = phe_utils.fractionize(
+ value=(abs(m) % self.cs.plaintext_modulo),
+ modulo=self.cs.plaintext_modulo,
+ precision=self.precision,
+ )
  dividend_encrypted = self.cs.encrypt(plaintext=dividend)
+ abs_dividend_encrypted = self.cs.encrypt(plaintext=abs_dividend)
  divisor_encrypted = self.cs.encrypt(plaintext=divisor)
- c = EncryptedTensor(
+ c = Fraction(
  dividend=dividend_encrypted,
  divisor=divisor_encrypted,
- sign=sign_encrypted,
+ abs_dividend=abs_dividend_encrypted,
+ sign=1 if m >= 0 else -1,
  )
  else:
  raise ValueError(f"unimplemented type - {type(m)}")
  encrypted_tensor.append(c)
- return EncryptedTensors(encrypted_tensor=encrypted_tensor)
+ return EncryptedTensor(fractions=encrypted_tensor, cs=self.cs)
 
- def __decrypt_tensors(
- self, encrypted_tensor: EncryptedTensors
- ) -> Union[List[int], List[float]]:
+ def __decrypt_tensors(self, encrypted_tensor: EncryptedTensor) -> Union[List[int], List[float]]:
  """
  Decrypt a given encrypted tensor
  Args:
@@ -191,26 +205,15 @@ def __decrypt_tensors(
  List of plain tensors
  """
  plain_tensor = []
- for c in encrypted_tensor.encrypted_tensor:
- if isinstance(c, EncryptedTensor) is False:
- raise ValueError("Ciphertext items must be EncryptedTensor")
-
- encrypted_dividend = c.dividend
- encrypted_divisor = c.divisor
- encrypted_sign = c.sign
-
- dividend = self.cs.decrypt(ciphertext=encrypted_dividend)
- divisor = self.cs.decrypt(ciphertext=encrypted_divisor)
- sign = self.cs.decrypt(ciphertext=encrypted_sign)
-
- if sign == self.cs.plaintext_modulo - 1:
- sign = -1
- elif sign == 1:
- sign = 1
- else:
- raise ValueError("this cannot be true!")
-
- m = sign * (dividend / divisor)
+ for c in encrypted_tensor.fractions:
+ if isinstance(c, Fraction) is False:
+ raise ValueError("Ciphertext items must be type of Fraction")
+
+ sign = c.sign
+ abs_dividend = self.cs.decrypt(ciphertext=c.abs_dividend)
+ # dividend = self.cs.decrypt(ciphertext=c.dividend)
+ divisor = self.cs.decrypt(ciphertext=c.divisor)
+ m = sign * abs_dividend / divisor
 
  plain_tensor.append(m)
  return plain_tensor

lightphe/commons/phe_utils.py

@@ -1,9 +1,10 @@
-from typing import Union, Tuple
+from typing import Union, Tuple, Optional
+from decimal import Decimal, getcontext
 from lightphe.commons.logger import Logger
 
 logger = Logger(module="lightphe/commons/phe_utils.py")
 
-# pylint: disable=no-else-return
+# pylint: disable=no-else-return, no-else-break
 
 
 def parse_int(value: Union[int, float], modulo: int) -> int:
@@ -22,10 +23,32 @@ def parse_int(value: Union[int, float], modulo: int) -> int:
  return result
 
 
-def fractionize(value: float, modulo: int) -> Tuple[int, int]:
- decimal_places = len(str(value).split(".")[1])
- scaling_factor = 10**decimal_places
- integer_value = int(value * scaling_factor) % modulo
+def fractionize(value: float, modulo: int, precision: Optional[int] = None) -> Tuple[int, int]:
+ getcontext().prec = 50
+
+ if precision is None:
+ decimal_places = len(str(value).split(".")[1])
+ scaling_factor = 10**decimal_places
+ else:
+ scaling_factor = 10**precision
+
+ while True:
+ integer_value = int(Decimal(value) * Decimal(scaling_factor)) % modulo
+
+ if precision is None:
+ break
+
+ if scaling_factor > 10**precision:
+ # If scaling factor is too large, discard excess part of integer value
+ integer_value = int(integer_value / (10 ** (scaling_factor - 10**precision)))
+ break
+ elif scaling_factor < 10 ** (precision - 1):
+ # If scaling factor is too small, multiply dividend and divisor 10 times
+ value *= 10
+ scaling_factor *= 10
+ else:
+ break
+
  logger.debug(f"{integer_value}*{scaling_factor}^-1 mod {modulo}")
  return integer_value, scaling_factor
 

lightphe/models/Ciphertext.py

@@ -102,15 +102,7 @@ def __rmul__(self, constant: Union[int, float]) -> "Ciphertext":
  Returns
  scalar multiplication of ciphertext
  """
- if self.cs.keys.get("public_key") is None:
- raise ValueError("You must have public key to perform scalar multiplication")
-
- if isinstance(constant, float):
- constant = phe_utils.parse_int(value=constant, modulo=self.cs.plaintext_modulo)
-
- # Handle multiplication with a constant on the right
- result = self.cs.multiply_by_contant(ciphertext=self.value, constant=constant)
- return Ciphertext(algorithm_name=self.algorithm_name, keys=self.keys, value=result)
+ return self.__mul__(other=constant)
 
  def __xor__(self, other: "Ciphertext") -> "Ciphertext":
  """