Added batch mgs signer

New sub type of msg signer allows sending signing requests in batches
in on messaging message

Author: midnightercz

mkdocs/docs/user-guide/user-guide.md

@@ -8,11 +8,5 @@ Before you start reading further:
 Bellow you find details about supported signatures and signers. How they work and how you can validate
 signatures produces by them.
 
-## Clearsign
-
-### Messaging signer
-
-Messaging signer works as client which communicates with the server via messaging bus. User data
-to be signed are wrapped into signing requests and sent to the server. The server replies with 
-signed requests which are composed from the original signing request and the signature. Signature
-is base64 encoded clearsign of user data.
+* [[clear-sign|Clearsign]]
+* [[container-signing|Containers Signing]]

mkdocs/mkdocs.yml

@@ -11,6 +11,7 @@ nav:
       - 'user-guide/config.md'
       - 'user-guide/cli-commands.md'
       - 'user-guide/container-signing.md'
+      - 'user-guide/clear-sign.md'
       - 'user-guide/signers/cosign-signer.md'
       - 'user-guide/signers/msg-signer.md'
   - 'developer/developer-guide.md'

requirements.txt

@@ -1,5 +1,20 @@
-requests
-typing-extensions
-mock
-
-
+#
+# This file is autogenerated by pip-compile with Python 3.12
+# by the following command:
+#
+#    pip-compile
+#
+certifi==2025.6.15
+    # via requests
+charset-normalizer==3.4.2
+    # via requests
+idna==3.10
+    # via requests
+mock==5.2.0
+    # via -r requirements.in
+requests==2.32.4
+    # via -r requirements.in
+typing-extensions==4.14.1
+    # via -r requirements.in
+urllib3==2.5.0
+    # via requests

src/pubtools/sign/conf/conf.py

@@ -7,7 +7,7 @@
 
 
 class MsgSignerSchema(ma.Schema):
-    """Radas signer configuration schema."""
+    """Msg signer configuration schema."""
 
     messaging_brokers = ma.fields.List(ma.fields.String(), required=True)
     messaging_cert_key = ma.fields.String(required=True)
@@ -24,6 +24,12 @@ class MsgSignerSchema(ma.Schema):
     key_aliases = ma.fields.Dict(required=False, keys=ma.fields.String(), values=ma.fields.String())
 
 
+class MsgBatchSignerSchema(MsgSignerSchema):
+    """Msg batch signer configuration schema."""
+
+    chunk_size = ma.fields.Integer(required=False)
+
+
 class CosignSignerSchema(ma.Schema):
     """Cosign signer configuration schema."""
 
@@ -46,6 +52,7 @@ class ConfigSchema(ma.Schema):
     """pubtools-sign configuration schema."""
 
     msg_signer = ma.fields.Nested(MsgSignerSchema)
+    msg_batch_signer = ma.fields.Nested(MsgBatchSignerSchema)
     cosign_signer = ma.fields.Nested(CosignSignerSchema)
 
 

src/pubtools/sign/signers/msgsigner.py

@@ -423,7 +423,7 @@ def clear_sign(self: MsgSigner, operation: ClearSignOperation) -> SigningResults
         return signing_results
 
     @staticmethod
-    def create_manifest_claim_message(signature_key: str, digest: str, reference: str) -> str:
+    def create_manifest_claim_message(digest: str, reference: str) -> str:
         """Create manifest claim for container signing.
 
         See below for the specification for the manifest claim that is created here
@@ -447,6 +447,26 @@ def create_manifest_claim_message(signature_key: str, digest: str, reference: st
         }
         return base64.b64encode(json.dumps(manifest_claim).encode("latin1")).decode("latin1")
 
+    def _prepare_messages(self, operation: ContainerSignOperation) -> List[MsgMessage]:
+        fargs = []
+        for digest, reference in zip(operation.digests, operation.references):
+            repo = reference.split("/", 1)[1].split(":")[0]
+            fargs.append(
+                FData(
+                    args=[
+                        self.create_manifest_claim_message(digest=digest, reference=reference),
+                        repo,
+                        operation,
+                        SignRequestType.CONTAINER,
+                    ],
+                    kwargs={
+                        "extra_attrs": {"pub_task_id": operation.task_id, "manifest_digest": digest}
+                    },
+                )
+            )
+        ret = run_in_parallel(self._create_msg_message, fargs)
+        return list(ret.values())
+
     def container_sign(self: MsgSigner, operation: ContainerSignOperation) -> SigningResults:
         """Run container signing operation.
 
@@ -469,26 +489,9 @@ def container_sign(self: MsgSigner, operation: ContainerSignOperation) -> Signin
 
         LOG.info(f"Container sign operation for {len(operation.digests)}")
 
-        fargs = []
-        for digest, reference in zip(operation.digests, operation.references):
-            repo = reference.split("/", 1)[1].split(":")[0]
-            fargs.append(
-                FData(
-                    args=[
-                        self.create_manifest_claim_message(
-                            signing_key, digest=digest, reference=reference
-                        ),
-                        repo,
-                        operation,
-                        SignRequestType.CONTAINER,
-                    ],
-                    kwargs={
-                        "extra_attrs": {"pub_task_id": operation.task_id, "manifest_digest": digest}
-                    },
-                )
-            )
-        ret = run_in_parallel(self._create_msg_message, fargs)
-        for n, message in ret.items():
+        ret = self._prepare_messages(operation)
+
+        for message in ret:
             message_to_data[message.body["request_id"]] = message
             messages.append(message)
 
@@ -605,16 +608,179 @@ def container_sign(self: MsgSigner, operation: ContainerSignOperation) -> Signin
                 signer_results.error_message += f"{error.name} : {error.description}\n"
             return signing_results
 
+        print("ALL MESSAGES", len(all_messages))
+        print("ALL MESSAGES", all_messages)
         operation_result = ContainerSignResult(
             signing_key=operation.signing_key, results=[""] * len(all_messages), failed=False
         )
         for recv_id, _received in recvc.recv.items():
+            print("RECV_ID", recv_id)
+            print("RECEIVED", _received)
             operation_result.failed = True if _received[0]["msg"]["errors"] else False
             operation_result.results[all_messages.index(message_to_data[recv_id])] = _received
         signing_results.operation_result = operation_result
         return signing_results
 
 
+class MsgBatchSigner(MsgSigner):
+    """Messaging batch signer class."""
+
+    _signer_config_key: str = "msg_batch_signer"
+
+    chunk_size: int = field(
+        init=False,
+        metadata={
+            "description": "Identify how many signing claims should be send in one message",
+            "sample": 10,
+        },
+    )
+
+    SUPPORTED_OPERATIONS: ClassVar[List[Type[SignOperation]]] = [
+        ContainerSignOperation,
+    ]
+
+    def _construct_signing_batch_message(
+        self: Self,
+        claims: List[str],
+        signing_keys: List[str],
+        repo: str,
+        signing_key_names: List[str] = [],
+        extra_attrs: Optional[Dict[str, Any]] = None,
+        sig_type: str = SignRequestType.CONTAINER,
+    ) -> dict[str, Any]:
+        data_attr = "claims" if sig_type == SignRequestType.CONTAINER else "data"
+        _extra_attrs = extra_attrs or {}
+        processed_claims = [
+            {
+                "claim_file": claim,
+                "sig_keyname": signing_key_names,
+                "sig_key_id": signing_keys,
+                "manifest_digest": digest,
+            }
+            for claim, digest in zip(claims, _extra_attrs.get("manifest_digest", ""))
+        ]
+        message = {
+            data_attr: processed_claims,
+            "request_id": str(uuid.uuid4()),
+            "created": isodate_now(),
+            "requested_by": self.creator,
+            "repo": repo,
+        }
+        _extra_attrs.pop("manifest_digest", None)
+        message.update(_extra_attrs)
+        return message
+
+    def _create_msg_batch_message(
+        self: Self,
+        data: List[str],
+        repo: str,
+        operation: SignOperation,
+        sig_type: SignRequestType,
+        extra_attrs: Optional[Dict[str, Any]] = None,
+    ) -> MsgMessage:
+        if operation.signing_key in self.key_aliases:
+            signing_key = self.key_aliases[operation.signing_key]
+            LOG.info(f"Using signing key alias {signing_key} for {operation.signing_key}")
+        else:
+            signing_key = operation.signing_key
+
+        extra_attrs = extra_attrs or {}
+        headers = self._construct_headers(sig_type, extra_attrs=extra_attrs)
+        if isinstance(operation, ContainerSignOperation):
+            extra_attrs["manifest_digest"] = operation.digests
+        ret = MsgMessage(
+            headers=headers,
+            body=self._construct_signing_batch_message(
+                data,
+                [signing_key],
+                repo,
+                signing_key_names=[operation.signing_key_name],
+                extra_attrs=extra_attrs,
+                sig_type=sig_type.value,
+            ),
+            address=self.topic_send_to.format(
+                **dict(list(asdict(self).items()) + list(asdict(operation).items()))
+            ),
+        )
+        LOG.debug(f"Construted message with request_id {ret.body['request_id']}")
+        return ret
+
+    def _prepare_messages(self: Self, operation: ContainerSignOperation) -> List[MsgMessage]:
+        messages: List[MsgMessage] = []
+        repo_groups: Dict[str, Dict[str, List[str]]] = {}
+        for digest, reference in zip(operation.digests, operation.references):
+            repo = reference.split("/", 1)[1].split(":")[0]
+            if repo not in repo_groups:
+                repo_groups[repo] = cast(dict[str, list[str]], {"digests": [], "references": []})
+            repo_groups[repo]["digests"].append(digest)
+            repo_groups[repo]["references"].append(reference)
+
+        batch_data: List[FData] = []
+        for repo, group in repo_groups.items():
+            claims = []
+            digests = []
+
+            for digest, reference in zip(group["digests"], group["references"]):
+                claims.append(
+                    self.create_manifest_claim_message(digest=digest, reference=reference)
+                )
+                digests.append(digest)
+                if len(claims) >= self.chunk_size:
+                    fdata = FData(
+                        args=[claims, repo, operation, SignRequestType.CONTAINER],
+                        kwargs={
+                            "extra_attrs": {
+                                "pub_task_id": operation.task_id,
+                                "manifest_digest": digests,
+                            }
+                        },
+                    )
+                    batch_data.append(fdata)
+                    claims = []
+                    digests = []
+            if claims:
+                fdata = FData(
+                    args=[claims, repo, operation, SignRequestType.CONTAINER],
+                    kwargs={
+                        "extra_attrs": {
+                            "pub_task_id": operation.task_id,
+                            "manifest_digest": digests,
+                        }
+                    },
+                )
+                batch_data.append(fdata)
+
+            ret = run_in_parallel(self._create_msg_batch_message, batch_data)
+            messages.extend(ret.values())
+        return messages
+
+    def load_config(self: Self, config_data: Dict[str, Any]) -> None:
+        """Load configuration of messaging signer.
+
+        Arguments:
+            config_data (dict): configuration data to load
+        """
+        self.messaging_brokers = config_data["msg_batch_signer"]["messaging_brokers"]
+        self.messaging_cert_key = os.path.expanduser(
+            config_data["msg_batch_signer"]["messaging_cert_key"]
+        )
+        self.messaging_ca_cert = os.path.expanduser(
+            config_data["msg_batch_signer"]["messaging_ca_cert"]
+        )
+        self.topic_send_to = config_data["msg_batch_signer"]["topic_send_to"]
+        self.topic_listen_to = config_data["msg_batch_signer"]["topic_listen_to"]
+        self.environment = config_data["msg_batch_signer"]["environment"]
+        self.service = config_data["msg_batch_signer"]["service"]
+        self.message_id_key = config_data["msg_batch_signer"]["message_id_key"]
+        self.retries = config_data["msg_batch_signer"]["retries"]
+        self.send_retries = config_data["msg_batch_signer"]["send_retries"]
+        self.log_level = config_data["msg_batch_signer"]["log_level"]
+        self.timeout = config_data["msg_batch_signer"]["timeout"]
+        self.creator = self._get_cert_subject_cn()
+        self.key_aliases = config_data["msg_batch_signer"].get("key_aliases", {})
+        self.chunk_size = config_data["msg_batch_signer"]["chunk_size"]
+
+
 def msg_clear_sign(
     inputs: List[str],
     signing_key: str = "",
@@ -669,9 +835,14 @@ def msg_container_sign(
     digest: list[str] = [],
     reference: list[str] = [],
     requester: str = "",
+    signer_type: str = "single",
 ) -> Dict[str, Any]:
     """Run containersign operation with cli arguments."""
-    msg_signer = MsgSigner()
+    if signer_type == "single":
+        msg_signer = MsgSigner()
+    elif signer_type == "batch":
+        msg_signer = MsgBatchSigner()
+
     config = _get_config_file(config_file)
     msg_signer.load_config(load_config(os.path.expanduser(config)))
     if requester:
@@ -814,6 +985,9 @@ def msg_clear_sign_main(
     default="INFO",
     help="Set log level",
 )
+@click.option(
+    "--signer-type", type=click.Choice(["single", "batch"]), default="single", help="Signer type"
+)
 def msg_container_sign_main(
     signing_key: str = "",
     signing_key_name: str = "",
@@ -824,6 +998,7 @@ def msg_container_sign_main(
     requester: str = "",
     raw: bool = False,
     log_level: str = "INFO",
+    signer_type: str = "single",
 ) -> None:
     """Entry point method for containersign operation.
 
@@ -851,6 +1026,7 @@ def msg_container_sign_main(
         digest=digest,
         reference=reference,
         requester=requester,
+        signer_type=signer_type,
     )
     if not raw:
         click.echo(json.dumps(ret))

src/pubtools/sign/utils.py

@@ -107,7 +107,7 @@ class FData:
 
 
 def run_in_parallel(
-    func: Callable[..., Any], data: List[FData], threads: int = 10
+    func: Callable[..., Any], data: Iterable[FData], threads: int = 10
 ) -> Dict[Any, Any]:
     """Run method on data in parallel.
 

tests/conftest.py

@@ -16,6 +16,10 @@
     f_config_msg_signer_ok,  # noqa: F401
     f_config_msg_signer_ok2,  # noqa: F401
     f_config_msg_signer_aliases,  # noqa: F401
+    f_msg_batch_signer,  # noqa: F401
+    f_config_msg_batch_signer_ok,  # noqa: F401
+    f_config_msg_batch_signer_aliases,  # noqa: F401
+    f_config_msg_batch_signer_ok2,  # noqa: F401
 )  # noqa: F401
 from .conftest_cosignsig import (  # noqa: F401
     f_cosign_signer,  # noqa: F401