cai.utils package¶

Submodules¶

cai.utils.binary module¶

Binary Tools.

This module is used to build Binary tools.

Copyright: Copyright (C) 2021-2021 cscs181
License: AGPL-3.0 or later. See LICENSE for detail.

class cai.utils.binary.BasePacket¶

基类：bytearray

Base Packet Class.

Provides basic operations like reading, writing by offset.

Inherit from bytearray

classmethod build(*data: Union[bytes, cai.utils.binary.BasePacket]) → cai.utils.binary.P¶

Build new packet and write data into it.

参数: *data (Union[bytes, Packet]) – Data to write
返回: Current Packet
返回类型: Packet

write(*data: Union[bytes, cai.utils.binary.BasePacket]) → cai.utils.binary.P¶

Write data into current packet.

参数: *data (Union[bytes, Packet]) – Data to write
返回: Current Packet
返回类型: Packet

write_with_length(*data: Union[bytes, cai.utils.binary.BasePacket], offset: int = 0) → cai.utils.binary.P¶

Write data into current packet with 4-byte length.

参数

*data (Union[bytes, Packet]) – Data to write
offset (int) – Length offset

返回

Current Packet

返回类型

Packet

unpack(format: Union[bytes, str]) → Tuple[Any, ...]¶

Unpack all data from current packet.

参数: format (Union[bytes, str]) – Struct format.
返回: Unpacked data tuple.
返回类型: Tuple[Any, …]

unpack_from(format: Union[bytes, str], offset: int = 0) → Tuple[Any, ...]¶

Unpack data from current packet with given offset.

参数

format (Union[bytes, str]) – Struct format.
offset (int, optional) – Data offset. Defaults to 0.

返回

Unpacked data.

返回类型

Tuple[Any, …]

read_int8(offset: int = 0) → INT8¶

read_uint8(offset: int = 0) → UINT8¶

read_int16(offset: int = 0) → INT16¶

read_uint16(offset: int = 0) → UINT16¶

read_int32(offset: int = 0) → INT32¶

read_uint32(offset: int = 0) → UINT32¶

read_int64(offset: int = 0) → INT64¶

read_uint64(offset: int = 0) → UINT64¶

read_byte(offset: int = 0) → BYTE¶

read_bytes(n: int, offset: int = 0) → BYTES¶

read_string(offset: int = 0) → STRING¶

class cai.utils.binary.Packet(*args, **kwargs)¶

基类：cai.utils.binary.BasePacket

Packet Class for extracting data more efficiently.

Support PEP646 typing hints. Using pyright or pylance for type checking.

Example

>>> packet = Packet(bytes.fromhex("01000233000000"))
>>> packet.start().int8().uint16().bytes(4).execute()
(1, 2, b'3')

start(offset: int = 0)¶

bool()¶

int8()¶

uint8()¶

int16()¶

uint16()¶

int32()¶

uint32()¶

int64()¶

uint64()¶

float()¶

double()¶

byte()¶

bytes(length: int)¶

bytes_with_length(head_bytes: int, offset: int = 0)¶

string(head_bytes: int, offset: int = 0, encoding: str = 'utf-8')¶

offset(offset: int)¶

remain()¶

execute()¶

cai.utils.coroutine module¶

Coroutine Related Tools

This module is used to build coroutine related tools.

Copyright: Copyright (C) 2021-2021 cscs181
License: AGPL-3.0 or later. See LICENSE for detail.

class cai.utils.coroutine.ContextManager(coro: Coroutine[cai.utils.coroutine.TY, cai.utils.coroutine.TS, cai.utils.coroutine.TR])¶

基类：collections.abc.Coroutine, Generic[cai.utils.coroutine.TY, cai.utils.coroutine.TS, cai.utils.coroutine.TR]

send(value: cai.utils.coroutine.TS) → cai.utils.coroutine.TY¶: Send a value into the coroutine. Return next yielded value or raise StopIteration.

throw(typ, val=None, tb=None)¶: Raise an exception in the coroutine. Return next yielded value or raise StopIteration.

close()¶: Raise GeneratorExit inside coroutine.

cai.utils.crypto module¶

Crypto Tool.

This module is used to encrypt data using ECDH or Session ticket encryption.

Copyright: Copyright (C) 2021-2021 cscs181
License: AGPL-3.0 or later. See LICENSE for detail.

class cai.utils.crypto.ECDH¶

基类：object

id = 135¶

svr_public_key = <cryptography.hazmat.backends.openssl.ec._EllipticCurvePublicKey object>¶

client_private_key = <cryptography.hazmat.backends.openssl.ec._EllipticCurvePrivateKey object>¶

client_public_key = b'\x04\xf9\xc0\x97\xbdgF\xcf\x16f\xd6}Ab\xfdV\x18nB\x1b\xd83t|\xadW\xa5\xcd\xc3#\xc2M\xd2\x16*\x11\xef\x12\xb0\xa9\xb9xSY/\x90\x9bi\x04f\x82\xa1c\xad\x82\xdbR\xbe\xb6a\xfb1\x88\xcf\x9b'¶

share_key = b'\xf3!\xe3\xb7c:\xd2\xea;\x84I~]\xa7\x19\xcf'¶

classmethod encrypt(data: Union[bytes, cai.utils.binary.Packet], key: Union[bytes, cai.utils.binary.Packet]) → cai.utils.binary.Packet¶

class cai.utils.crypto.EncryptSession(ticket: bytes)¶

基类：object

id = 69¶

encrypt(data: Union[bytes, cai.utils.binary.Packet], key: Union[bytes, cai.utils.binary.Packet]) → cai.utils.binary.Packet¶

cai.utils.dataclass module¶

Dataclass Related Tools

This module is used to build dataclass related tools.

Copyright: Copyright (C) 2021-2021 cscs181
License: AGPL-3.0 or later. See LICENSE for detail.

class cai.utils.dataclass.JsonableDataclass(**kwargs)¶

基类：object

to_dict(only_json: bool = False) → Dict[str, Optional[Union[Dict[str, Optional[Union[Dict[str, JSON], List[JSON], str, int, float, bool]]], List[Optional[Union[Dict[str, JSON], List[JSON], str, int, float, bool]]], str, int, float, bool]]]¶

classmethod from_dict(kvs: Dict[str, Any]) → cai.utils.dataclass.T¶

to_json(only_json: bool = True, *, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, cls=None, indent=None, separators=None, default=None, sort_keys=False, **kw) → str¶

classmethod from_json(json_str: str, *, cls=None, object_hook=None, parse_float=None, parse_int=None, parse_constant=None, object_pairs_hook=None, **kw) → cai.utils.dataclass.T¶

to_file(fp: IO[str], only_json: bool = True, *, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, cls=None, indent=None, separators=None, default=None, sort_keys=False, **kw) → None¶

classmethod from_file(fp: IO[str], *, cls=None, object_hook=None, parse_float=None, parse_int=None, parse_constant=None, object_pairs_hook=None, **kw) → cai.utils.dataclass.T¶

cai.utils.future module¶

class cai.utils.future.FutureStore¶

基类：Generic[cai.utils.future.KT, cai.utils.future.VT]

store_seq(seq: cai.utils.future.KT) → asyncio.Future[VT]¶

store_result(seq: cai.utils.future.KT, result: cai.utils.future.VT)¶

pop_seq(seq: cai.utils.future.KT) → asyncio.Future[VT]¶

add_callback(seq: cai.utils.future.KT, callback: Callable[[asyncio.Future[VT]], Any])¶

remove_callback(seq: cai.utils.future.KT, callback: Callable[[asyncio.Future[VT]], Any]) → int¶

done(seq: cai.utils.future.KT) → bool¶

result(seq: cai.utils.future.KT) → cai.utils.future.VT¶

cancel(seq: cai.utils.future.KT) → bool¶

cancel_all()¶

exception(seq: cai.utils.future.KT) → Optional[BaseException]¶

async fetch(seq: cai.utils.future.KT, timeout: Optional[float] = None) → cai.utils.future.VT¶

cai.utils.jce module¶

JCE Related Tools

This module is used to build JCE related tools including packaging and serialization.

License: AGPL-3.0 or later. See LICENSE for detail.

class cai.utils.jce.RequestPacket(*, version: jce.types.INT16 = 0, pkg_type: jce.types.BYTE = b'\x00', msg_type: jce.types.INT32 = 0, req_id: jce.types.INT32 = 0, servant_name: jce.types.STRING, func_name: jce.types.STRING, buffer: jce.types.BYTES = b'', timeout: jce.types.INT32 = 0, context: jce.types.MAP = {}, status: jce.types.MAP = {})¶

基类：jce.types.JceStruct

注解

Source: com.qq.taf.RequestPacket

version: jce.types.INT16¶

pkg_type: jce.types.BYTE¶

msg_type: jce.types.INT32¶

req_id: jce.types.INT32¶

servant_name: jce.types.STRING¶

func_name: jce.types.STRING¶

buffer: jce.types.BYTES¶

timeout: jce.types.INT32¶

context: jce.types.MAP¶

status: jce.types.MAP¶

class cai.utils.jce.RequestPacketVersion2(*, version: jce.types.INT16 = 2, pkg_type: jce.types.BYTE = b'\x00', msg_type: jce.types.INT32 = 0, req_id: jce.types.INT32 = 0, servant_name: jce.types.STRING, func_name: jce.types.STRING, buffer: jce.types.BYTES = b'', timeout: jce.types.INT32 = 0, context: jce.types.MAP = {}, status: jce.types.MAP = {}, data: jce.types.MAP[jce.types.STRING, jce.types.MAP[jce.types.STRING, jce.types.JceType]] = None)¶

基类：cai.utils.jce.RequestPacket

注解

Source: com.qq.jce.wup.OldUniAttribute

version: jce.types.INT16¶

data: Optional[jce.types.MAP[jce.types.STRING, jce.types.MAP[jce.types.STRING, jce.types.JceType]]]¶

encode(with_length: bool = False) → bytes¶

classmethod decode(data: bytes, **extra) → cai.utils.jce.RequestPacketVersion2¶

class cai.utils.jce.RequestPacketVersion3(*, version: jce.types.INT16 = 3, pkg_type: jce.types.BYTE = b'\x00', msg_type: jce.types.INT32 = 0, req_id: jce.types.INT32 = 0, servant_name: jce.types.STRING, func_name: jce.types.STRING, buffer: jce.types.BYTES = b'', timeout: jce.types.INT32 = 0, context: jce.types.MAP = {}, status: jce.types.MAP = {}, data: jce.types.MAP[jce.types.STRING, jce.types.JceType] = None)¶

基类：cai.utils.jce.RequestPacket

注解

Source: com.qq.jce.wup.UniAttribute

version: jce.types.INT16¶

data: Optional[jce.types.MAP[jce.types.STRING, jce.types.JceType]]¶

encode(with_length: bool = False) → bytes¶

classmethod decode(data: bytes, **extra) → cai.utils.jce.RequestPacketVersion3¶