turbo_broccoli.custom.secret

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