Add support mandatory signed components

aggrec/aggregates.py

@@ -65,6 +65,9 @@
     "Signature-Input",
 ]
 
+
+REQUIRED_SIGNED_COMPONENTS = set(["content-length", "content-type", "content-digest"])
+
 router = APIRouter()
 
 
@@ -219,8 +222,15 @@ async def create_aggregate(
 ):
     span = trace.get_current_span()
 
+    required_signed_components = (
+        REQUIRED_SIGNED_COMPONENTS | {"aggregate-interval"} if aggregate_interval else REQUIRED_SIGNED_COMPONENTS
+    )
+
     with tracer.start_as_current_span("http_request_verifier"):
-        http_request_verifier = RequestVerifier(key_resolver=request.app.key_resolver)
+        http_request_verifier = RequestVerifier(
+            key_resolver=request.app.key_resolver,
+            required_signed_components=required_signed_components,
+        )
         res = await http_request_verifier.verify(request)
 
     creator = res.parameters.get("keyid")

aggrec/helpers.py

@@ -1,5 +1,6 @@
 import hashlib
 import logging
+from collections.abc import Iterable
 from datetime import UTC, datetime, timedelta
 from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network, ip_address
 
@@ -51,9 +52,11 @@ def __init__(
         self,
         key_resolver: KeyResolver,
         algorithm: HTTPSignatureAlgorithm | None = None,
+        required_signed_components: Iterable[str] | None = None,
     ):
         self.algorithm = algorithm or DEFAULT_SIGNATURE_ALGORITHM
         self.http_key_resolver = CustomHTTPSignatureKeyResolver(key_resolver)
+        self.covered_components = set([f'"{component}"' for component in (required_signed_components or [])])
         self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)
 
     async def verify_content_digest(self, result: VerifyResult, request: Request):
@@ -110,6 +113,13 @@ async def verify(self, request: Request) -> VerifyResult:
             self.logger.warning(msg, extra=logger_extra, exc_info=exc)
             raise HTTPException(status.HTTP_400_BAD_REQUEST, msg) from exc
 
+        if self.covered_components:
+            for result in results:
+                if not all(header in result.covered_components for header in self.covered_components):
+                    msg = "Missing required headers in signature"
+                    self.logger.warning(msg, extra=logger_extra)
+                    raise HTTPException(status.HTTP_401_UNAUTHORIZED, msg)
+
         for result in results:
             try:
                 await self.verify_content_digest(result, request)

tests/test_http_signatures.py

@@ -8,13 +8,9 @@
 import httpx
 import pytest
 from cryptography.hazmat.primitives.asymmetric import ed25519
-from http_message_signatures import (
-    HTTPMessageSigner,
-    HTTPSignatureAlgorithm,
-    HTTPSignatureKeyResolver,
-    algorithms,
-)
+from http_message_signatures import HTTPMessageSigner, HTTPSignatureAlgorithm, HTTPSignatureKeyResolver, algorithms
 from starlette.datastructures import Headers
+from starlette.exceptions import HTTPException
 from starlette.requests import Request
 
 from aggrec.helpers import RequestVerifier
@@ -91,7 +87,16 @@ async def _test_http_signatures(algorithm: HTTPSignatureAlgorithm):
 
     key_resolver = TestHTTPSignatureKeyResolver(key_id=key_id, algorithm=algorithm)
     signer = HTTPMessageSigner(signature_algorithm=algorithm, key_resolver=key_resolver)
-    verifier = RequestVerifier(algorithm=algorithm, key_resolver=key_resolver)
+    verifier = RequestVerifier(
+        algorithm=algorithm,
+        key_resolver=key_resolver,
+        required_signed_components=["content-type", "content-digest", "content-length"],
+    )
+    verifier2 = RequestVerifier(
+        algorithm=algorithm,
+        key_resolver=key_resolver,
+        required_signed_components=["user-agent"],
+    )
 
     signer.sign(
         req,
@@ -113,6 +118,10 @@ async def _test_http_signatures(algorithm: HTTPSignatureAlgorithm):
     result = await verifier.verify(request)
     print(result)
 
+    with pytest.raises(HTTPException):
+        result = await verifier2.verify(request)
+        print(result)
+
 
 @pytest.mark.asyncio
 async def test_http_signatures_rsa_pkcs1_sha256():