""" The MIT License (MIT) Copyright (c) 2015-present Rapptz Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. Some documentation to refer to: - Our main web socket (mWS) sends opcode 4 with a guild ID and channel ID. - The mWS receives VOICE_STATE_UPDATE and VOICE_SERVER_UPDATE. - We pull the session_id from VOICE_STATE_UPDATE. - We pull the token, endpoint and server_id from VOICE_SERVER_UPDATE. - Then we initiate the voice web socket (vWS) pointing to the endpoint. - We send opcode 0 with the user_id, server_id, session_id and token using the vWS. - The vWS sends back opcode 2 with an ssrc, port, modes(array) and hearbeat_interval. - We send a UDP discovery packet to endpoint:port and receive our IP and our port in LE. - Then we send our IP and port via vWS with opcode 1. - When that's all done, we receive opcode 4 from the vWS. - Finally we can transmit data to endpoint:port. """ from __future__ import annotations import asyncio from dataclasses import dataclass import socket import logging import struct import threading from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING, Tuple from . import opus, utils from .backoff import ExponentialBackoff from .gateway import * from .errors import ClientException, ConnectionClosed from .player import AudioPlayer, AudioSource from .utils import MISSING if TYPE_CHECKING: from .client import Client from .guild import Guild from .state import ConnectionState from .user import ClientUser from .opus import Encoder from . import abc from .types.voice import ( GuildVoiceState as GuildVoiceStatePayload, VoiceServerUpdate as VoiceServerUpdatePayload, SupportedModes, ) has_nacl: bool try: import nacl.secret # type: ignore has_nacl = True except ImportError: has_nacl = False __all__ = ( 'VoiceProtocol', 'VoiceClient', ) _log = logging.getLogger(__name__) class VoiceProtocol: """A class that represents the Discord voice protocol. This is an abstract class. The library provides a concrete implementation under :class:`VoiceClient`. This class allows you to implement a protocol to allow for an external method of sending voice, such as Lavalink_ or a native library implementation. These classes are passed to :meth:`abc.Connectable.connect `. .. _Lavalink: https://github.com/freyacodes/Lavalink Parameters ------------ client: :class:`Client` The client (or its subclasses) that started the connection request. channel: :class:`abc.Connectable` The voice channel that is being connected to. """ def __init__(self, client: Client, channel: abc.Connectable) -> None: self.client: Client = client self.channel: abc.Connectable = channel async def on_voice_state_update(self, data: GuildVoiceStatePayload) -> None: """|coro| An abstract method that is called when the client's voice state has changed. This corresponds to ``VOICE_STATE_UPDATE``. Parameters ------------ data: :class:`dict` The raw `voice state payload`__. .. _voice_state_update_payload: https://discord.com/developers/docs/resources/voice#voice-state-object __ voice_state_update_payload_ """ raise NotImplementedError async def on_voice_server_update(self, data: VoiceServerUpdatePayload) -> None: """|coro| An abstract method that is called when initially connecting to voice. This corresponds to ``VOICE_SERVER_UPDATE``. Parameters ------------ data: :class:`dict` The raw `voice server update payload`__. .. _voice_server_update_payload: https://discord.com/developers/docs/topics/gateway#voice-server-update-voice-server-update-event-fields __ voice_server_update_payload_ """ raise NotImplementedError async def connect(self, *, timeout: float, reconnect: bool) -> None: """|coro| An abstract method called when the client initiates the connection request. When a connection is requested initially, the library calls the constructor under ``__init__`` and then calls :meth:`connect`. If :meth:`connect` fails at some point then :meth:`disconnect` is called. Within this method, to start the voice connection flow it is recommended to use :meth:`Guild.change_voice_state` to start the flow. After which, :meth:`on_voice_server_update` and :meth:`on_voice_state_update` will be called. The order that these two are called is unspecified. Parameters ------------ timeout: :class:`float` The timeout for the connection. reconnect: :class:`bool` Whether reconnection is expected. """ raise NotImplementedError async def disconnect(self, *, force: bool) -> None: """|coro| An abstract method called when the client terminates the connection. See :meth:`cleanup`. Parameters ------------ force: :class:`bool` Whether the disconnection was forced. """ raise NotImplementedError def cleanup(self) -> None: """This method *must* be called to ensure proper clean-up during a disconnect. It is advisable to call this from within :meth:`disconnect` when you are completely done with the voice protocol instance. This method removes it from the internal state cache that keeps track of currently alive voice clients. Failure to clean-up will cause subsequent connections to report that it's still connected. """ key_id, _ = self.channel._get_voice_client_key() self.client._connection._remove_voice_client(key_id) class Player: def __init__(self, client: VoiceClient) -> None: self.client = client self.loop: asyncio.AbstractEventLoop = client.loop self.encoder: Encoder = MISSING self._player: AudioPlayer = MISSING def send(self, data: bytes, encode: bool = True) -> None: """Sends an audio packet composed of the data. You must be connected to play audio. Parameters ---------- data: :class:`bytes` The :term:`py:bytes-like object` denoting PCM or Opus voice data. encode: :class:`bool` Indicates if ``data`` should be encoded into Opus. Raises ------- ClientException You are not connected. opus.OpusError Encoding the data failed. """ if encode: data = self.encoder.encode(data, self.encoder.SAMPLES_PER_FRAME) self.client.send_audio_packet(data) @property def ws(self) -> DiscordVoiceWebSocket: return self.client.ws @property def source(self) -> Optional[AudioSource]: """Optional[:class:`AudioSource`]: The audio source being played, if playing. This property can also be used to change the audio source currently being played. """ return self._player.source if self._player else None @source.setter def source(self, value) -> None: if not isinstance(value, AudioSource): raise TypeError('Expected AudioSource not {0.__class__.__name__}'.format(value)) if self._player is None: raise ValueError('Not playing anything') self._player._set_source(value) @property def playing(self) -> bool: return self.is_playing() def is_playing(self) -> bool: """Indicates if we're currently playing audio.""" return self._player is not None and self._player.is_playing() @property def paused(self) -> bool: return self.is_paused() def is_paused(self) -> bool: """Indicates if we're playing audio, but if we're paused.""" return self._player is not None and self._player.is_paused() def play( self, source: AudioSource, *, after: Callable[[Optional[Exception]], Any] = None ) -> None: """Plays an :class:`AudioSource`. The finalizer, ``after`` is called after the source has been exhausted or an error occurred. If an error happens while the audio player is running, the exception is caught and the audio player is then stopped. If no after callback is passed, any caught exception will be displayed as if it were raised. Parameters ----------- source: :class:`AudioSource` The audio source we're reading from. after: Callable[[:class:`Exception`], Any] The finalizer that is called after the stream is exhausted. This function must have a single parameter, ``error``, that denotes an optional exception that was raised during playing. Raises ------- ClientException Already playing audio or not connected. TypeError Source is not a :class:`AudioSource` or after is not a callable. OpusNotLoaded Source is not opus encoded and opus is not loaded. """ if not self.client.is_connected(): raise ClientException('Not connected to voice') if self.is_playing(): raise ClientException('Already playing audio') if not isinstance(source, AudioSource): raise TypeError(f'source must be an AudioSource not {source.__class__.__name__}') if not self.encoder and not source.is_opus(): self.encoder = opus.Encoder() self._player = AudioPlayer(source, self, after=after) self._player.start() def pause(self) -> None: """Pauses the audio playing.""" if self._player: self._player.pause() def resume(self) -> None: """Resumes the audio playing.""" if self._player: self._player.resume() def stop(self) -> None: """Stops playing audio.""" if self._player: self._player.stop() self._player = None class Listener: def __init__(self, client: VoiceClient) -> None: self.client = client self.loop: asyncio.AbstractEventLoop = client.loop self.decoder = None self._listener = None @property def ws(self) -> DiscordVoiceWebSocket: return self.client.ws @property def listening(self) -> bool: return self.is_playing() def is_listening(self) -> bool: """Indicates if we're currently listening.""" return self._listener is not None and self._listener.is_listening() @property def paused(self) -> bool: return self.is_paused() def is_paused(self) -> bool: """Indicates if we're listening, but we're paused.""" return self._listener is not None and self._listener.is_paused() def listen(self, sink, *, callback=None) -> None: if not self.client.is_connected(): raise ClientException('Not connected to voice') if self.is_listening(): raise ClientException('Already listening') if not isinstance(sink, AudioSink): raise TypeError(f'sink must an AudioSink not {sink.__class__.__name__}') class VoiceClient(VoiceProtocol): """Represents a Discord voice connection. You do not create these, you typically get them from e.g. :meth:`VoiceChannel.connect`. Warning -------- In order to use PCM based AudioSources, you must have the opus library installed on your system and loaded through :func:`opus.load_opus`. Otherwise, your AudioSources must be opus encoded (e.g. using :class:`FFmpegOpusAudio`) or the library will not be able to transmit audio. Attributes ----------- session_id: :class:`str` The voice connection session ID. token: :class:`str` The voice connection token. endpoint: :class:`str` The endpoint we are connecting to. channel: :class:`abc.Connectable` The voice channel connected to. loop: :class:`asyncio.AbstractEventLoop` The event loop that the voice client is running on. """ endpoint_ip: str voice_port: int secret_key: List[int] ssrc: int def __init__(self, client: Client, channel: abc.Connectable): if not has_nacl: raise RuntimeError("PyNaCl library needed in order to use voice") super().__init__(client, channel) state = client._connection self.token: str = MISSING self.socket = MISSING self.loop: asyncio.AbstractEventLoop = state.loop self._state: ConnectionState = state # This will be used in the threads self._connected: threading.Event = threading.Event() self._handshaking: bool = False self._potentially_reconnecting: bool = False self._voice_state_complete: asyncio.Event = asyncio.Event() self._voice_server_complete: asyncio.Event = asyncio.Event() self.mode: str = MISSING self._connections: int = 0 self.sequence: int = 0 self.timestamp: int = 0 self.player = Player(self) self.listener = Listener(self) self.timeout: float = 0 self._runner: asyncio.Task = MISSING self._lite_nonce: int = 0 self.ws: DiscordVoiceWebSocket = MISSING self.idrcs: Dict[int, int] = {} self.ssids: Dict[int, int] = {} warn_nacl = not has_nacl supported_modes: Tuple[SupportedModes, ...] = ( 'xsalsa20_poly1305_lite', 'xsalsa20_poly1305_suffix', 'xsalsa20_poly1305', ) @property def ssrc(self) -> int: """:class:`str`: Our ssrc.""" return self.idrcs.get(self.user.id) @ssrc.setter def ssrc(self, value): self.idrcs[self.user.id] = value self.ssids[value] = self.user.id @property def guild(self) -> Optional[Guild]: """Optional[:class:`Guild`]: The guild we're connected to, if applicable.""" return getattr(self.channel, 'guild', None) @property def user(self) -> ClientUser: """:class:`ClientUser`: The user connected to voice (i.e. ourselves).""" return self._state.user # Connection related async def on_voice_state_update(self, data: GuildVoiceStatePayload) -> None: self.session_id = data['session_id'] channel_id = data['channel_id'] if not self._handshaking or self._potentially_reconnecting: # If we're done handshaking then we just need to update ourselves # If we're potentially reconnecting due to a 4014, then we need to differentiate # a channel move and an actual force disconnect if channel_id is None: # We're being disconnected so cleanup await self.disconnect() else: guild = self.guild if guild is not None: self.channel = channel_id and guild.get_channel(int(channel_id)) # type: ignore else: self.channel = channel_id and self._state._get_private_channel(int(channel_id)) # type: ignore else: self._voice_state_complete.set() async def on_voice_server_update(self, data: VoiceServerUpdatePayload) -> None: if self._voice_server_complete.is_set(): _log.info('Ignoring extraneous voice server update.') return self.token = data.get('token') self.server_id = server_id = utils._get_as_snowflake(data, 'guild_id') if server_id is None: self.server_id = utils._get_as_snowflake(data, 'channel_id') endpoint = data.get('endpoint') if endpoint is None or self.token is None: _log.warning('Awaiting endpoint... This requires waiting. ' \ 'If timeout occurred considering raising the timeout and reconnecting.') return self.endpoint, _, _ = endpoint.rpartition(':') if self.endpoint.startswith('wss://'): self.endpoint = self.endpoint[6:] # Shouldn't ever be there... self.endpoint_ip = MISSING self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.socket.setblocking(False) if not self._handshaking: # If we're not handshaking then we need to terminate our previous connection to the websocket await self.ws.close(4000) return self._voice_server_complete.set() async def voice_connect(self) -> None: if self.guild: await self.guild.change_voice_state(channel=self.channel) else: await self._state.client.change_voice_state(channel=self.channel) async def voice_disconnect(self) -> None: _log.info('The voice handshake is being terminated for channel ID %s (guild ID %s).', self.channel.id, getattr(self.guild, 'id', None)) if self.guild: await self.guild.change_voice_state(channel=None) else: await self._state.client.change_voice_state(channel=None) def prepare_handshake(self) -> None: self._voice_state_complete.clear() self._voice_server_complete.clear() self._handshaking = True _log.info('Starting voice handshake (connection attempt %d)...', self._connections + 1) self._connections += 1 def finish_handshake(self) -> None: _log.info('Voice handshake complete. Endpoint found: %s.', self.endpoint) self._handshaking = False self._voice_server_complete.clear() self._voice_state_complete.clear() async def connect_websocket(self, resume=False) -> DiscordVoiceWebSocket: ws = await DiscordVoiceWebSocket.from_client(self, resume=resume) self._connected.clear() while ws.secret_key is None: await ws.poll_event() self._connected.set() return ws async def connect(self, *, reconnect: bool, timeout: float) ->None: _log.info('Connecting to voice...') self.timeout = timeout for i in range(5): self.prepare_handshake() # This has to be created before we start the flow. futures = [ self._voice_state_complete.wait(), self._voice_server_complete.wait(), ] # Start the connection flow await self.voice_connect() try: await utils.sane_wait_for(futures, timeout=timeout) except asyncio.TimeoutError: await self.disconnect(force=True) raise self.finish_handshake() try: self.ws = await self.connect_websocket() break except (ConnectionClosed, asyncio.TimeoutError): if reconnect: _log.exception('Failed to connect to voice. Retrying...') await asyncio.sleep(1 + i * 2.0) await self.voice_disconnect() continue else: await self.disconnect(force=True) raise if self._runner is MISSING: self._runner = self.loop.create_task(self.poll_voice_ws(reconnect)) async def potential_reconnect(self) -> bool: # Attempt to stop the player thread from playing early self._connected.clear() self.prepare_handshake() self._potentially_reconnecting = True try: # We only care about VOICE_SERVER_UPDATE since VOICE_STATE_UPDATE can come before we get disconnected await asyncio.wait_for(self._voice_server_complete.wait(), timeout=self.timeout) except asyncio.TimeoutError: self._potentially_reconnecting = False await self.disconnect(force=True) return False self.finish_handshake() self._potentially_reconnecting = False try: self.ws = await self.connect_websocket() except (ConnectionClosed, asyncio.TimeoutError): return False else: return True async def potential_resume(self) -> bool: # Attempt to stop the player thread from playing early self._connected.clear() self._potentially_reconnecting = True try: self.ws = await self.connect_websocket(resume=True) except (ConnectionClosed, asyncio.TimeoutError): return False # Reconnect normally if RESUME failed else: return True finally: self._potentially_reconnecting = False @property def latency(self) -> float: """:class:`float`: Latency between a HEARTBEAT and a HEARTBEAT_ACK in seconds. This could be referred to as the Discord Voice WebSocket latency and is an analogue of user's voice latencies as seen in the Discord client. .. versionadded:: 1.4 """ ws = self.ws return float("inf") if not ws else ws.latency @property def average_latency(self) -> float: """:class:`float`: Average of most recent 20 HEARTBEAT latencies in seconds. .. versionadded:: 1.4 """ ws = self.ws return float("inf") if not ws else ws.average_latency async def poll_voice_ws(self, reconnect: bool) -> None: backoff = ExponentialBackoff() while True: try: await self.ws.poll_event() except (ConnectionClosed, asyncio.TimeoutError) as exc: if isinstance(exc, ConnectionClosed): if exc.code == 1000: _log.info('Disconnecting from voice normally, close code %d.', exc.code) await self.disconnect() break if exc.code == 4015: _log.info('Disconnected from voice (close code %d), potentially RESUMEing...', exc.code) successful = await self.potential_resume() if not successful: _log.info('RESUME was unsuccessful, disconnecting from voice normally...') await self.disconnect() break else: continue if exc.code == 4014: _log.info('Disconnected from voice by force (close code %d)... potentially reconnecting.', exc.code) successful = await self.potential_reconnect() if not successful: _log.info('Reconnect was unsuccessful, disconnecting from voice normally...') await self.disconnect() break else: continue if not reconnect: await self.disconnect() raise retry = backoff.delay() _log.exception('Disconnected from voice... Reconnecting in %.2fs.', retry) self._connected.clear() await asyncio.sleep(retry) try: await self.connect(reconnect=True, timeout=self.timeout) except asyncio.TimeoutError: # We've retried 5 times, let's continue the loop _log.warning('Could not connect to voice. Retrying...') continue async def disconnect(self, *, force: bool = False) -> None: """|coro| Disconnects this voice client from voice. """ if not force and not self.is_connected(): return self.player.stop() # TODO: Stop listener self._connected.clear() try: if self.ws: await self.ws.close() await self.voice_disconnect() finally: self.cleanup() if self.socket: self.socket.close() async def move_to(self, channel: abc.Snowflake) -> None: """|coro| Moves you to a different voice channel. Parameters ----------- channel: :class:`abc.Snowflake` The channel to move to. Must be a :class:`Connectable`. """ if self.guild: await self.guild.change_voice_state(channel=channel) else: await self._state.client.change_voice_state(channel=channel) @property def connected(self) -> bool: return self.is_connected() def is_connected(self) -> bool: """Indicates if the voice client is connected to voice.""" return self._connected.is_set() # Audio related def _flip_ssrc(self, query) -> Optional[int]: value = self.idrcs.get(query) if value is None: value = self.ssids.get(query) return value def _set_ssrc(self, user_id, ssrc) -> None: self.idrcs[user_id] = ssrc self.ssids[ssrc] = user_id def _checked_add(self, attr, value, limit) -> None: val = getattr(self, attr) if val + value > limit: setattr(self, attr, 0) else: setattr(self, attr, val + value) @staticmethod def _strip_header(data) -> bytes: if data[0] == 0xbe and data[1] == 0xde and len(data) > 4: _, length = struct.unpack_from('>HH', data) offset = 4 + length * 4 data = data[offset:] return data def _get_voice_packet(self, data) -> bytes: header = bytearray(12) # Formulate RTP header header[0] = 0x80 header[1] = 0x78 struct.pack_into('>H', header, 2, self.sequence) struct.pack_into('>I', header, 4, self.timestamp) struct.pack_into('>I', header, 8, self.ssrc) encrypt_packet = getattr(self, '_encrypt_' + self.mode) return encrypt_packet(header, data) def _encrypt_xsalsa20_poly1305(self, header: bytes, data) -> bytes: box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce = bytearray(24) nonce[:12] = header return header + box.encrypt(bytes(data), bytes(nonce)).ciphertext def _decrypt_xsalsa20_poly1305(self, header, data): box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce = bytearray(24) nonce[:12] = header return self._strip_header(box.decrypt(bytes(data), bytes(nonce))) def _encrypt_xsalsa20_poly1305_suffix(self, header: bytes, data) -> bytes: box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce = nacl.utils.random(nacl.secret.SecretBox.NONCE_SIZE) return header + box.encrypt(bytes(data), nonce).ciphertext + nonce def _decrypt_xsalsa20_poly1305_suffix(self, header, data): box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce_size = nacl.secret.SecretBox.NONCE_SIZE nonce = data[-nonce_size:] return self._strip_header(box.decrypt(bytes(data[:-nonce_size]), nonce)) def _encrypt_xsalsa20_poly1305_lite(self, header: bytes, data) -> bytes: box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce = bytearray(24) nonce[:4] = struct.pack('>I', self._lite_nonce) self._checked_add('_lite_nonce', 1, 4294967295) return header + box.encrypt(bytes(data), bytes(nonce)).ciphertext + nonce[:4] def _decrypt_xsalsa20_poly1305_lite(self, header, data): box = nacl.secret.SecretBox(bytes(self.secret_key)) nonce = bytearray(24) nonce[:4] = data[-4:] data = data[:-4] return self._strip_header(box.decrypt(bytes(data), bytes(nonce))) def play(self, source: AudioSource, *, after: Callable[[Optional[Exception]], Any]=None) -> None: """Plays an :class:`AudioSource`. The finalizer, ``after`` is called after the source has been exhausted or an error occurred. If an error happens while the audio player is running, the exception is caught and the audio player is then stopped. If no after callback is passed, any caught exception will be displayed as if it were raised. Parameters ----------- source: :class:`AudioSource` The audio source we're reading from. after: Callable[[Optional[:class:`Exception`]], Any] The finalizer that is called after the stream is exhausted. This function must have a single parameter, ``error``, that denotes an optional exception that was raised during playing. Raises ------- ClientException Already playing audio or not connected. TypeError Source is not a :class:`AudioSource` or after is not a callable. OpusNotLoaded Source is not opus encoded and opus is not loaded. """ return self.player.play(source, after=after) def listen(self, *args, **kwargs): return self.listener.listen(*args, **kwargs) @property def source(self) -> Optional[AudioSource]: """Optional[:class:`AudioSource`]: The audio source being played, if playing. This property can also be used to change the audio source currently being played. """ return self.player.source @source.setter def source(self, value: AudioSource) -> None: self.player.source = value @property def sink(self) -> Optional[AudioSink]: """Optional[:class:`AudioSink`]: Where received audio is being sent. This property can also be used to change the value. """ return self.listener.sink @sink.setter def sink(self, value): self.listener.sink = value #if not isinstance(value, AudioSink): #raise TypeError('Expected AudioSink not {value.__class__.__name__}') #if self._recorder is None: #raise ValueError('Not listening') def send_audio_packet(self, data: bytes) -> None: """Sends an audio packet composed of the data. You must be connected to play audio. Parameters ---------- data: :class:`bytes` The :term:`py:bytes-like object` denoting Opus voice data. Raises ------- ClientException You are not connected. opus.OpusError Encoding the data failed. """ self._checked_add('sequence', 1, 65535) packet = self._get_voice_packet(data) try: self.socket.sendto(packet, (self.endpoint_ip, self.voice_port)) except BlockingIOError: _log.warning('A packet has been dropped (seq: %s, timestamp: %s).', self.sequence, self.timestamp) self._checked_add('timestamp', opus.Encoder.SAMPLES_PER_FRAME, 4294967295)