81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
import json
|
|
from typing import Dict
|
|
|
|
from plumbum import local, ProcessExecutionError
|
|
|
|
|
|
class CasWrapper:
|
|
"""
|
|
The python wrapper around binary `cas`
|
|
from Codenotary Community Attestation Service
|
|
"""
|
|
|
|
binary_name = 'cas'
|
|
|
|
def __init__(
|
|
self,
|
|
cas_api_key: str,
|
|
cas_signer_id: str,
|
|
):
|
|
if self.binary_name not in local:
|
|
raise FileNotFoundError(
|
|
'Binary CAS is not found in PATH on the machine',
|
|
)
|
|
self._cas_api_key = cas_api_key
|
|
self._cas_signer_id = cas_signer_id
|
|
with local.env(
|
|
CAS_API_KEY=self._cas_api_key,
|
|
SIGNER_ID=self._cas_signer_id
|
|
):
|
|
self._cas = local['cas']
|
|
self._cas['login']()
|
|
|
|
def notarize(self, local_git_repo_path: str) -> str:
|
|
"""
|
|
Wrapper around `cas notarize`
|
|
:param local_git_repo_path: path to a local Git repo
|
|
:return: hash of notarized commit
|
|
:rtype: str
|
|
"""
|
|
command = self._cas[
|
|
'notarize',
|
|
local_git_repo_path,
|
|
'-o',
|
|
'json',
|
|
]
|
|
result_of_execution = command()
|
|
return json.loads(result_of_execution)['hash']
|
|
|
|
def authenticate(
|
|
self,
|
|
local_path: str,
|
|
metadata: Dict = None,
|
|
):
|
|
"""
|
|
Wrapper around `cas authenticate`
|
|
:param local_path: path to a local Git repo
|
|
(should be started from `git://`)
|
|
or to a single local file
|
|
:param metadata: additional metadata
|
|
:return: true if a commit is trusted, vice versa - false
|
|
:rtype: bool
|
|
"""
|
|
command = self._cas[
|
|
'authenticate',
|
|
local_path,
|
|
'-o',
|
|
'json',
|
|
]
|
|
if metadata is not None:
|
|
for key, value in metadata.items():
|
|
command = command[
|
|
'-a',
|
|
f'{key}={value}',
|
|
]
|
|
try:
|
|
result_of_execution = command()
|
|
except ProcessExecutionError:
|
|
# in case if commit is untrusted
|
|
result_of_execution = command(retcode=1)
|
|
return not bool(json.loads(result_of_execution)['status'])
|