Core

pyvider.rpcplugin.handshake.core

Core handshake functionality for the RPC plugin system.

This module contains the primary handshake configuration, validation, building, and parsing logic.

Classes

HandshakeConfig

Attributes:

Name	Type	Description
magic_cookie_key	str	The expected environment key for the handshake cookie.
magic_cookie_value	str	The expected handshake cookie value.
protocol_versions	list[int]	A list of protocol versions supported by the server.
supported_transports	list[str]	A list of supported transport types (e.g. "tcp", "unix").

Functions

build_handshake_response async

build_handshake_response(
    plugin_version: int,
    transport_name: str,
    transport: TransportT,
    server_cert: Certificate | None = None,
    port: int | None = None,
) -> str

CORE_VERSION|PLUGIN_VERSION|NETWORK|ADDRESS|PROTOCOL|TLS_CERT

Note: For TCP transport, the ADDRESS 127.0.0.1 is standard for same-host plugin communication, ensuring the plugin host connects to the plugin locally. The actual listening interface might be broader (e.g., 0.0.0.0), but the handshake communicates 127.0.0.1 for the host to connect to.

Parameters:

Name	Type	Description	Default
plugin_version	int	The version of the plugin.	required
transport_name	str	The name of the transport ("tcp" or "unix").	required
transport	TransportT	The transport instance.	required
server_cert	Certificate | None	Optional server certificate for TLS.	None
port	int | None	Optional port number, required for TCP transport.	None

Returns:

Type	Description
str	The constructed handshake response string.

Raises:

Type	Description
ValueError	If required parameters are missing (e.g., port for TCP).
TransportError	If an unsupported transport type is given.
Exception	Propagates exceptions from underlying operations.

Source code in pyvider/rpcplugin/handshake/core.py

@resilient(
    context={"operation": "build_handshake_response", "component": "handshake"},
    log_errors=True,
)
async def build_handshake_response(
    plugin_version: int,
    transport_name: str,
    transport: TransportT,
    server_cert: Certificate | None = None,
    port: int | None = None,
) -> str:
    """
    CORE_VERSION|PLUGIN_VERSION|NETWORK|ADDRESS|PROTOCOL|TLS_CERT

    Note: For TCP transport, the ADDRESS `127.0.0.1` is standard for same-host
    plugin communication, ensuring the plugin host connects to the plugin
    locally. The actual listening interface might be broader (e.g., `0.0.0.0`),
    but the handshake communicates `127.0.0.1` for the host to connect to.

    Args:
        plugin_version: The version of the plugin.
        transport_name: The name of the transport ("tcp" or "unix").
        transport: The transport instance.
        server_cert: Optional server certificate for TLS.
        port: Optional port number, required for TCP transport.

    Returns:
        The constructed handshake response string.

    Raises:
        ValueError: If required parameters are missing (e.g., port for TCP).
        TransportError: If an unsupported transport type is given.
        Exception: Propagates exceptions from underlying operations.
    """
    if _tracer:
        with _tracer.start_as_current_span("rpc.handshake.build_response") as span:
            span.set_attribute("transport", transport_name)
            span.set_attribute("plugin_version", plugin_version)
            span.set_attribute("has_cert", server_cert is not None)
            return await _build_handshake_response_impl(
                plugin_version, transport_name, transport, server_cert, port
            )
    else:
        return await _build_handshake_response_impl(
            plugin_version, transport_name, transport, server_cert, port
        )

is_valid_handshake_parts

is_valid_handshake_parts(
    parts: list[str],
) -> TypeGuard[list[str]]

Ensures it contains exactly 6 parts and the first two parts are digits.

Source code in pyvider/rpcplugin/handshake/core.py

def is_valid_handshake_parts(parts: list[str]) -> TypeGuard[list[str]]:
    """
    Ensures it contains exactly 6 parts and the first two parts are digits.
    """
    return len(parts) == 6 and parts[0].isdigit() and parts[1].isdigit()

parse_handshake_response

parse_handshake_response(
    response: str,
) -> tuple[int, int, str, str, str, str | None]

(📡🔍 Handshake Parsing) Parses the handshake response string. Expected Format: CORE_VERSION|PLUGIN_VERSION|NETWORK|ADDRESS|PROTOCOL|TLS_CERT

Parameters:

Name	Type	Description	Default
response	str	The handshake response string to parse.	required

Returns:

Type	Description
tuple[int, int, str, str, str, str | None]	A tuple containing: - core_version (int) - plugin_version (int) - network (str) - address (str) - protocol (str) - server_cert (str | None)

Raises:

Type	Description
HandshakeError	If parsing fails or the format is invalid.
ValueError	If parts of the handshake string are invalid (e.g., non-integer versions).

Source code in pyvider/rpcplugin/handshake/core.py

@resilient(
    context={"operation": "parse_handshake_response", "component": "handshake"},
    log_errors=True,
)
def parse_handshake_response(
    response: str,
) -> tuple[int, int, str, str, str, str | None]:
    """
    (📡🔍 Handshake Parsing) Parses the handshake response string.
    Expected Format: CORE_VERSION|PLUGIN_VERSION|NETWORK|ADDRESS|PROTOCOL|TLS_CERT

    Args:
        response: The handshake response string to parse.

    Returns:
        A tuple containing:
            - core_version (int)
            - plugin_version (int)
            - network (str)
            - address (str)
            - protocol (str)
            - server_cert (str | None)

    Raises:
        HandshakeError: If parsing fails or the format is invalid.
        ValueError: If parts of the handshake string are invalid
                    (e.g., non-integer versions).
    """
    if _tracer:
        with _tracer.start_as_current_span("rpc.handshake.parse_response") as span:
            # Only set response_length if response is a string
            if isinstance(response, str):
                span.set_attribute("response_length", len(response))
            return _parse_handshake_response_impl(response, span)
    else:
        return _parse_handshake_response_impl(response, None)

validate_magic_cookie

validate_magic_cookie(
    magic_cookie_key: (
        str | None | _SentinelType
    ) = _SENTINEL_INSTANCE,
    magic_cookie_value: (
        str | None | _SentinelType
    ) = _SENTINEL_INSTANCE,
    magic_cookie: (
        str | None | _SentinelType
    ) = _SENTINEL_INSTANCE,
) -> None

Validates the magic cookie.

If a parameter is omitted (i.e. remains as the sentinel), its value is read from rpcplugin_config. However, if the caller explicitly passes None, that is treated as missing and an error is raised.

Parameters:

Name	Type	Description	Default
magic_cookie_key	str | None | _SentinelType	The environment key for the magic cookie.	_SENTINEL_INSTANCE
magic_cookie_value	str | None | _SentinelType	The expected value of the magic cookie.	_SENTINEL_INSTANCE
magic_cookie	str | None | _SentinelType	The actual cookie value provided.	_SENTINEL_INSTANCE

Raises:

Type	Description
HandshakeError	If cookie validation fails.

Source code in pyvider/rpcplugin/handshake/core.py

@resilient(
    context={"operation": "validate_magic_cookie", "component": "handshake"},
    log_errors=True,
)
def validate_magic_cookie(
    magic_cookie_key: str | None | _SentinelType = _SENTINEL_INSTANCE,
    magic_cookie_value: str | None | _SentinelType = _SENTINEL_INSTANCE,
    magic_cookie: str | None | _SentinelType = _SENTINEL_INSTANCE,
) -> None:
    """
    Validates the magic cookie.

    If a parameter is omitted (i.e. remains as the sentinel),
    its value is read from rpcplugin_config. However, if the caller
    explicitly passes None, that is treated as missing and an error is raised.

    Args:
        magic_cookie_key: The environment key for the magic cookie.
        magic_cookie_value: The expected value of the magic cookie.
        magic_cookie: The actual cookie value provided.

    Raises:
        HandshakeError: If cookie validation fails.
    """
    if _tracer:
        with _tracer.start_as_current_span("rpc.handshake.validate_cookie") as span:
            span.set_attribute("component", "handshake")
            _validate_magic_cookie_impl(magic_cookie_key, magic_cookie_value, magic_cookie)
    else:
        _validate_magic_cookie_impl(magic_cookie_key, magic_cookie_value, magic_cookie)