Server API¶

SpindleX's server framework allows building custom SSH and SFTP server implementations.

SSH Server¶

`spindlex.server.ssh_server` ¶

SSH Server Implementation

Base class for SSH server implementations providing client authentication, channel management, and server-side SSH operations.

Classes¶

`SSHServer` ¶

Base SSH server implementation.

Provides hooks for authentication, authorization, and channel management that can be overridden to implement custom SSH server behavior.

Source code in spindlex/server/ssh_server.py

class SSHServer:
    """
    Base SSH server implementation.

    Provides hooks for authentication, authorization, and channel management
    that can be overridden to implement custom SSH server behavior.
    """

    def __init__(self) -> None:
        """Initialize SSH server with default settings."""
        self._server_key: Optional[PKey] = None
        self._transports: list[Transport] = []
        self._authenticated_users: dict[str, bool] = {}
        self._lock = threading.Lock()
        self._logger = get_logger("spindlex.server.ssh_server")

    def set_server_key(self, server_key: PKey) -> None:
        """
        Set the server's host key.

        Args:
            server_key: Server's private key for host authentication
        """
        self._server_key = server_key

    def get_server_key(self) -> Optional[PKey]:
        """
        Get the server's host key.

        Returns:
            Server's private key or None if not set
        """
        return self._server_key

    def start_server(
        self, sock: socket.socket, timeout: Optional[float] = None
    ) -> Transport:
        """
        Start server-side transport and handshake handling.

        Args:
            sock: Connected client socket
            timeout: Handshake timeout in seconds

        Returns:
            Transport instance for the connection

        Raises:
            TransportException: If server key not set or handshake fails
        """
        if self._server_key is None:
            raise TransportException("Server key must be set before starting server")

        # Create transport for this connection
        transport = Transport(sock)

        # Set server interface before starting so auth requests are never
        # received without an interface in place (MED-15)
        transport.set_server_interface(self)

        # Start server-side transport
        transport.start_server(self._server_key, timeout)

        with self._lock:
            self._transports.append(transport)
        return transport

    def check_auth_password(self, username: str, password: str) -> int:
        """
        Check password authentication.

        Override this method to implement custom password authentication logic.
        Default implementation rejects all password authentication attempts.

        Args:
            username: Username attempting authentication
            password: Password provided for authentication

        Returns:
            Authentication result code:
            - AUTH_SUCCESSFUL: Authentication successful
            - AUTH_FAILED: Authentication failed
            - AUTH_PARTIAL: Partial authentication (more methods required)
        """
        # Default implementation rejects all password authentication
        return AUTH_FAILED

    def check_auth_publickey(self, username: str, key: PKey) -> int:
        """
        Check public key authentication.

        Override this method to implement custom public key authentication logic.
        Default implementation rejects all public key authentication attempts.

        Args:
            username: Username attempting authentication
            key: Public key for authentication

        Returns:
            Authentication result code:
            - AUTH_SUCCESSFUL: Authentication successful
            - AUTH_FAILED: Authentication failed
            - AUTH_PARTIAL: Partial authentication (more methods required)
        """
        # Default implementation rejects all public key authentication
        return AUTH_FAILED

    def check_auth_keyboard_interactive(self, username: str, submethods: str) -> int:
        """
        Check keyboard-interactive authentication.

        Override this method to implement custom keyboard-interactive authentication.
        Default implementation rejects all keyboard-interactive authentication attempts.

        Args:
            username: Username attempting authentication
            submethods: Comma-separated list of submethods

        Returns:
            Authentication result code:
            - AUTH_SUCCESSFUL: Authentication successful
            - AUTH_FAILED: Authentication failed
            - AUTH_PARTIAL: Partial authentication (more methods required)
        """
        # Default implementation rejects all keyboard-interactive authentication
        return AUTH_FAILED

    def get_allowed_auths(self, username: str) -> list[str]:
        """
        Get list of allowed authentication methods for a user.

        Override this method to customize allowed authentication methods per user.
        Default implementation allows password and publickey authentication.

        Args:
            username: Username requesting authentication methods

        Returns:
            List of allowed authentication method names
        """
        return ["password", "publickey"]

    def check_auth_gssapi_with_mic(
        self, username: str, gss_authenticated: int, cc_file: str
    ) -> int:
        """
        Check GSSAPI authentication with MIC.

        Override this method to implement GSSAPI authentication.
        Default implementation rejects all GSSAPI authentication attempts.

        Args:
            username: Username attempting authentication
            gss_authenticated: GSSAPI authentication status
            cc_file: Credentials cache file

        Returns:
            Authentication result code
        """
        # Default implementation rejects GSSAPI authentication
        return AUTH_FAILED

    def check_port_forward_request(self, address: str, port: int) -> bool:
        """
        Check if port forwarding request should be allowed.
        Override this to implement custom forwarding policies.

        Args:
            address: Address to bind to
            port: Port to bind to

        Returns:
            True if request is allowed, False otherwise
        """
        # Default implementation rejects all port forwarding requests
        return False

    def check_port_forward_cancel_request(self, address: str, port: int) -> bool:
        """
        Check if port forwarding cancel request should be allowed.

        Args:
            address: Address to cancel forwarding for
            port: Port to cancel forwarding for

        Returns:
            True if request is allowed, False otherwise
        """
        # Usually safe to allow cancellation if they could request it
        return True

    def check_channel_request(self, kind: str, chanid: int) -> int:
        """
        Check channel creation request.

        Override this method to implement custom channel authorization logic.
        Default implementation allows session channels and rejects others.

        Args:
            kind: Type of channel requested (e.g., "session", "direct-tcpip")
            chanid: Channel ID for the request

        Returns:
            SSH channel open result code:
            - 0: Success (SSH_OPEN_CONNECT_SUCCESS)
            - 1: Administratively prohibited
            - 2: Connect failed
            - 3: Unknown channel type
            - 4: Resource shortage
        """
        # Default implementation allows session channels
        if kind == CHANNEL_SESSION:
            return 0  # SSH_OPEN_CONNECT_SUCCESS
        else:
            return SSH_OPEN_UNKNOWN_CHANNEL_TYPE

    def check_channel_exec_request(self, channel: Channel, command: bytes) -> bool:
        """
        Check command execution request.

        Override this method to implement custom command execution authorization.
        Default implementation rejects all command execution requests.

        Args:
            channel: Channel for command execution
            command: Command to be executed

        Returns:
            True if command execution is allowed, False otherwise
        """
        # Default implementation rejects all command execution
        return False

    def check_channel_shell_request(self, channel: Channel) -> bool:
        """
        Check shell access request.

        Override this method to implement custom shell access authorization.
        Default implementation rejects all shell access requests.

        Args:
            channel: Channel for shell access

        Returns:
            True if shell access is allowed, False otherwise
        """
        # Default implementation rejects all shell access
        return False

    def check_channel_subsystem_request(self, channel: Channel, name: str) -> bool:
        """
        Check subsystem request.

        Override this method to implement custom subsystem authorization.
        Default implementation rejects all subsystem requests.

        Args:
            channel: Channel for subsystem
            name: Name of the subsystem (e.g., "sftp")

        Returns:
            True if subsystem access is allowed, False otherwise
        """
        # Default implementation rejects all subsystem requests
        return False

    def check_channel_pty_request(
        self,
        channel: Channel,
        term: str,
        width: int,
        height: int,
        pixelwidth: int,
        pixelheight: int,
        modes: bytes,
    ) -> bool:
        """
        Check PTY allocation request.

        Override this method to implement custom PTY allocation authorization.
        Default implementation allows PTY allocation.

        Args:
            channel: Channel requesting PTY
            term: Terminal type (e.g., "xterm")
            width: Terminal width in characters
            height: Terminal height in characters
            pixelwidth: Terminal width in pixels
            pixelheight: Terminal height in pixels
            modes: Terminal modes

        Returns:
            True if PTY allocation is allowed, False otherwise
        """
        # Default implementation allows PTY allocation
        return True

    def check_channel_window_change_request(
        self,
        channel: Channel,
        width: int,
        height: int,
        pixelwidth: int,
        pixelheight: int,
    ) -> bool:
        """
        Check window change request.

        Override this method to implement custom window change authorization.
        Default implementation allows window changes.

        Args:
            channel: Channel requesting window change
            width: New terminal width in characters
            height: New terminal height in characters
            pixelwidth: New terminal width in pixels
            pixelheight: New terminal height in pixels

        Returns:
            True if window change is allowed, False otherwise
        """
        # Default implementation allows window changes
        return True

    def check_channel_x11_request(
        self,
        channel: Channel,
        single_connection: bool,
        auth_protocol: str,
        auth_cookie: bytes,
        screen_number: int,
    ) -> bool:
        """
        Check X11 forwarding request.

        Override this method to implement custom X11 forwarding authorization.
        Default implementation rejects X11 forwarding.

        Args:
            channel: Channel requesting X11 forwarding
            single_connection: Whether to allow only single connection
            auth_protocol: X11 authentication protocol
            auth_cookie: X11 authentication cookie
            screen_number: X11 screen number

        Returns:
            True if X11 forwarding is allowed, False otherwise
        """
        # Default implementation rejects X11 forwarding
        return False

    def check_channel_env_request(
        self, channel: Channel, name: str, value: str
    ) -> bool:
        """
        Check environment variable setting request.

        Override this method to implement custom environment variable authorization.
        Default implementation rejects environment variable setting.

        Args:
            channel: Channel requesting environment variable
            name: Environment variable name
            value: Environment variable value

        Returns:
            True if environment variable setting is allowed, False otherwise
        """
        # Default implementation rejects environment variable setting
        return False

    def get_banner(self) -> Optional[str]:
        """
        Get authentication banner message.

        Override this method to provide a custom banner message displayed
        to clients before authentication.

        Returns:
            Banner message string or None for no banner
        """
        return None

    def check_global_request(self, kind: str, msg: Any) -> bool:
        """
        Check global request.

        Override this method to implement custom global request handling.
        Default implementation rejects all global requests.

        Args:
            kind: Type of global request
            msg: Request message data

        Returns:
            True if global request is allowed, False otherwise
        """
        # Default implementation rejects all global requests
        return False

    # Server-side channel management methods

    def get_active_channels(self) -> list[Channel]:
        """
        Get list of all active channels across all connections.

        Returns:
            List of active Channel instances
        """
        channels = []
        with self._lock:
            # Clean up inactive transports
            self._transports = [t for t in self._transports if t.active]
            for transport in self._transports:
                channels.extend(list(transport._channels.values()))
        return channels

    def get_channel_count(self) -> int:
        """
        Get total number of active channels across all connections.

        Returns:
            Number of active channels
        """
        count = 0
        with self._lock:
            self._transports = [t for t in self._transports if t.active]
            for transport in self._transports:
                count += len(transport._channels)
        return count

    def close_channel(self, channel: Channel) -> None:
        """
        Close a specific channel.

        Args:
            channel: Channel to close
        """
        try:
            channel.close()
        except Exception as e:
            # Log errors during channel close
            self._logger.debug(f"Error during channel close: {e}")

    def close_all_channels(self) -> None:
        """Close all active channels."""
        channels = self.get_active_channels()
        for channel in channels:
            self.close_channel(channel)

    def is_channel_authorized(self, channel: Channel, username: str) -> bool:
        """
        Check if a channel is authorized for a specific user.

        Override this method to implement custom channel authorization logic.
        Default implementation allows all channels for authenticated users.

        Args:
            channel: Channel to check
            username: Username to check authorization for

        Returns:
            True if channel is authorized for the user
        """
        # Default implementation allows all channels for authenticated users
        return (
            username in self._authenticated_users
            and self._authenticated_users[username]
        )

    def on_channel_opened(self, channel: Channel) -> None:
        """
        Called when a new channel is opened.

        Override this method to implement custom channel open handling.

        Args:
            channel: Newly opened channel
        """
        # Default implementation does nothing
        pass

    def on_channel_closed(self, channel: Channel) -> None:
        """
        Called when a channel is closed.

        Override this method to implement custom channel close handling.

        Args:
            channel: Closed channel
        """
        # Default implementation does nothing
        pass

    def on_authentication_successful(self, username: str, method: str) -> None:
        """
        Called when authentication is successful.

        Override this method to implement custom authentication success handling.

        Args:
            username: Successfully authenticated username
            method: Authentication method used
        """
        with self._lock:
            self._authenticated_users[username] = True

    def on_authentication_failed(self, username: str, method: str) -> None:
        """
        Called when authentication fails.

        Override this method to implement custom authentication failure handling.

        Args:
            username: Username that failed authentication
            method: Authentication method that failed
        """
        # Default implementation does nothing
        pass

Methods:¶

`init()` ¶

Initialize SSH server with default settings.

Source code in spindlex/server/ssh_server.py

def __init__(self) -> None:
    """Initialize SSH server with default settings."""
    self._server_key: Optional[PKey] = None
    self._transports: list[Transport] = []
    self._authenticated_users: dict[str, bool] = {}
    self._lock = threading.Lock()
    self._logger = get_logger("spindlex.server.ssh_server")

`check_auth_gssapi_with_mic(username, gss_authenticated, cc_file)` ¶

Check GSSAPI authentication with MIC.

Override this method to implement GSSAPI authentication. Default implementation rejects all GSSAPI authentication attempts.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username attempting authentication	required
`gss_authenticated`	`int`	GSSAPI authentication status	required
`cc_file`	`str`	Credentials cache file	required

Returns:

Type	Description
`int`	Authentication result code

Source code in spindlex/server/ssh_server.py

def check_auth_gssapi_with_mic(
    self, username: str, gss_authenticated: int, cc_file: str
) -> int:
    """
    Check GSSAPI authentication with MIC.

    Override this method to implement GSSAPI authentication.
    Default implementation rejects all GSSAPI authentication attempts.

    Args:
        username: Username attempting authentication
        gss_authenticated: GSSAPI authentication status
        cc_file: Credentials cache file

    Returns:
        Authentication result code
    """
    # Default implementation rejects GSSAPI authentication
    return AUTH_FAILED

`check_auth_keyboard_interactive(username, submethods)` ¶

Check keyboard-interactive authentication.

Override this method to implement custom keyboard-interactive authentication. Default implementation rejects all keyboard-interactive authentication attempts.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username attempting authentication	required
`submethods`	`str`	Comma-separated list of submethods	required

Returns:

Type	Description
`int`	Authentication result code:
`int`	AUTH_SUCCESSFUL: Authentication successful
`int`	AUTH_FAILED: Authentication failed
`int`	AUTH_PARTIAL: Partial authentication (more methods required)

Source code in spindlex/server/ssh_server.py

def check_auth_keyboard_interactive(self, username: str, submethods: str) -> int:
    """
    Check keyboard-interactive authentication.

    Override this method to implement custom keyboard-interactive authentication.
    Default implementation rejects all keyboard-interactive authentication attempts.

    Args:
        username: Username attempting authentication
        submethods: Comma-separated list of submethods

    Returns:
        Authentication result code:
        - AUTH_SUCCESSFUL: Authentication successful
        - AUTH_FAILED: Authentication failed
        - AUTH_PARTIAL: Partial authentication (more methods required)
    """
    # Default implementation rejects all keyboard-interactive authentication
    return AUTH_FAILED

`check_auth_password(username, password)` ¶

Check password authentication.

Override this method to implement custom password authentication logic. Default implementation rejects all password authentication attempts.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username attempting authentication	required
`password`	`str`	Password provided for authentication	required

Returns:

Type	Description
`int`	Authentication result code:
`int`	AUTH_SUCCESSFUL: Authentication successful
`int`	AUTH_FAILED: Authentication failed
`int`	AUTH_PARTIAL: Partial authentication (more methods required)

Source code in spindlex/server/ssh_server.py

def check_auth_password(self, username: str, password: str) -> int:
    """
    Check password authentication.

    Override this method to implement custom password authentication logic.
    Default implementation rejects all password authentication attempts.

    Args:
        username: Username attempting authentication
        password: Password provided for authentication

    Returns:
        Authentication result code:
        - AUTH_SUCCESSFUL: Authentication successful
        - AUTH_FAILED: Authentication failed
        - AUTH_PARTIAL: Partial authentication (more methods required)
    """
    # Default implementation rejects all password authentication
    return AUTH_FAILED

`check_auth_publickey(username, key)` ¶

Check public key authentication.

Override this method to implement custom public key authentication logic. Default implementation rejects all public key authentication attempts.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username attempting authentication	required
`key`	`PKey`	Public key for authentication	required

Returns:

Type	Description
`int`	Authentication result code:
`int`	AUTH_SUCCESSFUL: Authentication successful
`int`	AUTH_FAILED: Authentication failed
`int`	AUTH_PARTIAL: Partial authentication (more methods required)

Source code in spindlex/server/ssh_server.py

def check_auth_publickey(self, username: str, key: PKey) -> int:
    """
    Check public key authentication.

    Override this method to implement custom public key authentication logic.
    Default implementation rejects all public key authentication attempts.

    Args:
        username: Username attempting authentication
        key: Public key for authentication

    Returns:
        Authentication result code:
        - AUTH_SUCCESSFUL: Authentication successful
        - AUTH_FAILED: Authentication failed
        - AUTH_PARTIAL: Partial authentication (more methods required)
    """
    # Default implementation rejects all public key authentication
    return AUTH_FAILED

`check_channel_env_request(channel, name, value)` ¶

Check environment variable setting request.

Override this method to implement custom environment variable authorization. Default implementation rejects environment variable setting.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel requesting environment variable	required
`name`	`str`	Environment variable name	required
`value`	`str`	Environment variable value	required

Returns:

Type	Description
`bool`	True if environment variable setting is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_env_request(
    self, channel: Channel, name: str, value: str
) -> bool:
    """
    Check environment variable setting request.

    Override this method to implement custom environment variable authorization.
    Default implementation rejects environment variable setting.

    Args:
        channel: Channel requesting environment variable
        name: Environment variable name
        value: Environment variable value

    Returns:
        True if environment variable setting is allowed, False otherwise
    """
    # Default implementation rejects environment variable setting
    return False

`check_channel_exec_request(channel, command)` ¶

Check command execution request.

Override this method to implement custom command execution authorization. Default implementation rejects all command execution requests.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel for command execution	required
`command`	`bytes`	Command to be executed	required

Returns:

Type	Description
`bool`	True if command execution is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_exec_request(self, channel: Channel, command: bytes) -> bool:
    """
    Check command execution request.

    Override this method to implement custom command execution authorization.
    Default implementation rejects all command execution requests.

    Args:
        channel: Channel for command execution
        command: Command to be executed

    Returns:
        True if command execution is allowed, False otherwise
    """
    # Default implementation rejects all command execution
    return False

`check_channel_pty_request(channel, term, width, height, pixelwidth, pixelheight, modes)` ¶

Check PTY allocation request.

Override this method to implement custom PTY allocation authorization. Default implementation allows PTY allocation.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel requesting PTY	required
`term`	`str`	Terminal type (e.g., "xterm")	required
`width`	`int`	Terminal width in characters	required
`height`	`int`	Terminal height in characters	required
`pixelwidth`	`int`	Terminal width in pixels	required
`pixelheight`	`int`	Terminal height in pixels	required
`modes`	`bytes`	Terminal modes	required

Returns:

Type	Description
`bool`	True if PTY allocation is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_pty_request(
    self,
    channel: Channel,
    term: str,
    width: int,
    height: int,
    pixelwidth: int,
    pixelheight: int,
    modes: bytes,
) -> bool:
    """
    Check PTY allocation request.

    Override this method to implement custom PTY allocation authorization.
    Default implementation allows PTY allocation.

    Args:
        channel: Channel requesting PTY
        term: Terminal type (e.g., "xterm")
        width: Terminal width in characters
        height: Terminal height in characters
        pixelwidth: Terminal width in pixels
        pixelheight: Terminal height in pixels
        modes: Terminal modes

    Returns:
        True if PTY allocation is allowed, False otherwise
    """
    # Default implementation allows PTY allocation
    return True

`check_channel_request(kind, chanid)` ¶

Check channel creation request.

Override this method to implement custom channel authorization logic. Default implementation allows session channels and rejects others.

Parameters:

Name	Type	Description	Default
`kind`	`str`	Type of channel requested (e.g., "session", "direct-tcpip")	required
`chanid`	`int`	Channel ID for the request	required

Returns:

Type	Description
`int`	SSH channel open result code:
`int`	0: Success (SSH_OPEN_CONNECT_SUCCESS)
`int`	1: Administratively prohibited
`int`	2: Connect failed
`int`	3: Unknown channel type
`int`	4: Resource shortage

Source code in spindlex/server/ssh_server.py

def check_channel_request(self, kind: str, chanid: int) -> int:
    """
    Check channel creation request.

    Override this method to implement custom channel authorization logic.
    Default implementation allows session channels and rejects others.

    Args:
        kind: Type of channel requested (e.g., "session", "direct-tcpip")
        chanid: Channel ID for the request

    Returns:
        SSH channel open result code:
        - 0: Success (SSH_OPEN_CONNECT_SUCCESS)
        - 1: Administratively prohibited
        - 2: Connect failed
        - 3: Unknown channel type
        - 4: Resource shortage
    """
    # Default implementation allows session channels
    if kind == CHANNEL_SESSION:
        return 0  # SSH_OPEN_CONNECT_SUCCESS
    else:
        return SSH_OPEN_UNKNOWN_CHANNEL_TYPE

`check_channel_shell_request(channel)` ¶

Check shell access request.

Override this method to implement custom shell access authorization. Default implementation rejects all shell access requests.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel for shell access	required

Returns:

Type	Description
`bool`	True if shell access is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_shell_request(self, channel: Channel) -> bool:
    """
    Check shell access request.

    Override this method to implement custom shell access authorization.
    Default implementation rejects all shell access requests.

    Args:
        channel: Channel for shell access

    Returns:
        True if shell access is allowed, False otherwise
    """
    # Default implementation rejects all shell access
    return False

`check_channel_subsystem_request(channel, name)` ¶

Check subsystem request.

Override this method to implement custom subsystem authorization. Default implementation rejects all subsystem requests.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel for subsystem	required
`name`	`str`	Name of the subsystem (e.g., "sftp")	required

Returns:

Type	Description
`bool`	True if subsystem access is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_subsystem_request(self, channel: Channel, name: str) -> bool:
    """
    Check subsystem request.

    Override this method to implement custom subsystem authorization.
    Default implementation rejects all subsystem requests.

    Args:
        channel: Channel for subsystem
        name: Name of the subsystem (e.g., "sftp")

    Returns:
        True if subsystem access is allowed, False otherwise
    """
    # Default implementation rejects all subsystem requests
    return False

`check_channel_window_change_request(channel, width, height, pixelwidth, pixelheight)` ¶

Check window change request.

Override this method to implement custom window change authorization. Default implementation allows window changes.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel requesting window change	required
`width`	`int`	New terminal width in characters	required
`height`	`int`	New terminal height in characters	required
`pixelwidth`	`int`	New terminal width in pixels	required
`pixelheight`	`int`	New terminal height in pixels	required

Returns:

Type	Description
`bool`	True if window change is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_window_change_request(
    self,
    channel: Channel,
    width: int,
    height: int,
    pixelwidth: int,
    pixelheight: int,
) -> bool:
    """
    Check window change request.

    Override this method to implement custom window change authorization.
    Default implementation allows window changes.

    Args:
        channel: Channel requesting window change
        width: New terminal width in characters
        height: New terminal height in characters
        pixelwidth: New terminal width in pixels
        pixelheight: New terminal height in pixels

    Returns:
        True if window change is allowed, False otherwise
    """
    # Default implementation allows window changes
    return True

`check_channel_x11_request(channel, single_connection, auth_protocol, auth_cookie, screen_number)` ¶

Check X11 forwarding request.

Override this method to implement custom X11 forwarding authorization. Default implementation rejects X11 forwarding.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel requesting X11 forwarding	required
`single_connection`	`bool`	Whether to allow only single connection	required
`auth_protocol`	`str`	X11 authentication protocol	required
`auth_cookie`	`bytes`	X11 authentication cookie	required
`screen_number`	`int`	X11 screen number	required

Returns:

Type	Description
`bool`	True if X11 forwarding is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_channel_x11_request(
    self,
    channel: Channel,
    single_connection: bool,
    auth_protocol: str,
    auth_cookie: bytes,
    screen_number: int,
) -> bool:
    """
    Check X11 forwarding request.

    Override this method to implement custom X11 forwarding authorization.
    Default implementation rejects X11 forwarding.

    Args:
        channel: Channel requesting X11 forwarding
        single_connection: Whether to allow only single connection
        auth_protocol: X11 authentication protocol
        auth_cookie: X11 authentication cookie
        screen_number: X11 screen number

    Returns:
        True if X11 forwarding is allowed, False otherwise
    """
    # Default implementation rejects X11 forwarding
    return False

`check_global_request(kind, msg)` ¶

Check global request.

Override this method to implement custom global request handling. Default implementation rejects all global requests.

Parameters:

Name	Type	Description	Default
`kind`	`str`	Type of global request	required
`msg`	`Any`	Request message data	required

Returns:

Type	Description
`bool`	True if global request is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_global_request(self, kind: str, msg: Any) -> bool:
    """
    Check global request.

    Override this method to implement custom global request handling.
    Default implementation rejects all global requests.

    Args:
        kind: Type of global request
        msg: Request message data

    Returns:
        True if global request is allowed, False otherwise
    """
    # Default implementation rejects all global requests
    return False

`check_port_forward_cancel_request(address, port)` ¶

Check if port forwarding cancel request should be allowed.

Parameters:

Name	Type	Description	Default
`address`	`str`	Address to cancel forwarding for	required
`port`	`int`	Port to cancel forwarding for	required

Returns:

Type	Description
`bool`	True if request is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_port_forward_cancel_request(self, address: str, port: int) -> bool:
    """
    Check if port forwarding cancel request should be allowed.

    Args:
        address: Address to cancel forwarding for
        port: Port to cancel forwarding for

    Returns:
        True if request is allowed, False otherwise
    """
    # Usually safe to allow cancellation if they could request it
    return True

`check_port_forward_request(address, port)` ¶

Check if port forwarding request should be allowed. Override this to implement custom forwarding policies.

Parameters:

Name	Type	Description	Default
`address`	`str`	Address to bind to	required
`port`	`int`	Port to bind to	required

Returns:

Type	Description
`bool`	True if request is allowed, False otherwise

Source code in spindlex/server/ssh_server.py

def check_port_forward_request(self, address: str, port: int) -> bool:
    """
    Check if port forwarding request should be allowed.
    Override this to implement custom forwarding policies.

    Args:
        address: Address to bind to
        port: Port to bind to

    Returns:
        True if request is allowed, False otherwise
    """
    # Default implementation rejects all port forwarding requests
    return False

`close_all_channels()` ¶

Close all active channels.

Source code in spindlex/server/ssh_server.py

def close_all_channels(self) -> None:
    """Close all active channels."""
    channels = self.get_active_channels()
    for channel in channels:
        self.close_channel(channel)

`close_channel(channel)` ¶

Close a specific channel.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel to close	required

Source code in spindlex/server/ssh_server.py

def close_channel(self, channel: Channel) -> None:
    """
    Close a specific channel.

    Args:
        channel: Channel to close
    """
    try:
        channel.close()
    except Exception as e:
        # Log errors during channel close
        self._logger.debug(f"Error during channel close: {e}")

`get_active_channels()` ¶

Get list of all active channels across all connections.

Returns:

Type	Description
`list[Channel]`	List of active Channel instances

Source code in spindlex/server/ssh_server.py

def get_active_channels(self) -> list[Channel]:
    """
    Get list of all active channels across all connections.

    Returns:
        List of active Channel instances
    """
    channels = []
    with self._lock:
        # Clean up inactive transports
        self._transports = [t for t in self._transports if t.active]
        for transport in self._transports:
            channels.extend(list(transport._channels.values()))
    return channels

`get_allowed_auths(username)` ¶

Get list of allowed authentication methods for a user.

Override this method to customize allowed authentication methods per user. Default implementation allows password and publickey authentication.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username requesting authentication methods	required

Returns:

Type	Description
`list[str]`	List of allowed authentication method names

Source code in spindlex/server/ssh_server.py

def get_allowed_auths(self, username: str) -> list[str]:
    """
    Get list of allowed authentication methods for a user.

    Override this method to customize allowed authentication methods per user.
    Default implementation allows password and publickey authentication.

    Args:
        username: Username requesting authentication methods

    Returns:
        List of allowed authentication method names
    """
    return ["password", "publickey"]

`get_banner()` ¶

Get authentication banner message.

Override this method to provide a custom banner message displayed to clients before authentication.

Returns:

Type	Description
`Optional[str]`	Banner message string or None for no banner

Source code in spindlex/server/ssh_server.py

def get_banner(self) -> Optional[str]:
    """
    Get authentication banner message.

    Override this method to provide a custom banner message displayed
    to clients before authentication.

    Returns:
        Banner message string or None for no banner
    """
    return None

`get_channel_count()` ¶

Get total number of active channels across all connections.

Returns:

Type	Description
`int`	Number of active channels

Source code in spindlex/server/ssh_server.py

def get_channel_count(self) -> int:
    """
    Get total number of active channels across all connections.

    Returns:
        Number of active channels
    """
    count = 0
    with self._lock:
        self._transports = [t for t in self._transports if t.active]
        for transport in self._transports:
            count += len(transport._channels)
    return count

`get_server_key()` ¶

Get the server's host key.

Returns:

Type	Description
`Optional[PKey]`	Server's private key or None if not set

Source code in spindlex/server/ssh_server.py

def get_server_key(self) -> Optional[PKey]:
    """
    Get the server's host key.

    Returns:
        Server's private key or None if not set
    """
    return self._server_key

`is_channel_authorized(channel, username)` ¶

Check if a channel is authorized for a specific user.

Override this method to implement custom channel authorization logic. Default implementation allows all channels for authenticated users.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Channel to check	required
`username`	`str`	Username to check authorization for	required

Returns:

Type	Description
`bool`	True if channel is authorized for the user

Source code in spindlex/server/ssh_server.py

def is_channel_authorized(self, channel: Channel, username: str) -> bool:
    """
    Check if a channel is authorized for a specific user.

    Override this method to implement custom channel authorization logic.
    Default implementation allows all channels for authenticated users.

    Args:
        channel: Channel to check
        username: Username to check authorization for

    Returns:
        True if channel is authorized for the user
    """
    # Default implementation allows all channels for authenticated users
    return (
        username in self._authenticated_users
        and self._authenticated_users[username]
    )

`on_authentication_failed(username, method)` ¶

Called when authentication fails.

Override this method to implement custom authentication failure handling.

Parameters:

Name	Type	Description	Default
`username`	`str`	Username that failed authentication	required
`method`	`str`	Authentication method that failed	required

Source code in spindlex/server/ssh_server.py

def on_authentication_failed(self, username: str, method: str) -> None:
    """
    Called when authentication fails.

    Override this method to implement custom authentication failure handling.

    Args:
        username: Username that failed authentication
        method: Authentication method that failed
    """
    # Default implementation does nothing
    pass

`on_authentication_successful(username, method)` ¶

Called when authentication is successful.

Override this method to implement custom authentication success handling.

Parameters:

Name	Type	Description	Default
`username`	`str`	Successfully authenticated username	required
`method`	`str`	Authentication method used	required

Source code in spindlex/server/ssh_server.py

def on_authentication_successful(self, username: str, method: str) -> None:
    """
    Called when authentication is successful.

    Override this method to implement custom authentication success handling.

    Args:
        username: Successfully authenticated username
        method: Authentication method used
    """
    with self._lock:
        self._authenticated_users[username] = True

`on_channel_closed(channel)` ¶

Called when a channel is closed.

Override this method to implement custom channel close handling.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Closed channel	required

Source code in spindlex/server/ssh_server.py

def on_channel_closed(self, channel: Channel) -> None:
    """
    Called when a channel is closed.

    Override this method to implement custom channel close handling.

    Args:
        channel: Closed channel
    """
    # Default implementation does nothing
    pass

`on_channel_opened(channel)` ¶

Called when a new channel is opened.

Override this method to implement custom channel open handling.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	Newly opened channel	required

Source code in spindlex/server/ssh_server.py

def on_channel_opened(self, channel: Channel) -> None:
    """
    Called when a new channel is opened.

    Override this method to implement custom channel open handling.

    Args:
        channel: Newly opened channel
    """
    # Default implementation does nothing
    pass

`set_server_key(server_key)` ¶

Set the server's host key.

Parameters:

Name	Type	Description	Default
`server_key`	`PKey`	Server's private key for host authentication	required

Source code in spindlex/server/ssh_server.py

def set_server_key(self, server_key: PKey) -> None:
    """
    Set the server's host key.

    Args:
        server_key: Server's private key for host authentication
    """
    self._server_key = server_key

`start_server(sock, timeout=None)` ¶

Start server-side transport and handshake handling.

Parameters:

Name	Type	Description	Default
`sock`	`socket`	Connected client socket	required
`timeout`	`Optional[float]`	Handshake timeout in seconds	`None`

Returns:

Type	Description
`Transport`	Transport instance for the connection

Raises:

Type	Description
`TransportException`	If server key not set or handshake fails

Source code in spindlex/server/ssh_server.py

def start_server(
    self, sock: socket.socket, timeout: Optional[float] = None
) -> Transport:
    """
    Start server-side transport and handshake handling.

    Args:
        sock: Connected client socket
        timeout: Handshake timeout in seconds

    Returns:
        Transport instance for the connection

    Raises:
        TransportException: If server key not set or handshake fails
    """
    if self._server_key is None:
        raise TransportException("Server key must be set before starting server")

    # Create transport for this connection
    transport = Transport(sock)

    # Set server interface before starting so auth requests are never
    # received without an interface in place (MED-15)
    transport.set_server_interface(self)

    # Start server-side transport
    transport.start_server(self._server_key, timeout)

    with self._lock:
        self._transports.append(transport)
    return transport

`SSHServerManager` ¶

SSH Server Manager for handling multiple client connections.

Manages server lifecycle, multi-client connections, and resource cleanup.

Source code in spindlex/server/ssh_server.py

class SSHServerManager:
    """
    SSH Server Manager for handling multiple client connections.

    Manages server lifecycle, multi-client connections, and resource cleanup.
    """

    def __init__(
        self,
        server_interface: SSHServer,
        server_key: PKey,
        bind_address: str = "0.0.0.0",  # nosec B104
        port: int = 22,
    ) -> None:
        """
        Initialize SSH server manager.

        Args:
            server_interface: SSHServer instance for handling client requests
            server_key: Server's private key for host authentication
            bind_address: Address to bind server socket
            port: Port to bind server socket
        """
        self._server_interface = server_interface
        self._server_key = server_key
        self._bind_address = bind_address
        self._port = port

        # Ensure server interface has the host key immediately
        self._server_interface.set_server_key(self._server_key)

        self._server_socket: Optional[socket.socket] = None
        self._running = False
        self._connections: dict[str, Transport] = {}  # connection_id -> transport
        self._connection_threads: dict[str, threading.Thread] = {}
        self._lock = threading.RLock()
        self._accept_thread: Optional[threading.Thread] = None

        # Connection limits and timeouts
        self._max_connections = 100
        self._connection_timeout = 30.0
        self._auth_timeout = 30.0

        # Statistics
        self._total_connections = 0
        self._active_connections = 0
        self._failed_connections = 0
        self._logger = get_logger("spindlex.server.ssh_server")

    def set_max_connections(self, max_connections: int) -> None:
        """
        Set maximum number of concurrent connections.

        Args:
            max_connections: Maximum number of concurrent connections
        """
        self._max_connections = max_connections

    def set_connection_timeout(self, timeout: float) -> None:
        """
        Set connection timeout.

        Args:
            timeout: Connection timeout in seconds
        """
        self._connection_timeout = timeout

    def set_auth_timeout(self, timeout: float) -> None:
        """
        Set authentication timeout.

        Args:
            timeout: Authentication timeout in seconds
        """
        self._auth_timeout = timeout

    def start_server(self) -> None:
        """
        Start SSH server and begin accepting connections.

        Raises:
            TransportException: If server fails to start
        """
        with self._lock:
            if self._running:
                raise TransportException("Server is already running")

            try:
                addr_info = socket.getaddrinfo(
                    self._bind_address,
                    self._port,
                    socket.AF_UNSPEC,
                    socket.SOCK_STREAM,
                    0,
                    socket.AI_PASSIVE,
                )
                if not addr_info:
                    raise TransportException(
                        f"Could not resolve bind address: {self._bind_address}"
                    )

                # Use the first available address info
                af, socktype, proto, canonname, sa = addr_info[0]

                # Create and bind server socket
                self._server_socket = socket.socket(af, socktype, proto)
                self._server_socket.setsockopt(
                    socket.SOL_SOCKET, socket.SO_REUSEADDR, 1
                )

                # For IPv6, try to enable dual-stack if bind_address is empty or ::
                if af == socket.AF_INET6:
                    try:
                        self._server_socket.setsockopt(
                            socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0
                        )
                    except (AttributeError, OSError):
                        pass

                self._server_socket.bind(sa)
                self._server_socket.listen(socket.SOMAXCONN)

                self._running = True

                # Start accept thread
                self._accept_thread = threading.Thread(
                    target=self._accept_connections,
                    name="SSHServer-Accept",
                    daemon=True,
                )
                self._accept_thread.start()

            except Exception as e:
                self._cleanup_server_socket()
                raise TransportException(f"Failed to start SSH server: {e}") from e

    def stop_server(self) -> None:
        """
        Stop SSH server and close all connections.
        """
        with self._lock:
            if not self._running:
                return

            self._running = False

        # Close server socket to stop accepting new connections
        self._cleanup_server_socket()

        # Wait for accept thread to finish
        if self._accept_thread and self._accept_thread.is_alive():
            self._accept_thread.join(timeout=5.0)

        # Close all active connections
        self._close_all_connections()

    def _accept_connections(self) -> None:
        """Accept incoming connections in a loop."""
        while self._running:
            try:
                if self._server_socket is None:
                    break

                # Accept new connection
                client_socket, client_address = self._server_socket.accept()

                # Check connection limits
                with self._lock:
                    if len(self._connections) >= self._max_connections:
                        client_socket.close()
                        continue

                    self._total_connections += 1
                    # Handle IPv6 address tuples (host, port, flowinfo, scopeid)
                    client_host = client_address[0]
                    client_port = client_address[1]
                    connection_id = (
                        f"{client_host}:{client_port}:{self._total_connections}"
                    )

                # Handle connection in separate thread
                connection_thread = threading.Thread(
                    target=self._handle_connection,
                    args=(client_socket, client_address, connection_id),
                    name=f"SSHServer-{connection_id}",
                    daemon=True,
                )

                with self._lock:
                    self._connection_threads[connection_id] = connection_thread

                connection_thread.start()

            except Exception as e:
                if self._running:
                    self._logger.error(f"Error accepting connection: {e}")
                    time.sleep(0.1)  # Avoid tight loop on persistent error
                else:
                    # Server is shutting down
                    break  # type: ignore[unreachable]

    def _handle_connection(
        self, client_socket: socket.socket, client_address: tuple, connection_id: str
    ) -> None:
        """
        Handle individual client connection.

        Args:
            client_socket: Client socket
            client_address: Client address tuple
            connection_id: Unique connection identifier
        """
        transport = None
        try:
            with self._lock:
                self._active_connections += 1

            # Set socket timeout
            client_socket.settimeout(self._connection_timeout)

            # Start server transport
            transport = self._server_interface.start_server(
                client_socket, self._auth_timeout
            )

            with self._lock:
                self._connections[connection_id] = transport

            # Keep connection alive and process messages
            while transport.active:
                try:
                    # _pump will handle internal messages (auth, service, channels)
                    # and queue any other messages for other threads to pick up.
                    transport._pump()
                except (socket.timeout, TransportException):
                    # Check if still active
                    if not transport.active:
                        break  # type: ignore[unreachable]
                    continue
                except Exception as e:
                    self._logger.debug(
                        f"Connection loop error for {connection_id}: {e}"
                    )
                    break

        except Exception as e:
            self._logger.error(
                f"Error handling connection {connection_id}: {e}", exc_info=True
            )
            with self._lock:
                self._failed_connections += 1

        finally:
            # Cleanup connection
            self._cleanup_connection(connection_id, transport, client_socket)

    def _cleanup_connection(
        self,
        connection_id: str,
        transport: Optional[Transport],
        client_socket: socket.socket,
    ) -> None:
        """
        Clean up connection resources.

        Args:
            connection_id: Connection identifier
            transport: Transport instance (may be None)
            client_socket: Client socket
        """
        try:
            # Close transport
            if transport:
                transport.close()

            # Close socket
            client_socket.close()

        except Exception as e:
            # Ignore cleanup errors
            self._logger.debug(f"Cleanup error: {e}")
        finally:
            with self._lock:
                # Remove from active connections
                self._connections.pop(connection_id, None)
                self._connection_threads.pop(connection_id, None)
                self._active_connections = max(0, self._active_connections - 1)

    def _close_all_connections(self) -> None:
        """Close all active connections."""
        connections_to_close = []

        with self._lock:
            connections_to_close = list(self._connections.items())

        # Close connections outside of lock to avoid deadlock
        for _connection_id, transport in connections_to_close:
            try:
                transport.close()
            except Exception as e:
                self._logger.debug(f"Server socket close error: {e}")

        # Wait for connection threads to finish
        threads_to_join = []
        with self._lock:
            threads_to_join = list(self._connection_threads.values())

        for thread in threads_to_join:
            if thread.is_alive():
                thread.join(timeout=5.0)

        with self._lock:
            self._connections.clear()
            self._connection_threads.clear()
            self._active_connections = 0

    def _cleanup_server_socket(self) -> None:
        """Clean up server socket."""
        if self._server_socket:
            try:
                self._server_socket.close()
            except Exception as e:
                self._logger.debug(f"Server socket close error: {e}")
            finally:
                self._server_socket = None

    def is_running(self) -> bool:
        """
        Check if server is running.

        Returns:
            True if server is running
        """
        return self._running

    def get_connection_count(self) -> int:
        """
        Get number of active connections.

        Returns:
            Number of active connections
        """
        with self._lock:
            return len(self._connections)

    def get_connection_stats(self) -> dict[str, int]:
        """
        Get connection statistics.

        Returns:
            Dictionary with connection statistics
        """
        with self._lock:
            return {
                "total_connections": self._total_connections,
                "active_connections": self._active_connections,
                "failed_connections": self._failed_connections,
                "max_connections": self._max_connections,
            }

    def get_active_connections(self) -> list[str]:
        """
        Get list of active connection IDs.

        Returns:
            List of active connection identifiers
        """
        with self._lock:
            return list(self._connections.keys())

    def close_connection(self, connection_id: str) -> bool:
        """
        Close a specific connection.

        Args:
            connection_id: Connection identifier to close

        Returns:
            True if connection was found and closed
        """
        with self._lock:
            transport = self._connections.get(connection_id)
            if transport:
                try:
                    transport.close()
                    return True
                except Exception as e:
                    self._logger.debug(f"Error during transport close: {e}")

        return False

Methods:¶

`init(server_interface, server_key, bind_address='0.0.0.0', port=22)` ¶

Initialize SSH server manager.

Parameters:

Name	Type	Description	Default
`server_interface`	`SSHServer`	SSHServer instance for handling client requests	required
`server_key`	`PKey`	Server's private key for host authentication	required
`bind_address`	`str`	Address to bind server socket	`'0.0.0.0'`
`port`	`int`	Port to bind server socket	`22`

Source code in spindlex/server/ssh_server.py

def __init__(
    self,
    server_interface: SSHServer,
    server_key: PKey,
    bind_address: str = "0.0.0.0",  # nosec B104
    port: int = 22,
) -> None:
    """
    Initialize SSH server manager.

    Args:
        server_interface: SSHServer instance for handling client requests
        server_key: Server's private key for host authentication
        bind_address: Address to bind server socket
        port: Port to bind server socket
    """
    self._server_interface = server_interface
    self._server_key = server_key
    self._bind_address = bind_address
    self._port = port

    # Ensure server interface has the host key immediately
    self._server_interface.set_server_key(self._server_key)

    self._server_socket: Optional[socket.socket] = None
    self._running = False
    self._connections: dict[str, Transport] = {}  # connection_id -> transport
    self._connection_threads: dict[str, threading.Thread] = {}
    self._lock = threading.RLock()
    self._accept_thread: Optional[threading.Thread] = None

    # Connection limits and timeouts
    self._max_connections = 100
    self._connection_timeout = 30.0
    self._auth_timeout = 30.0

    # Statistics
    self._total_connections = 0
    self._active_connections = 0
    self._failed_connections = 0
    self._logger = get_logger("spindlex.server.ssh_server")

`close_connection(connection_id)` ¶

Close a specific connection.

Parameters:

Name	Type	Description	Default
`connection_id`	`str`	Connection identifier to close	required

Returns:

Type	Description
`bool`	True if connection was found and closed

Source code in spindlex/server/ssh_server.py

def close_connection(self, connection_id: str) -> bool:
    """
    Close a specific connection.

    Args:
        connection_id: Connection identifier to close

    Returns:
        True if connection was found and closed
    """
    with self._lock:
        transport = self._connections.get(connection_id)
        if transport:
            try:
                transport.close()
                return True
            except Exception as e:
                self._logger.debug(f"Error during transport close: {e}")

    return False

`get_active_connections()` ¶

Get list of active connection IDs.

Returns:

Type	Description
`list[str]`	List of active connection identifiers

Source code in spindlex/server/ssh_server.py

def get_active_connections(self) -> list[str]:
    """
    Get list of active connection IDs.

    Returns:
        List of active connection identifiers
    """
    with self._lock:
        return list(self._connections.keys())

`get_connection_count()` ¶

Get number of active connections.

Returns:

Type	Description
`int`	Number of active connections

Source code in spindlex/server/ssh_server.py

def get_connection_count(self) -> int:
    """
    Get number of active connections.

    Returns:
        Number of active connections
    """
    with self._lock:
        return len(self._connections)

`get_connection_stats()` ¶

Get connection statistics.

Returns:

Type	Description
`dict[str, int]`	Dictionary with connection statistics

Source code in spindlex/server/ssh_server.py

def get_connection_stats(self) -> dict[str, int]:
    """
    Get connection statistics.

    Returns:
        Dictionary with connection statistics
    """
    with self._lock:
        return {
            "total_connections": self._total_connections,
            "active_connections": self._active_connections,
            "failed_connections": self._failed_connections,
            "max_connections": self._max_connections,
        }

`is_running()` ¶

Check if server is running.

Returns:

Type	Description
`bool`	True if server is running

Source code in spindlex/server/ssh_server.py

def is_running(self) -> bool:
    """
    Check if server is running.

    Returns:
        True if server is running
    """
    return self._running

`set_auth_timeout(timeout)` ¶

Set authentication timeout.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Authentication timeout in seconds	required

Source code in spindlex/server/ssh_server.py

def set_auth_timeout(self, timeout: float) -> None:
    """
    Set authentication timeout.

    Args:
        timeout: Authentication timeout in seconds
    """
    self._auth_timeout = timeout

`set_connection_timeout(timeout)` ¶

Set connection timeout.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Connection timeout in seconds	required

Source code in spindlex/server/ssh_server.py

def set_connection_timeout(self, timeout: float) -> None:
    """
    Set connection timeout.

    Args:
        timeout: Connection timeout in seconds
    """
    self._connection_timeout = timeout

`set_max_connections(max_connections)` ¶

Set maximum number of concurrent connections.

Parameters:

Name	Type	Description	Default
`max_connections`	`int`	Maximum number of concurrent connections	required

Source code in spindlex/server/ssh_server.py

def set_max_connections(self, max_connections: int) -> None:
    """
    Set maximum number of concurrent connections.

    Args:
        max_connections: Maximum number of concurrent connections
    """
    self._max_connections = max_connections

`start_server()` ¶

Start SSH server and begin accepting connections.

Raises:

Type	Description
`TransportException`	If server fails to start

Source code in spindlex/server/ssh_server.py

def start_server(self) -> None:
    """
    Start SSH server and begin accepting connections.

    Raises:
        TransportException: If server fails to start
    """
    with self._lock:
        if self._running:
            raise TransportException("Server is already running")

        try:
            addr_info = socket.getaddrinfo(
                self._bind_address,
                self._port,
                socket.AF_UNSPEC,
                socket.SOCK_STREAM,
                0,
                socket.AI_PASSIVE,
            )
            if not addr_info:
                raise TransportException(
                    f"Could not resolve bind address: {self._bind_address}"
                )

            # Use the first available address info
            af, socktype, proto, canonname, sa = addr_info[0]

            # Create and bind server socket
            self._server_socket = socket.socket(af, socktype, proto)
            self._server_socket.setsockopt(
                socket.SOL_SOCKET, socket.SO_REUSEADDR, 1
            )

            # For IPv6, try to enable dual-stack if bind_address is empty or ::
            if af == socket.AF_INET6:
                try:
                    self._server_socket.setsockopt(
                        socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0
                    )
                except (AttributeError, OSError):
                    pass

            self._server_socket.bind(sa)
            self._server_socket.listen(socket.SOMAXCONN)

            self._running = True

            # Start accept thread
            self._accept_thread = threading.Thread(
                target=self._accept_connections,
                name="SSHServer-Accept",
                daemon=True,
            )
            self._accept_thread.start()

        except Exception as e:
            self._cleanup_server_socket()
            raise TransportException(f"Failed to start SSH server: {e}") from e

`stop_server()` ¶

Stop SSH server and close all connections.

Source code in spindlex/server/ssh_server.py

def stop_server(self) -> None:
    """
    Stop SSH server and close all connections.
    """
    with self._lock:
        if not self._running:
            return

        self._running = False

    # Close server socket to stop accepting new connections
    self._cleanup_server_socket()

    # Wait for accept thread to finish
    if self._accept_thread and self._accept_thread.is_alive():
        self._accept_thread.join(timeout=5.0)

    # Close all active connections
    self._close_all_connections()

Functions:¶

SFTP Server¶

`spindlex.server.sftp_server` ¶

SFTP Server Implementation

Provides server-side SFTP functionality with file system operations and customizable authorization hooks.

Classes¶

`SFTPHandle` ¶

SFTP file handle for managing open files.

Represents an open file or directory on the server side.

Source code in spindlex/server/sftp_server.py

class SFTPHandle:
    """
    SFTP file handle for managing open files.

    Represents an open file or directory on the server side.
    """

    def __init__(
        self,
        handle_id: bytes,
        path: str,
        flags: int,
        file_obj: Optional[BinaryIO] = None,
    ) -> None:
        """
        Initialize SFTP handle.

        Args:
            handle_id: Unique handle identifier
            path: File/directory path
            flags: Open flags
            file_obj: File object for file handles (None for directory handles)
        """
        self.handle_id = handle_id
        self.path = path
        self.flags = flags
        self.file_obj = file_obj
        self.is_directory = file_obj is None
        self.position = 0
        self.dir_entries: Optional[list[tuple]] = None
        self.dir_index = 0

    def read(self, length: int) -> bytes:
        """
        Read data from file handle.

        Args:
            length: Number of bytes to read

        Returns:
            Read data

        Raises:
            SFTPError: If handle is not readable or read fails
        """
        if self.is_directory:
            raise SFTPError("Cannot read from directory handle", SSH_FX_FAILURE)

        if not (self.flags & SSH_FXF_READ):
            raise SFTPError("Handle not open for reading", SSH_FX_PERMISSION_DENIED)

        if self.file_obj is None:
            raise SFTPError("File object not available", SSH_FX_FAILURE)

        try:
            return self.file_obj.read(length)
        except (OSError, ValueError) as e:
            raise SFTPError(f"Read failed: {e}", SSH_FX_FAILURE)

    def write(self, data: bytes) -> int:
        """
        Write data to file handle.

        Args:
            data: Data to write

        Returns:
            Number of bytes written

        Raises:
            SFTPError: If handle is not writable or write fails
        """
        if self.is_directory:
            raise SFTPError("Cannot write to directory handle", SSH_FX_FAILURE)

        if not (self.flags & (SSH_FXF_WRITE | SSH_FXF_APPEND)):
            raise SFTPError("Handle not open for writing", SSH_FX_PERMISSION_DENIED)

        if self.file_obj is None:
            raise SFTPError("File object not available", SSH_FX_FAILURE)

        try:
            return self.file_obj.write(data)
        except (OSError, ValueError) as e:
            raise SFTPError(f"Write failed: {e}", SSH_FX_FAILURE)

    def seek(self, offset: int) -> None:
        """
        Seek to position in file.

        Args:
            offset: Byte offset to seek to

        Raises:
            SFTPError: If seek fails
        """
        if self.is_directory:
            raise SFTPError("Cannot seek in directory handle", SSH_FX_FAILURE)

        if self.file_obj is None:
            raise SFTPError("File object not available", SSH_FX_FAILURE)

        try:
            self.file_obj.seek(offset)
            self.position = offset
        except (OSError, ValueError) as e:
            raise SFTPError(f"Seek failed: {e}", SSH_FX_FAILURE)

    def close(self) -> None:
        """Close the handle and cleanup resources."""
        if self.file_obj:
            try:
                self.file_obj.close()
            except OSError:
                pass
            finally:
                self.file_obj = None

Methods:¶

`init(handle_id, path, flags, file_obj=None)` ¶

Initialize SFTP handle.

Parameters:

Name	Type	Description	Default
`handle_id`	`bytes`	Unique handle identifier	required
`path`	`str`	File/directory path	required
`flags`	`int`	Open flags	required
`file_obj`	`Optional[BinaryIO]`	File object for file handles (None for directory handles)	`None`

Source code in spindlex/server/sftp_server.py

def __init__(
    self,
    handle_id: bytes,
    path: str,
    flags: int,
    file_obj: Optional[BinaryIO] = None,
) -> None:
    """
    Initialize SFTP handle.

    Args:
        handle_id: Unique handle identifier
        path: File/directory path
        flags: Open flags
        file_obj: File object for file handles (None for directory handles)
    """
    self.handle_id = handle_id
    self.path = path
    self.flags = flags
    self.file_obj = file_obj
    self.is_directory = file_obj is None
    self.position = 0
    self.dir_entries: Optional[list[tuple]] = None
    self.dir_index = 0

`close()` ¶

Close the handle and cleanup resources.

Source code in spindlex/server/sftp_server.py

def close(self) -> None:
    """Close the handle and cleanup resources."""
    if self.file_obj:
        try:
            self.file_obj.close()
        except OSError:
            pass
        finally:
            self.file_obj = None

`read(length)` ¶

Read data from file handle.

Parameters:

Name	Type	Description	Default
`length`	`int`	Number of bytes to read	required

Returns:

Type	Description
`bytes`	Read data

Raises:

Type	Description
`SFTPError`	If handle is not readable or read fails

Source code in spindlex/server/sftp_server.py

def read(self, length: int) -> bytes:
    """
    Read data from file handle.

    Args:
        length: Number of bytes to read

    Returns:
        Read data

    Raises:
        SFTPError: If handle is not readable or read fails
    """
    if self.is_directory:
        raise SFTPError("Cannot read from directory handle", SSH_FX_FAILURE)

    if not (self.flags & SSH_FXF_READ):
        raise SFTPError("Handle not open for reading", SSH_FX_PERMISSION_DENIED)

    if self.file_obj is None:
        raise SFTPError("File object not available", SSH_FX_FAILURE)

    try:
        return self.file_obj.read(length)
    except (OSError, ValueError) as e:
        raise SFTPError(f"Read failed: {e}", SSH_FX_FAILURE)

`seek(offset)` ¶

Seek to position in file.

Parameters:

Name	Type	Description	Default
`offset`	`int`	Byte offset to seek to	required

Raises:

Type	Description
`SFTPError`	If seek fails

Source code in spindlex/server/sftp_server.py

def seek(self, offset: int) -> None:
    """
    Seek to position in file.

    Args:
        offset: Byte offset to seek to

    Raises:
        SFTPError: If seek fails
    """
    if self.is_directory:
        raise SFTPError("Cannot seek in directory handle", SSH_FX_FAILURE)

    if self.file_obj is None:
        raise SFTPError("File object not available", SSH_FX_FAILURE)

    try:
        self.file_obj.seek(offset)
        self.position = offset
    except (OSError, ValueError) as e:
        raise SFTPError(f"Seek failed: {e}", SSH_FX_FAILURE)

`write(data)` ¶

Write data to file handle.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	Data to write	required

Returns:

Type	Description
`int`	Number of bytes written

Raises:

Type	Description
`SFTPError`	If handle is not writable or write fails

Source code in spindlex/server/sftp_server.py

def write(self, data: bytes) -> int:
    """
    Write data to file handle.

    Args:
        data: Data to write

    Returns:
        Number of bytes written

    Raises:
        SFTPError: If handle is not writable or write fails
    """
    if self.is_directory:
        raise SFTPError("Cannot write to directory handle", SSH_FX_FAILURE)

    if not (self.flags & (SSH_FXF_WRITE | SSH_FXF_APPEND)):
        raise SFTPError("Handle not open for writing", SSH_FX_PERMISSION_DENIED)

    if self.file_obj is None:
        raise SFTPError("File object not available", SSH_FX_FAILURE)

    try:
        return self.file_obj.write(data)
    except (OSError, ValueError) as e:
        raise SFTPError(f"Write failed: {e}", SSH_FX_FAILURE)

`SFTPServer` ¶

Base SFTP server implementation.

Provides hooks for file system operations that can be overridden to implement custom SFTP server behavior and authorization.

Source code in spindlex/server/sftp_server.py

class SFTPServer:
    """
    Base SFTP server implementation.

    Provides hooks for file system operations that can be overridden
    to implement custom SFTP server behavior and authorization.
    """

    def __init__(
        self, channel: "Channel", root_path: str = "/", start_thread: bool = True
    ) -> None:
        """
        Initialize SFTP server with channel and root path.

        Args:
            channel: SSH channel for SFTP communication
            root_path: Root directory for SFTP operations (default: "/")
            start_thread: Whether to start the message processing thread (default: True)
        """
        self._channel = channel
        # Resolve the root once, up front: every subsequent path check compares
        # realpath(candidate) against this canonical root, so a symlinked root
        # cannot be sidestepped later by replacing it mid-session.
        self._root_path = os.path.realpath(os.path.abspath(root_path))
        self._handles: dict[bytes, SFTPHandle] = {}
        self._handle_counter = 0
        self._handle_lock = threading.Lock()
        self._logger = logging.getLogger(__name__)
        self._client_version: Optional[int] = None
        self._client_extensions: dict[str, str] = {}

        if start_thread:
            # Start SFTP session in a separate thread to avoid blocking
            self._thread = threading.Thread(
                target=self._run_server,
                name=f"SFTPServer-{channel.channel_id}",
                daemon=True,
            )
            self._thread.start()

    def _run_server(self) -> None:
        """Run the SFTP server session."""
        try:
            self._start_sftp_session()
        except (OSError, struct.error, SSHException) as e:
            self._logger.error(f"SFTP server session error: {e}")
            self.close()

    def _start_sftp_session(self) -> None:
        """
        Start SFTP session and handle version negotiation.

        Raises:
            SFTPError: If SFTP initialization fails
        """
        try:
            # Bound recv() waits so the SFTP thread exits promptly when the
            # client disconnects instead of blocking for the full socket timeout.
            self._channel.settimeout(30.0)

            # Wait for client init message
            init_msg = self._receive_message()
            if not isinstance(init_msg, SFTPInitMessage):
                raise SFTPError("Expected SFTP init message", SSH_FX_BAD_MESSAGE)

            self._client_version = init_msg.version

            # Send version response
            version_msg = SFTPVersionMessage(SFTP_VERSION, {})
            self._send_message(version_msg)

            self._logger.debug(
                f"SFTP session started, client version: {self._client_version}"
            )

            # Start message processing loop
            self._process_messages()

        except (OSError, struct.error, SSHException) as e:
            self._logger.error(f"SFTP session initialization failed: {e}")
            if isinstance(e, SFTPError):
                raise
            raise SFTPError(f"SFTP initialization failed: {e}") from e

    def _generate_handle(self) -> bytes:
        """
        Generate unique handle identifier.

        Returns:
            Unique handle bytes
        """
        with self._handle_lock:
            self._handle_counter += 1
            return f"handle_{self._handle_counter}".encode()

    def _send_message(self, message: SFTPMessage) -> None:
        """
        Send SFTP message over channel.

        Args:
            message: SFTP message to send

        Raises:
            SFTPError: If message sending fails
        """
        try:
            data = message.pack()
            self._channel.send(data)
        except (OSError, struct.error, SSHException) as e:
            raise SFTPError(f"Failed to send SFTP message: {e}") from e

    def _receive_message(self) -> SFTPMessage:
        """
        Receive SFTP message from channel.

        Returns:
            Received SFTP message

        Raises:
            SFTPError: If message receiving fails
        """
        try:
            # Read message length first (4 bytes)
            length_data = self._channel.recv_exactly(4)
            msg_length = int.from_bytes(length_data, "big")

            # Read message content (msg_length bytes)
            payload = self._channel.recv_exactly(msg_length)
            msg_data = length_data + payload

            return SFTPMessage.unpack(msg_data)
        except (OSError, struct.error, ValueError, SSHException) as e:
            raise SFTPError(f"Failed to receive SFTP message: {e}") from e

    def _process_messages(self) -> None:
        """
        Process incoming SFTP messages in a loop.

        Handles all SFTP protocol messages and dispatches to appropriate handlers.
        """
        message = None
        while True:
            try:
                message = self._receive_message()
                self._handle_message(message)
            except (
                ConnectionResetError,
                ConnectionAbortedError,
                BrokenPipeError,
                EOFError,
            ) as e:
                self._logger.debug(f"SFTP session ended: {e}")
                break
            except (OSError, struct.error, SSHException) as e:
                self._logger.error(f"Error processing SFTP message: {e}")
                try:
                    if (
                        message is not None
                        and hasattr(message, "request_id")
                        and message.request_id is not None
                    ):
                        error_msg = SFTPStatusMessage(
                            message.request_id, SSH_FX_FAILURE, str(e)
                        )
                        self._send_message(error_msg)
                except (OSError, SFTPError):
                    pass
                break

    def _handle_message(self, message: SFTPMessage) -> None:
        """
        Handle individual SFTP message.

        Args:
            message: SFTP message to handle
        """
        # Dispatch based on message type
        if isinstance(message, SFTPInitMessage):
            self._handle_init(message)
        elif isinstance(message, SFTPOpenMessage):
            self._handle_open(message)
        elif isinstance(message, SFTPCloseMessage):
            self._handle_close(message)
        elif isinstance(message, SFTPReadMessage):
            self._handle_read(message)
        elif isinstance(message, SFTPWriteMessage):
            self._handle_write(message)
        elif isinstance(message, SFTPStatMessage):
            self._handle_stat(message)
        elif isinstance(message, SFTPLStatMessage):
            self._handle_lstat(message)
        elif isinstance(message, SFTPFStatMessage):
            self._handle_fstat(message)
        elif isinstance(message, SFTPSetStatMessage):
            self._handle_setstat(message)
        elif isinstance(message, SFTPOpenDirMessage):
            self._handle_opendir(message)
        elif isinstance(message, SFTPReadDirMessage):
            self._handle_readdir(message)
        elif isinstance(message, SFTPMkdirMessage):
            self._handle_mkdir(message)
        elif isinstance(message, SFTPRmdirMessage):
            self._handle_rmdir(message)
        elif isinstance(message, SFTPRemoveMessage):
            self._handle_remove(message)
        elif isinstance(message, SFTPRenameMessage):
            self._handle_rename(message)
        elif isinstance(message, SFTPRealPathMessage):
            self._handle_realpath(message)
        else:
            # Unsupported operation
            if hasattr(message, "request_id") and message.request_id is not None:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_OP_UNSUPPORTED, "Operation not supported"
                )
                self._send_message(error_msg)

    def _handle_init(self, message: SFTPInitMessage) -> None:
        """Handle SFTP init message (should not happen after session start)."""
        # This should not happen after session initialization
        pass

    def _resolve_path(self, path: str) -> str:
        """
        Resolve relative path to absolute path within root.

        Args:
            path: Path to resolve

        Returns:
            Absolute path within server root

        Raises:
            SFTPError: If path is outside root directory
        """
        # Reject NUL bytes outright - they can truncate paths in some native
        # APIs and have no legitimate use in SFTP paths.
        if "\x00" in path:
            raise SFTPError("Invalid path", SSH_FX_PERMISSION_DENIED)

        # Normalize SFTP path (always use forward slashes in SFTP)
        path = path.replace("\\", "/")

        # Check if this is an SFTP absolute path (starts with /)
        if path.startswith("/"):
            # Strip leading slash and treat as relative to root
            path = path.lstrip("/")

        # Join with root path
        full_path = os.path.normpath(os.path.join(self._root_path, path))

        # Fully resolve path (resolve symlinks and ..)
        resolved = os.path.realpath(full_path)
        root_real = self._root_path  # already realpath()'d in __init__

        # Check if resolved path is within root. Using prefix comparison with a
        # trailing separator avoids the classic /var/www vs /var/wwwbad bypass
        # and sidesteps commonpath's ValueError on mixed drives (Windows).
        resolved_norm = os.path.normcase(resolved)
        root_norm = os.path.normcase(root_real)
        root_with_sep = root_norm.rstrip(os.sep) + os.sep

        if resolved_norm != root_norm and not resolved_norm.startswith(root_with_sep):
            raise SFTPError("Path outside root directory", SSH_FX_PERMISSION_DENIED)

        return resolved

    def _path_to_attrs(self, path: str) -> SFTPAttributes:
        """
        Convert file system path to SFTP attributes.

        Args:
            path: File system path

        Returns:
            SFTPAttributes object

        Raises:
            SFTPError: If stat fails
        """
        try:
            st = os.stat(path)
            attrs = SFTPAttributes()

            attrs.flags = (
                SSH_FILEXFER_ATTR_SIZE
                | SSH_FILEXFER_ATTR_PERMISSIONS
                | SSH_FILEXFER_ATTR_ACMODTIME
                | SSH_FILEXFER_ATTR_UIDGID
            )
            attrs.size = st.st_size
            attrs.permissions = st.st_mode
            attrs.atime = int(st.st_atime)
            attrs.mtime = int(st.st_mtime)
            attrs.uid = st.st_uid
            attrs.gid = st.st_gid

            return attrs
        except OSError as e:
            if e.errno == errno.ENOENT:
                raise SFTPError("No such file or directory", SSH_FX_NO_SUCH_FILE)
            else:
                raise SFTPError(f"Stat failed: {e}", SSH_FX_FAILURE)

    # Message handlers
    def _handle_open(self, message: SFTPOpenMessage) -> None:
        """Handle file open request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.filename)

            # Check authorization
            if message.pflags & (SSH_FXF_WRITE | SSH_FXF_APPEND | SSH_FXF_CREAT):
                if not self.check_file_access(resolved_path, "w"):
                    error_msg = SFTPStatusMessage(
                        message.request_id,
                        SSH_FX_PERMISSION_DENIED,
                        "Read access denied",
                    )
                    self._send_message(error_msg)
                    return

            else:
                if not self.check_file_access(resolved_path, "r"):
                    error_msg = SFTPStatusMessage(
                        message.request_id,
                        SSH_FX_PERMISSION_DENIED,
                        "Read access denied",
                    )
                    self._send_message(error_msg)
                    return

            # Determine file mode
            mode = ""
            if message.pflags & SSH_FXF_READ and message.pflags & SSH_FXF_WRITE:
                mode = "r+b"
            elif message.pflags & SSH_FXF_WRITE:
                if message.pflags & SSH_FXF_CREAT:
                    if message.pflags & SSH_FXF_EXCL:
                        mode = "xb"  # Exclusive create
                    elif message.pflags & SSH_FXF_TRUNC:
                        mode = "wb"  # Create or truncate
                    else:
                        mode = "ab"  # Create or append
                else:
                    mode = "r+b"  # Write to existing file
            elif message.pflags & SSH_FXF_APPEND:
                mode = "ab"
            else:
                mode = "rb"  # Read only

            # Open file
            file_obj = None
            try:
                file_obj = open(resolved_path, mode)

                # Create handle
                handle_id = self._generate_handle()
                from typing import BinaryIO, cast

                handle = SFTPHandle(
                    handle_id,
                    resolved_path,
                    message.pflags,
                    file_obj=cast(BinaryIO, file_obj),
                )

                with self._handle_lock:
                    if len(self._handles) >= MAX_SFTP_HANDLES:
                        file_obj.close()
                        error_msg = SFTPStatusMessage(
                            message.request_id,
                            SSH_FX_FAILURE,
                            "Too many open handles",
                        )
                        self._send_message(error_msg)
                        return

                    self._handles[handle_id] = handle

                # Send handle response
                handle_msg = SFTPHandleMessage(message.request_id, handle_id)
                self._send_message(handle_msg)

            except OSError as e:
                if file_obj:
                    try:
                        file_obj.close()
                    except OSError:
                        pass
                if isinstance(e, FileNotFoundError):
                    error_msg = SFTPStatusMessage(
                        int(message.request_id), SSH_FX_NO_SUCH_FILE, "File not found"
                    )
                elif isinstance(e, PermissionError):
                    error_msg = SFTPStatusMessage(
                        int(message.request_id),
                        SSH_FX_PERMISSION_DENIED,
                        "Permission denied",
                    )
                elif isinstance(e, FileExistsError):
                    error_msg = SFTPStatusMessage(
                        int(message.request_id), SSH_FX_FAILURE, "File already exists"
                    )
                else:
                    error_msg = SFTPStatusMessage(
                        int(message.request_id), SSH_FX_FAILURE, f"Open failed: {e}"
                    )
                self._send_message(error_msg)
                return

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, ValueError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_close(self, message: SFTPCloseMessage) -> None:
        """Handle file close request."""
        assert message.request_id is not None
        try:
            with self._handle_lock:
                handle = self._handles.get(message.handle)
                if handle is None:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Invalid handle"
                    )
                    self._send_message(error_msg)
                    return

                # Close and remove handle
                handle.close()
                del self._handles[message.handle]

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except (OSError, SSHException) as e:
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_read(self, message: SFTPReadMessage) -> None:
        """Handle file read request."""
        assert message.request_id is not None
        try:
            with self._handle_lock:
                handle = self._handles.get(message.handle)
                if handle is None:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Invalid handle"
                    )
                    self._send_message(error_msg)
                    return

            # Seek to requested offset
            handle.seek(message.offset)

            # Read data (limit to max read size)
            read_length = min(message.length, SFTP_MAX_READ_SIZE)
            data = handle.read(read_length)

            if len(data) == 0:
                # End of file
                status_msg = SFTPStatusMessage(message.request_id, SSH_FX_EOF, "")
                self._send_message(status_msg)
            else:
                # Send data response
                data_msg = SFTPDataMessage(message.request_id, data)
                self._send_message(data_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, ValueError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_write(self, message: SFTPWriteMessage) -> None:
        """Handle file write request."""
        assert message.request_id is not None
        try:
            with self._handle_lock:
                handle = self._handles.get(message.handle)
                if handle is None:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Invalid handle"
                    )
                    self._send_message(error_msg)
                    return

            # Seek to requested offset
            handle.seek(message.offset)

            # Write data
            handle.write(message.data)

            # Flush to ensure data is written
            if handle.file_obj:
                handle.file_obj.flush()

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, ValueError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_stat(self, message: SFTPStatMessage) -> None:
        """Handle stat request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            if not self.check_file_access(resolved_path, "r"):
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_PERMISSION_DENIED, "Access denied"
                )
                self._send_message(error_msg)
                return

            # Get file attributes
            attrs = self._path_to_attrs(resolved_path)

            # Send attributes response
            attrs_msg = SFTPAttrsMessage(message.request_id, attrs)
            self._send_message(attrs_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_lstat(self, message: SFTPLStatMessage) -> None:
        """Handle lstat request (don't follow symlinks)."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            if not self.check_file_access(resolved_path, "r"):
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_PERMISSION_DENIED, "Access denied"
                )
                self._send_message(error_msg)
                return

            # Get file attributes (lstat doesn't follow symlinks)
            try:
                st = os.lstat(resolved_path)
                attrs = SFTPAttributes()

                attrs.flags = (
                    SSH_FILEXFER_ATTR_SIZE
                    | SSH_FILEXFER_ATTR_PERMISSIONS
                    | SSH_FILEXFER_ATTR_ACMODTIME
                    | SSH_FILEXFER_ATTR_UIDGID
                )
                attrs.size = st.st_size
                attrs.permissions = st.st_mode
                attrs.atime = int(st.st_atime)
                attrs.mtime = int(st.st_mtime)
                attrs.uid = st.st_uid
                attrs.gid = st.st_gid

            except OSError as e:
                if e.errno == errno.ENOENT:
                    raise SFTPError("No such file or directory", SSH_FX_NO_SUCH_FILE)
                else:
                    raise SFTPError(f"Lstat failed: {e}", SSH_FX_FAILURE)

            # Send attributes response
            attrs_msg = SFTPAttrsMessage(message.request_id, attrs)
            self._send_message(attrs_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_fstat(self, message: SFTPFStatMessage) -> None:
        """Handle fstat request (get attributes of open file)."""
        assert message.request_id is not None
        try:
            with self._handle_lock:
                handle = self._handles.get(message.handle)
                if handle is None:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Invalid handle"
                    )
                    self._send_message(error_msg)
                    return

            # Get file attributes for the open file
            attrs = self._path_to_attrs(handle.path)

            # Send attributes response
            attrs_msg = SFTPAttrsMessage(message.request_id, attrs)
            self._send_message(attrs_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_setstat(self, message: SFTPSetStatMessage) -> None:
        """Handle setstat request (set file attributes)."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            if not self.check_file_access(resolved_path, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_PERMISSION_DENIED, "Write access denied"
                )
                self._send_message(error_msg)
                return

            attrs = message.attrs

            # Set permissions
            if (
                attrs.flags & SSH_FILEXFER_ATTR_PERMISSIONS
                and attrs.permissions is not None
            ):
                try:
                    os.chmod(resolved_path, attrs.permissions)
                except OSError as e:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, f"Chmod failed: {e}"
                    )
                    self._send_message(error_msg)
                    return

            # Set access and modification times
            if (
                attrs.flags & SSH_FILEXFER_ATTR_ACMODTIME
                and attrs.atime is not None
                and attrs.mtime is not None
            ):
                try:
                    os.utime(resolved_path, (attrs.atime, attrs.mtime))
                except OSError as e:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, f"Utime failed: {e}"
                    )
                    self._send_message(error_msg)
                    return

            # Set ownership (if supported and authorized)
            if attrs.flags & SSH_FILEXFER_ATTR_UIDGID:
                try:
                    if hasattr(os, "chown"):
                        uid = attrs.uid if attrs.uid is not None else -1
                        gid = attrs.gid if attrs.gid is not None else -1
                        os.chown(resolved_path, uid, gid)
                except (OSError, AttributeError):
                    # chown may not be supported on all platforms
                    # or user may not have permission
                    pass

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_opendir(self, message: SFTPOpenDirMessage) -> None:
        """Handle directory open request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            if not self.check_directory_access(resolved_path, "r"):
                assert message.request_id is not None
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Read access denied",
                )
                self._send_message(error_msg)
                return

            # Check if path is a directory
            if not os.path.isdir(resolved_path):
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_NO_SUCH_FILE, "Not a directory"
                )
                self._send_message(error_msg)
                return

            # Read directory entries
            try:
                entries = []
                for name in os.listdir(resolved_path):
                    entry_path = os.path.join(resolved_path, name)
                    try:
                        attrs = self._path_to_attrs(entry_path)
                        # Create long name (ls -l style)
                        longname = self._format_longname(name, attrs)
                        entries.append((name, longname, attrs))
                    except (OSError, SFTPError):
                        # Skip entries we can't stat
                        continue

            except OSError:
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Read access denied",
                )
                self._send_message(error_msg)
                return

            # Create directory handle
            handle_id = self._generate_handle()
            handle = SFTPHandle(handle_id, resolved_path, 0)  # Directory handle
            handle.dir_entries = entries
            handle.dir_index = 0

            with self._handle_lock:
                self._handles[handle_id] = handle

            # Send handle response
            handle_msg = SFTPHandleMessage(message.request_id, handle_id)
            self._send_message(handle_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_readdir(self, message: SFTPReadDirMessage) -> None:
        """Handle directory read request."""
        assert message.request_id is not None
        try:
            with self._handle_lock:
                handle = self._handles.get(message.handle)
                if handle is None:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Invalid handle"
                    )
                    self._send_message(error_msg)
                    return

                if not handle.is_directory:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Handle is not a directory"
                    )
                    self._send_message(error_msg)
                    return

            # Check if we've reached the end of directory
            if handle.dir_entries is None or handle.dir_index >= len(
                handle.dir_entries
            ):
                status_msg = SFTPStatusMessage(message.request_id, SSH_FX_EOF, "")
                self._send_message(status_msg)
                return

            # Return a batch of entries (limit to avoid large messages)
            batch_size = 50  # Reasonable batch size
            start_index = handle.dir_index
            end_index = min(start_index + batch_size, len(handle.dir_entries))

            batch_entries = handle.dir_entries[start_index:end_index]
            handle.dir_index = end_index

            # Send name response
            name_msg = SFTPNameMessage(message.request_id, batch_entries)
            self._send_message(name_msg)

        except (OSError, SSHException) as e:
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_mkdir(self, message: SFTPMkdirMessage) -> None:
        """Handle directory creation request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            parent_dir = os.path.dirname(resolved_path)
            if not self.check_directory_access(parent_dir, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Read access denied",
                )
                self._send_message(error_msg)
                return

            # Get permissions from attributes or use default
            mode = self.get_directory_permissions(resolved_path)
            if (
                message.attrs.flags & SSH_FILEXFER_ATTR_PERMISSIONS
                and message.attrs.permissions is not None
            ):
                mode = message.attrs.permissions

            # Create directory
            try:
                os.mkdir(resolved_path, mode)
            except FileExistsError:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_FAILURE, "Directory already exists"
                )
                self._send_message(error_msg)
                return
            except OSError as e:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_FAILURE, f"Mkdir failed: {e}"
                )
                self._send_message(error_msg)
                return

            # Set additional attributes if specified
            if message.attrs.flags & SSH_FILEXFER_ATTR_UIDGID:
                try:
                    if hasattr(os, "chown"):
                        uid = message.attrs.uid if message.attrs.uid is not None else -1
                        gid = message.attrs.gid if message.attrs.gid is not None else -1
                        os.chown(resolved_path, uid, gid)
                except (OSError, AttributeError):
                    # chown may not be supported or user may not have permission
                    pass

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_rmdir(self, message: SFTPRmdirMessage) -> None:
        """Handle directory removal request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Check authorization
            parent_dir = os.path.dirname(resolved_path)
            if not self.check_directory_access(parent_dir, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Read access denied",
                )
                self._send_message(error_msg)
                return

            # Remove directory
            try:
                os.rmdir(resolved_path)
            except FileNotFoundError:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_NO_SUCH_FILE, "Directory not found"
                )
                self._send_message(error_msg)
                return
            except OSError as e:
                if e.errno == 39:  # Directory not empty
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, "Directory not empty"
                    )
                else:
                    error_msg = SFTPStatusMessage(
                        message.request_id, SSH_FX_FAILURE, f"Rmdir failed: {e}"
                    )
                self._send_message(error_msg)
                return

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_remove(self, message: SFTPRemoveMessage) -> None:
        """Handle file removal request."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.filename)

            # Check authorization
            if not self.check_file_access(resolved_path, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Read access denied",
                )
                self._send_message(error_msg)
                return

            # Remove file
            try:
                os.unlink(resolved_path)
            except FileNotFoundError:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_NO_SUCH_FILE, "File not found"
                )
                self._send_message(error_msg)
                return
            except OSError as e:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_FAILURE, f"Remove failed: {e}"
                )
                self._send_message(error_msg)
                return

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_rename(self, message: SFTPRenameMessage) -> None:
        """Handle file rename request."""
        assert message.request_id is not None
        try:
            # Resolve and validate paths
            old_path = self._resolve_path(message.oldpath)
            new_path = self._resolve_path(message.newpath)

            # Check authorization for both paths
            if not self.check_file_access(old_path, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Write access denied",
                )
                self._send_message(error_msg)
                return

            new_parent = os.path.dirname(new_path)
            if not self.check_directory_access(new_parent, "w"):
                error_msg = SFTPStatusMessage(
                    message.request_id,
                    SSH_FX_PERMISSION_DENIED,
                    "Destination directory write access denied",
                )
                self._send_message(error_msg)
                return

            # Rename file
            try:
                os.rename(old_path, new_path)
            except FileNotFoundError:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_NO_SUCH_FILE, "Source file not found"
                )
                self._send_message(error_msg)
                return
            except OSError as e:
                error_msg = SFTPStatusMessage(
                    message.request_id, SSH_FX_FAILURE, f"Rename failed: {e}"
                )
                self._send_message(error_msg)
                return

            # Send success response
            status_msg = SFTPStatusMessage(message.request_id, SSH_FX_OK, "")
            self._send_message(status_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _handle_realpath(self, message: SFTPRealPathMessage) -> None:
        """Handle realpath request (resolve path)."""
        assert message.request_id is not None
        try:
            # Resolve and validate path
            resolved_path = self._resolve_path(message.path)

            # Convert back to relative path from root
            relative_path = os.path.relpath(resolved_path, self._root_path)
            if relative_path == ".":
                relative_path = "/"
            elif not relative_path.startswith("/"):
                relative_path = "/" + relative_path

            # Create attributes for the path (if it exists)
            try:
                attrs = self._path_to_attrs(resolved_path)
                longname = self._format_longname(os.path.basename(relative_path), attrs)
            except SFTPError:
                # Path doesn't exist, create minimal attributes
                attrs = SFTPAttributes()
                longname = os.path.basename(relative_path)

            # Send name response with single entry
            names = [(relative_path, longname, attrs)]
            name_msg = SFTPNameMessage(int(message.request_id), names)
            self._send_message(name_msg)

        except SFTPError as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(
                message.request_id, e.status_code or SSH_FX_FAILURE, str(e)
            )
            self._send_message(error_msg)
        except (OSError, SSHException) as e:
            assert message.request_id is not None
            error_msg = SFTPStatusMessage(message.request_id, SSH_FX_FAILURE, str(e))
            self._send_message(error_msg)

    def _format_longname(self, filename: str, attrs: SFTPAttributes) -> str:
        """
        Format long name for directory listings (ls -l style).

        Args:
            filename: File name
            attrs: File attributes

        Returns:
            Formatted long name string
        """
        # File type and permissions
        if attrs.permissions is not None:
            mode = attrs.permissions
            if stat.S_ISDIR(mode):
                type_char = "d"
            elif stat.S_ISLNK(mode):
                type_char = "l"
            elif stat.S_ISREG(mode):
                type_char = "-"
            else:
                type_char = "?"

            # Permission bits
            perms = (
                ("r" if mode & stat.S_IRUSR else "-")
                + ("w" if mode & stat.S_IWUSR else "-")
                + ("x" if mode & stat.S_IXUSR else "-")
                + ("r" if mode & stat.S_IRGRP else "-")
                + ("w" if mode & stat.S_IWGRP else "-")
                + ("x" if mode & stat.S_IXGRP else "-")
                + ("r" if mode & stat.S_IROTH else "-")
                + ("w" if mode & stat.S_IWOTH else "-")
                + ("x" if mode & stat.S_IXOTH else "-")
            )
            mode_str = type_char + perms
        else:
            mode_str = "----------"

        # Number of links (hardcoded to 1)
        nlink = 1

        # Owner and group (use numeric IDs)
        uid = attrs.uid if attrs.uid is not None else 0
        gid = attrs.gid if attrs.gid is not None else 0

        # File size
        size = attrs.size if attrs.size is not None else 0

        # Modification time (simplified format)
        if attrs.mtime is not None:
            import time

            mtime_str = time.strftime("%b %d %H:%M", time.localtime(attrs.mtime))
        else:
            mtime_str = "Jan  1 00:00"

        return (
            f"{mode_str} {nlink:3d} {uid:8d} {gid:8d} {size:8d} {mtime_str} {filename}"
        )

    # Authorization hooks - can be overridden by subclasses

    def check_file_access(self, path: str, mode: str) -> bool:
        """
        Check if file access is authorized.

        Override this method to implement custom file access authorization.

        Args:
            path: File path to check
            mode: Access mode ('r', 'w', 'x')

        Returns:
            True if access is authorized
        """
        # Default implementation allows all access
        return True

    def check_directory_access(self, path: str, mode: str) -> bool:
        """
        Check if directory access is authorized.

        Override this method to implement custom directory access authorization.

        Args:
            path: Directory path to check
            mode: Access mode ('r', 'w', 'x')

        Returns:
            True if access is authorized
        """
        # Default implementation allows all access
        return True

    def get_file_permissions(self, path: str) -> int:
        """
        Get file permissions for new files.

        Override this method to customize file permissions.

        Args:
            path: File path

        Returns:
            File permissions (octal mode)
        """
        # Default permissions: 644 (rw-r--r--)
        return 0o644

    def get_directory_permissions(self, path: str) -> int:
        """
        Get directory permissions for new directories.

        Override this method to customize directory permissions.

        Args:
            path: Directory path

        Returns:
            Directory permissions (octal mode)
        """
        # Default permissions: 755 (rwxr-xr-x)
        return 0o755

    def close(self) -> None:
        """Close SFTP server and cleanup resources."""
        # Close all open handles
        with self._handle_lock:
            for handle in self._handles.values():
                handle.close()
            self._handles.clear()

        # Close channel
        if self._channel:
            try:
                self._channel.close()
            except (OSError, SSHException):
                pass

Methods:¶

`init(channel, root_path='/', start_thread=True)` ¶

Initialize SFTP server with channel and root path.

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	SSH channel for SFTP communication	required
`root_path`	`str`	Root directory for SFTP operations (default: "/")	`'/'`
`start_thread`	`bool`	Whether to start the message processing thread (default: True)	`True`

Source code in spindlex/server/sftp_server.py

def __init__(
    self, channel: "Channel", root_path: str = "/", start_thread: bool = True
) -> None:
    """
    Initialize SFTP server with channel and root path.

    Args:
        channel: SSH channel for SFTP communication
        root_path: Root directory for SFTP operations (default: "/")
        start_thread: Whether to start the message processing thread (default: True)
    """
    self._channel = channel
    # Resolve the root once, up front: every subsequent path check compares
    # realpath(candidate) against this canonical root, so a symlinked root
    # cannot be sidestepped later by replacing it mid-session.
    self._root_path = os.path.realpath(os.path.abspath(root_path))
    self._handles: dict[bytes, SFTPHandle] = {}
    self._handle_counter = 0
    self._handle_lock = threading.Lock()
    self._logger = logging.getLogger(__name__)
    self._client_version: Optional[int] = None
    self._client_extensions: dict[str, str] = {}

    if start_thread:
        # Start SFTP session in a separate thread to avoid blocking
        self._thread = threading.Thread(
            target=self._run_server,
            name=f"SFTPServer-{channel.channel_id}",
            daemon=True,
        )
        self._thread.start()

`check_directory_access(path, mode)` ¶

Check if directory access is authorized.

Override this method to implement custom directory access authorization.

Parameters:

Name	Type	Description	Default
`path`	`str`	Directory path to check	required
`mode`	`str`	Access mode ('r', 'w', 'x')	required

Returns:

Type	Description
`bool`	True if access is authorized

Source code in spindlex/server/sftp_server.py

def check_directory_access(self, path: str, mode: str) -> bool:
    """
    Check if directory access is authorized.

    Override this method to implement custom directory access authorization.

    Args:
        path: Directory path to check
        mode: Access mode ('r', 'w', 'x')

    Returns:
        True if access is authorized
    """
    # Default implementation allows all access
    return True

`check_file_access(path, mode)` ¶

Check if file access is authorized.

Override this method to implement custom file access authorization.

Parameters:

Name	Type	Description	Default
`path`	`str`	File path to check	required
`mode`	`str`	Access mode ('r', 'w', 'x')	required

Returns:

Type	Description
`bool`	True if access is authorized

Source code in spindlex/server/sftp_server.py

def check_file_access(self, path: str, mode: str) -> bool:
    """
    Check if file access is authorized.

    Override this method to implement custom file access authorization.

    Args:
        path: File path to check
        mode: Access mode ('r', 'w', 'x')

    Returns:
        True if access is authorized
    """
    # Default implementation allows all access
    return True

`close()` ¶

Close SFTP server and cleanup resources.

Source code in spindlex/server/sftp_server.py

def close(self) -> None:
    """Close SFTP server and cleanup resources."""
    # Close all open handles
    with self._handle_lock:
        for handle in self._handles.values():
            handle.close()
        self._handles.clear()

    # Close channel
    if self._channel:
        try:
            self._channel.close()
        except (OSError, SSHException):
            pass

`get_directory_permissions(path)` ¶

Get directory permissions for new directories.

Override this method to customize directory permissions.

Parameters:

Name	Type	Description	Default
`path`	`str`	Directory path	required

Returns:

Type	Description
`int`	Directory permissions (octal mode)

Source code in spindlex/server/sftp_server.py

def get_directory_permissions(self, path: str) -> int:
    """
    Get directory permissions for new directories.

    Override this method to customize directory permissions.

    Args:
        path: Directory path

    Returns:
        Directory permissions (octal mode)
    """
    # Default permissions: 755 (rwxr-xr-x)
    return 0o755

`get_file_permissions(path)` ¶

Get file permissions for new files.

Override this method to customize file permissions.

Parameters:

Name	Type	Description	Default
`path`	`str`	File path	required

Returns:

Type	Description
`int`	File permissions (octal mode)

Source code in spindlex/server/sftp_server.py

def get_file_permissions(self, path: str) -> int:
    """
    Get file permissions for new files.

    Override this method to customize file permissions.

    Args:
        path: File path

    Returns:
        File permissions (octal mode)
    """
    # Default permissions: 644 (rw-r--r--)
    return 0o644

Server API¶

SSH Server¶

spindlex.server.ssh_server ¶

Classes¶

SSHServer ¶

Methods:¶

__init__() ¶

check_auth_gssapi_with_mic(username, gss_authenticated, cc_file) ¶

check_auth_keyboard_interactive(username, submethods) ¶

check_auth_password(username, password) ¶

check_auth_publickey(username, key) ¶

check_channel_env_request(channel, name, value) ¶

check_channel_exec_request(channel, command) ¶

check_channel_pty_request(channel, term, width, height, pixelwidth, pixelheight, modes) ¶

check_channel_request(kind, chanid) ¶

check_channel_shell_request(channel) ¶

check_channel_subsystem_request(channel, name) ¶

check_channel_window_change_request(channel, width, height, pixelwidth, pixelheight) ¶

check_channel_x11_request(channel, single_connection, auth_protocol, auth_cookie, screen_number) ¶

check_global_request(kind, msg) ¶

check_port_forward_cancel_request(address, port) ¶

check_port_forward_request(address, port) ¶

close_all_channels() ¶

close_channel(channel) ¶

get_active_channels() ¶

get_allowed_auths(username) ¶

get_banner() ¶

get_channel_count() ¶

get_server_key() ¶

is_channel_authorized(channel, username) ¶

on_authentication_failed(username, method) ¶

on_authentication_successful(username, method) ¶

on_channel_closed(channel) ¶

on_channel_opened(channel) ¶

set_server_key(server_key) ¶

start_server(sock, timeout=None) ¶

SSHServerManager ¶

Methods:¶

__init__(server_interface, server_key, bind_address='0.0.0.0', port=22) ¶

close_connection(connection_id) ¶

get_active_connections() ¶

get_connection_count() ¶

get_connection_stats() ¶

is_running() ¶

set_auth_timeout(timeout) ¶

set_connection_timeout(timeout) ¶

set_max_connections(max_connections) ¶

start_server() ¶

stop_server() ¶

Functions:¶

SFTP Server¶

spindlex.server.sftp_server ¶

Classes¶

SFTPHandle ¶

Methods:¶

__init__(handle_id, path, flags, file_obj=None) ¶

close() ¶

read(length) ¶

seek(offset) ¶

write(data) ¶

SFTPServer ¶

Methods:¶

__init__(channel, root_path='/', start_thread=True) ¶

check_directory_access(path, mode) ¶

check_file_access(path, mode) ¶

close() ¶

get_directory_permissions(path) ¶

get_file_permissions(path) ¶

`spindlex.server.ssh_server` ¶

`SSHServer` ¶

`init()` ¶

`check_auth_gssapi_with_mic(username, gss_authenticated, cc_file)` ¶

`check_auth_keyboard_interactive(username, submethods)` ¶

`check_auth_password(username, password)` ¶

`check_auth_publickey(username, key)` ¶

`check_channel_env_request(channel, name, value)` ¶

`check_channel_exec_request(channel, command)` ¶

`check_channel_pty_request(channel, term, width, height, pixelwidth, pixelheight, modes)` ¶

`check_channel_request(kind, chanid)` ¶

`check_channel_shell_request(channel)` ¶

`check_channel_subsystem_request(channel, name)` ¶

`check_channel_window_change_request(channel, width, height, pixelwidth, pixelheight)` ¶

`check_channel_x11_request(channel, single_connection, auth_protocol, auth_cookie, screen_number)` ¶

`check_global_request(kind, msg)` ¶

`check_port_forward_cancel_request(address, port)` ¶

`check_port_forward_request(address, port)` ¶

`close_all_channels()` ¶

`close_channel(channel)` ¶

`get_active_channels()` ¶

`get_allowed_auths(username)` ¶

`get_banner()` ¶

`get_channel_count()` ¶

`get_server_key()` ¶

`is_channel_authorized(channel, username)` ¶

`on_authentication_failed(username, method)` ¶

`on_authentication_successful(username, method)` ¶

`on_channel_closed(channel)` ¶

`on_channel_opened(channel)` ¶

`set_server_key(server_key)` ¶

`start_server(sock, timeout=None)` ¶

`SSHServerManager` ¶

`init(server_interface, server_key, bind_address='0.0.0.0', port=22)` ¶

`close_connection(connection_id)` ¶

`get_active_connections()` ¶

`get_connection_count()` ¶

`get_connection_stats()` ¶

`is_running()` ¶

`set_auth_timeout(timeout)` ¶

`set_connection_timeout(timeout)` ¶

`set_max_connections(max_connections)` ¶

`start_server()` ¶

`stop_server()` ¶

`spindlex.server.sftp_server` ¶

`SFTPHandle` ¶

`init(handle_id, path, flags, file_obj=None)` ¶

`close()` ¶

`read(length)` ¶

`seek(offset)` ¶

`write(data)` ¶

`SFTPServer` ¶

`init(channel, root_path='/', start_thread=True)` ¶

`check_directory_access(path, mode)` ¶

`check_file_access(path, mode)` ¶

`close()` ¶

`get_directory_permissions(path)` ¶

`get_file_permissions(path)` ¶