Skip to content

signing

frequenz.client.base.signing ¤

An Interceptor that adds HMAC signature of the metadata fields to a gRPC call.

Classes¤

frequenz.client.base.signing.SigningInterceptorUnaryStream ¤

Bases: UnaryStreamClientInterceptor

An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call.

Source code in src/frequenz/client/base/signing.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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):
        """Create an instance of the interceptor.

        Args:
            options: The options for signing the message.
        """
        self._secret = options.secret.encode()

    async def intercept_unary_stream(
        self,
        continuation: Callable[
            [ClientCallDetails, Any], UnaryStreamCall[object, object]
        ],
        client_call_details: ClientCallDetails,
        request: Any,
    ) -> AsyncIterable[object] | UnaryStreamCall[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).
        """
        _add_hmac(
            self._secret,
            client_call_details,
            int(time.time()),
            secrets.token_bytes(16),
        )
        return await continuation(client_call_details, request)  # type: ignore
Functions¤
__init__ ¤
__init__(options: SigningOptions)

Create an instance of the interceptor.

PARAMETER DESCRIPTION
options

The options for signing the message.

TYPE: SigningOptions

Source code in src/frequenz/client/base/signing.py 
124
125
126
127
128
129
130
def __init__(self, options: SigningOptions):
    """Create an instance of the interceptor.

    Args:
        options: The options for signing the message.
    """
    self._secret = options.secret.encode()
intercept_unary_stream async ¤
intercept_unary_stream(
    continuation: Callable[
        [ClientCallDetails, Any],
        UnaryStreamCall[object, object],
    ],
    client_call_details: ClientCallDetails,
    request: Any,
) -> (
    AsyncIterable[object] | UnaryStreamCall[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.

PARAMETER DESCRIPTION
continuation

The next interceptor in the chain.

TYPE: Callable[[ClientCallDetails, Any], UnaryStreamCall[object, object]]

client_call_details

The call details.

TYPE: ClientCallDetails

request

The request object.

TYPE: Any

RETURNS DESCRIPTION
AsyncIterable[object] | UnaryStreamCall[object, object]

The response object (this implementation does not modify the response).
Source code in src/frequenz/client/base/signing.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
async def intercept_unary_stream(
    self,
    continuation: Callable[
        [ClientCallDetails, Any], UnaryStreamCall[object, object]
    ],
    client_call_details: ClientCallDetails,
    request: Any,
) -> AsyncIterable[object] | UnaryStreamCall[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).
    """
    _add_hmac(
        self._secret,
        client_call_details,
        int(time.time()),
        secrets.token_bytes(16),
    )
    return await continuation(client_call_details, request)  # type: ignore

frequenz.client.base.signing.SigningInterceptorUnaryUnary ¤

Bases: UnaryUnaryClientInterceptor

An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call.

Source code in src/frequenz/client/base/signing.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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):
        """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).
        """
        _add_hmac(
            self._secret,
            client_call_details,
            int(time.time()),
            secrets.token_bytes(16),
        )
        return await continuation(client_call_details, request)
Functions¤
__init__ ¤
__init__(options: SigningOptions)

Create an instance of the interceptor.

PARAMETER DESCRIPTION
options

The options for signing the message.

TYPE: SigningOptions

Source code in src/frequenz/client/base/signing.py 
83
84
85
86
87
88
89
def __init__(self, options: SigningOptions):
    """Create an instance of the interceptor.

    Args:
        options: The options for signing the message.
    """
    self._secret = options.secret.encode()
intercept_unary_unary async ¤
intercept_unary_unary(
    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.

PARAMETER DESCRIPTION
continuation

The next interceptor in the chain.

TYPE: Callable[[ClientCallDetails, object], UnaryUnaryCall[object, object]]

client_call_details

The call details.

TYPE: ClientCallDetails

request

The request object.

TYPE: object

RETURNS DESCRIPTION
object

The response object (this implementation does not modify the response).
Source code in src/frequenz/client/base/signing.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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).
    """
    _add_hmac(
        self._secret,
        client_call_details,
        int(time.time()),
        secrets.token_bytes(16),
    )
    return await continuation(client_call_details, request)

frequenz.client.base.signing.SigningOptions dataclass ¤

Options for message signing of messages.

Source code in src/frequenz/client/base/signing.py 
71
72
73
74
75
76
@dataclasses.dataclass(frozen=True)
class SigningOptions:
    """Options for message signing of messages."""

    secret: str
    """The secret to sign the message with."""
Attributes¤
secret instance-attribute ¤
secret: str

The secret to sign the message with.