Source code for pycape.function_ref

"""A structured set of metadata representing a deployed Cape function.

A :class:`FunctionRef` is intended to capture any/all metadata related to a Cape
function. The metadata is generally user-supplied, provided to them with the output of
the Cape CLI's ``deploy`` command.

Note: We recommend using the :meth:`pycape.Cape.function` method to create
a ``FunctionRef``.

**Usage**

::

    fid = "asdf231lkg1324afdg"
    fchecksum = str(b"2l1h21jhgb2k1jh3".hex())
    fref = FunctionRef(fid, fchecksum)
    token = pycape.Token("eY12shd...")

    cape = Cape()
    cape.connect(fref, token)
"""
from __future__ import annotations

import json
import logging
import os
import pathlib
from typing import Optional
from typing import Union

_logger = logging.getLogger("pycape")


[docs]class FunctionRef: """A reference to a Cape function. Args: id: String denoting the function ID of the deployed Cape function. Typically given with the output of the Cape CLI's ``deploy`` command. name: String denoting the name of the deployed Cape function. Must be of the form ``USER/FUNC_NAME`` where ``USER`` is the Github username of the Cape user and ``FUNC_NAME`` is the name they gave for the function at deploy-time. checksum: Optional string denoting the checksum of the deployed Cape function. If supplied as part of a ``FunctionRef``, the :class:`~pycape.cape.Cape` client will verify that enclave responses includes a matching checksum whenever the ``FunctionRef`` is included in Cape requests. """ def __init__( self, id: Optional[str] = None, name: Optional[str] = None, checksum: Optional[str] = None, ): id_ = id if id_ is None and name is None: raise ValueError( "Must provide one of `id` or `name` arguments, found None for both." ) if id_ is not None and not isinstance(id_, str): raise TypeError(f"Function id must be a string, found {type(id_)}.") self._id = id_ self._user = None self._name = None if name is not None: if not isinstance(name, str): raise TypeError(f"Function name must be a string, found {type(id_)}.") terms = name.split("/") if len(terms) != 2: raise ValueError( "Function name must be of form '<username>/<function_name>', " f"found '{name}'." ) self._user, self._name = terms if checksum is not None and not isinstance(checksum, str): raise TypeError( f"Function checksum must be a string, found {type(checksum)}." ) self._checksum = checksum def __repr__(self): return ( f"{self.__class__.__name__}(\n" f" name={self.full_name},\n" f" id={self.id},\n" f" checksum={self.checksum},\n" f")" ) @property def id(self): return self._id @property def checksum(self): return self._checksum @property def user(self): return self._user @property def name(self): return self._name @property def full_name(self): if self.user is not None and self.name is not None: return f"{self.user}/{self.name}"
[docs] @classmethod def from_json(cls, function_json: Union[str, os.PathLike]) -> FunctionRef: """Construct a :class:`~.function_ref.FunctionRef` from a JSON string or file. Args: function_json: a JSON string or filepath containing function ID and optional function checksum. Returns: A :class:`~.function_ref.FunctionRef` representing the deployed Cape function. Raises: ValueError: if the json file doesn't exist, or the json is missing a ``function_id`` key-value. TypeError: if ``function_json`` is neither Path-like nor str. """ if isinstance(function_json, pathlib.Path): function_config = _try_load_json_file(function_json) if function_config is None: raise ValueError(f"JSON file not found @ {str(function_json)}") elif isinstance(function_json, str): # try to treat function_json as filepath str json_path = pathlib.Path(function_json) function_config = _try_load_json_file(json_path) # if file not found, treat function_json as json str function_config = function_config or json.loads(function_json) else: raise TypeError( "The function_json argument expects a json string or " f"a path to a json file, found: {type(function_json)}." ) function_id = function_config.get("function_id") function_name = function_config.get("function_name") if function_id is None and function_name is None: raise ValueError( "Function JSON must have either function_id or function_name values, " "found neither." ) # warn user when they have a deprecated function token function_token = function_config.get("function_token") if function_token is not None: _logger.warn( "Ignoring function_token in FunctionRef json. Function tokens have " "been removed. Instead, request a Personal Access Token from the " "function owner and pass it to Cape.run. More info at " "https://docs.capeprivacy.com/reference/user-tokens." ) function_checksum = function_config.get("function_checksum") return cls(function_id, function_name, function_checksum)
[docs] def to_json(self, path: Optional[Union[str, os.PathLike]] = None) -> Optional[str]: """Write this :class:`~.function_ref.FunctionRef` to a JSON string or file. Args: path: Optional file path to write the resulting JSON to. Returns: If ``path`` is None, a string with this :class:`~.function_ref.FunctionRef` as a JSON struct. """ fn_ref_dict = {} if self.id is not None: fn_ref_dict["function_id"] = self.id if self.user is not None and self.name is not None: fn_ref_dict["function_name"] = f"{self.user}/{self.name}" if self.checksum is not None: fn_ref_dict["function_checksum"] = self.checksum if path is None: return json.dumps(fn_ref_dict) with open(path, "w") as f: json.dump(fn_ref_dict, f)
def _try_load_json_file(json_file: pathlib.Path): if json_file.exists(): with open(json_file, "r") as f: json_output = json.load(f) return json_output