JwtPlugin

JwtPlugin

JwtPlugin(verifier)

Bases: BasePlugin

JWT interception plugin.

Patches jwt.encode and jwt.decode at the module level. Uses reference counting so nested sandboxes work correctly.

SECURITY: The key parameter is intentionally excluded from interaction details to prevent secret keys from appearing in test assertion output.

Source code in src/tripwire/plugins/jwt_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._queues: dict[str, deque[JwtMockConfig]] = {}
    self._registry_lock: threading.Lock = threading.Lock()

mock_encode

mock_encode(*, returns, raises=None, required=True)

Register a mock for jwt.encode().

Source code in src/tripwire/plugins/jwt_plugin.py

def mock_encode(
    self,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for jwt.encode()."""
    config = JwtMockConfig(
        operation="encode", returns=returns, raises=raises, required=required
    )
    with self._registry_lock:
        if "encode" not in self._queues:
            self._queues["encode"] = deque()
        self._queues["encode"].append(config)

mock_decode

mock_decode(*, returns, raises=None, required=True)

Register a mock for jwt.decode().

Source code in src/tripwire/plugins/jwt_plugin.py

def mock_decode(
    self,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for jwt.decode()."""
    config = JwtMockConfig(
        operation="decode", returns=returns, raises=raises, required=required
    )
    with self._registry_lock:
        if "decode" not in self._queues:
            self._queues["decode"] = deque()
        self._queues["decode"].append(config)

install_patches

install_patches()

Install jwt.encode and jwt.decode patches.

Source code in src/tripwire/plugins/jwt_plugin.py

def install_patches(self) -> None:
    """Install jwt.encode and jwt.decode patches."""
    if not _JWT_AVAILABLE:
        raise ImportError(
            "Install python-tripwire[jwt] to use JwtPlugin: pip install python-tripwire[jwt]"
        )
    JwtPlugin._original_encode = jwt_lib.encode
    JwtPlugin._original_decode = jwt_lib.decode
    setattr(jwt_lib, "encode", _patched_encode)
    setattr(jwt_lib, "decode", _patched_decode)

restore_patches

restore_patches()

Restore original jwt.encode and jwt.decode.

Source code in src/tripwire/plugins/jwt_plugin.py

def restore_patches(self) -> None:
    """Restore original jwt.encode and jwt.decode."""
    if JwtPlugin._original_encode is not None:
        jwt_lib.encode = JwtPlugin._original_encode
        JwtPlugin._original_encode = None
    if JwtPlugin._original_decode is not None:
        jwt_lib.decode = JwtPlugin._original_decode
        JwtPlugin._original_decode = None

assert_encode

assert_encode(*, payload, algorithm, extra_kwargs=None, **extra)

Assert the next jwt.encode() interaction.

Source code in src/tripwire/plugins/jwt_plugin.py

def assert_encode(
    self, *, payload: dict[str, Any], algorithm: str | None,
    extra_kwargs: dict[str, Any] | None = None,
    **extra: Any,  # noqa: ANN401
) -> None:
    """Assert the next jwt.encode() interaction."""
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    sentinel = _JwtSentinel("encode")
    actual_extra_kwargs = extra_kwargs if extra_kwargs is not None else {}
    _get_test_verifier_or_raise().assert_interaction(
        sentinel, payload=payload, algorithm=algorithm,
        extra_kwargs=actual_extra_kwargs, **extra,
    )

assert_decode

assert_decode(*, token, algorithms, options=None, **extra)

Assert the next jwt.decode() interaction.

Source code in src/tripwire/plugins/jwt_plugin.py

def assert_decode(  # noqa: ANN401
    self, *, token: str | bytes, algorithms: Any,  # noqa: ANN401
    options: Any = None, **extra: Any,  # noqa: ANN401
) -> None:
    """Assert the next jwt.decode() interaction."""
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    sentinel = _JwtSentinel("decode")
    _get_test_verifier_or_raise().assert_interaction(
        sentinel, token=token, algorithms=algorithms, options=options, **extra
    )