Refactor multisig support

Author: kislyuk

README.rst

@@ -5,6 +5,12 @@ http-message-signatures: An implementation of RFC 9421, the IETF HTTP Message Si
 `RFC 9421 HTTP Message Signatures <https://datatracker.ietf.org/doc/rfc9421/>`_ standard in
 Python.
 
+.. admonition:: Security considerations
+
+ It is recommended that you read and understand
+ `section 7 of the RFC, Security Considerations <https://www.rfc-editor.org/rfc/rfc9421#name-security-considerations>`_
+ before using this library.
+
 Installation
 ------------
 ::

http_message_signatures/signatures.py

@@ -167,27 +167,29 @@ def validate_created_and_expires(self, sig_input, max_age=None):
             if self._parse_integer_timestamp(sig_input.params["created"], field_name="created") + max_age < min_time:
                 raise InvalidSignature(f"Signature age exceeds maximum allowable age {max_age}")
 
-    def _get_sig_input_and_signature(self, message, expect_label):
+    def _get_sig_inputs_and_signatures(self, message, *, expect_tag=None, expect_label=None):
         sig_inputs = self._parse_dict_header("Signature-Input", message.headers)
         signatures = self._parse_dict_header("Signature", message.headers)
         if len(sig_inputs) == 0 or len(signatures) == 0:
             raise InvalidSignature("No signatures found in the message")
-        if (len(sig_inputs) > 1 or len(signatures) > 1) and expect_label is None:
-            raise InvalidSignature("Multiple signatures found and no label specified")
+        if (len(sig_inputs) > 1 or len(signatures) > 1) and expect_tag is None and expect_label is None:
+            raise InvalidSignature("Multiple signatures found and no tag or label specified")
         if sig_inputs.keys() != signatures.keys():
             raise InvalidSignature("Signature-Input and Signature headers have different labels")
+        n_sigs = 0
         for label, sig_input in sig_inputs.items():
-            if expect_label is not None and label != expect_label:
-                continue
             if label not in signatures:
                 raise InvalidSignature(f'Signature missing expected label "{label}"')
-            return label, sig_input, signatures[label]
-        raise InvalidSignature(f'Signature-Input does not contain expected label "{expect_label}"')
+            if expect_tag is not None and sig_input.params.get("tag") != expect_tag:
+                continue
+            if expect_label is not None and label != expect_label:
+                continue
+            yield label, sig_input, signatures[label]
+            n_sigs += 1
+        if n_sigs == 0:
+            raise InvalidSignature("No signatures found matching the expected tag or label")
 
-    def verify(
-        self, message, *, max_age: datetime.timedelta = datetime.timedelta(days=1), expect_label: str | None = None
-    ) -> List[VerifyResult]:
-        label, sig_input, signature = self._get_sig_input_and_signature(message, expect_label)
+    def _verify_one(self, *, label, sig_input, signature, message, max_age):
         self.validate_created_and_expires(sig_input, max_age=max_age)
         if "alg" in sig_input.params:
             if sig_input.params["alg"] != self.signature_algorithm.algorithm_id:
@@ -208,12 +210,28 @@ def verify(
             verifier.verify(signature=raw_signature, message=sig_base.encode())
         except Exception as e:
             raise InvalidSignature(e) from e
-        return [
-            VerifyResult(
-                label=label,
-                algorithm=self.signature_algorithm,
-                covered_components=sig_elements,
-                parameters=dict(sig_params_node.params),
-                body=None,
+        return VerifyResult(
+            label=label,
+            algorithm=self.signature_algorithm,
+            covered_components=sig_elements,
+            parameters=dict(sig_params_node.params),
+            body=None,
+        )
+
+    def verify(
+        self,
+        message,
+        *,
+        max_age: datetime.timedelta = datetime.timedelta(days=1),
+        expect_tag: str | None = None,
+        expect_label: str | None = None,
+    ) -> List[VerifyResult]:
+        verify_results = []
+        for label, sig_input, signature in self._get_sig_inputs_and_signatures(
+            message, expect_tag=expect_tag, expect_label=expect_label
+        ):
+            verify_result = self._verify_one(
+                label=label, sig_input=sig_input, signature=signature, message=message, max_age=max_age
             )
-        ]
+            verify_results.append(verify_result)
+        return verify_results

pyproject.toml

@@ -51,3 +51,6 @@ profile = "black"
 
 [tool.ruff]
 line-length = 120
+
+[lint]
+unfixable = ["F401"]

test/test.py

@@ -7,7 +7,9 @@
 import os
 import sys
 import unittest
+from contextlib import contextmanager
 from datetime import datetime, timedelta
+from unittest.mock import patch
 
 import requests
 from cryptography.hazmat.primitives.serialization import (
@@ -36,6 +38,17 @@
 )
 
 
+@contextmanager
+def patch_time(dt):
+    class MockDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return dt
+
+    with patch("http_message_signatures.signatures.datetime.datetime", MockDateTime):
+        yield
+
+
 class MyHTTPSignatureKeyResolver(HTTPSignatureKeyResolver):
     known_pem_keys = {"test-key-rsa", "test-key-rsa-pss", "test-key-ecc-p256", "test-key-ed25519"}
 
@@ -77,20 +90,20 @@ def setUp(self):
         self.key_resolver = MyHTTPSignatureKeyResolver()
         self.max_age = timedelta(weeks=90000)
 
-    def verify(self, verifier, message, max_age=None, expect_label=None):
+    def verify(self, verifier, message, max_age=None, expect_tag=None, expect_label=None):
         if max_age is None:
             max_age = self.max_age
         m = copy.deepcopy(message)
         m.headers["Signature"] = m.headers["Signature"][:8] + m.headers["Signature"][8:].upper()
         with self.assertRaises(InvalidSignature):
-            verifier.verify(m, max_age=max_age, expect_label=expect_label)
+            verifier.verify(m, max_age=max_age, expect_tag=expect_tag, expect_label=expect_label)
         m.headers["Signature"] = m.headers["Signature"].upper()
         with self.assertRaisesRegex(InvalidSignature, "Malformed structured header field"):
-            verifier.verify(m, max_age=max_age, expect_label=expect_label)
+            verifier.verify(m, max_age=max_age, expect_tag=expect_tag, expect_label=expect_label)
         del m.headers["Signature"]
         with self.assertRaisesRegex(InvalidSignature, 'Expected "Signature" header field to be present'):
-            verifier.verify(m, max_age=max_age, expect_label=expect_label)
-        return verifier.verify(message, max_age=max_age, expect_label=expect_label)
+            verifier.verify(m, max_age=max_age, expect_tag=expect_tag, expect_label=expect_label)
+        return verifier.verify(message, max_age=max_age, expect_tag=expect_tag, expect_label=expect_label)
 
     def test_http_message_signatures_B21(self):
         signer = HTTPMessageSigner(signature_algorithm=RSA_PSS_SHA512, key_resolver=self.key_resolver)
@@ -345,6 +358,7 @@ def test_multiple_signatures(self):
             created=datetime.fromtimestamp(1618884480),
             expires=datetime.fromtimestamp(1618884540),
             label="proxy_sig",
+            tag=None,
             append_if_signature_exists=False,
         )
         signer2 = HTTPMessageSigner(signature_algorithm=RSA_V1_5_SHA256, key_resolver=self.key_resolver)
@@ -363,13 +377,26 @@ def test_multiple_signatures(self):
             "sig1=:X5spyd6CFnAG5QnDyHfqoSNICd+BUP4LYMz2Q0JXlb//4Ijpzp+kve2w4NIyqeAuM7jTDX+sNalzA8ESSaHD3A==:, proxy_sig=:S6ZzPXSdAMOPjN/6KXfXWNO/f7V6cHm7BXYUh3YD/fRad4BCaRZxP+JH+8XY1I6+8Cy+CM5g92iHgxtRPz+MjniOaYmdkDcnL9cCpXJleXsOckpURl49GwiyUpZ10KHgOEe11sx3G2gxI8S0jnxQB+Pu68U9vVcasqOWAEObtNKKZd8tSFu7LB5YAv0RAGhB8tmpv7sFnIm9y+7X5kXQfi8NMaZaA8i2ZHwpBdg7a6CMfwnnrtflzvZdXAsD3LH2TwevU+/PBPv0B6NMNk93wUs/vfJvye+YuI87HU38lZHowtznbLVdp770I6VHR6WfgS9ddzirrswsE1w5o0LV/g==:",
         )
         self.assertIn("sig1", self.test_request.headers["Signature-Input"])
-        with self.assertRaisesRegex(InvalidSignature, "Multiple signatures found and no label specified"):
+        with self.assertRaisesRegex(InvalidSignature, "Multiple signatures found and no tag or label specified"):
             verifier.verify(self.test_request)
         verifier2 = HTTPMessageVerifier(signature_algorithm=RSA_V1_5_SHA256, key_resolver=self.key_resolver)
         with self.assertRaisesRegex(InvalidSignature, 'Signature "expires" parameter is set to a time in the past'):
             self.verify(verifier2, self.test_request, expect_label="proxy_sig")
-        verifier2.validate_created_and_expires = lambda *args, **kwargs: None
-        self.verify(verifier2, self.test_request, expect_label="proxy_sig")
+
+        with patch_time(datetime.fromtimestamp(1618884500)):
+            res = self.verify(verifier2, self.test_request, expect_label="proxy_sig")
+            self.assertEqual(len(res), 1)
+            self.assertEqual(res[0].label, "proxy_sig")
+
+        signer2_args.update(label="my-label", tag="my-tag")
+        signer2.sign(self.test_request, **signer2_args)
+        with self.assertRaisesRegex(InvalidSignature, "No signatures found matching the expected tag or label"):
+            self.verify(verifier2, self.test_request, expect_tag="test")
+        with patch_time(datetime.fromtimestamp(1618884500)):
+            res = self.verify(verifier2, self.test_request, expect_tag="my-tag")
+            self.assertEqual(len(res), 1)
+            self.assertEqual(res[0].label, "my-label")
+            self.assertEqual(res[0].parameters["tag"], "my-tag")
 
     def test_query_parameters(self):
         signer = HTTPMessageSigner(signature_algorithm=HMAC_SHA256, key_resolver=self.key_resolver)