Module glean.metrics.jwe

Expand source code
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.


from typing import List, Optional
import json


from .. import _ffi
from .._dispatcher import Dispatcher
from ..testing import ErrorType


from .lifetime import Lifetime


class JweData:
    """
    A representation of a JWE value.
    """

    def __init__(
        self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
    ):

        self.header = header
        self.key = key
        self.init_vector = init_vector
        self.cipher_text = cipher_text
        self.auth_tag = auth_tag


class JweMetricType:
    """
    This implements the developer facing API for recording JWE metrics.

    Instances of this class type are automatically generated by
    `glean.load_metrics`, allowing developers to record values that were
    previously registered in the metrics.yaml file.

    The string API exposes the `JweMetricType.set` and
    `JweMetricType.setWithCompactRepresentation` methods,
    which take care of validating the input data
    and making sure that limits are enforced.
    """

    def __init__(
        self,
        disabled: bool,
        category: str,
        lifetime: Lifetime,
        name: str,
        send_in_pings: List[str],
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_jwe_metric(
            _ffi.ffi_encode_string(category),
            _ffi.ffi_encode_string(name),
            _ffi.ffi_encode_vec_string(send_in_pings),
            len(send_in_pings),
            lifetime.value,
            disabled,
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_jwe_metric(self._handle)

    def set_with_compact_representation(self, value: str):
        """
        Set to the specified JWE value.

        Args:
            value (str): the compact representation of a JWE value.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set_with_compact_representation(
                self._handle, _ffi.ffi_encode_string(value)
            )

    def set(
        self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
    ) -> None:
        """
        Build a JWE value from its elements and set to it.

        Args:
            header (str): A variable-size JWE protected header.
            key (str): A variable-size encrypted key.
                This can be an empty octet sequence.
            init_vector (str): A fixed-size, 96-bit, base64 encoded Jwe initialization vector.
                If not required by the encryption algorithm, can be an empty octet sequence.
            cipher_text (str): The variable-size base64 encoded cipher text.
            auth_tag (str): A fixed-size, 132-bit, base64 encoded authentication tag.
                Can be an empty octet sequence.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set(
                self._handle,
                _ffi.ffi_encode_string(header),
                _ffi.ffi_encode_string(key),
                _ffi.ffi_encode_string(init_vector),
                _ffi.ffi_encode_string(cipher_text),
                _ffi.ffi_encode_string(auth_tag),
            )

    def test_has_value(self, ping_name: Optional[str] = None) -> bool:
        """
        Tests whether a value is stored for the metric for testing purposes
        only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            has_value (bool): True if the metric value exists.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        return bool(
            _ffi.lib.glean_jwe_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> JweData:
        """
        Returns the stored value for testing purposes only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (JweData): value of the stored metric.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        if not self.test_has_value(ping_name):
            raise ValueError("metric has no value")

        json_payload = json.loads(
            _ffi.ffi_decode_string(
                _ffi.lib.glean_jwe_test_get_value_as_json_string(
                    self._handle, _ffi.ffi_encode_string(ping_name)
                )
            )
        )

        return JweData(
            json_payload["header"],
            json_payload["key"],
            json_payload["init_vector"],
            json_payload["cipher_text"],
            json_payload["auth_tag"],
        )

    def test_get_compact_representation(self, ping_name: Optional[str] = None) -> str:
        """
        Returns the stored value as the compact representation
        for testing purposes only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (JweData): value of the stored metric.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        if not self.test_has_value(ping_name):
            raise ValueError("metric has no value")

        return _ffi.ffi_decode_string(
            _ffi.lib.glean_jwe_test_get_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_num_recorded_errors(
        self, error_type: ErrorType, ping_name: Optional[str] = None
    ) -> int:
        """
        Returns the number of errors recorded for the given metric.

        Args:
            error_type (ErrorType): The type of error recorded.
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            num_errors (int): The number of errors recorded for the metric for
                the given error type.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        return _ffi.lib.glean_jwe_test_get_num_recorded_errors(
            self._handle,
            error_type.value,
            _ffi.ffi_encode_string(ping_name),
        )


__all__ = ["JweMetricType"]

Classes

class JweMetricType (disabled: bool, category: str, lifetime: Lifetime, name: str, send_in_pings: List[str])

This implements the developer facing API for recording JWE metrics.

Instances of this class type are automatically generated by load_metrics(), allowing developers to record values that were previously registered in the metrics.yaml file.

The string API exposes the JweMetricType.set() and JweMetricType.setWithCompactRepresentation methods, which take care of validating the input data and making sure that limits are enforced.

Expand source code
class JweMetricType:
    """
    This implements the developer facing API for recording JWE metrics.

    Instances of this class type are automatically generated by
    `glean.load_metrics`, allowing developers to record values that were
    previously registered in the metrics.yaml file.

    The string API exposes the `JweMetricType.set` and
    `JweMetricType.setWithCompactRepresentation` methods,
    which take care of validating the input data
    and making sure that limits are enforced.
    """

    def __init__(
        self,
        disabled: bool,
        category: str,
        lifetime: Lifetime,
        name: str,
        send_in_pings: List[str],
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_jwe_metric(
            _ffi.ffi_encode_string(category),
            _ffi.ffi_encode_string(name),
            _ffi.ffi_encode_vec_string(send_in_pings),
            len(send_in_pings),
            lifetime.value,
            disabled,
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_jwe_metric(self._handle)

    def set_with_compact_representation(self, value: str):
        """
        Set to the specified JWE value.

        Args:
            value (str): the compact representation of a JWE value.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set_with_compact_representation(
                self._handle, _ffi.ffi_encode_string(value)
            )

    def set(
        self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
    ) -> None:
        """
        Build a JWE value from its elements and set to it.

        Args:
            header (str): A variable-size JWE protected header.
            key (str): A variable-size encrypted key.
                This can be an empty octet sequence.
            init_vector (str): A fixed-size, 96-bit, base64 encoded Jwe initialization vector.
                If not required by the encryption algorithm, can be an empty octet sequence.
            cipher_text (str): The variable-size base64 encoded cipher text.
            auth_tag (str): A fixed-size, 132-bit, base64 encoded authentication tag.
                Can be an empty octet sequence.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set(
                self._handle,
                _ffi.ffi_encode_string(header),
                _ffi.ffi_encode_string(key),
                _ffi.ffi_encode_string(init_vector),
                _ffi.ffi_encode_string(cipher_text),
                _ffi.ffi_encode_string(auth_tag),
            )

    def test_has_value(self, ping_name: Optional[str] = None) -> bool:
        """
        Tests whether a value is stored for the metric for testing purposes
        only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            has_value (bool): True if the metric value exists.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        return bool(
            _ffi.lib.glean_jwe_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> JweData:
        """
        Returns the stored value for testing purposes only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (JweData): value of the stored metric.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        if not self.test_has_value(ping_name):
            raise ValueError("metric has no value")

        json_payload = json.loads(
            _ffi.ffi_decode_string(
                _ffi.lib.glean_jwe_test_get_value_as_json_string(
                    self._handle, _ffi.ffi_encode_string(ping_name)
                )
            )
        )

        return JweData(
            json_payload["header"],
            json_payload["key"],
            json_payload["init_vector"],
            json_payload["cipher_text"],
            json_payload["auth_tag"],
        )

    def test_get_compact_representation(self, ping_name: Optional[str] = None) -> str:
        """
        Returns the stored value as the compact representation
        for testing purposes only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (JweData): value of the stored metric.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        if not self.test_has_value(ping_name):
            raise ValueError("metric has no value")

        return _ffi.ffi_decode_string(
            _ffi.lib.glean_jwe_test_get_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_num_recorded_errors(
        self, error_type: ErrorType, ping_name: Optional[str] = None
    ) -> int:
        """
        Returns the number of errors recorded for the given metric.

        Args:
            error_type (ErrorType): The type of error recorded.
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            num_errors (int): The number of errors recorded for the metric for
                the given error type.
        """
        if ping_name is None:
            ping_name = self._send_in_pings[0]

        return _ffi.lib.glean_jwe_test_get_num_recorded_errors(
            self._handle,
            error_type.value,
            _ffi.ffi_encode_string(ping_name),
        )

Methods

def set(self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str) ‑> None

Build a JWE value from its elements and set to it.

Args

header : str
A variable-size JWE protected header.
key : str
A variable-size encrypted key. This can be an empty octet sequence.
init_vector : str
A fixed-size, 96-bit, base64 encoded Jwe initialization vector. If not required by the encryption algorithm, can be an empty octet sequence.
cipher_text : str
The variable-size base64 encoded cipher text.
auth_tag : str
A fixed-size, 132-bit, base64 encoded authentication tag. Can be an empty octet sequence.
Expand source code
def set(
    self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
) -> None:
    """
    Build a JWE value from its elements and set to it.

    Args:
        header (str): A variable-size JWE protected header.
        key (str): A variable-size encrypted key.
            This can be an empty octet sequence.
        init_vector (str): A fixed-size, 96-bit, base64 encoded Jwe initialization vector.
            If not required by the encryption algorithm, can be an empty octet sequence.
        cipher_text (str): The variable-size base64 encoded cipher text.
        auth_tag (str): A fixed-size, 132-bit, base64 encoded authentication tag.
            Can be an empty octet sequence.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_jwe_set(
            self._handle,
            _ffi.ffi_encode_string(header),
            _ffi.ffi_encode_string(key),
            _ffi.ffi_encode_string(init_vector),
            _ffi.ffi_encode_string(cipher_text),
            _ffi.ffi_encode_string(auth_tag),
        )
def set_with_compact_representation(self, value: str)

Set to the specified JWE value.

Args

value : str
the compact representation of a JWE value.
Expand source code
def set_with_compact_representation(self, value: str):
    """
    Set to the specified JWE value.

    Args:
        value (str): the compact representation of a JWE value.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_jwe_set_with_compact_representation(
            self._handle, _ffi.ffi_encode_string(value)
        )
def test_get_compact_representation(self, ping_name: Optional[str] = None) ‑> str

Returns the stored value as the compact representation for testing purposes only.

Args

ping_name : str
(default: first value in send_in_pings) The name of the ping to retrieve the metric for.

Returns

value (JweData): value of the stored metric.

Expand source code
def test_get_compact_representation(self, ping_name: Optional[str] = None) -> str:
    """
    Returns the stored value as the compact representation
    for testing purposes only.

    Args:
        ping_name (str): (default: first value in send_in_pings) The name
            of the ping to retrieve the metric for.

    Returns:
        value (JweData): value of the stored metric.
    """
    if ping_name is None:
        ping_name = self._send_in_pings[0]

    if not self.test_has_value(ping_name):
        raise ValueError("metric has no value")

    return _ffi.ffi_decode_string(
        _ffi.lib.glean_jwe_test_get_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
def test_get_num_recorded_errors(self, error_type: ErrorType, ping_name: Optional[str] = None) ‑> int

Returns the number of errors recorded for the given metric.

Args

error_type : ErrorType
The type of error recorded.
ping_name : str
(default: first value in send_in_pings) The name of the ping to retrieve the metric for.

Returns

num_errors (int): The number of errors recorded for the metric for the given error type.

Expand source code
def test_get_num_recorded_errors(
    self, error_type: ErrorType, ping_name: Optional[str] = None
) -> int:
    """
    Returns the number of errors recorded for the given metric.

    Args:
        error_type (ErrorType): The type of error recorded.
        ping_name (str): (default: first value in send_in_pings) The name
            of the ping to retrieve the metric for.

    Returns:
        num_errors (int): The number of errors recorded for the metric for
            the given error type.
    """
    if ping_name is None:
        ping_name = self._send_in_pings[0]

    return _ffi.lib.glean_jwe_test_get_num_recorded_errors(
        self._handle,
        error_type.value,
        _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name: Optional[str] = None) ‑> glean.metrics.jwe.JweData

Returns the stored value for testing purposes only.

Args

ping_name : str
(default: first value in send_in_pings) The name of the ping to retrieve the metric for.

Returns

value (JweData): value of the stored metric.

Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> JweData:
    """
    Returns the stored value for testing purposes only.

    Args:
        ping_name (str): (default: first value in send_in_pings) The name
            of the ping to retrieve the metric for.

    Returns:
        value (JweData): value of the stored metric.
    """
    if ping_name is None:
        ping_name = self._send_in_pings[0]

    if not self.test_has_value(ping_name):
        raise ValueError("metric has no value")

    json_payload = json.loads(
        _ffi.ffi_decode_string(
            _ffi.lib.glean_jwe_test_get_value_as_json_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )
    )

    return JweData(
        json_payload["header"],
        json_payload["key"],
        json_payload["init_vector"],
        json_payload["cipher_text"],
        json_payload["auth_tag"],
    )
def test_has_value(self, ping_name: Optional[str] = None) ‑> bool

Tests whether a value is stored for the metric for testing purposes only.

Args

ping_name : str
(default: first value in send_in_pings) The name of the ping to retrieve the metric for.

Returns

has_value (bool): True if the metric value exists.

Expand source code
def test_has_value(self, ping_name: Optional[str] = None) -> bool:
    """
    Tests whether a value is stored for the metric for testing purposes
    only.

    Args:
        ping_name (str): (default: first value in send_in_pings) The name
            of the ping to retrieve the metric for.

    Returns:
        has_value (bool): True if the metric value exists.
    """
    if ping_name is None:
        ping_name = self._send_in_pings[0]

    return bool(
        _ffi.lib.glean_jwe_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )