"""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.
**Usage**
::
fid = "asdf231lkg1324afdg"
fchecksum = str(b"2l1h21jhgb2k1jh3".hex())
fref = FunctionRef(fid, fchecksum)
cape = Cape()
cape.connect(fref)
"""
from __future__ import annotations
import json
import os
import pathlib
from typing import Optional
from typing import Union
[docs]class FunctionRef:
"""A reference to a Cape function.
Args:
id: Required string denoting the function ID of the deployed Cape function.
Typically given with the output of the Cape CLI's ``deploy`` command.
token: Required string containing a Cape function token generated by the Cape
CLI during ``cape token``.
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: str,
token: str,
checksum: Optional[str] = None,
):
id_ = id
if not isinstance(id_, str):
raise TypeError(f"Function id must be a string, found {type(id_)}.")
if not isinstance(token, str):
raise TypeError(f"Function token must be a string, found {type(token)}.")
if checksum is not None and not isinstance(checksum, str):
raise TypeError(
f"Function checksum must be a string, found {type(checksum)}."
)
self._id = id_
self._checksum = checksum
self._token = token
@property
def id(self):
return self._id
@property
def checksum(self):
return self._checksum
@property
def token(self):
return self._token
[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, token, and
optional checksum, e.g. as generated by the Cape CLI ``token`` command.
Returns:
A :class:`~.function_ref.FunctionRef` representing the deployed Cape
function.
Raises:
ValueError: if the json token file doesn't exist or, the token file
doesn't contain a `function_id` or a `function_token`.
"""
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")
if function_id is None:
raise ValueError("Couldn't find a `function_id` in the token file provided")
function_token = function_config.get("function_token")
if function_token is None:
raise ValueError(
"Couldn't find a `function_token` in the token file provided"
)
function_checksum = function_config.get("function_checksum")
return cls(function_id, function_token, 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 = {
"function_id": self._id,
"function_token": self._token,
"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
