import json
import os
import pathlib
import random
import re
import string
import subprocess
from typing import Optional
from typing import Union
from pycape import _config as cape_config
from pycape import function_ref as fref
from pycape import token as tkn
from pycape.cape import _synchronizer
[docs]@_synchronizer
async def deploy(
deploy_path: Union[str, os.PathLike],
url: Optional[str] = None,
public: bool = False,
) -> fref.FunctionRef:
"""Deploy a directory or a zip file containing a Cape function declared in
an app.py script.
This method calls `cape deploy` and `cape token` from the Cape CLI to deploy
a Cape function then returns a `~.function_ref.FunctionRef` representing
the deployed function. This `~.function_ref.FunctionRef` will hold a function ID or
function name, and a function checksum. Note that the ``deploy_path`` has to point
to a directory or a zip file containing a Cape function declared in an app.py file
and the size of its content is currently limited to 1GB.
Args:
deploy_path: A path pointing to a directory or a zip file containing
a Cape function declared in an app.py script.
url: The Cape platform's websocket URL, which is responsible for forwarding
client requests to the proper enclave instances. If None, tries to load
value from the ``CAPE_ENCLAVE_HOST`` environment variable. If no such
variable value is supplied, defaults to ``"https://app.capeprivacy.com"``.
public: Boolean determining if the function should be publicly accessible or
not.
Returns:
A :class:`~.function_ref.FunctionRef` representing the deployed Cape
function.
Raises:
RuntimeError: if the websocket response or the enclave attestation doc is
malformed, if the function path is not pointing to a directory
or a zip file, if folder size exceeds 1GB, or if the Cape CLI cannot
be found on the device.
"""
url = url or cape_config.ENCLAVE_HOST
deploy_path = pathlib.Path(deploy_path)
cmd_deploy = f"cape deploy {deploy_path} -u {url} -o json"
if public:
cmd_deploy += " --public"
out_deploy, err_deploy = _call_cape_cli(cmd_deploy)
err_deploy = err_deploy.decode()
out_deploy = out_deploy.decode()
err_deploy = err_deploy.split("\n")
error_output = None
# Parse err_deploy to get potential errors
for msg in err_deploy:
if "Error" in msg:
error_output = msg
error_msg = error_output.partition("Error:")[2]
raise RuntimeError(f"Cape deploy error - {error_msg}")
# Parse out_deploy to get function ID and function checksum
out_deploy = json.loads(out_deploy.split("\n")[0])
function_id = out_deploy.get("function_id")
function_checksum = out_deploy.get("function_checksum")
if function_id is None:
raise RuntimeError(
f"Function ID not found in 'cape.deploy' response: \n{err_deploy}"
)
return fref.FunctionRef(id=function_id, checksum=function_checksum)
[docs]@_synchronizer
async def token(
name: Optional[str] = None,
description: Optional[str] = None,
expiry: Optional[str] = None,
) -> tkn.Token:
"""Generate a Personal Access Token.
This method calls `cape token` from the Cape CLI to generate a Personal Access
Token. Tokens can be created statically (long expiration and bundled with your
application) or created dynamically (short-lived) and have an owner-specified
expiration. This PAT is required along with the function name or function ID when
calling a Cape function.
Args:
name: Optional name for the token.
description: Optional description for the token.
expiry: Amount of time in seconds until the function token expires. Defaults to
1h.
Returns:
A :class:`~.token.Token` representing the PAT.
Raises:
RuntimeError: if the Cape CLI cannot be found on the device, if the CLI failed
to generate the token, or if PyCape failed to parse the token from the CLI's
output.
"""
if name is None:
token_suffix = "".join(random.choices(string.ascii_lowercase, k=8))
name = f"pycape-deploy-{token_suffix}"
if description is None:
description = '"An ephemeral token generated by pycape.experimental.cli."'
cmd_token = f"cape token create -n {name} -d {description}"
if expiry:
cmd_token += f" -e {expiry}"
out_token, err_token = _call_cape_cli(cmd_token)
err_token = err_token.decode()
out_token = out_token.decode()
err_lines = err_token.split("\n")
error_output = None
# Parse err_token to get potential errors
for i in err_lines:
if "Error" in i:
error_output = i
error_msg = error_output.partition("Error:")[2]
raise RuntimeError(f"Cape token error - {error_msg}")
# Parse out_token to get function token
token_match = re.match("Success! Your token: (.*)", out_token)
if token_match is None:
raise RuntimeError(
"Cape token error - could not parse token output: "
f"\n{out_token}\n{err_lines}"
)
function_token = token_match.group(1)
if function_token is None:
raise RuntimeError(
"Function token not found in 'cape.token' response: "
f"\n{out_token}\n{err_lines}"
)
return tkn.Token(function_token)
def _check_if_cape_cli_available():
exitcode, output = subprocess.getstatusoutput("cape")
if exitcode != 0:
raise RuntimeError(
f"Please make sure Cape CLI is installed on your device: {output}"
)
def _call_cape_cli(cape_cmd):
_check_if_cape_cli_available()
proc = subprocess.Popen(
cape_cmd,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
out, err = proc.communicate()
return out, err
