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 
112
113
114
115
116
117
118
119
120
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
class SigningInterceptorUnaryStream(UnaryStreamClientInterceptor):  # type: ignore[type-arg]
    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""

    def __init__(self, secret: str):
        """Create an instance of the interceptor.

        Args:
            secret: The secret used for signing the message.
        """
        self._secret = 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__(secret: str)

Create an instance of the interceptor.

PARAMETER DESCRIPTION
secret

The secret used for signing the message.

TYPE: str

Source code in src/frequenz/client/base/signing.py 
115
116
117
118
119
120
121
def __init__(self, secret: str):
    """Create an instance of the interceptor.

    Args:
        secret: The secret used for signing the message.
    """
    self._secret = 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 
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
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 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 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
class SigningInterceptorUnaryUnary(UnaryUnaryClientInterceptor):  # type: ignore[type-arg]
    """An Interceptor that adds HMAC authentication of the metadata fields to a gRPC call."""

    def __init__(self, secret: str):
        """Create an instance of the interceptor.

        Args:
            secret: The secret used for signing the message.
        """
        self._secret = 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__(secret: str)

Create an instance of the interceptor.

PARAMETER DESCRIPTION
secret

The secret used for signing the message.

TYPE: str

Source code in src/frequenz/client/base/signing.py 
74
75
76
77
78
79
80
def __init__(self, secret: str):
    """Create an instance of the interceptor.

    Args:
        secret: The secret used for signing the message.
    """
    self._secret = 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 
 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
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)