Implement HMAC for metadata

This implements a metadata-based hmac. Note that this implementation
only uses the metadata as input for the HMAC.

Signed-off-by: Florian Wagner <[email protected]>

Author: florian-wagner-frequenz

RELEASE_NOTES.md

@@ -1,5 +1,9 @@
 # Frequenz Client Base Library Release Notes
 
+## Features
+
+* Added support for HMAC signing of client messages
+
 ## Upgrading
 
 * Updated `protobuf` dependency range: changed from `>=4.21.6, <6` to `>=5.29.2, <7`

src/frequenz/client/base/authentication.py

@@ -0,0 +1,77 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""An Interceptor that adds the API key to a gRPC call."""
+
+import dataclasses
+from typing import Callable
+
+from grpc.aio import (
+    ClientCallDetails,
+    Metadata,
+    UnaryUnaryCall,
+    UnaryUnaryClientInterceptor,
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class AuthenticationOptions:
+    """Options for authenticating to the endpoint."""
+
+    api_key: str
+    """The API key to authenticate with."""
+
+
+# There is an issue in gRPC that causes the type to be unspecifieable correctly here.
+class AuthenticationInterceptor(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
+    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
+
+    def __init__(self, options: AuthenticationOptions):
+        """Create an instance of the interceptor.
+
+        Args:
+            options: The options for authenticating to the endpoint.
+        """
+        self._key = options.api_key
+
+    async def intercept_unary_unary(
+        self,
+        continuation: Callable[
+            [ClientCallDetails, object], UnaryUnaryCall[object, object]
+        ],
+        client_call_details: ClientCallDetails,
+        request: object,
+    ) -> object:
+        """Intercept the call to add HMAC authentication to the metadata fields.
+
+        This is a known method from the base class that is overridden.
+
+        Args:
+            continuation: The next interceptor in the chain.
+            client_call_details: The call details.
+            request: The request object.
+
+        Returns:
+            The response object (this implementation does not modify the response).
+        """
+        self.add_auth_header(
+            client_call_details,
+        )
+        return await continuation(client_call_details, request)
+
+    def add_auth_header(
+        self,
+        client_call_details: ClientCallDetails,
+    ) -> None:
+        """Add the API key as a metadata field to the call.
+
+        The API key is used by the later sign interceptor to calculate the HMAC.
+        In addition it is used as a first layer of authentication by the server.
+
+        Args:
+            client_call_details: The call details.
+        """
+        if client_call_details.metadata is None:
+            client_call_details.metadata = Metadata()
+
+        client_call_details.metadata["x-key"] = self._key

src/frequenz/client/base/channel.py

@@ -10,7 +10,15 @@
 from urllib.parse import parse_qs, urlparse
 
 from grpc import ssl_channel_credentials
-from grpc.aio import Channel, insecure_channel, secure_channel
+from grpc.aio import (
+    Channel,
+    ClientInterceptor,
+    insecure_channel,
+    secure_channel,
+)
+
+from .authentication import AuthenticationInterceptor, AuthenticationOptions
+from .signing import SigningInterceptor, SigningOptions
 
 
 @dataclasses.dataclass(frozen=True)
@@ -69,6 +77,12 @@ class ChannelOptions:
     keep_alive: KeepAliveOptions = KeepAliveOptions()
     """HTTP2 keep-alive options for the channel."""
 
+    sign: SigningOptions | None = None
+    """Signing options for the channel."""
+
+    auth: AuthenticationOptions | None = None
+    """Authentication options for the channel."""
+
 
 def parse_grpc_uri(
     uri: str,
@@ -177,6 +191,17 @@ def parse_grpc_uri(
         else None
     )
 
+    interceptors: list[ClientInterceptor] = []
+    if defaults.auth is not None:
+        interceptors.append(
+            AuthenticationInterceptor(options=defaults.auth)  # type: ignore[arg-type]
+        )
+
+    if defaults.sign is not None:
+        interceptors.append(
+            SigningInterceptor(options=defaults.sign)  # type: ignore[arg-type]
+        )
+
     ssl = defaults.ssl.enabled if options.ssl is None else options.ssl
     if ssl:
         return secure_channel(
@@ -199,8 +224,9 @@ def parse_grpc_uri(
                 ),
             ),
             channel_options,
+            interceptors=interceptors,
         )
-    return insecure_channel(target, channel_options)
+    return insecure_channel(target, channel_options, interceptors=interceptors)
 
 
 def _to_bool(value: str) -> bool:

src/frequenz/client/base/client.py

@@ -8,7 +8,10 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, Generic, Self, TypeVar, overload
 
-from grpc.aio import AioRpcError, Channel
+from grpc.aio import (
+    AioRpcError,
+    Channel,
+)
 
 from .channel import ChannelOptions, parse_grpc_uri
 from .exception import ApiClientError, ClientNotConnected

src/frequenz/client/base/signing.py

@@ -0,0 +1,101 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""An Interceptor that adds HMAC signature of the metadata fields to a gRPC call."""
+
+import dataclasses
+import hmac
+import logging
+import secrets
+import time
+from base64 import urlsafe_b64encode
+from typing import Any, Callable
+
+from grpc.aio import (
+    ClientCallDetails,
+    UnaryUnaryCall,
+    UnaryUnaryClientInterceptor,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass(frozen=True)
+class SigningOptions:
+    """Options for message signing of messages."""
+
+    secret: str
+    """The secret to sign the message with."""
+
+
+# There is an issue in gRPC that causes the type to be unspecifieable correctly here.
+class SigningInterceptor(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
+    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
+
+    def __init__(self, options: SigningOptions):
+        """Create an instance of the interceptor.
+
+        Args:
+            options: The options for signing the message.
+        """
+        self._secret = options.secret.encode()
+
+    async def intercept_unary_unary(
+        self,
+        continuation: Callable[
+            [ClientCallDetails, object], UnaryUnaryCall[object, object]
+        ],
+        client_call_details: ClientCallDetails,
+        request: object,
+    ) -> object:
+        """Intercept the call to add HMAC authentication to the metadata fields.
+
+        This is a known method from the base class that is overridden.
+
+        Args:
+            continuation: The next interceptor in the chain.
+            client_call_details: The call details.
+            request: The request object.
+
+        Returns:
+            The response object (this implementation does not modify the response).
+        """
+        self.add_hmac(
+            client_call_details,
+            int(time.time()).to_bytes(8, "big"),
+            secrets.token_bytes(16),
+        )
+        return await continuation(client_call_details, request)
+
+    def add_hmac(
+        self, client_call_details: ClientCallDetails, ts: bytes, nonce: bytes
+    ) -> None:
+        """Add the HMAC authentication to the metadata fields of the call details.
+
+        The extra headers are directly added to the client_call details.
+
+        Args:
+            client_call_details: The call details.
+            ts: The timestamp to use for the HMAC.
+            nonce: The nonce to use for the HMAC.
+        """
+        if client_call_details.metadata is None:
+            _logger.error(
+                "No metadata found, cannot extract an api key. Therefore, cannot sign the request."
+            )
+            return
+
+        key: Any = client_call_details.metadata.get("x-key")
+        if key is None:
+            _logger.error("No key found in metadata, cannot sign the request.")
+            return
+        hmac_obj = hmac.new(self._secret, digestmod="sha256")
+        hmac_obj.update(key.encode())
+        hmac_obj.update(ts)
+        hmac_obj.update(nonce)
+
+        hmac_obj.update(client_call_details.method.encode())
+
+        client_call_details.metadata["x-ts"] = ts
+        client_call_details.metadata["x-nonce"] = nonce
+        client_call_details.metadata["x-hmac"] = urlsafe_b64encode(hmac_obj.digest())

tests/test_authentication.py

@@ -0,0 +1,28 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the Interceptor functionalities."""
+
+from unittest import mock
+
+from frequenz.client.base.authentication import (
+    AuthenticationInterceptor,
+    AuthenticationOptions,
+)
+
+
+async def test_auth_interceptor() -> None:
+    """Test that the Auth Interceptor adds the correct header."""
+    auth: AuthenticationOptions = AuthenticationOptions(api_key="my_key")
+    auth_interceptor: AuthenticationInterceptor = AuthenticationInterceptor(
+        options=auth
+    )
+
+    metadata: dict[str, str] = {}
+
+    client_call_details = mock.MagicMock(method="my_rpc")
+    client_call_details.metadata = metadata
+
+    auth_interceptor.add_auth_header(client_call_details)
+
+    assert metadata["x-key"] == "my_key"

tests/test_channel.py

@@ -315,11 +315,14 @@ def test_parse_uri_ok(  # pylint: disable=too-many-locals
             certificate_chain=expected_certificate_chain,
         )
         secure_channel_mock.assert_called_once_with(
-            expected_target, expected_credentials, expected_channel_options
+            expected_target,
+            expected_credentials,
+            expected_channel_options,
+            interceptors=[],
         )
     else:
         insecure_channel_mock.assert_called_once_with(
-            expected_target, expected_channel_options
+            expected_target, expected_channel_options, interceptors=[]
         )
 
 

tests/test_client.py

@@ -12,7 +12,11 @@
 import pytest_mock
 
 from frequenz.client.base.channel import ChannelOptions, SslOptions
-from frequenz.client.base.client import BaseApiClient, StubT, call_stub_method
+from frequenz.client.base.client import (
+    BaseApiClient,
+    StubT,
+    call_stub_method,
+)
 from frequenz.client.base.exception import ClientNotConnected, UnknownError
 
 

tests/test_signing.py

@@ -0,0 +1,28 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the Interceptor functionalities."""
+
+from unittest import mock
+
+from frequenz.client.base.signing import (
+    SigningInterceptor,
+    SigningOptions,
+)
+
+
+async def test_sign_interceptor() -> None:
+    """Test that the HMAC is calculated correctly so that it will match the value of the server."""
+    sign: SigningOptions = SigningOptions(secret="my_secret")
+    sign_interceptor: SigningInterceptor = SigningInterceptor(options=sign)
+
+    metadata: dict[str, str | bytes] = {"x-key": "my_key"}
+
+    client_call_details = mock.MagicMock(method="my_rpc")
+    client_call_details.metadata = metadata
+
+    sign_interceptor.add_hmac(client_call_details, b"1634567890", b"123456789")
+
+    assert metadata["x-hmac"] == "NJDvrkRZhOPekn5AvPiaJsYTJYCgnLzA-LQFC2D7GNE=".encode(
+        "utf-8"
+    )