Module mudproto.charset

Charset protocol.

Class CharsetMixIn(TelnetInterface)

A charset mix in class for the Telnet protocol.

Source code in mudproto/charset.py

class CharsetMixIn(TelnetInterface):
	"""A charset mix in class for the Telnet protocol."""

	def __init__(self, *args: Any, **kwargs: Any) -> None:
		"""
		Defines the constructor.

		Args:
			*args: Positional arguments to be passed to the parent constructor.
			**kwargs: Key-word only arguments to be passed to the parent constructor.
		"""
		super().__init__(*args, **kwargs)
		self.subnegotiation_map[CHARSET] = self.on_charset
		self._charsets: tuple[bytes, ...] = (b"US-ASCII",)
		self._charset: bytes = self._charsets[0]

	@property
	def charset(self) -> str:
		"""The currently used character set."""
		return str(self._charset, "us-ascii")

	def negotiate_charset(self, name: bytes | str) -> None:
		"""
		Negotiates changing the character set.

		Args:
			name: The name of the character set to use.
		"""
		separator: bytes = b";"
		if not isinstance(name, str):
			name = str(name, "us-ascii")
		try:
			target = codecs.lookup(name).name
		except LookupError:
			logger.warning(f"'{name}' not a valid codec")
			return
		for item in self._charsets:
			if target == codecs.lookup(str(item, "us-ascii")).name:
				logger.debug(f"Tell peer we would like to use the {item!r} charset.")
				self.request_negotiation(CHARSET, CHARSET_REQUEST + separator + item)
				return
		logger.warning(f"Could not find any charsets which target '{target}'")

	@staticmethod
	def parse_supported_charsets(response: bytes) -> tuple[bytes, ...]:
		"""
		Parses the supported character sets from peer.

		Args:
			response: The response from peer, containing the supported character sets.

		Returns:
			The character sets supported by peer, with duplicate aliases removed.
		"""
		charsets: list[bytes] = []
		names: set[str] = set()
		separator, response = response[:1], response[1:]
		for item in response.split(separator):
			with suppress(LookupError):
				name = codecs.lookup(str(item, "us-ascii")).name
				if name not in names:
					charsets.append(item)
					names.add(name)
		return tuple(charsets)

	def on_charset(self, data: bytes) -> None:
		"""
		Called when a charset subnegotiation is received.

		Args:
			data: The payload.
		"""
		status, response = data[:1], data[1:]
		if status == CHARSET_REQUEST:
			self._charsets = self.parse_supported_charsets(response)
			logger.debug(f"Peer responds: Supported charsets: {self._charsets!r}.")
			self.negotiate_charset(self._charset)
		elif status == CHARSET_ACCEPTED:
			logger.debug(f"Peer responds: Charset {response!r} accepted.")
			self._charset = response
		elif status == CHARSET_REJECTED:
			logger.warning("Peer responds: Charset rejected.")
		else:
			logger.warning(f"Unknown charset negotiation response from peer: {data!r}")
			self.wont(CHARSET)

	def on_enable_local(self, option: bytes) -> bool:  # NOQA: D102
		if option == CHARSET:
			logger.debug("Charset negotiation enabled.")
			return True
		return bool(super().on_enable_local(option))  # pragma: no cover

	def on_disable_local(self, option: bytes) -> None:  # NOQA: D102
		if option == CHARSET:
			logger.debug("Charset negotiation disabled.")
			return
		super().on_disable_local(option)  # type: ignore[safe-super]  # pragma: no cover

Attribute charset: str property readonly

The currently used character set.

Attribute is_client: bool inherited property readonly

True if acting as a client, False otherwise.

Attribute is_server: bool inherited property readonly

True if acting as a server, False otherwise.

Method __init__(self, *args, **kwargs) special

Defines the constructor.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments to be passed to the parent constructor.	()
**kwargs	Any	Key-word only arguments to be passed to the parent constructor.	{}

Source code in mudproto/charset.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
	"""
	Defines the constructor.

	Args:
		*args: Positional arguments to be passed to the parent constructor.
		**kwargs: Key-word only arguments to be passed to the parent constructor.
	"""
	super().__init__(*args, **kwargs)
	self.subnegotiation_map[CHARSET] = self.on_charset
	self._charsets: tuple[bytes, ...] = (b"US-ASCII",)
	self._charset: bytes = self._charsets[0]

Method do(self, option) inherited

Requests that the peer enable an option.

Parameters:

Name	Type	Description	Default
option	bytes	The option to enable.	required

Source code in mudproto/charset.py

@abstractmethod
def do(self, option: bytes) -> None:
	"""
	Requests that the peer enable an option.

	Args:
		option: The option to enable.
	"""

Method dont(self, option) inherited

Requests that the peer disable an option.

Parameters:

Name	Type	Description	Default
option	bytes	The option to disable.	required

Source code in mudproto/charset.py

@abstractmethod
def dont(self, option: bytes) -> None:
	"""
	Requests that the peer disable an option.

	Args:
		option: The option to disable.
	"""

Method get_option_state(self, option) inherited

Gets the state of a Telnet option.

Parameters:

Name	Type	Description	Default
option	bytes	The option to get state.	required

Returns:

Type	Description
_OptionState	An object containing the option state.

Source code in mudproto/charset.py

@abstractmethod
def get_option_state(self, option: bytes) -> _OptionState:
	"""
	Gets the state of a Telnet option.

	Args:
		option: The option to get state.

	Returns:
		An object containing the option state.
	"""

Method negotiate_charset(self, name)

Negotiates changing the character set.

Parameters:

Name	Type	Description	Default
name	bytes | str	The name of the character set to use.	required

Source code in mudproto/charset.py

def negotiate_charset(self, name: bytes | str) -> None:
	"""
	Negotiates changing the character set.

	Args:
		name: The name of the character set to use.
	"""
	separator: bytes = b";"
	if not isinstance(name, str):
		name = str(name, "us-ascii")
	try:
		target = codecs.lookup(name).name
	except LookupError:
		logger.warning(f"'{name}' not a valid codec")
		return
	for item in self._charsets:
		if target == codecs.lookup(str(item, "us-ascii")).name:
			logger.debug(f"Tell peer we would like to use the {item!r} charset.")
			self.request_negotiation(CHARSET, CHARSET_REQUEST + separator + item)
			return
	logger.warning(f"Could not find any charsets which target '{target}'")

Method on_charset(self, data)

Called when a charset subnegotiation is received.

Parameters:

Name	Type	Description	Default
data	bytes	The payload.	required

Source code in mudproto/charset.py

def on_charset(self, data: bytes) -> None:
	"""
	Called when a charset subnegotiation is received.

	Args:
		data: The payload.
	"""
	status, response = data[:1], data[1:]
	if status == CHARSET_REQUEST:
		self._charsets = self.parse_supported_charsets(response)
		logger.debug(f"Peer responds: Supported charsets: {self._charsets!r}.")
		self.negotiate_charset(self._charset)
	elif status == CHARSET_ACCEPTED:
		logger.debug(f"Peer responds: Charset {response!r} accepted.")
		self._charset = response
	elif status == CHARSET_REJECTED:
		logger.warning("Peer responds: Charset rejected.")
	else:
		logger.warning(f"Unknown charset negotiation response from peer: {data!r}")
		self.wont(CHARSET)

Method on_command(self, command, option) inherited

Called when a 1 or 2 byte command is received.

Parameters:

Name	Type	Description	Default
command	bytes	The first byte in a 1 or 2 byte negotiation sequence.	required
option	bytes | None	The second byte in a 2 byte negotiation sequence or None.	required

Source code in mudproto/charset.py

@abstractmethod
def on_command(self, command: bytes, option: bytes | None) -> None:
	"""
	Called when a 1 or 2 byte command is received.

	Args:
		command: The first byte in a 1 or 2 byte negotiation sequence.
		option: The second byte in a 2 byte negotiation sequence or None.
	"""

Method on_connection_lost(self) inherited

Called by disconnect when a connection to peer has been lost.

Source code in mudproto/charset.py

@abstractmethod
def on_connection_lost(self) -> None:
	"""Called by `disconnect` when a connection to peer has been lost."""

Method on_connection_made(self) inherited

Called by connect when a connection to peer has been established.

Source code in mudproto/charset.py

@abstractmethod
def on_connection_made(self) -> None:
	"""Called by `connect` when a connection to peer has been established."""

Method on_data_received(self, data) inherited

Called by parse when data is received.

Parameters:

Name	Type	Description	Default
data	bytes	The received data.	required

Source code in mudproto/charset.py

@abstractmethod
def on_data_received(self, data: bytes) -> None:
	"""
	Called by `parse` when data is received.

	Args:
		data: The received data.
	"""
	self._receiver(data)

Method on_disable_local(self, option)

Disables a locally managed option.

This method is called before we disable a locally enabled option, in order to perform any necessary cleanup.

Note

If on_enable_local is overridden, this method must be overridden as well.

Parameters:

Name	Type	Description	Default
option	bytes	The option being disabled.	required

Source code in mudproto/charset.py

def on_disable_local(self, option: bytes) -> None:  # NOQA: D102
	if option == CHARSET:
		logger.debug("Charset negotiation disabled.")
		return
	super().on_disable_local(option)  # type: ignore[safe-super]  # pragma: no cover

Method on_disable_remote(self, option) inherited

Disables a remotely managed option.

This method is called when peer disables a remotely enabled option, in order to perform any necessary cleanup on our end.

Note

If on_enable_remote is overridden, this method must be overridden as well.

Parameters:

Name	Type	Description	Default
option	bytes	The option being disabled.	required

Source code in mudproto/charset.py

@abstractmethod
def on_disable_remote(self, option: bytes) -> None:
	"""
	Disables a remotely managed option.

	This method is called when peer disables a remotely enabled option,
	in order to perform any necessary cleanup on our end.

	Note:
		If on_enable_remote is overridden, this method must be overridden as well.

	Args:
		option: The option being disabled.
	"""
	raise NotImplementedError(f"Don't know how to disable remote Telnet option {option!r}")

Method on_enable_local(self, option)

Called to accept or reject the request for us to manage the option.

Parameters:

Name	Type	Description	Default
option	bytes	The option that peer requests us to handle.	required

Returns:

Type	Description
bool	True if we will handle the option, False otherwise.

Source code in mudproto/charset.py

def on_enable_local(self, option: bytes) -> bool:  # NOQA: D102
	if option == CHARSET:
		logger.debug("Charset negotiation enabled.")
		return True
	return bool(super().on_enable_local(option))  # pragma: no cover

Method on_enable_remote(self, option) inherited

Called to accept or reject the request for peer to manage the option.

Parameters:

Name	Type	Description	Default
option	bytes	The option that peer wants to handle.	required

Returns:

Type	Description
bool	True if we will allow peer to handle the option, False otherwise.

Source code in mudproto/charset.py

@abstractmethod
def on_enable_remote(self, option: bytes) -> bool:
	"""
	Called to accept or reject the request for peer to manage the option.

	Args:
		option: The option that peer wants to handle.

	Returns:
		True if we will allow peer to handle the option, False otherwise.
	"""
	return False  # Reject all options by default.

Method on_option_enabled(self, option) inherited

Called after an option has been fully enabled.

Parameters:

Name	Type	Description	Default
option	bytes	The option that has been enabled.	required

Source code in mudproto/charset.py

@abstractmethod
def on_option_enabled(self, option: bytes) -> None:
	"""
	Called after an option has been fully enabled.

	Args:
		option: The option that has been enabled.
	"""

Method on_subnegotiation(self, option, data) inherited

Called when a subnegotiation is received.

Parameters:

Name	Type	Description	Default
option	bytes	The subnegotiation option.	required
data	bytes	The payload.	required

Source code in mudproto/charset.py

@abstractmethod
def on_subnegotiation(self, option: bytes, data: bytes) -> None:
	"""
	Called when a subnegotiation is received.

	Args:
		option: The subnegotiation option.
		data: The payload.
	"""

Method on_unhandled_command(self, command, option) inherited

Called for commands for which no handler is installed.

Parameters:

Name	Type	Description	Default
command	bytes	The first byte in a 1 or 2 byte negotiation sequence.	required
option	bytes | None	The second byte in a 2 byte negotiation sequence or None.	required

Source code in mudproto/charset.py

@abstractmethod
def on_unhandled_command(self, command: bytes, option: bytes | None) -> None:
	"""
	Called for commands for which no handler is installed.

	Args:
		command: The first byte in a 1 or 2 byte negotiation sequence.
		option: The second byte in a 2 byte negotiation sequence or None.
	"""

Method on_unhandled_subnegotiation(self, option, data) inherited

Called for subnegotiations for which no handler is installed.

Parameters:

Name	Type	Description	Default
option	bytes	The subnegotiation option.	required
data	bytes	The payload.	required

Source code in mudproto/charset.py

@abstractmethod
def on_unhandled_subnegotiation(self, option: bytes, data: bytes) -> None:
	"""
	Called for subnegotiations for which no handler is installed.

	Args:
		option: The subnegotiation option.
		data: The payload.
	"""

Method parse_supported_charsets(response) staticmethod

Parses the supported character sets from peer.

Parameters:

Name	Type	Description	Default
response	bytes	The response from peer, containing the supported character sets.	required

Returns:

Type	Description
tuple[bytes, ...]	The character sets supported by peer, with duplicate aliases removed.

Source code in mudproto/charset.py

@staticmethod
def parse_supported_charsets(response: bytes) -> tuple[bytes, ...]:
	"""
	Parses the supported character sets from peer.

	Args:
		response: The response from peer, containing the supported character sets.

	Returns:
		The character sets supported by peer, with duplicate aliases removed.
	"""
	charsets: list[bytes] = []
	names: set[str] = set()
	separator, response = response[:1], response[1:]
	for item in response.split(separator):
		with suppress(LookupError):
			name = codecs.lookup(str(item, "us-ascii")).name
			if name not in names:
				charsets.append(item)
				names.add(name)
	return tuple(charsets)

Method request_negotiation(self, option, data) inherited

Sends a subnegotiation message to the peer.

Parameters:

Name	Type	Description	Default
option	bytes	The subnegotiation option.	required
data	bytes	The payload.	required

Source code in mudproto/charset.py

@abstractmethod
def request_negotiation(self, option: bytes, data: bytes) -> None:
	"""
	Sends a subnegotiation message to the peer.

	Args:
		option: The subnegotiation option.
		data: The payload.
	"""

Method will(self, option) inherited

Indicates our willingness to enable an option.

Parameters:

Name	Type	Description	Default
option	bytes	The option to accept.	required

Source code in mudproto/charset.py

@abstractmethod
def will(self, option: bytes) -> None:
	"""
	Indicates our willingness to enable an option.

	Args:
		option: The option to accept.
	"""

Method wont(self, option) inherited

Indicates we are not willing to enable an option.

Parameters:

Name	Type	Description	Default
option	bytes	The option to reject.	required

Source code in mudproto/charset.py

@abstractmethod
def wont(self, option: bytes) -> None:
	"""
	Indicates we are not willing to enable an option.

	Args:
		option: The option to reject.
	"""

Method write(self, data) inherited

Writes data to peer.

Parameters:

Name	Type	Description	Default
data	bytes	The bytes to be written.	required

Source code in mudproto/charset.py

def write(self, data: bytes) -> None:
	"""
	Writes data to peer.

	Args:
		data: The bytes to be written.
	"""
	self._writer(data)