component_framework

44class CorruptStateError(ComponentError):
45    """Raised when inbound state fails signature verification.
46
47    Covers tampered payloads, tampered/invalid MACs, malformed or truncated
48    tokens, unsigned input (plain JSON or raw dicts) while signing is
49    enabled, and tokens signed with an unknown key.
50    """

 64class StateSigner:
 65    """HMAC-SHA256 signer/verifier for serialized component state.
 66
 67    Key resolution order:
 68        1. Keys set via :meth:`configure` (including an explicit ``None`` to
 69           disable signing regardless of the environment).
 70        2. The ``STATE_SIGNING_KEY`` environment variable (comma-separated
 71           values enable rotation).
 72        3. Nothing configured: signing is disabled (legacy pass-through) and
 73           a one-time warning is logged on first serialization.
 74
 75    Rotation: the first key signs, all keys verify. To rotate, prepend the
 76    new key, keep the old key until in-flight client states have expired,
 77    then drop it.
 78    """
 79
 80    _keys: ClassVar[tuple[bytes, ...] | None] = None
 81    _configured: ClassVar[bool] = False
 82    _warned_unsigned: ClassVar[bool] = False
 83
 84    # ---------- Configuration ----------
 85
 86    @classmethod
 87    def configure(cls, secret: SecretInput) -> None:
 88        """Configure signing keys.
 89
 90        Args:
 91            secret: A single key (``str`` or ``bytes``), a sequence of keys
 92                for rotation (first signs, all verify), or ``None`` to
 93                explicitly disable signing (overrides the environment
 94                variable).
 95
 96        Raises:
 97            ValueError: If a key is empty or an empty sequence is given.
 98        """
 99        cls._keys = None if secret is None else cls._normalize(secret)
100        cls._configured = True
101        cls._warned_unsigned = False
102
103    @classmethod
104    def reset(cls) -> None:
105        """Reset to the unconfigured default (env-var fallback).
106
107        Intended for tests and application teardown.
108        """
109        cls._keys = None
110        cls._configured = False
111        cls._warned_unsigned = False
112
113    @classmethod
114    def enabled(cls) -> bool:
115        """Return True when at least one signing key is available."""
116        return bool(cls._resolve_keys())
117
118    @classmethod
119    def _normalize(cls, secret: str | bytes | Sequence[str | bytes]) -> tuple[bytes, ...]:
120        """Normalize a secret (or sequence of secrets) into key bytes."""
121        raw: Sequence[str | bytes]
122        raw = [secret] if isinstance(secret, (str, bytes)) else list(secret)
123        if not raw:
124            raise ValueError("StateSigner.configure() requires at least one signing key")
125        keys: list[bytes] = []
126        for item in raw:
127            key = item.encode("utf-8") if isinstance(item, str) else bytes(item)
128            if not key:
129                raise ValueError("Signing keys must be non-empty")
130            keys.append(key)
131        return tuple(keys)
132
133    @classmethod
134    def _resolve_keys(cls) -> tuple[bytes, ...] | None:
135        """Resolve active keys from explicit config or the environment."""
136        if cls._configured:
137            return cls._keys
138        env_value = os.environ.get(ENV_VAR, "")
139        parts = [part.strip() for part in env_value.split(",") if part.strip()]
140        if not parts:
141            return None
142        return tuple(part.encode("utf-8") for part in parts)
143
144    # ---------- Signing / verification ----------
145
146    @classmethod
147    def sign(cls, payload_json: str) -> str:
148        """Sign a JSON payload string into a ``cfs1`` token.
149
150        Args:
151            payload_json: The serialized (JSON) state to protect.
152
153        Returns:
154            A ``cfs1.<payload_b64>.<mac_b64>`` token string.
155
156        Raises:
157            ComponentError: If no signing key is configured.
158        """
159        keys = cls._resolve_keys()
160        if not keys:
161            raise ComponentError("StateSigner.sign() called but no signing key is configured")
162        payload_b64 = _b64url_encode(payload_json.encode("utf-8"))
163        mac = hmac.new(keys[0], cls._signing_input(payload_b64), hashlib.sha256).digest()
164        return f"{TOKEN_VERSION}.{payload_b64}.{_b64url_encode(mac)}"
165
166    @classmethod
167    def verify(cls, token: str) -> str:
168        """Verify a ``cfs1`` token and return the embedded JSON payload.
169
170        All configured keys are tried (rotation support), using
171        :func:`hmac.compare_digest` for constant-time comparison.
172
173        Args:
174            token: The inbound token string.
175
176        Returns:
177            The verified JSON payload string.
178
179        Raises:
180            CorruptStateError: If the token is malformed, unsigned, tampered
181                with, or signed with an unknown key.
182            ComponentError: If no signing key is configured.
183        """
184        keys = cls._resolve_keys()
185        if not keys:
186            raise ComponentError("StateSigner.verify() called but no signing key is configured")
187
188        if not isinstance(token, str):
189            raise CorruptStateError(
190                f"Signed state token must be a string, got {type(token).__name__}"
191            )
192
193        parts = token.split(".")
194        if len(parts) != 3 or parts[0] != TOKEN_VERSION:
195            raise CorruptStateError("State token is malformed or unsigned")
196        _, payload_b64, mac_b64 = parts
197
198        try:
199            provided_mac = _b64url_decode(mac_b64)
200        except (binascii.Error, ValueError) as e:
201            raise CorruptStateError("State token MAC is not valid base64") from e
202
203        signing_input = cls._signing_input(payload_b64)
204        for key in keys:
205            expected_mac = hmac.new(key, signing_input, hashlib.sha256).digest()
206            if hmac.compare_digest(expected_mac, provided_mac):
207                try:
208                    return _b64url_decode(payload_b64).decode("utf-8")
209                except (binascii.Error, ValueError, UnicodeDecodeError) as e:
210                    raise CorruptStateError("State token payload is not decodable") from e
211
212        raise CorruptStateError("State signature verification failed")
213
214    @staticmethod
215    def _signing_input(payload_b64: str) -> bytes:
216        """Build the MAC input, binding the version prefix to the payload."""
217        return f"{TOKEN_VERSION}.{payload_b64}".encode("ascii")
218
219    # ---------- Diagnostics ----------
220
221    @classmethod
222    def warn_unsigned_once(cls) -> None:
223        """Log a one-time prominent warning that state is unsigned."""
224        if cls._warned_unsigned:
225            return
226        cls._warned_unsigned = True
227        logger.warning(
228            "SECURITY: component state signing is DISABLED — client-carried "
229            "state is unsigned and can be tampered with. Set the "
230            "%s environment variable or call StateSigner.configure() "
231            "with a secret key before deploying to production.",
232            ENV_VAR,
233        )