Add signer abstract class

Signed-off-by: Zelin Hao <zelinhao@amazon.com>

src/sign_workflow/sign_artifacts.py

@@ -15,6 +15,7 @@
 from manifests.build_manifest import BuildManifest
 from sign_workflow.signer_pgp import SignerPGP
 from sign_workflow.signer_windows import SignerWindows
+from sign_workflow.signers import Signers
 
 
 class SignArtifacts:
@@ -31,7 +32,7 @@ def __init__(self, target: Path, components: List[str], artifact_type: str, sign
         self.artifact_type = artifact_type
         self.signature_type = signature_type
         self.platform = platform
-        self.__signer_type__(platform)
+        self.signer = Signers.create(platform)
 
     @abstractmethod
     def __sign__(self) -> None:
@@ -47,12 +48,12 @@ def __sign_artifacts__(self, artifacts: List[str], basepath: Path) -> None:
     def __sign_artifact__(self, artifact: str, basepath: Path) -> None:
         self.signer.sign_artifact(artifact, basepath, self.signature_type)
 
-    @classmethod
-    def __signer_type__(self, platform: str) -> None:
-        if (platform == "windows"):
-            self.signer = SignerWindows()
-        if (platform == "linux"):
-            self.signer = SignerPGP()
+    # @classmethod
+    # def __signer_type__(self, platform: str) -> None:
+    #     if (platform == "windows"):
+    #         self.signer = SignerWindows()
+    #     if (platform == "linux"):
+    #         self.signer = SignerPGP()
 
     @classmethod
     def __signer_class__(self, path: Path) -> Type[Any]:

src/sign_workflow/signer.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python
+
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+import logging
+import os
+from pathlib import Path
+from typing import List
+from abc import ABC, abstractmethod
+
+from git.git_repository import GitRepository
+
+"""
+This class is responsible for signing an artifact using the OpenSearch-signer-client and verifying its signature.
+The signed artifacts will be found in the same location as the original artifacts.
+"""
+
+
+class Signer(ABC):
+    git_repo: GitRepository
+
+    def __init__(self) -> None:
+        self.git_repo = GitRepository(self.get_repo_url(), "HEAD", working_subdirectory="src")
+        self.git_repo.execute("./bootstrap")
+        self.git_repo.execute("rm config.cfg")
+
+    def sign_artifact(self, artifact: str, basepath: Path, signature_type: str) -> None:
+        if not self.is_valid_file_type(artifact):
+            logging.info(f"Skipping signing of file {artifact}")
+            return
+        self.generate_signature_and_verify(artifact, basepath, signature_type)
+
+    def sign_artifacts(self, artifacts: List[str], basepath: Path, signature_type: str) -> None:
+        for artifact in artifacts:
+            if not self.is_valid_file_type(artifact):
+                logging.info(f"Skipping signing of file {artifact}")
+                continue
+            self.generate_signature_and_verify(artifact, basepath, signature_type)
+
+    @abstractmethod
+    def generate_signature_and_verify(self, artifact: str, basepath: Path, signature_type: str) -> None:
+        pass
+
+    @abstractmethod
+    def is_valid_file_type(self, file_name: str) -> bool:
+        pass
+
+    def get_repo_url(self) -> str:
+        if "GITHUB_TOKEN" in os.environ:
+            return "https://${GITHUB_TOKEN}@github.com/opensearch-project/opensearch-signer-client.git"
+        return "https://github.com/opensearch-project/opensearch-signer-client.git"
+
+    def __remove_existing_signature__(self, signature_file: str) -> None:
+        if os.path.exists(signature_file):
+            logging.warning(f"Removing existing signature file {signature_file}")
+            os.remove(signature_file)
+
+    @abstractmethod
+    def sign(self, filename: str, signature_type: str) -> None:
+        pass
+
+    @abstractmethod
+    def verify(self, filename: str) -> None:
+        pass

src/sign_workflow/signer_pgp.py

@@ -6,42 +6,23 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-import logging
 import os
 from pathlib import Path
-from typing import List
 
 from git.git_repository import GitRepository
+from sign_workflow.signer import Signer
 
 """
 This class is responsible for signing an artifact using the OpenSearch-signer-client and verifying its signature.
 The signed artifacts will be found in the same location as the original artifacts.
 """
 
 
-class SignerPGP:
+class SignerPGP(Signer):
     git_repo: GitRepository
 
     ACCEPTED_FILE_TYPES = [".zip", ".jar", ".war", ".pom", ".module", ".tar.gz", ".whl", ".crate", ".rpm"]
 
-    def __init__(self) -> None:
-        self.git_repo = GitRepository(self.get_repo_url(), "HEAD", working_subdirectory="src")
-        self.git_repo.execute("./bootstrap")
-        self.git_repo.execute("rm config.cfg")
-
-    def sign_artifact(self, artifact: str, basepath: Path, signature_type: str) -> None:
-        if not self.is_valid_file_type(artifact):
-            logging.info(f"Skipping signing of file {artifact}")
-            return
-        self.generate_signature_and_verify(artifact, basepath, signature_type)
-
-    def sign_artifacts(self, artifacts: List[str], basepath: Path, signature_type: str) -> None:
-        for artifact in artifacts:
-            if not self.is_valid_file_type(artifact):
-                logging.info(f"Skipping signing of file {artifact}")
-                continue
-            self.generate_signature_and_verify(artifact, basepath, signature_type)
-
     def generate_signature_and_verify(self, artifact: str, basepath: Path, signature_type: str) -> None:
         location = os.path.join(basepath, artifact)
         self.sign(location, signature_type)
@@ -52,16 +33,6 @@ def is_valid_file_type(self, file_name: str) -> bool:
             file_name.endswith(x) for x in SignerPGP.ACCEPTED_FILE_TYPES
         )
 
-    def get_repo_url(self) -> str:
-        if "GITHUB_TOKEN" in os.environ:
-            return "https://${GITHUB_TOKEN}@github.com/opensearch-project/opensearch-signer-client.git"
-        return "https://github.com/opensearch-project/opensearch-signer-client.git"
-
-    def __remove_existing_signature__(self, signature_file: str) -> None:
-        if os.path.exists(signature_file):
-            logging.warning(f"Removing existing signature file {signature_file}")
-            os.remove(signature_file)
-
     def sign(self, filename: str, signature_type: str) -> None:
         signature_file = filename + signature_type
         self.__remove_existing_signature__(signature_file)

src/sign_workflow/signer_windows.py

@@ -6,42 +6,23 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-import logging
 import os
 from pathlib import Path
-from typing import List
 
 from git.git_repository import GitRepository
+from sign_workflow.signer import Signer
 
 """
 This class is responsible for signing an artifact using the OpenSearch-signer-client and verifying its signature.
 The signed artifacts will be found in the subfolder called signed under the origin location as the original artifacts.
 """
 
 
-class SignerWindows:
+class SignerWindows(Signer):
     git_repo: GitRepository
 
     ACCEPTED_FILE_TYPES = [".msi", ".exe", ".dll", ".sys", ".ps1", ".psm1", ".psd1", ".cat", ".zip"]
 
-    def __init__(self) -> None:
-        self.git_repo = GitRepository(self.get_repo_url(), "HEAD", working_subdirectory="src")
-        self.git_repo.execute("./bootstrap")
-        self.git_repo.execute("rm config.cfg")
-
-    def sign_artifact(self, artifact: str, basepath: Path, signature_type: str) -> None:
-        if not self.is_valid_file_type(artifact):
-            logging.info(f"Skipping signing of file {artifact}")
-            return
-        self.generate_signature_and_verify(artifact, basepath, signature_type)
-
-    def sign_artifacts(self, artifacts: List[str], basepath: Path, signature_type: str) -> None:
-        for artifact in artifacts:
-            if not self.is_valid_file_type(artifact):
-                logging.info(f"Skipping signing of file {artifact}")
-                continue
-            self.generate_signature_and_verify(artifact, basepath, signature_type)
-
     def generate_signature_and_verify(self, artifact: str, basepath: Path, signature_type: str) -> None:
         self.sign(artifact, basepath, signature_type)
 
@@ -50,16 +31,6 @@ def is_valid_file_type(self, file_name: str) -> bool:
             file_name.endswith(x) for x in SignerWindows.ACCEPTED_FILE_TYPES
         )
 
-    def get_repo_url(self) -> str:
-        if "GITHUB_TOKEN" in os.environ:
-            return "https://${GITHUB_TOKEN}@github.com/opensearch-project/opensearch-signer-client.git"
-        return "https://github.com/opensearch-project/opensearch-signer-client.git"
-
-    def __remove_existing_signature__(self, signature_file: str) -> None:
-        if os.path.exists(signature_file):
-            logging.warning(f"Removing existing signature file {signature_file}")
-            os.remove(signature_file)
-
     def sign(self, artifact: str, basepath: Path, signature_type: str) -> None:
         filename = os.path.join(basepath, artifact)
         signed_prefix = "signed_"

src/sign_workflow/signers.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python
+
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+from typing import Any, List, Type, Union
+
+from sign_workflow.signer_pgp import SignerPGP
+from sign_workflow.signer_windows import SignerWindows
+from sign_workflow.signer import Signer
+
+"""
+This class is responsible for signing an artifact using the OpenSearch-signer-client and verifying its signature.
+The signed artifacts will be found in the same location as the original artifacts.
+"""
+
+
+class Signers:
+    TYPES = {
+        "windows": SignerWindows,
+        "linux": SignerPGP,
+    }
+
+    @classmethod
+    def from_platform(cls, platform: str) -> Signer:
+        klass = cls.TYPES.get(platform, None)
+        if not klass:
+            raise ValueError(f"Unsupported type of platform for signing: {platform}")
+        return klass  # type: ignore[return-value]
+
+    @classmethod
+    def create(cls, platform: str) -> Signer:
+        klass = cls.from_platform(platform)
+        return klass()  # type: ignore[no-any-return, operator]