diff options
Diffstat (limited to 'kvmd/apps/vnc/rfb')
| -rw-r--r-- | kvmd/apps/vnc/rfb/__init__.py | 11 | ||||
| -rw-r--r-- | kvmd/apps/vnc/rfb/crypto.py | 4 | ||||
| -rw-r--r-- | kvmd/apps/vnc/rfb/encodings.py | 14 | ||||
| -rw-r--r-- | kvmd/apps/vnc/rfb/stream.py | 7 |
4 files changed, 12 insertions, 24 deletions
diff --git a/kvmd/apps/vnc/rfb/__init__.py b/kvmd/apps/vnc/rfb/__init__.py index 059d4d1d..f715f9b7 100644 --- a/kvmd/apps/vnc/rfb/__init__.py +++ b/kvmd/apps/vnc/rfb/__init__.py @@ -23,9 +23,6 @@ import asyncio import ssl -from typing import Tuple -from typing import List -from typing import Dict from typing import Callable from typing import Coroutine @@ -66,7 +63,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute width: int, height: int, name: str, - vnc_passwds: List[str], + vnc_passwds: list[str], vencrypt: bool, none_auth_only: bool, ) -> None: @@ -103,7 +100,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute finally: await aiotools.shield_fg(self.__cleanup(tasks)) - async def __cleanup(self, tasks: List[asyncio.Task]) -> None: + async def __cleanup(self, tasks: list[asyncio.Task]) -> None: for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) @@ -150,7 +147,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute async def _on_ext_key_event(self, code: int, state: bool) -> None: raise NotImplementedError - async def _on_pointer_event(self, buttons: Dict[str, bool], wheel: Dict[str, int], move: Dict[str, int]) -> None: + async def _on_pointer_event(self, buttons: dict[str, bool], wheel: dict[str, int], move: dict[str, int]) -> None: raise NotImplementedError async def _on_cut_event(self, text: str) -> None: @@ -232,7 +229,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute # ===== async def __handshake_security(self) -> None: - sec_types: Dict[int, Tuple[str, Callable]] = {} + sec_types: dict[int, tuple[str, Callable]] = {} if self.__vencrypt and self.__rfb_version > 3: sec_types[19] = ("VeNCrypt", self.__handshake_security_vencrypt) if self.__none_auth_only: diff --git a/kvmd/apps/vnc/rfb/crypto.py b/kvmd/apps/vnc/rfb/crypto.py index 854dceb3..8953bef5 100644 --- a/kvmd/apps/vnc/rfb/crypto.py +++ b/kvmd/apps/vnc/rfb/crypto.py @@ -22,8 +22,6 @@ import os -from typing import List - import passlib.crypto.des @@ -43,7 +41,7 @@ def rfb_encrypt_challenge(challenge: bytes, passwd: bytes) -> bytes: def _make_key(passwd: bytes) -> bytes: passwd = (passwd + b"\0" * 8)[:8] - key: List[int] = [] + key: list[int] = [] for ch in passwd: btgt = 0 for index in range(8): diff --git a/kvmd/apps/vnc/rfb/encodings.py b/kvmd/apps/vnc/rfb/encodings.py index de51dde5..e153b5e8 100644 --- a/kvmd/apps/vnc/rfb/encodings.py +++ b/kvmd/apps/vnc/rfb/encodings.py @@ -22,10 +22,6 @@ import dataclasses -from typing import List -from typing import Dict -from typing import FrozenSet -from typing import Union from typing import Any @@ -45,13 +41,13 @@ class RfbEncodings: H264 = 50 # Open H.264 Encoding -def _make_meta(variants: Union[int, FrozenSet[int]]) -> Dict: +def _make_meta(variants: (int | frozenset[int])) -> dict: return {"variants": (frozenset([variants]) if isinstance(variants, int) else variants)} @dataclasses.dataclass(frozen=True) class RfbClientEncodings: # pylint: disable=too-many-instance-attributes - encodings: FrozenSet[int] + encodings: frozenset[int] has_resize: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RESIZE)) # noqa: E224 has_rename: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RENAME)) # noqa: E224 @@ -63,8 +59,8 @@ class RfbClientEncodings: # pylint: disable=too-many-instance-attributes has_h264: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.H264)) # noqa: E224 - def get_summary(self) -> List[str]: - summary: List[str] = [f"encodings -- {sorted(self.encodings)}"] + def get_summary(self) -> list[str]: + summary: list[str] = [f"encodings -- {sorted(self.encodings)}"] for field in dataclasses.fields(self): if field.name != "encodings": found = ", ".join(map(str, sorted(map(int, self.__get_found(field))))) @@ -80,7 +76,7 @@ class RfbClientEncodings: # pylint: disable=too-many-instance-attributes def __set_value(self, key: str, value: Any) -> None: object.__setattr__(self, key, value) - def __get_found(self, field: dataclasses.Field) -> FrozenSet[int]: + def __get_found(self, field: dataclasses.Field) -> frozenset[int]: return self.encodings.intersection(field.metadata["variants"]) def __get_tight_jpeg_quality(self) -> int: diff --git a/kvmd/apps/vnc/rfb/stream.py b/kvmd/apps/vnc/rfb/stream.py index ee42436e..dfa6cb13 100644 --- a/kvmd/apps/vnc/rfb/stream.py +++ b/kvmd/apps/vnc/rfb/stream.py @@ -24,9 +24,6 @@ import asyncio import ssl import struct -from typing import Tuple -from typing import Union - from .... import aiotools from .errors import RfbConnectionError @@ -57,7 +54,7 @@ class RfbClientStream: except (ConnectionError, asyncio.IncompleteReadError) as err: raise RfbConnectionError(f"Can't read {msg}", err) - async def _read_struct(self, msg: str, fmt: str) -> Tuple[int, ...]: + async def _read_struct(self, msg: str, fmt: str) -> tuple[int, ...]: assert len(fmt) > 1 try: fmt = f">{fmt}" @@ -73,7 +70,7 @@ class RfbClientStream: # ===== - async def _write_struct(self, msg: str, fmt: str, *values: Union[int, bytes], drain: bool=True) -> None: + async def _write_struct(self, msg: str, fmt: str, *values: (int | bytes), drain: bool=True) -> None: try: if not fmt: for value in values: |