From cd9caa1cbb38d8a29d61e5886d4a4dd65e954773 Mon Sep 17 00:00:00 2001
From: AlgoVoi <chopmob@gmail.com>
Date: Wed, 15 Jul 2026 13:38:37 +0100
Subject: [PATCH] fix(merchant-backend): make signature verifier RFC
 9421-conformant

---
 .../app/security/signature_verification.py    | 237 ++++++++++++------
 1 file changed, 158 insertions(+), 79 deletions(-)

diff --git a/merchant-backend/app/security/signature_verification.py b/merchant-backend/app/security/signature_verification.py
index acafe35..ce6533a 100644
--- a/merchant-backend/app/security/signature_verification.py
+++ b/merchant-backend/app/security/signature_verification.py
@@ -8,13 +8,13 @@
 
 import re
 import time
-import json
 from typing import Dict, Optional, Tuple
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.primitives.asymmetric import rsa, padding
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from cryptography.exceptions import InvalidSignature
 import base64
-import hashlib
 
 publicKey = """-----BEGIN PUBLIC KEY-----
 MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAysHJFJ9uoVvU1sH2x3TV
@@ -26,6 +26,22 @@
 LQIDAQAB
 -----END PUBLIC KEY-----"""
 
+# Replay cache to prevent nonce reuse (in production this would be a shared/
+# persistent store; an in-memory map suffices for the single-process reference
+# merchant backend). Each nonce is retained until the expiry of the signature
+# that used it: while the signature is still valid the nonce blocks a replay,
+# and once it expires the timestamp check rejects it regardless, so the entry
+# is purged. Keying retention to the signature's own expiry (rather than a fixed
+# TTL) closes the replay window that opens when a signature outlives the TTL.
+_nonce_cache: Dict[str, int] = {}
+
+
+def _purge_expired_nonces(now: int) -> None:
+    for nonce, expires_at in list(_nonce_cache.items()):
+        if expires_at < now:
+            del _nonce_cache[nonce]
+
+
 class SignatureVerifier:
     def __init__(self):
         # In production, these would be loaded from secure storage/config
@@ -39,10 +55,10 @@ def __init__(self):
                 "name": "Sample Payment Directory"
             }
         }
-    
+
     def _load_public_key(self, agent_name: str):
         """Load public key for the agent. In production, load from secure storage."""
-        
+
         if agent_name == "example":
             return serialization.load_pem_public_key(publicKey.encode("utf-8"))
         elif agent_name == "sample":
@@ -51,124 +67,187 @@ def _load_public_key(self, agent_name: str):
         else:
             raise ValueError(f"Unknown agent name: {agent_name}")
 
-    
     def parse_signature_headers(self, signature_agent: str, signature_input: str, signature: str) -> Optional[Dict]:
-        """Parse the signature headers and extract components."""
+        """Parse the RFC 9421 signature headers and extract components.
+
+        Accepts the standard RFC 9421 wire form used by the CDN proxy reference,
+        e.g.:
+            Signature-Input: sig2=("@authority" "@path"); created=...; expires=...;
+                             keyid="..."; alg="rsa-pss-sha256"; nonce="..."; tag="..."
+            Signature:       sig2=:<base64>:
+        The covered-component list is a space separated list of individually
+        quoted identifiers (RFC 9421 sec 2.1), the signature parameters are
+        unordered, and the label (sig1/sig2/...) is arbitrary.
+        """
         try:
-            # Parse Signature-Agent
             agent_url = signature_agent.strip('"')
-            
-            # Parse Signature-Input
-            signature_input_pattern = r'sig1=\("([^"]+)"\);\s*nonce="([^"]+)";\s*created=(\d+);\s*expires=(\d+);\s*keyid="([^"]+)";\s*tag="([^"]+)"'
-            match = re.match(signature_input_pattern, signature_input.strip())
-            
-            if not match:
+
+            # <label>=(<components>); <params...>
+            input_match = re.match(r'^\s*(\w+)=\(([^)]*)\);\s*(.+)$', signature_input.strip())
+            if not input_match:
                 return None
-            
-            signature_params, nonce, created, expires, keyid, tag = match.groups()
-            
-            # Parse Signature
-            signature_pattern = r'sig1=:([^:]+):'
-            sig_match = re.match(signature_pattern, signature.strip())
-            
+            label, components_blob, attr_string = input_match.groups()
+
+            # Covered components: split on whitespace and strip quotes. This
+            # accepts the RFC 9421 individually-quoted form ("a" "b") as the
+            # CDN proxy emits.
+            signature_params = [p.strip('"') for p in components_blob.split() if p.strip()]
+
+            # Signature parameters are unordered structured-field members.
+            attributes: Dict[str, object] = {}
+            for key, value in re.findall(r'(\w+)=("[^"]*"|\d+)', attr_string):
+                if value.startswith('"') and value.endswith('"'):
+                    attributes[key] = value[1:-1]
+                else:
+                    attributes[key] = int(value)
+
+            if "created" not in attributes or "expires" not in attributes:
+                return None
+
+            # keyId (camelCase, as the CDN proxy emits) or keyid.
+            keyid = attributes.get("keyId") or attributes.get("keyid")
+            algorithm = str(attributes.get("alg", "rsa-pss-sha256")).lower()
+
+            # Signature value: <label>=:<base64>:
+            sig_match = re.match(r'^\s*(\w+)=:([^:]+):\s*$', signature.strip())
             if not sig_match:
                 return None
-            
-            signature_value = sig_match.group(1)
-            
+            signature_value = sig_match.group(2)
+
             return {
+                "label": label,
                 "agent_url": agent_url,
-                "signature_params": signature_params.split(" "),
-                "nonce": nonce,
-                "created": int(created),
-                "expires": int(expires),
+                "signature_params": signature_params,
+                "nonce": attributes.get("nonce"),
+                "created": int(attributes["created"]),
+                "expires": int(attributes["expires"]),
                 "keyid": keyid,
-                "tag": tag,
-                "signature": signature_value
+                "algorithm": algorithm,
+                "tag": attributes.get("tag"),
+                "signature": signature_value,
+                # The verbatim Signature-Input value is required to reconstruct
+                # the @signature-params line of the signature base.
+                "signature_input_raw": signature_input.strip(),
             }
         except Exception as e:
             print(f"Error parsing signature headers: {e}")
             return None
-    
+
     def verify_signature(self, parsed_data: Dict, request_data: Dict) -> Tuple[bool, str]:
         """Verify the signature against the request data."""
         try:
             agent_url = parsed_data["agent_url"]
-            
+
             # Check if agent is trusted
             if agent_url not in self.trusted_agents:
                 return False, f"Unknown agent: {agent_url}"
-            
-            # Check timestamp validity
+
+            # Check timestamp validity (allow small clock skew on created, as the
+            # CDN proxy does, to tolerate minor client/server drift).
             current_time = int(time.time())
-            if current_time < parsed_data["created"]:
+            if current_time + 60 < parsed_data["created"]:
                 return False, "Signature created in the future"
-            
             if current_time > parsed_data["expires"]:
                 return False, "Signature expired"
-            
-            # Build signature string
+
+            # Replay protection: a nonce is required and may be seen only once.
+            nonce = parsed_data.get("nonce")
+            if not nonce:
+                return False, "Missing nonce"
+            _purge_expired_nonces(current_time)
+            if nonce in _nonce_cache:
+                return False, "Replay detected: nonce already used"
+
+            # Build the RFC 9421 signature base and verify it BEFORE recording
+            # the nonce, so an invalid signature cannot burn a valid nonce.
             signature_string = self._build_signature_string(
                 parsed_data["signature_params"],
                 request_data,
-                parsed_data["nonce"],
-                parsed_data["created"],
-                parsed_data["expires"]
+                parsed_data["signature_input_raw"],
             )
-            
-            # Verify signature
+
             public_key = self.trusted_agents[agent_url]["public_key"]
             signature_bytes = base64.b64decode(parsed_data["signature"])
-            
+            algorithm = parsed_data.get("algorithm", "rsa-pss-sha256")
+
             try:
-                public_key.verify(
-                    signature_bytes,
-                    signature_string.encode('utf-8'),
-                    padding.PSS(
-                        mgf=padding.MGF1(hashes.SHA256()),
-                        salt_length=padding.PSS.MAX_LENGTH
-                    ),
-                    hashes.SHA256()
-                )
-                return True, f"Verified agent: {self.trusted_agents[agent_url]['name']}"
+                if algorithm == "rsa-pss-sha256":
+                    if not isinstance(public_key, RSAPublicKey):
+                        return False, "Key does not match rsa-pss-sha256 algorithm"
+                    public_key.verify(
+                        signature_bytes,
+                        signature_string.encode("utf-8"),
+                        padding.PSS(
+                            mgf=padding.MGF1(hashes.SHA256()),
+                            salt_length=padding.PSS.MAX_LENGTH,
+                        ),
+                        hashes.SHA256(),
+                    )
+                elif algorithm == "ed25519":
+                    if not isinstance(public_key, Ed25519PublicKey):
+                        return False, "Key does not match ed25519 algorithm"
+                    public_key.verify(signature_bytes, signature_string.encode("utf-8"))
+                else:
+                    return False, f"Unsupported algorithm: {algorithm}"
             except InvalidSignature:
                 return False, "Invalid signature"
-                
+
+            # Signature is valid: record the nonce (retained until the signature
+            # expires) to block replays within its validity window.
+            _nonce_cache[nonce] = parsed_data["expires"]
+            return True, f"Verified agent: {self.trusted_agents[agent_url]['name']}"
+
         except Exception as e:
             return False, f"Verification error: {str(e)}"
-    
-    def _build_signature_string(self, params: list, request_data: Dict, nonce: str, created: int, expires: int) -> str:
-        """Build the signature string from the parameters."""
-        signature_parts = []
-        
+
+    def _build_signature_string(self, params: list, request_data: Dict, signature_input_raw: str) -> str:
+        """Build the RFC 9421 signature base string.
+
+        Mirrors the CDN proxy reference (buildRFC9421SignatureString): each
+        covered component is emitted as `"<name>": <value>`, and the base ends
+        with the mandatory `"@signature-params"` line (RFC 9421 sec 2.5) carrying
+        the verbatim signature parameters. Without that final line the base can
+        never match a conformant signer's, so every RFC 9421 signature would be
+        rejected.
+        """
+        components = []
+
         for param in params:
             if param == "@authority":
-                signature_parts.append(f'"@authority": "{request_data.get("authority", "")}"')
+                components.append(f'"@authority": {request_data.get("authority", "")}')
             elif param == "@path":
-                signature_parts.append(f'"@path": "{request_data.get("path", "")}"')
-            elif param == "directory-agent":
-                signature_parts.append(f'"directory-agent": "{request_data.get("directory-agent", "")}"')
-            elif param == "query-param":
-                signature_parts.append(f'"query-param": "{request_data.get("query-param", "")}"')
-        
-        signature_parts.extend([
-            f'"nonce": "{nonce}"',
-            f'"created": {created}',
-            f'"expires": {expires}'
-        ])
-        
-        return "\n".join(signature_parts)
-    
+                components.append(f'"@path": {request_data.get("path", "")}')
+            elif param == "content-type":
+                content_type = request_data.get("contentType") or request_data.get("content-type") or "application/json"
+                components.append(f'"content-type": {content_type}')
+            elif param == "host":
+                host = request_data.get("host") or request_data.get("authority", "")
+                components.append(f'"host": {host}')
+            else:
+                # Custom covered header (e.g. directory-agent, query-param).
+                value = request_data.get(param)
+                if value:
+                    components.append(f'"{param}": {value}')
+
+        # @signature-params is the verbatim parameters portion of Signature-Input
+        # (the label and leading '=' removed), added as the final line.
+        signature_params = signature_input_raw
+        eq_index = signature_params.find("=")
+        if eq_index != -1:
+            signature_params = signature_params[eq_index + 1:]
+        components.append(f'"@signature-params": {signature_params}')
+
+        return "\n".join(components)
+
     def is_trusted_agent(self, signature_agent: str, signature_input: str, signature: str, request_data: Dict) -> Tuple[bool, str]:
         """Main method to verify if the request is from a trusted agent."""
-        # Parse headers
         parsed_data = self.parse_signature_headers(signature_agent, signature_input, signature)
-        
+
         if not parsed_data:
             return False, "Invalid signature format"
-        
-        # Verify signature
+
         return self.verify_signature(parsed_data, request_data)
 
+
 # Global instance
 signature_verifier = SignatureVerifier()