turbo_broccoli.custom.secret

  1# pylint: disable=missing-class-docstring
  2"""Serialize secrets"""
  3
  4import json
  5from typing import Any, NoReturn
  6
  7from nacl.secret import SecretBox
  8
  9from turbo_broccoli.context import Context
 10from turbo_broccoli.exceptions import TypeNotSupported
 11
 12
 13class Secret:
 14    """
 15    A wrapper for a basic Python variable whose value is considered to be
 16    secret. Similar API as [`pydantic`'s secret
 17    types](https://pydantic-docs.helpmanual.io/usage/types/#secret-types)
 18    """
 19
 20    _value: Any
 21
 22    def __eq__(self, __o: object) -> bool:
 23        return False
 24
 25    def __init__(self, value: Any) -> None:
 26        self._value = value
 27
 28    def __ne__(self, __o: object) -> bool:
 29        return False
 30
 31    def __repr__(self) -> str:
 32        return "--REDACTED--"
 33
 34    def __str__(self) -> str:
 35        return "--REDACTED--"
 36
 37    def get_secret_value(self) -> Any:
 38        """Self-explanatory"""
 39        return self._value
 40
 41
 42class LockedSecret(Secret):
 43    """
 44    Represented a secret that could not be decrypted because the shared key was
 45    not provided. The `get_secret_value` method always raises a `RuntimeError`.
 46    """
 47
 48    def __init__(self) -> None:
 49        super().__init__(None)
 50
 51    def get_secret_value(self) -> NoReturn:
 52        raise RuntimeError("Cannot get the secret value of a locked secret")
 53
 54
 55class SecretDict(Secret):
 56    def __init__(self, value: dict) -> None:
 57        super().__init__(value)
 58
 59
 60class SecretFloat(Secret):
 61    def __init__(self, value: float) -> None:
 62        super().__init__(value)
 63
 64
 65class SecretInt(Secret):
 66    def __init__(self, value: int) -> None:
 67        super().__init__(value)
 68
 69
 70class SecretList(Secret):
 71    def __init__(self, value: list) -> None:
 72        super().__init__(value)
 73
 74
 75class SecretStr(Secret):
 76    def __init__(self, value: str) -> None:
 77        super().__init__(value)
 78
 79
 80def _from_json_v2(dct: dict, ctx: Context) -> Any:
 81    if ctx.nacl_shared_key is None:
 82        return LockedSecret()
 83    box = SecretBox(ctx.nacl_shared_key)
 84    return json.loads(box.decrypt(dct["data"]).decode("utf-8"))
 85
 86
 87# pylint: disable=missing-function-docstring
 88def from_json(dct: dict, ctx: Context) -> Any:
 89    ctx.raise_if_nodecode("bytes")
 90    decoders = {
 91        # 1: _from_json_v1,  # Use turbo_broccoli v3
 92        2: _from_json_v2,
 93    }
 94    obj = decoders[dct["__version__"]](dct, ctx)
 95    if isinstance(obj, LockedSecret):
 96        return obj
 97    types = {
 98        dict: SecretDict,
 99        float: SecretFloat,
100        int: SecretInt,
101        list: SecretList,
102        str: SecretStr,
103    }
104    return types[type(obj)](obj)
105
106
107def to_json(obj: Secret, ctx: Context) -> dict:
108    """
109    Encrypts a JSON **string representation** of a secret document into a
110    new JSON document with the following structure:
111
112    ```py
113    {
114        "__type__": "secret",
115        "__version__": 2,
116        "data": <encrypted bytes>,
117    }
118    ```
119    """
120    if not isinstance(obj, Secret):
121        raise TypeNotSupported()
122    if ctx.nacl_shared_key is None:
123        raise RuntimeError(
124            "Attempting to serialize a secret type but no shared key is set. "
125            "Either set `nacl_shared_key` when constructing the encoding "
126            "torbo_broccoli.context.Context, or set the TB_SHARED_KEY "
127            "environment variable."
128        )
129    box = SecretBox(ctx.nacl_shared_key)
130    return {
131        "__type__": "secret",
132        "__version__": 2,
133        "data": box.encrypt(
134            json.dumps(obj.get_secret_value()).encode("utf-8")
135        ),
136    }