Added optional signing_key_name to msg signer messages

Signed-off-by: Jindrich Luza <jluza@redhat.com>

Author: midnightercz

src/pubtools/sign/operations/base.py

@@ -2,7 +2,7 @@
 
 from abc import ABC, abstractmethod
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import ClassVar, Dict, Any, Type
 from typing_extensions import Self
 
@@ -15,6 +15,18 @@ class SignOperation(ABC):
 
     ResultType: ClassVar[OperationResult]
     signing_key: str
+    signing_key_name: str = field(
+        default="",
+        metadata={
+            "description": "Signing key name which should be used for signing",
+            "sample": "key1",
+        },
+    )
+
+    def __post_init__(self) -> None:
+        """Post init method."""
+        if not hasattr(self, "signing_key_name"):
+            self.signing_key_name = ""
 
     @classmethod
     def doc_arguments(cls: Type[Self]) -> Dict[str, Any]:

src/pubtools/sign/operations/clearsign.py

@@ -20,31 +20,35 @@ class ClearSignOperation(SignOperation):
             "description": "Signing data",
             "required": "true",
             "sample": ["input1", "input2"],
-        }
+        },
+        default_factory=list,
     )
     signing_key: str = field(
         metadata={
             "type": "str",
             "description": "Signing key short id which should be used for signing",
             "required": "true",
             "sample": "123",
-        }
+        },
+        default="",
     )
     task_id: str = field(
         metadata={
             "type": "str",
             "description": "Usually pub task id, serves as identifier for in signing request",
             "required": "true",
             "sample": "1",
-        }
+        },
+        default="",
     )
     repo: str = field(
         metadata={
             "type": "str",
             "description": "Repository name",
             "required": "true",
             "sample": "repo",
-        }
+        },
+        default="",
     )
     requester: str = field(
         metadata={"description": "Requester of the signing operation"}, default=""

src/pubtools/sign/operations/containersign.py

@@ -14,13 +14,21 @@ class ContainerSignOperation(SignOperation):
     """ContainersSignOperation model class."""
 
     ResultType: ClassVar[OperationResult]
-    digests: List[str] = field(metadata={"description": "List of digest to sign"})
-    references: List[str] = field(metadata={"description": "List of references to sign"})
+    digests: List[str] = field(
+        metadata={"description": "List of digest to sign"}, default_factory=list
+    )
+    references: List[str] = field(
+        metadata={"description": "List of references to sign"}, default_factory=list
+    )
     signing_key: str = field(
-        metadata={"description": "Signing key short id which should be used for signing"}
+        metadata={"description": "Signing key short id which should be used for signing"},
+        default="",
     )
     task_id: str = field(
-        metadata={"description": "Usually pub task id, serves as identifier for in signing request"}
+        metadata={
+            "description": "Usually pub task id, serves as identifier for in signing request"
+        },
+        default="",
     )
     identity_references: List[str] = field(
         metadata={"description": "List of references to sign"}, default_factory=list
@@ -34,6 +42,7 @@ def to_dict(self) -> dict[str, Any]:
         return dict(
             digests=self.digests,
             references=self.references,
+            signing_key_name=self.signing_key_name,
             signing_key=self.signing_key,
             task_id=self.task_id,
             requester=self.requester,

src/pubtools/sign/signers/msgsigner.py

@@ -36,6 +36,7 @@
 
 
 LOG = logging.getLogger("pubtools.sign.signers.msgsigner")
+LOG.setLevel(logging.INFO)
 
 
 class SignRequestType(str, enum.Enum):
@@ -169,6 +170,7 @@ def _construct_signing_message(
         claim: str,
         signing_key: str,
         repo: str,
+        signing_key_name: str = "",
         extra_attrs: Optional[Dict[str, Any]] = None,
         sig_type: str = SignRequestType.CONTAINER,
     ) -> dict[str, Any]:
@@ -182,6 +184,8 @@ def _construct_signing_message(
             "requested_by": self.creator,
             "repo": repo,
         }
+        if signing_key_name:
+            message["sig_keyname"] = signing_key_name
         message.update(_extra_attrs)
         return message
 
@@ -217,6 +221,7 @@ def _create_msg_message(
                 data,
                 signing_key,
                 repo,
+                signing_key_name=operation.signing_key_name,
                 extra_attrs=extra_attrs,
                 sig_type=sig_type.value,
             ),
@@ -625,6 +630,7 @@ def msg_clear_sign(
 
 def msg_container_sign(
     signing_key: str = "",
+    signing_key_name: str = "",
     task_id: str = "",
     config_file: str = "",
     digest: list[str] = [],
@@ -642,6 +648,7 @@ def msg_container_sign(
         digests=digest,
         references=reference,
         signing_key=signing_key,
+        signing_key_name=signing_key_name,
         task_id=task_id,
         requester=requester,
     )
@@ -727,6 +734,11 @@ def msg_clear_sign_main(
     required=True,
     help="8 characters key fingerprint of key which should be used for signing",
 )
+@click.option(
+    "--signing-key-name",
+    required=False,
+    help="signing key name",
+)
 @click.option("--task-id", required=True, help="Task id identifier (usually pub task-id)")
 @click.option("--config-file", default=CONFIG_PATHS[0], help="path to the config file")
 @click.option(
@@ -759,6 +771,7 @@ def msg_clear_sign_main(
 )
 def msg_container_sign_main(
     signing_key: str = "",
+    signing_key_name: str = "",
     task_id: str = "",
     config_file: str = "",
     digest: List[str] = [],
@@ -775,6 +788,7 @@ def msg_container_sign_main(
 
     ret = msg_container_sign(
         signing_key=signing_key,
+        signing_key_name=signing_key_name,
         task_id=task_id,
         config_file=config_file,
         digest=digest,

tests/test_msg_signer.py

@@ -52,6 +52,33 @@ def test_msg_container_sign(f_msg_signer, f_config_msg_signer_ok):
     print(result.stdout)
     assert result.exit_code == 0, result.output
 
+def test_msg_container_sign_keyname(f_msg_signer, f_config_msg_signer_ok):
+    f_msg_signer.return_value.sign.return_value.signer_results.to_dict.return_value = {
+        "status": "ok"
+    }
+    f_msg_signer.return_value.sign.return_value.operation_result.results = []
+    f_msg_signer.return_value.sign.return_value.operation_result.signing_key = ""
+    f_msg_signer.return_value.sign.return_value.operation.to_dict.return_value = {}
+    result = CliRunner().invoke(
+        msg_container_sign_main,
+        [
+            "--signing-key",
+            "test-signing-key",
+            "--signing-key-name",
+            "test-signing-key-name",
+            "--digest",
+            "some-digest",
+            "--reference",
+            "some-reference",
+            "--task-id",
+            "1",
+            "--config-file",
+            f_config_msg_signer_ok,
+        ],
+    )
+    print(result.stdout)
+    assert result.exit_code == 0, result.output
+
 
 def test_msg_container_sign_requester(f_msg_signer, f_config_msg_signer_ok):
     f_msg_signer.return_value.sign.return_value.signer_results.to_dict.return_value = {
@@ -408,9 +435,31 @@ def test__construct_signing_message(f_config_msg_signer_ok):
     with patch("uuid.uuid4", return_value="1234-5678-abcd-efgh"):
         with patch("pubtools.sign.signers.msgsigner.isodate_now") as patched_date:
             patched_date.return_value = "created-date-Z"
-            ret = signer._construct_signing_message("some-claim", "some-signing-key", "repo", {})
+            ret = signer._construct_signing_message(
+                "some-claim", "some-signing-key", "repo", extra_attrs={}, signing_key_name=""
+            )
+            assert ret == {
+                "sig_key_id": "ning-key",
+                "claim_file": "some-claim",
+                "request_id": "1234-5678-abcd-efgh",
+                "created": "created-date-Z",
+                "requested_by": "pubtools-sign-test",
+                "repo": "repo",
+            }
+
+def test__construct_signing_message_signing_key_name(f_config_msg_signer_ok):
+    signer = MsgSigner()
+    signer.load_config(load_config(f_config_msg_signer_ok))
+    with patch("uuid.uuid4", return_value="1234-5678-abcd-efgh"):
+        with patch("pubtools.sign.signers.msgsigner.isodate_now") as patched_date:
+            patched_date.return_value = "created-date-Z"
+            ret = signer._construct_signing_message(
+                "some-claim", "some-signing-key", "repo", extra_attrs={},
+                signing_key_name="test-signing-key-name"
+            )
             assert ret == {
                 "sig_key_id": "ning-key",
+                "sigkey_name": "test-signing-key-name",
                 "claim_file": "some-claim",
                 "request_id": "1234-5678-abcd-efgh",
                 "created": "created-date-Z",
@@ -595,6 +644,7 @@ def test_clear_sign_aliases(patched_uuid, f_config_msg_signer_aliases):
                     "repo",
                     extra_attrs={"pub_task_id": "1"},
                     sig_type="clearsign_signature",
+                    signing_key_name="",
                 )
 
             assert res == SigningResults(
@@ -755,6 +805,7 @@ def test_container_sign_alias(patched_uuid, f_config_msg_signer_aliases, f_clien
                     "abcde1245",
                     "namespace/repo",
                     extra_attrs={"pub_task_id": "1", "manifest_digest": "sha256:abcdefg"},
+                    signing_key_name="",
                     sig_type="container_signature",
                 )
 

tests/test_sign_operations.py

@@ -11,6 +11,9 @@ def test_containersign_operation_doc_argument():
             "references": {"description": "List of references to sign"},
             "requester": {"description": "Requester of the signing operation"},
             "signing_key": {"description": "Signing key short id which should be used for signing"},
+            "signing_key_name": {
+                "description": "Signing key name which should be used for signing"
+            },
             "task_id": {
                 "description": "Usually pub task id, serves as identifier for in signing request"
             },
@@ -20,6 +23,7 @@ def test_containersign_operation_doc_argument():
             "digests": "",
             "references": "",
             "signing_key": "",
+            "signing_key_name": "key1",
             "task_id": "",
             "requester": "",
             "identity_references": "",
@@ -36,6 +40,9 @@ def test_clearsign_operation_doc_argument():
                 "description": "Signing key short id which should be used for signing",
                 "required": "true",
             },
+            "signing_key_name": {
+                "description": "Signing key name which should be used for signing"
+            },
             "task_id": {
                 "type": "str",
                 "description": "Usually pub task id, serves as identifier for in signing request",
@@ -51,6 +58,7 @@ def test_clearsign_operation_doc_argument():
         "examples": {
             "inputs": ["input1", "input2"],
             "signing_key": "123",
+            "signing_key_name": "key1",
             "task_id": "1",
             "requester": "",
             "repo": "repo",
@@ -63,12 +71,14 @@ def test_container_sign_to_dict():
         digests=["digest"],
         references=["references"],
         signing_key="sig-key",
+        signing_key_name="sig-key-name",
         task_id="task-id",
         requester="requester",
     ).to_dict() == dict(
         digests=["digest"],
         references=["references"],
         signing_key="sig-key",
+        signing_key_name="sig-key-name",
         task_id="task-id",
         requester="requester",
     )