Merge pull request #66 from gardenlinux/remove-signing

Remove signing

Author: Malte Münch

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python_gardenlinux_lib"
-version = "0.3.0"
+version = "0.4.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,7 +19,6 @@ 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,122 +1,4 @@
 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
-from cryptography.hazmat.backends import default_backend
-from cryptography.exceptions import InvalidSignature
-from cryptography import x509
-import base64
-
-
-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)
-
-        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(
-                    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(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):

src/python_gardenlinux_lib/oras/crypto.py

@@ -31,7 +31,6 @@
 from python_gardenlinux_lib.oras.crypto import (
     calculate_sha256,
     verify_sha256,
-    Signer,
 )
 from python_gardenlinux_lib.oras.defaults import (
     annotation_signature_key,
@@ -136,7 +135,6 @@ class GlociRegistry(Registry):
     def __init__(
         self,
         container_name: str,
-        signer: Signer,
         insecure: bool = False,
         token: Optional[str] = None,
         config_path: Optional[str] = None,
@@ -146,7 +144,6 @@ def __init__(
         self.container_name = container_name
         self.registry_url = self.container.registry
         self.config_path = config_path
-        self.signer = signer
         if not token:
             logger.info("No Token provided.")
         else:
@@ -168,7 +165,6 @@ def get_manifest_json(
         get_manifest = f"{self.prefix}://{container.manifest_url()}"
         response = self.do_request(get_manifest, "GET", headers=headers)
         self._check_200_response(response)
-        self.verify_manifest_signature(response.json())
         return response
 
     @ensure_container
@@ -252,7 +248,6 @@ def get_manifest_meta_data_by_cname(
                 and manifest_meta["annotations"]["architecture"] == arch
                 and manifest_meta["platform"]["os.version"] == version
             ):
-                self.verify_manifest_meta_signature(manifest_meta)
                 return manifest_meta
 
         return None
@@ -278,7 +273,6 @@ def get_manifest_by_digest(
         self._check_200_response(response)
         manifest = response.json()
         verify_sha256(digest, response.content)
-        self.verify_manifest_signature(manifest)
         jsonschema.validate(manifest, schema=oras_manifest_schema)
         return manifest
 
@@ -367,8 +361,6 @@ def attach_layer(
             self.container, cname, version, architecture
         )
 
-        self.verify_manifest_signature(manifest)
-
         layer = self.create_layer(file_path, cname, version, architecture, media_type)
         self._check_200_response(self.upload_blob(file_path, self.container, layer))
 
@@ -384,7 +376,6 @@ def attach_layer(
         new_manifest_metadata["size"] = self.get_manifest_size(manifest_container)
         new_manifest_metadata["platform"] = NewPlatform(architecture, version)
 
-        self.sign_manifest_entry(new_manifest_metadata, version, architecture, cname)
         new_index = self.update_index(old_manifest_digest, new_manifest_metadata)
         self._check_200_response(self.upload_index(new_index))
 
@@ -535,6 +526,7 @@ def push_image_manifest(
         :param str build_artifacts_dir: directory where the build artifacts are located
         :param str feature_set: the expanded list of the included features of this manifest. It will be set in the
         manifest itself and in the index entry for this manifest
+        :returns the digest of the pushed manifest
         """
 
         # TODO: construct oci_artifacts default data
@@ -594,6 +586,8 @@ def push_image_manifest(
             f"{self.container_name}-{cname}-{architecture}"
         )
 
+        local_digest = f"sha256:{hashlib.sha256(json.dumps(manifest_image).encode('utf-8')).hexdigest()}"
+
         self._check_200_response(
             self.upload_manifest(manifest_image, manifest_container)
         )
@@ -603,13 +597,14 @@ def push_image_manifest(
         attach_state(metadata_annotations, "")
         metadata_annotations["feature_set"] = feature_set
         manifest_digest = self.get_digest(manifest_container)
+        if manifest_digest != local_digest:
+            raise ValueError("local and remotely calculated digests do not match")
         manifest_index_metadata = NewManifestMetadata(
             manifest_digest,
             self.get_manifest_size(manifest_container),
             metadata_annotations,
             NewPlatform(architecture, version),
         )
-        self.sign_manifest_entry(manifest_index_metadata, version, architecture, cname)
 
         old_manifest_meta_data = self.get_manifest_meta_data_by_cname(
             self.container, cname, version, architecture
@@ -624,7 +619,7 @@ def push_image_manifest(
         self._check_200_response(self.upload_index(new_index))
 
         print(f"Successfully pushed {self.container}")
-        return response
+        return local_digest
 
     def create_layer(
         self,
@@ -639,9 +634,6 @@ def create_layer(
         layer["annotations"] = {
             oras.defaults.annotation_title: os.path.basename(file_path),
         }
-        self.sign_layer(
-            layer, cname, version, architecture, checksum_sha256, media_type
-        )
         return layer
 
     def push_from_tar(self, architecture: str, version: str, cname: str, tar: str):
@@ -667,7 +659,7 @@ def push_from_tar(self, architecture: str, version: str, cname: str, tar: str):
                             break
                     file.close()
 
-            self.push_image_manifest(
+            digest = self.push_image_manifest(
                 architecture, cname, version, tmpdir, oci_metadata, features
             )
         except Exception as e:
@@ -677,6 +669,7 @@ def push_from_tar(self, architecture: str, version: str, cname: str, tar: str):
             exit(1)
         shutil.rmtree(tmpdir, ignore_errors=True)
         print("removed tmp files.")
+        return digest
 
 
 def extract_tar(tar: str, tmpdir: str):

src/python_gardenlinux_lib/oras/registry.py

@@ -3,7 +3,6 @@
 import pytest
 import os
 
-from python_gardenlinux_lib.oras.crypto import LocalSigner
 from python_gardenlinux_lib.oras.registry import GlociRegistry
 from python_gardenlinux_lib.features import parse_features
 
@@ -34,13 +33,7 @@ def test_push_example(version, cname, arch):
         cname, version, arch, GARDENLINUX_ROOT_DIR_EXAMPLE
     )
     container_name = f"{CONTAINER_NAME_ZOT_EXAMPLE}:{version}"
-    signer = LocalSigner(
-        private_key_file_path="cert/oci-sign.key",
-        public_key_file_path="cert/oci-sign.crt",
-    )
-    a_registry = GlociRegistry(
-        container_name=container_name, insecure=True, signer=signer
-    )
+    a_registry = GlociRegistry(container_name=container_name, insecure=True)
     features = parse_features.get_features(cname, GARDENLINUX_ROOT_DIR_EXAMPLE)
     a_registry.push_image_manifest(
         arch,