Merge pull request #61 from gardenlinux/enable-kms

enable kms

Author: Malte Münch

.github/workflows/pytests.yml

@@ -34,5 +34,5 @@ jobs:
         run: |
           export GLOCI_REGISTRY_USERNAME="gardenlinux"
           export GLOCI_REGISTRY_TOKEN="invalid"
-          poetry run pytest
+          poetry run pytest -k "not kms"
 

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python_gardenlinux_lib"
-version = "0.2.0"
+version = "0.3.0"
 description = "Contains tools to work with the features directory of gardenlinux, for example deducting dependencies from feature sets or validating cnames"
 authors = ["Garden Linux Maintainers <[email protected]>"]
 license = "Apache-2.0"
@@ -19,6 +19,7 @@ jsonschema = "^4.23.0"
 oras = { git  = "https://github.com/oras-project/oras-py.git", rev="caf8db5b279382335fbb1f6d7402ed9b73618d37" }
 python-dotenv = "^1.0.1"
 cryptography = "^43.0.0"
+boto3 = "1.35.30"
 
 
 [tool.poetry.group.dev.dependencies]

pyproject.toml

@@ -1,4 +1,9 @@
 import hashlib
+import logging
+import os
+
+import boto3
+from babel.messages import Message
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.serialization import load_pem_private_key
@@ -8,32 +13,84 @@
 import base64
 
 
-def sign_data(data_str: str, private_key_file_path: str) -> str:
-    with open(private_key_file_path, "rb") as key_file:
-        private_key = load_pem_private_key(
-            key_file.read(), password=None, backend=default_backend()
-        )
+class Signer:
+    """
+    Signer
+
+    abstract class defining the methods for signing and verifying data
+    """
+
+    def sign_data(self, data_str: str) -> str:
+        pass
+
+    def verify_signature(self, data_str: str, signature: str):
+        pass
+
+
+class KMSSigner(Signer):
+    """
+    KMSSigner
+
+    implementation of Signer interface using AWS KMS
+    """
+
+    def __init__(self, arn: str):
+        self.kms_client = boto3.client("kms")
+        # check if client works
+        key_info = self.kms_client.describe_key(KeyId=arn)
 
-    signature = private_key.sign(
-        data_str.encode("utf-8"),
-        padding.PSS(
-            mgf=padding.MGF1(
-                hashes.SHA256()
-            ),  # Mask generation function based on SHA-256
-            salt_length=padding.PSS.MAX_LENGTH,  # Maximum salt length
-        ),
-        hashes.SHA256(),
-    )
-    return base64.b64encode(signature).decode("utf-8")
-
-
-def verify_signature(data_str: str, signature: str, public_key_file_path: str):
-    with open(public_key_file_path, "rb") as cert_file:
-        cert = x509.load_pem_x509_certificate(cert_file.read(), default_backend())
-        public_key = cert.public_key()
-    try:
-        public_key.verify(
-            base64.b64decode(signature),
+        if "SIGN" not in key_info["KeyMetadata"]["KeyUsage"]:
+            raise Exception("Key is missing the sign property")
+        if "VERIFY" not in key_info["KeyMetadata"]["KeyUsage"]:
+            raise Exception("Key is missing the verify property")
+        if "RSASSA_PSS_SHA_256" not in key_info["KeyMetadata"]["SigningAlgorithms"]:
+            raise Exception(
+                "Key is missing the required signing algorithm (RSASSA_PSS_SHA_256)"
+            )
+        # seems to be fine, use it
+        self.arn = arn
+
+    def sign_data(self, data_str: str) -> str:
+        signature = self.kms_client.sign(
+            KeyId=self.arn,
+            MessageType="RAW",
+            SigningAlgorithm="RSASSA_PSS_SHA_256",
+            Message=data_str.encode("utf-8"),
+        )["Signature"]
+        return base64.b64encode(signature).decode("utf-8")
+
+    def verify_signature(self, data_str: str, signature: str):
+        if not self.kms_client.verify(
+            KeyId=self.arn,
+            Message=data_str.encode("utf-8"),
+            MessageType="RAW",
+            Signature=base64.b64decode(signature),
+            SigningAlgorithm="RSASSA_PSS_SHA_256",
+        )["SignatureValid"]:
+            raise ValueError(f"Invalid Signature {signature} for data: {data_str}")
+
+
+class LocalSigner(Signer):
+    """
+    LocalSigner
+
+    implementation of Signer interface using local certificates
+
+    :param private_key_file_path
+    :param public_key_file_path
+    """
+
+    def __init__(self, private_key_file_path: str, public_key_file_path: str):
+        with open(private_key_file_path, "rb") as key_file:
+            self.private_key = load_pem_private_key(
+                key_file.read(), password=None, backend=default_backend()
+            )
+        with open(public_key_file_path, "rb") as cert_file:
+            cert = x509.load_pem_x509_certificate(cert_file.read(), default_backend())
+            self.public_key = cert.public_key()
+
+    def sign_data(self, data_str: str) -> str:
+        signature = self.private_key.sign(
             data_str.encode("utf-8"),
             padding.PSS(
                 mgf=padding.MGF1(
@@ -43,8 +100,23 @@ def verify_signature(data_str: str, signature: str, public_key_file_path: str):
             ),
             hashes.SHA256(),
         )
-    except InvalidSignature:
-        raise ValueError(f"Invalid Signature {signature} for data: {data_str}")
+        return base64.b64encode(signature).decode("utf-8")
+
+    def verify_signature(self, data_str: str, signature: str):
+        try:
+            self.public_key.verify(
+                base64.b64decode(signature),
+                data_str.encode("utf-8"),
+                padding.PSS(
+                    mgf=padding.MGF1(
+                        hashes.SHA256()
+                    ),  # Mask generation function based on SHA-256
+                    salt_length=padding.PSS.MAX_LENGTH,  # Maximum salt length
+                ),
+                hashes.SHA256(),
+            )
+        except InvalidSignature:
+            raise ValueError(f"Invalid Signature {signature} for data: {data_str}")
 
 
 def verify_sha256(checksum: str, data: bytes):