diff --git a/cas_wrapper.py b/cas_wrapper.py new file mode 100644 index 0000000..8533dc9 --- /dev/null +++ b/cas_wrapper.py @@ -0,0 +1,95 @@ +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_path: str, + metadata: Dict = None, + ) -> str: + """ + Wrapper around `cas notarize` + :param local_path: path to a local Git repo + :param metadata: additional metadata + :return: hash of notarized commit + :rtype: str + """ + command = self._cas[ + 'notarize', + local_path, + '-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() + return json.loads(result_of_execution)['hash'] + + def authenticate( + self, + local_path: str, + ): + """ + Wrapper around `cas authenticate` + :param local_path: path to a local Git repo + (should be started from `git://`) + or to a single local file + :return: true if a commit is trusted, vice versa - false + :rtype: bool + """ + command = self._cas[ + 'authenticate', + local_path, + '-o', + 'json', + ] + try: + with local.env( + CAS_API_KEY=self._cas_api_key, + SIGNER_ID=self._cas_signer_id + ): + result_of_execution = command() + 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) + return not bool(json.loads(result_of_execution)['status']) diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..d7c772d --- /dev/null +++ b/setup.py @@ -0,0 +1,26 @@ +from setuptools import setup + +setup( + name="cas_wrapper", + version="0.0.1", + author="Stepan Oksanichenko", + author_email="soksanichenko@almalinux.org", + description="The python wrapper around binary cas from " + "project Codenotary Community Attestation Service.", + url="https://git.almalinux.org/almalinux/cas_wrapper", + project_urls={ + "Bug Tracker": "https://git.almalinux.org/almalinux/cas_wrapper/issues", + }, + classifiers=[ + "Programming Language :: Python :: 3", + "License :: OSI Approved :: " + "GNU General Public License v3 or later (GPLv3+)", + "Operating System :: OS Independent", + ], + py_modules=['cas_wrapper'], + scripts=['cas_wrapper.py'], + install_requires=[ + 'plumbum>=1.7.2', + ], + python_requires=">=3.6", +)