cas_wrapper/cas_wrapper.py

195 lines
6.3 KiB
Python
Raw Normal View History

2022-06-20 10:20:00 +00:00
import json
import logging
import typing
2022-06-20 10:20:00 +00:00
from plumbum import local, ProcessExecutionError
class CasWrapper:
"""
The python wrapper around binary `cas`
from Codenotary Community Attestation Service
"""
binary_name = 'cas'
@classmethod
def _is_binary_present(cls):
if cls.binary_name not in local:
raise FileNotFoundError(
'Binary CAS is not found in PATH on the machine',
)
2022-06-20 10:20:00 +00:00
def __init__(
self,
cas_api_key: str,
cas_signer_id: str,
logger: logging.Logger = None,
2022-06-20 10:20:00 +00:00
):
self._is_binary_present()
2022-06-20 10:20:00 +00:00
self._cas_api_key = cas_api_key
self._cas_signer_id = cas_signer_id
self._cas = local['cas']
self._logger = logger
if self._logger is None:
self._logger = logging.getLogger()
@classmethod
def get_version(cls):
cls._is_binary_present()
command = local['cas']['--version']
version = command()[:-1].split().split('v')[1]
return version
def ensure_login(self):
2022-06-20 10:20:00 +00:00
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id,
2022-06-20 10:20:00 +00:00
):
2022-06-20 10:22:52 +00:00
self._cas['login']()
2022-06-20 10:20:00 +00:00
def notarize(
self,
local_path: str,
metadata: typing.Dict = None,
) -> str:
2022-06-20 10:20:00 +00:00
"""
Wrapper around `cas notarize`
:param local_path: path to a local Git repo
:param metadata: additional metadata
2022-06-20 10:20:00 +00:00
:return: hash of notarized commit
:rtype: str
"""
command = self._cas[
'notarize',
local_path,
2022-06-20 10:20:00 +00:00
'-o',
'json',
]
if metadata is not None:
for key, value in metadata.items():
command = command[
'-a',
f'{key}={value}',
]
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id
):
result_of_execution = command()
2022-06-20 10:20:00 +00:00
return json.loads(result_of_execution)['hash']
def authenticate(
self,
local_path: str,
return_json: bool = False,
use_hash: bool = False,
):
2022-06-20 10:20:00 +00:00
"""
Wrapper around `cas authenticate`
:param local_path: path to a local Git repo
(should be started from `git://`)
or to a single local file or hash
:param return_json: flag for return json response
:param use_hash: flag for authenticate by hash
2022-06-20 10:20:00 +00:00
:return: true if a commit is trusted, vice versa - false
or dict with result if return_json param is True
:rtype: bool or dict
2022-06-20 10:20:00 +00:00
"""
command_args = ['authenticate', local_path]
if use_hash:
command_args = ['authenticate', '--hash', local_path]
command_args.extend(('-o', 'json'))
command = self._cas[command_args]
2022-06-20 10:20:00 +00:00
try:
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id
):
result_of_execution = command()
2022-06-20 10:20:00 +00:00
except ProcessExecutionError:
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id
):
# in case if commit is untrusted
result_of_execution = command(retcode=1)
json_result = json.loads(result_of_execution)
if return_json:
return json_result
return not bool(json_result['status'])
def authenticate_source(
self,
local_path: str,
) -> typing.Tuple[bool, typing.Optional[str]]:
"""
Authenticates source by git path.
Returns authenticate result and source commit hash.
"""
is_authenticated = False
commit_cas_hash = None
self.ensure_login()
try:
result_json = self.authenticate(local_path, return_json=True)
is_authenticated = result_json['verified']
commit_cas_hash = result_json['hash']
# we can fall with ProcessExecutionError,
# because source can be not notarized
except ProcessExecutionError:
self._logger.exception('Cannot authenticate: %s', local_path)
return is_authenticated, commit_cas_hash
def authenticate_artifact(
self,
local_path: str,
use_hash: bool = False,
) -> bool:
"""
Authenticates artifact by artifact path or hash if `use_hash` is True.
Returns authenticate result.
"""
is_authenticated = False
self.ensure_login()
try:
is_authenticated = self.authenticate(
local_path,
use_hash=use_hash,
return_json=True,
)['verified']
# we can fall with ProcessExecutionError,
# because artifact can be not notarized
except ProcessExecutionError:
self._logger.exception('Cannot authenticate: %s', local_path)
return is_authenticated
def notarize_artifacts(
self,
artifact_paths: typing.List[str],
metadata: typing.Dict[str, typing.Any],
) -> typing.Tuple[bool, typing.Dict[str, str]]:
"""
Notarize artifacts by their paths.
Returns `True` if all artifacts was succesful notarizated
and dict with CAS hashes.
"""
all_artifacts_is_notarized = True
notarized_artifacts = {}
self.ensure_login()
# ALBS-576: We stopped doing this process in parallel due to the
# problems experienced and described in this CAS issue:
# https://github.com/codenotary/cas/issues/275
# Hence, we decided to go sequential here until the problem is
# resolved in CAS itself.
for artifact_path in artifact_paths:
try:
cas_artifact_hash = self.notarize(artifact_path, metadata)
except Exception:
self._logger.exception('Cannot notarize artifact: %s',
artifact_path)
all_artifacts_is_notarized = False
continue
notarized_artifacts[artifact_path] = cas_artifact_hash
return all_artifacts_is_notarized, notarized_artifacts