from __future__ import annotations
import json
import logging
import os
import sys
import tempfile
from abc import ABC, ABCMeta, abstractmethod
from collections.abc import Mapping
from os import path
import numpy as np
from pyfe4ai.core import config as configs, profiles
from pyfe4ai.core import versions
from pyfe4ai.core.status import warn_if_known_insecure
from pyfe4ai.core.constants import CryptoCONST
from pyfe4ai.core.exceptions import FEKeyError, FESchemeError, FEValidationError, warn_at_caller
from pyfe4ai.core.ids import as_id, check_id_list
logger = logging.getLogger(__name__)
#: Environment variable that sets the parameter cache directory (see
#: :func:`cache_dir`).
CACHE_DIR_ENV = "PYFE4AI_CACHE_DIR"
[docs]
def default_cache_dir() -> str:
"""The per-user cache directory of PyFE4AI on this platform.
``~/Library/Caches/pyfe4ai`` on macOS, ``%LOCALAPPDATA%\\pyfe4ai\\Cache``
on Windows, and ``$XDG_CACHE_HOME/pyfe4ai`` elsewhere (``~/.cache/pyfe4ai``
if ``XDG_CACHE_HOME`` is unset or not an absolute path) -- the
directories of ``platformdirs.user_cache_dir("pyfe4ai", appauthor=False)``.
"""
home = path.expanduser("~")
if sys.platform == "darwin":
return path.join(home, "Library", "Caches", "pyfe4ai")
if os.name == "nt":
base = os.environ.get("LOCALAPPDATA") or path.join(home, "AppData", "Local")
return path.join(base, "pyfe4ai", "Cache")
xdg = os.environ.get("XDG_CACHE_HOME", "")
return path.join(xdg if path.isabs(xdg) else path.join(home, ".cache"), "pyfe4ai")
[docs]
def cache_dir(explicit=None) -> str:
"""The parameter cache directory: *explicit* (the ``cache_dir`` entry of a
key generator's or crypto object's config) if given, else the
``PYFE4AI_CACHE_DIR`` environment variable if set and not empty, else
:func:`default_cache_dir`. A relative path is relative to the working
directory.
Key generators keep their records in ``<cache_dir>/authority/<scheme
type>/`` (see :class:`ParameterCacheMixin`). The cache is trusted: it must
not be writable by others.
Raises:
FEValidationError: If *explicit* is not a non-empty path.
"""
if explicit is not None:
if not isinstance(explicit, (str, os.PathLike)) or not os.fspath(explicit):
raise FEValidationError("cache_dir must be a non-empty path, got {!r}".format(explicit))
return os.fspath(explicit)
return os.environ.get(CACHE_DIR_ENV) or default_cache_dir()
[docs]
class ParameterCacheMixin:
"""Mixin that provides JSON-based parameter caching with validation.
Subclasses must set ``_scheme_type`` (a CryptoCONST string) and implement
``_param_verification``, ``_apply_parameters``, and
``_generate_and_save``. The records are kept in ``config_folder``, the
``authority`` folder of the cache directory (:func:`cache_dir`: the
``cache_dir`` entry of the config, ``PYFE4AI_CACHE_DIR`` or the per-user
cache directory).
"""
config_folder: str
_scheme_type: str
#: Set by key generators whose parameters include a DCR modulus: no check
#: can tell whether someone knows its factors, so under the ``standard``
#: profile such parameters are generated afresh, never loaded or stored.
_holds_dcr_modulus = False
def _load_parameters(self) -> None:
"""Load cached parameters from disk, or generate and save fresh ones.
Looks for :meth:`_param_file_name` (``param.json`` unless a key
generator keeps several kinds of record) in the scheme-specific
folder ``<config_folder>/<scheme type>``. If it parses, matches the configuration (per :meth:`_param_verification`)
and the applied values pass :meth:`_loaded_parameters_valid`, they are
used; otherwise (also for an unreadable or truncated file) fresh ones
are generated by :meth:`_generate_and_save`. The new file is written
to a temporary file and renamed, so an interrupted run never leaves a
partial record.
"""
folder = os.path.join(self.config_folder, self._scheme_type)
os.makedirs(folder, exist_ok=True)
param_file = os.path.join(folder, self._param_file_name())
if not versions.caches_record(self):
fd, tmp_file = tempfile.mkstemp(dir=folder, prefix=".param-", suffix=".json")
os.close(fd)
try:
self._generate_and_save(tmp_file)
finally:
os.unlink(tmp_file)
return
if os.path.exists(param_file):
logger.debug("load param file: %s", param_file)
try:
with open(param_file, "r", encoding="utf-8") as f:
param = json.load(f)
if (
isinstance(param, dict)
and self._param_verification(param)
and self._apply_and_check(param)
):
logger.debug("load param file successfully")
return
logger.error("param file is invalid, regenerate...")
warn_at_caller(
"the parameter record {} does not match this configuration (for example a record of an "
"earlier version): regenerated with new parameters and seeds; keys, public parameters and "
"ciphertexts made with the old record no longer match".format(param_file),
)
except (OSError, ValueError, KeyError, TypeError, IndexError) as exc:
logger.error("param file is unreadable (%s), regenerate...", exc)
warn_at_caller(
"the parameter record {} is unreadable ({}): regenerated with new parameters and seeds; keys, "
"public parameters and ciphertexts made with the old record no longer match".format(param_file, exc),
)
fd, tmp_file = tempfile.mkstemp(dir=folder, prefix=".param-", suffix=".json")
os.close(fd)
try:
self._generate_and_save(tmp_file)
os.replace(tmp_file, param_file)
finally:
if os.path.exists(tmp_file):
os.unlink(tmp_file)
def _param_file_name(self) -> str:
"""The record file of this configuration in the scheme's config folder:
``param.json``. Key generators whose configurations derive records of
different kinds (e.g. a prime and an RNS modulus) name them apart, so
that alternating configurations in one folder do not overwrite each
other's record."""
return "param.json"
def _apply_and_check(self, param: dict) -> bool:
self._apply_parameters(param)
return self._loaded_parameters_valid()
def _loaded_parameters_valid(self) -> bool:
"""Structural checks of parameters just loaded from the cache.
The default accepts them; the discrete-log and DCR families check
primality, subgroup orders and sizes (see
:func:`~pyfe4ai.primitives.groups.integer_groups.is_dl_group`).
"""
return True
@abstractmethod
def _param_verification(self, param: dict) -> bool:
"""Return True if *param* matches the current configuration."""
raise NotImplementedError
@abstractmethod
def _apply_parameters(self, param: dict) -> None:
"""Assign cryptographic attributes from the loaded *param* dict."""
raise NotImplementedError
@abstractmethod
def _generate_and_save(self, param_file: str) -> None:
"""Generate fresh parameters, assign them to *self*, and persist to *param_file*."""
raise NotImplementedError
class _ProfiledKeyGeneratorMeta(ABCMeta):
"""Converts the config to its scheme's config class and completes it with
its security profile before ``__init__`` runs (:mod:`pyfe4ai.core.config`).
The whole ``__init__`` chain of a key generator then sees the presets of
the profile; a config without ``security_profile`` keeps its values.
"""
def __call__(cls, *args, **kwargs):
# every config goes through the profile: any Mapping is resolved, and
# anything else (which could skip it) is refused
if args:
args = (cls._resolved(args[0]),) + args[1:]
elif "config" in kwargs:
kwargs["config"] = cls._resolved(kwargs["config"])
return super().__call__(*args, **kwargs)
def _resolved(cls, config) -> configs.SchemeConfig:
if not isinstance(config, Mapping):
raise FEValidationError(
"a key generator config must be a mapping, got {}".format(type(config).__name__)
)
scheme = profiles.scheme_of(cls)
return configs.resolve(configs.config_class(scheme)(config), scheme)
[docs]
class IPFEAbsKeyGenerator(metaclass=_ProfiledKeyGeneratorMeta):
"""Abstract base for all inner-product FE key generators.
Subclasses implement :meth:`setup` (generate keys), :meth:`get_public_parameters`,
:meth:`get_private_keys`, and :meth:`get_decryption_keys`.
Args:
config: Scheme configuration (a mapping, converted to the scheme's
config class of :mod:`pyfe4ai.core.config`). Expected keys vary by
scheme; common keys include ``sec_param``, ``eta``, ``n``, ``s``,
``lst_nid``, ``security_profile`` (see :mod:`pyfe4ai.core.profiles`) and
``cache_dir`` (the parameter cache directory, see
:func:`cache_dir`).
"""
def __init__(self, config: dict, **kwargs) -> None:
super().__init__()
warn_if_known_insecure(type(self).__module__)
self.security_profile = config.get("security_profile", profiles.TOY)
stat_param = config.get("stat_param")
if stat_param is not None and (
isinstance(stat_param, bool) or not isinstance(stat_param, int) or stat_param > profiles.MAX_STAT_PARAM
):
raise FEValidationError(
"stat_param must be an integer of at most {} (the MPFR Gaussians' guard bits cover "
"no more), got {!r}".format(profiles.MAX_STAT_PARAM, stat_param)
)
self.sec_param = config.get("sec_param", CryptoCONST.SEC_PARAM)
self.n = config.get("n", None)
self.s = config.get("s", None)
self.config_folder = path.join(cache_dir(config.get("cache_dir")), "authority")
if self.n is not None:
self.lst_nid = config.get(
"lst_nid", ["nid_{}".format(i) for i in range(self.n)]
)
# a repeated id would silently merge two clients' keys
check_id_list(self.lst_nid, "lst_nid")
if len(self.lst_nid) != self.n:
# parameters (bounds, noise widths, moduli) are sized for n slots
raise FEValidationError(
"lst_nid has {} clients but n={}".format(len(self.lst_nid), self.n)
)
[docs]
@abstractmethod
def setup(self) -> None:
"""Generate master public and secret keys for the scheme."""
raise NotImplementedError()
[docs]
@abstractmethod
def get_public_parameters(self, **kwargs) -> None | dict:
"""Return serialisable public parameters.
Returns:
A dict of public parameters, or ``None`` on error.
"""
raise NotImplementedError()
[docs]
@abstractmethod
def get_private_keys(self, nid: str, **kwargs) -> None | dict:
"""Return the private (encryption) keys for client *nid*.
Args:
nid: A client/node identifier.
Returns:
A dict of private keys, or ``None`` if *nid* is invalid.
"""
raise NotImplementedError()
[docs]
@abstractmethod
def get_decryption_keys(self, sid: str, **kwargs) -> None | dict:
"""Derive a functional decryption key for session *sid*.
Args:
sid: A session identifier.
**kwargs: Must include ``credentials`` with a ``fusion_weight`` entry.
Returns:
A dict containing the decryption key material.
Raises:
FEKeyError: If credentials are missing.
FEValidationError: If fusion weights are invalid.
"""
raise NotImplementedError()
[docs]
def decoded_layer(values, layer: int, scale: int = 1) -> np.ndarray:
"""Return the decrypted elements of one ndarray layer, divided by *scale*.
Args:
values: Array (or nested sequence) of decrypted integers; ``None``
marks an element that did not decrypt.
layer: Index of the layer, for the error message.
scale: Fixed-point scale of the plaintexts (``10^precision``).
Raises:
FESchemeError: For the first element that did not decrypt -- its
result is outside the decryption bound, or the key and the
ciphertext do not match. ``None`` is never turned into ``NaN``.
"""
values = np.asarray(values, dtype=object)
for index, value in np.ndenumerate(values):
if value is None:
raise FESchemeError(
"element {} of layer {} does not decrypt (result outside the decryption "
"bound, or key and ciphertext do not match)".format(index, layer)
)
return (values / scale).astype(float)
#: The ``FEKeyError`` text for an encryption key (``keys["sk"]``) that is not
#: a dict (raised by ``IPFEAbsCrypto._check_private_key``).
PRIVATE_KEY_NOT_A_DICT = "the encryption key (keys['sk']) is not a dict of its fields"
[docs]
class IPFEAbsCrypto(ABC):
"""Abstract base for all inner-product FE crypto operations.
Subclasses implement :meth:`encrypt_lst_ndarray`, :meth:`compute_lst_ndarray_ct`,
and :meth:`decrypt_lst_ndarray_ct`.
Args:
config: Runtime configuration dict containing at least ``keys``
(a dict with ``pp``, ``sk``, and/or ``dk`` sub-dicts) and
optionally ``id``, ``type``, and ``precision``.
"""
def __init__(self, config: dict, **kwargs) -> None:
super().__init__()
warn_if_known_insecure(type(self).__module__)
self.keys = config.get("keys", None)
if not self.keys:
raise FEKeyError("no keys provided to initialize crypto system")
self.id = as_id(config.get("id", None)) # a numpy integer id is the int it equals
self.type = config.get("type", None)
self.precision = config.get("precision", 0)
self.config_folder = path.join(cache_dir(config.get("cache_dir")), "crypto")
[docs]
@abstractmethod
def encrypt_lst_ndarray(self, lst_ndarray: list, **kwargs) -> list | None:
"""Encrypt a list of ndarray-like plaintexts.
Args:
lst_ndarray: List of plaintext arrays to encrypt.
Returns:
List of ciphertext dicts, or ``None`` on error.
"""
raise NotImplementedError()
[docs]
@abstractmethod
def compute_lst_ndarray_ct(self, dict_ndarray_ct: dict, **kwargs) -> list | None:
"""Aggregate ciphertexts from multiple clients.
Args:
dict_ndarray_ct: Mapping of client IDs to their ciphertext lists.
Returns:
Aggregated ciphertext list, or ``None`` on error.
"""
raise NotImplementedError()
[docs]
@abstractmethod
def decrypt_lst_ndarray_ct(self, dict_ndarray_ct: dict, **kwargs) -> list | None:
"""Decrypt aggregated ndarray ciphertexts.
Args:
dict_ndarray_ct: Mapping of client IDs to their ciphertext lists.
Returns:
One float array of decrypted results per layer.
Raises:
FESchemeError: If an element does not decrypt (see
:func:`~pyfe4ai.core.base.decoded_layer`).
"""
raise NotImplementedError()
def _has_public_parameters(self) -> bool:
"""Check whether public parameters are available."""
return "pp" in self.keys
def _has_private_keys(self) -> bool:
"""Check whether private (encryption) keys are available."""
return "sk" in self.keys
def _check_private_key(self) -> None:
"""Raise unless the encryption key ``keys["sk"]``, where given, is a
dict (a mapping): the encryption paths that read its fields with
``.get`` would fail on another value with a bare ``AttributeError`` or
``TypeError``.
Raises:
FEKeyError: :data:`PRIVATE_KEY_NOT_A_DICT` (``None``, a list, a
string, ...).
"""
if self._has_private_keys() and not isinstance(self.keys["sk"], Mapping):
raise FEKeyError(PRIVATE_KEY_NOT_A_DICT)
def _has_decryption_keys(self) -> bool:
"""Check whether decryption keys are available."""
return "dk" in self.keys