Utilities

This section shows general objects and helper functions that are used with this package.

Connections

class balderhub.rfb.lib.connections.RfbConnection(from_device: Type[Device] | None = None, to_device: Type[Device] | None = None, from_device_node_name: str | None = None, to_device_node_name: str | None = None)

Bases: Connection

special connection for rfb connections

Exceptions

class balderhub.rfb.lib.exceptions.RfbConnectionFailed

Bases: Exception

raised in case a rfb connection failed

class balderhub.rfb.lib.exceptions.RfbClientHandshakeSecurityTypeFailed

Bases: Exception

raised in case the security handshake failed

class balderhub.rfb.lib.exceptions.RfbClientSecResultFailed

Bases: Exception

raised in case the security result failed

Test Server

class balderhub.rfb.lib.utils.TcpThreadedServer(host: str, port: int)

Bases: Thread

Helper thread for a tcp server

CNN_HANDLER_THREAD_CLS

alias of TcpConnectionThread

handle_new_connection(sock, client_host, client_port)

callback method that handles a new incoming connection

Parameters:

  • sock – the connection socket

  • client_host – the remote client host

  • client_port – the remote client port

property host: str
Returns:

returns the host of the tcp server

property port: int
Returns:

returns the port of the tcp server

run()

the threaded method

shutdown(timeout: float)

Note

The timeout should be higher than 1, because the inner select uses a timeout of 1 second

Parameters:

timeout – max time this method will block

start() → None

starts the thread

property thread_exc: BaseException | None
Returns:

holds the exception that was thrown in the thread (or None if there was no exception in the thread yet)

wait_till_thread_loop_is_active(timeout=10)

waits for the thread loop to terminate

Parameters:

timeout – max time this method will block

class balderhub.rfb.lib.utils.TcpConnectionThread(server, sock: socket, address_info: Tuple[str, int])

Bases: Thread

Helper thread for every new tcp connection

property client_host: str
Returns:

returns the host of the connection client

property client_port: int
Returns:

returns the port of the connection client

handle_new_message(data: bytes)

method that will be called in case a new message arrived

handle_no_message()

callback method that will be triggered for every iteration at which no message was received

run()

the threaded method

property thread_exc: BaseException | None
Returns:

holds the exception that was thrown in the thread (or None if there was no exception in the thread yet)

trigger_stop()

will trigger the stop process for this thread

class balderhub.rfb.lib.utils.RfbServer(host: str, port: int)

Bases: TcpThreadedServer

threaded rfb test server

CNN_HANDLER_THREAD_CLS

alias of RfbConnectionThread

filter_available_security_types_for_next_upcoming_connection(available_types: List[int])

This method filters the available security types for the next upcoming connection. This makes it possible to specify which security types the server should support in the next connection.

Parameters:

available_types – a list of all remaining available security types

handle_new_connection(sock, client_host, client_port)

callback method that handles a new incoming connection

Parameters:

  • sock – the connection socket

  • client_host – the remote client host

  • client_port – the remote client port

register_cb_for_next_upcoming_connection(cb_key: str, cb_function: Callable[[RfbConnectionThread, bytes], None])

This method allows to register a custom callback for the next incoming connection.

Parameters:

  • cb_key – the callback key

  • cb_function – the callback function

class balderhub.rfb.lib.utils.RfbConnectionThread(server: RfbServer, sock: socket, address_info: Tuple[str, int])

Bases: TcpConnectionThread

PASSWORD = None

holds the password that should be used by the server

SECURITY_TYPE_CALLABLES: Dict[int, str] = {1: 'handshake-auth-none', 2: 'handshake-auth-vnc'}

holds the callbacks of all registered security callbacks (sends 0 if there are no!)

SUPPORTED_PROTOCOL_VERSIONS = [(3, 3), (3, 7), (3, 8)]

holds all supported protocol versions

cb_default_handshake_recv_protocol_version(received_data: bytes)

Default callback method that will be executed as soon as the client should answer with the selected protocol version

Parameters:

received_data – the received data from the previous message

cb_default_handshake_recv_selected_security_type(received_data: bytes)

Default callback method that will be executed as soon as the client should answer with the selected security type

Parameters:

received_data – the received data from the previous message

cb_default_handshake_send_available_security_types(received_data: bytes)

Default callback method that will be executed as soon as the server should send the available security types

Parameters:

received_data – the received data from the previous message

cb_default_handshake_send_protocol_version(received_data: bytes)

Default callback method that will be executed as soon as the server should send the protocol version

Parameters:

received_data – the received data from the previous message

cb_default_handshake_send_security_type_3_3(received_data: bytes)

Default callback method that will be executed if the server only allows the security type 3.3 and as soon as it needs to send it

Parameters:

received_data – the received data from the previous message

challenge

holds the send challenge or None if no challenge was sent to client

handle_new_message(data: bytes)

method that will be called in case a new message arrived

handle_no_message()

callback method that will be triggered for every iteration at which no message was received

handle_rfb_messages()

callback that will be executed for every incoming rfb message

register_new_expected_msg(byte_count: int | None, callback: str)

Method that registers a new expected message

Parameters:

  • byte_count – holds the count of bytes, the incoming message has

  • callback – holds the callback that should be executed as soon as bytes was received

run()

the threaded method

security_result

holds the security result of this connection

send_conn_failed_msg(msg: bytes | None = None, close_connection: bool = True)

This method sends a rfb failed message and terminates the connection thread (in case close_connection is True)

Parameters:

  • msg – the message text that should be send

  • close_connection – True if the thread should terminate after sending the error message

validate_password(encrypted_bytes: bytes)

RFB protocol for authentication requires client to encrypt challenge sent by server with password using DES method. However, bits in each byte of the password are put in reverse order before using it as encryption key (MSB is required).

Parameters:

encrypted_bytes – the received encrypted bytes

Returns:

True if the encrypted bytes are valid

Test Clients