Change interface for HMAC

This changes the interface for communicating the API key and the signing
secret. Specifically it removes them from "authentication options" and
moves them to their own parameters instead.
This allows using the default pattern for channel options in places that
previously did not touch it. This is desired as soon most clients will
require the API key and the signing secret to function properly, and we
want to keep the changes needed to them minimal.

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

Author: florian-wagner-frequenz

src/frequenz/client/base/authentication.py

@@ -3,7 +3,6 @@
 
 """An Interceptor that adds the API key to a gRPC call."""
 
-import dataclasses
 from typing import AsyncIterable, Callable
 
 from grpc.aio import (
@@ -35,25 +34,17 @@ def _add_auth_header(
     client_call_details.metadata["key"] = key
 
 
-@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 which means the type can not be specified correctly here.
 class AuthenticationInterceptorUnaryUnary(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
     """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
 
-    def __init__(self, options: AuthenticationOptions):
+    def __init__(self, api_key: str):
         """Create an instance of the interceptor.
 
         Args:
-            options: The options for authenticating to the endpoint.
+            api_key: The API key to send along for the request.
         """
-        self._key = options.api_key
+        self._key = api_key
 
     async def intercept_unary_unary(
         self,
@@ -83,13 +74,13 @@ async def intercept_unary_unary(
 class AuthenticationInterceptorUnaryStream(UnaryStreamClientInterceptor):  # type: ignore[type-arg]
     """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
 
-    def __init__(self, options: AuthenticationOptions):
+    def __init__(self, api_key: str):
         """Create an instance of the interceptor.
 
         Args:
-            options: The options for authenticating to the endpoint.
+            api_key: The API key to send along for the request.
         """
-        self._key = options.api_key
+        self._key = api_key
 
     async def intercept_unary_stream(
         self,

src/frequenz/client/base/channel.py

@@ -20,12 +20,10 @@
 from .authentication import (
     AuthenticationInterceptorUnaryStream,
     AuthenticationInterceptorUnaryUnary,
-    AuthenticationOptions,
 )
 from .signing import (
     SigningInterceptorUnaryStream,
     SigningInterceptorUnaryUnary,
-    SigningOptions,
 )
 
 
@@ -85,17 +83,13 @@ 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,
     /,
     defaults: ChannelOptions = ChannelOptions(),
+    key: str | None = None,
+    sign_secret: str | None = None,
 ) -> Channel:
     """Create a client channel from a URI.
 
@@ -133,6 +127,8 @@ def parse_grpc_uri(
         uri: The gRPC URI specifying the connection parameters.
         defaults: The default options use to create the channel when not specified in
             the URI.
+        key: The API key to use when connecting to the service.
+        sign_secret: The secret to use when creating message HMAC.
 
     Returns:
         A client channel object.
@@ -200,16 +196,16 @@ def parse_grpc_uri(
     )
 
     interceptors: list[ClientInterceptor] = []
-    if defaults.auth is not None:
+    if key is not None:
         interceptors += [
-            AuthenticationInterceptorUnaryUnary(options=defaults.auth),  # type: ignore [list-item]
-            AuthenticationInterceptorUnaryStream(options=defaults.auth),  # type: ignore [list-item]
+            AuthenticationInterceptorUnaryUnary(api_key=key),  # type: ignore [list-item]
+            AuthenticationInterceptorUnaryStream(api_key=key),  # type: ignore [list-item]
         ]
 
-    if defaults.sign is not None:
+    if sign_secret is not None:
         interceptors += [
-            SigningInterceptorUnaryUnary(options=defaults.sign),  # type: ignore [list-item]
-            SigningInterceptorUnaryStream(options=defaults.sign),  # type: ignore [list-item]
+            SigningInterceptorUnaryUnary(secret=sign_secret),  # type: ignore [list-item]
+            SigningInterceptorUnaryStream(secret=sign_secret),  # type: ignore [list-item]
         ]
 
     ssl = defaults.ssl.enabled if options.ssl is None else options.ssl

src/frequenz/client/base/signing.py

@@ -3,7 +3,6 @@
 
 """An Interceptor that adds HMAC signature of the metadata fields to a gRPC call."""
 
-import dataclasses
 import hmac
 import logging
 import secrets
@@ -68,25 +67,17 @@ def _add_hmac(
     )
 
 
-@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 which means the type can not be specified correctly here.
 class SigningInterceptorUnaryUnary(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
     """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
 
-    def __init__(self, options: SigningOptions):
+    def __init__(self, secret: str):
         """Create an instance of the interceptor.
 
         Args:
-            options: The options for signing the message.
+            secret: The secret used for signing the message.
         """
-        self._secret = options.secret.encode()
+        self._secret = secret.encode()
 
     async def intercept_unary_unary(
         self,
@@ -121,13 +112,13 @@ async def intercept_unary_unary(
 class SigningInterceptorUnaryStream(UnaryStreamClientInterceptor):  # type: ignore[type-arg]
     """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""
 
-    def __init__(self, options: SigningOptions):
+    def __init__(self, secret: str):
         """Create an instance of the interceptor.
 
         Args:
-            options: The options for signing the message.
+            secret: The secret used for signing the message.
         """
-        self._secret = options.secret.encode()
+        self._secret = secret.encode()
 
     async def intercept_unary_stream(
         self,