Library Internals¶

This document describes the internals of the library. It is intended for developers who want to contribute to the library.

pyquickshare¶

This fully documents the pyquickshare module. Including private members.

Quick Share implementation in Python.

class pyquickshare.ReceiveMode(value)

Bases: Enum

FILES = 2

TEXT = 3

WIFI = 1

class pyquickshare.ShareRequest(header: PayloadHeader, pin: str)

Bases: object

Parameters:

header (offline_wire_formats_pb2.PayloadTransferFrame.PayloadHeader)
pin (str)

async accept() → list[FileResult | TextResult | WifiResult]

Return type:: list[FileResult | TextResult | WifiResult]

async reject() → None

Return type:: None

pyquickshare._decrypt(frame: SecureMessage, keychain: Keychain) → OfflineFrame

Parameters:

frame (SecureMessage)
keychain (Keychain)

Return type:

OfflineFrame

pyquickshare._derive_endpoint_id_from_mac(mac: bytes) → bytes

Parameters:: mac (bytes)
Return type:: bytes

pyquickshare._generate_accept() → bytes

Return type:: bytes

pyquickshare._generate_connection_response() → OfflineFrame

Return type:: OfflineFrame

pyquickshare._generate_file_metadata(fp: str, id: int) → FileMetadata

Parameters:

fp (str)
id (int)

Return type:

FileMetadata

pyquickshare._generate_paired_key_encryption() → Frame

Return type:: Frame

async pyquickshare._handle_client(requests: Queue[ShareRequest], reader: StreamReader, writer: StreamWriter) → None

Parameters:

requests (Queue[ShareRequest])
reader (StreamReader)
writer (StreamWriter)

Return type:

None

async pyquickshare._handle_target(file: str, reader: StreamReader, writer: StreamWriter) → None

Parameters:

file (str)
reader (StreamReader)
writer (StreamWriter)

Return type:

None

async pyquickshare._iter_payload_messages(reader: asyncio.StreamReader, keychain: Keychain) → AsyncIterator[tuple[offline_wire_formats_pb2.PayloadTransferFrame.PayloadHeader, bytes]]

Parameters:

reader (asyncio.StreamReader)
keychain (Keychain)

Return type:

AsyncIterator[tuple[offline_wire_formats_pb2.PayloadTransferFrame.PayloadHeader, bytes]]

async pyquickshare._keep_alive(send: Callable[[bytes], Awaitable[None]]) → None

Parameters:: send (Callable[[bytes], Awaitable[None]])
Return type:: None

pyquickshare._make_send(writer: asyncio.StreamWriter, keychain: Keychain, sequence_number: Callable[[], int]) → Callable[[bytes], Awaitable[None]]

Parameters:

writer (asyncio.StreamWriter)
keychain (Keychain)
sequence_number (Callable[[], int])

Return type:

Callable[[bytes], Awaitable[None]]

pyquickshare._make_sequence_number() → Callable[[], int]

Return type:: Callable[[], int]

pyquickshare._mime_to_type(mime_type: str) → <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fd77a912850>

Parameters:: mime_type (str)
Return type:: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fd77a912850>

pyquickshare._payloadify(frame: bytes, keychain: Keychain, *, id: int, flags: int, type: offline_wire_formats_pb2.PayloadTransferFrame.PayloadHeader.PayloadType, file_name: str | None = None, offset: int = 0, total_size: int | None = None, sequence_number: Callable[[], int]) → bytes

Parameters:

frame (bytes)
keychain (Keychain)
id (int)
flags (int)
type (offline_wire_formats_pb2.PayloadTransferFrame.PayloadHeader.PayloadType)
file_name (str | None)
offset (int)
total_size (int | None)
sequence_number (Callable[[], int])

Return type:

bytes

pyquickshare._pick_mac_deterministically(interfaces: list[str]) → bytes

Parameters:: interfaces (list[str])
Return type:: bytes

async pyquickshare._send_file(*, file: str, writer: asyncio.StreamWriter, keychain: Keychain, sequence_number: Callable[[], int], id: int) → None

Parameters:

file (str)
writer (asyncio.StreamWriter)
keychain (Keychain)
sequence_number (Callable[[], int])
id (int)

Return type:

None

async pyquickshare._socket_server(port: int, requests: Queue[ShareRequest]) → None

Parameters:

port (int)
requests (Queue[ShareRequest])

Return type:

None

async pyquickshare._start_mdns_service(services: list[AsyncServiceInfo]) → None

Parameters:: services (list[AsyncServiceInfo])
Return type:: None

async pyquickshare.discover_services() → AsyncIterator[AsyncServiceInfo]

Discover services on the network.

Example

async for service in discover_services():
    print(service)

Return type:: AsyncIterator[AsyncServiceInfo]

pyquickshare.generate_enpoint_id() → bytes

Generate a random 4-byte endpoint ID.

Example

endpoint_id = generate_enpoint_id()
async for request in receive(endpoint_id=endpoint_id):
    ...

Returns:: The generated endpoint ID
Return type:: bytes

pyquickshare.parse_n(n: bytes) → tuple[bool, Type, str]

Parameters:: n (bytes)
Return type:: tuple[bool, Type, str]

async pyquickshare.receive(*, endpoint_id: bytes | None = None) → AsyncIterator[ShareRequest]

Receive something over Quick Share. Runs forever.

This function registers an mDNS service and opens a socket server to receive data. If firewalld is available, it temporarily reconfigures firewalld to allow incoming connections on the port.

Yields:: ShareRequest – A request to share something
Parameters:: endpoint_id (bytes | None)
Return type:: AsyncIterator[ShareRequest]

Example

async for request in receive():
    results = await request.accept()
    print(results)

async pyquickshare.send_to(service: AsyncServiceInfo, *, file: str) → None

Send a file to a service.

Parameters:

service (AsyncServiceInfo) – The service to send the file to
file (str) – The file to send

Return type:

None

pyquickshare.to_pin(bytes_: bytes) → str

Parameters:: bytes_ (bytes)
Return type:: str

firewalld¶

Module to command firewalld to open a port to allow incoming connections temporarily.

async pyquickshare.firewalld.get_proxy_object(bus: MessageBus, name: str, path: str, introspection: Any = None) → Any¶

Parameters:

bus (MessageBus)
name (str)
path (str)
introspection (Any)

Return type:

Any

async pyquickshare.firewalld.temporarily_open_port(interface: str, port: int = 8080) → None¶

Parameters:

interface (str)
port (int)

Return type:

None

ukey2¶

For some reason, Sphinx cannot seem to resolve some of the typehints from the protobuf generated code. That’s why some typehints may look odd.

class pyquickshare.ukey2.Keychain(decrypt_key, receive_hmac_key, encrypt_key, send_hmac_key, auth_string)¶

Bases: NamedTuple

Parameters:

decrypt_key (bytes)
receive_hmac_key (bytes)
encrypt_key (bytes)
send_hmac_key (bytes)
auth_string (bytes)

auth_string: bytes¶: Alias for field number 4

decrypt_key: bytes¶: Alias for field number 0

encrypt_key: bytes¶: Alias for field number 2

receive_hmac_key: bytes¶: Alias for field number 1

send_hmac_key: bytes¶: Alias for field number 3

pyquickshare.ukey2.decode_public_key(generic_key: GenericPublicKey) → EllipticCurvePublicKey¶

Decode a GenericPublicKey as an EllipticCurvePublicKey.

Parameters:: generic_key (securemessage_pb2.GenericPublicKey) – The public key to decode
Raises:: ValueError – If the key is not EC_P256
Returns:: The decoded public key
Return type:: ec.EllipticCurvePublicKey

pyquickshare.ukey2.derive_keys(m1: bytes, m2: bytes, private_key: EllipticCurvePrivateKey, peer_public_key: EllipticCurvePublicKey) → Keychain¶

Derive the keys for a session.

Parameters:

m1 (bytes) – The first message (CLIENT_INIT or SERVER_INIT)
m2 (bytes) – The second message (SERVER_INIT or CLIENT_INIT)
private_key (ec.EllipticCurvePrivateKey) – Our private key
peer_public_key (ec.EllipticCurvePublicKey) – The peer’s public key

Returns:

The derived keys

Return type:

Keychain

async pyquickshare.ukey2.do_client_key_exchange(reader: StreamReader, writer: StreamWriter) → Keychain | None¶

Perform a client key exchange.

Parameters:

reader (asyncio.StreamReader) – The reader
writer (asyncio.StreamWriter) – The writer

Returns:

The keychain, or None if the exchange failed

Return type:

Keychain | None

async pyquickshare.ukey2.do_server_key_exchange(reader: StreamReader, writer: StreamWriter) → Keychain | None¶

Perform a server key exchange.

Parameters:

reader (asyncio.StreamReader) – The reader
writer (asyncio.StreamWriter) – The writer

Returns:

The keychain, or None if the exchange failed

Return type:

Keychain | None

pyquickshare.ukey2.encode_public_key(public_key: EllipticCurvePublicKey) → GenericPublicKey¶

Encode an EllipticCurvePublicKey as a GenericPublicKey.

Parameters:: public_key (ec.EllipticCurvePublicKey) – The public key to encode
Returns:: The encoded public key
Return type:: securemessage_pb2.GenericPublicKey

pyquickshare.ukey2.from_twos_complement(data: bytes) → int¶

Convert a two’s complement byte array to an integer.

Parameters:: data (bytes) – The two’s complement byte array
Returns:: The integer
Return type:: int

pyquickshare.ukey2.make_alert(alert_type: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fd77af0e050>, error_message: str) → Ukey2Message¶

Construct an alert message.

Parameters:

alert_type (ukey_pb2.Ukey2Alert.AlertType) – The alert type
error_message (str) – The error message

Returns:

The constructed alert message

Return type:

ukey_pb2.Ukey2Message

async pyquickshare.ukey2.parse_client_finished(raw_message: bytes, commitment: CipherCommitment, writer: StreamWriter) → EllipticCurvePublicKey | None¶

Parse a CLIENT_FINISH message.

Parameters:

raw_message (bytes) – The raw CLIENT_FINISH message
commitment (ukey_pb2.Ukey2ClientInit.CipherCommitment) – The cipher commitment from the client
writer (asyncio.StreamWriter) – The writer to send alerts to

Returns:

The client’s public key, or None if the message is invalid

Return type:

ec.EllipticCurvePublicKey | None

async pyquickshare.ukey2.parse_client_init(ukey_client_init: Ukey2ClientInit, writer: StreamWriter) → tuple[str, CipherCommitment] | None¶

Parse a CLIENT_INIT message.

Parameters:

ukey_client_init (ukey_pb2.Ukey2ClientInit) – The CLIENT_INIT message
writer (asyncio.StreamWriter) – The writer to send alerts to

Returns:

The next protocol and cipher commitment, or None if the message is invalid

Return type:

tuple[str, ukey_pb2.Ukey2ClientInit.CipherCommitment] | None

async pyquickshare.ukey2.send_server_init(private_key: EllipticCurvePrivateKey, cipher_commitment: CipherCommitment, writer: StreamWriter) → bytes¶

Send a SERVER_INIT message.

Parameters:

private_key (ec.EllipticCurvePrivateKey) – The server’s private key
cipher_commitment (ukey_pb2.Ukey2ClientInit.CipherCommitment) – The cipher commitment from the client
writer (asyncio.StreamWriter) – The writer to send the message to

Returns:

The serialized SERVER_INIT message (for key derivation)

Return type:

bytes

pyquickshare.ukey2.swap_keychain(keychain: Keychain) → Keychain¶

Swap the perspective of a keychain.

Parameters:: keychain (Keychain) – The keychain to swap
Returns:: The swapped keychain
Return type:: Keychain

pyquickshare.ukey2.to_twos_complement(n: int) → bytes¶

Convert an integer to a two’s complement byte array.

Parameters:: n (int) – The integer to convert
Returns:: The two’s complement byte array
Return type:: bytes

async pyquickshare.ukey2.ukey_alert(*, alert_type: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fd77af0e050>, alert_message: str, writer: ~asyncio.streams.StreamWriter) → None¶

Send an alert message and close the connection.

Parameters:

alert_type (ukey_pb2.Ukey2Alert.AlertType) – The alert type
alert_message (str) – The alert message
writer (asyncio.StreamWriter) – The writer to send the alert to

Return type:

None

mdns¶

The mdns module provides an interface to the mDNS implementation on the system. It also handles BLE advertisement. Communication with the mDNS implementation and BlueZ is done via D-Bus.

class pyquickshare.mdns.send.AsyncRunner¶

Bases: object

async async_close() → None¶

Return type:: None

async async_display_service_info(zeroconf: Zeroconf, service_type: str, name: str) → None¶

Parameters:

zeroconf (Zeroconf)
service_type (str)
name (str)

Return type:

None

async_on_service_state_change(zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange) → None¶

Parameters:

zeroconf (Zeroconf)
service_type (str)
name (str)
state_change (ServiceStateChange)

Return type:

None

async async_run() → None¶

Return type:: None

pyquickshare.mdns.send.cleanup() → None¶

Return type:: None

async pyquickshare.mdns.send.discover_services(timeout: float = 10) → Queue[AsyncServiceInfo]¶

Parameters:: timeout (float)
Return type:: Queue[AsyncServiceInfo]

async pyquickshare.mdns.send.trigger_devices() → None¶

Return type:: None

class pyquickshare.mdns.receive.IPV4Runner¶

Bases: object

async register_services(infos: list[AsyncServiceInfo]) → None¶

Parameters:: infos (list[AsyncServiceInfo])
Return type:: None

async unregister_services(infos: list[AsyncServiceInfo]) → None¶

Parameters:: infos (list[AsyncServiceInfo])
Return type:: None

pyquickshare.mdns.receive.get_interface_mac(interface: str) → bytes¶

Parameters:: interface (str)
Return type:: bytes

pyquickshare.mdns.receive.get_interfaces() → list[str]¶

Return type:: list[str]

pyquickshare.mdns.receive.make_n(*, visible: bool, type: Type, name: bytes) → bytearray¶

Parameters:

visible (bool)
type (Type)
name (bytes)

Return type:

bytearray

async pyquickshare.mdns.receive.make_service(*, visible: bool, type_: Type, name: bytes, endpoint_id: bytes) → tuple[int, AsyncServiceInfo]¶

Parameters:

visible (bool)
type_ (Type)
name (bytes)
endpoint_id (bytes)

Return type:

tuple[int, AsyncServiceInfo]

pyquickshare.mdns.receive.make_service_name(endpoint_id: bytes) → bytearray¶

Parameters:: endpoint_id (bytes)
Return type:: bytearray

common¶

class pyquickshare.common.Type(value)¶

Bases: Enum

laptop = 3¶

phone = 1¶

tablet = 2¶

unknown = 0¶

pyquickshare.common.create_task(*args: Any, **kwargs: Any) → Task[Any]¶

Parameters:

args (Any)
kwargs (Any)

Return type:

Task[Any]

pyquickshare.common.from_url64(data: str) → bytes¶

Parameters:: data (str)
Return type:: bytes

async pyquickshare.common.read(reader: StreamReader) → bytes¶

Parameters:: reader (StreamReader)
Return type:: bytes

pyquickshare.common.safe_assert(condition: bool, message: str | None = None) → None¶

Parameters:

condition (bool)
message (str | None)

Return type:

None

pyquickshare.common.shutdown() → None¶

Return type:: None

pyquickshare.common.to_url64(data: bytes | bytearray) → str¶

Parameters:: data (bytes | bytearray)
Return type:: str