feat(voice): add aead_xchacha20_poly1305_rtpsize encryption mode, remove old modes (#1228)

Author: shiftinv

changelog/1228.feature.rst

@@ -0,0 +1 @@
+Add support for ``aead_xchacha20_poly1305_rtpsize`` encryption mode for voice connections, and remove deprecated ``xsalsa20_poly1305*`` modes.

changelog/1228.misc.rst

@@ -0,0 +1 @@
+Raise PyNaCl version requirement to ``v1.5.0``.

disnake/gateway.py

@@ -1038,7 +1038,7 @@ async def initial_connection(self, data: VoiceReadyPayload) -> None:
         state.port = struct.unpack_from(">H", recv, len(recv) - 2)[0]
         _log.debug("detected ip: %s port: %s", state.ip, state.port)
 
-        # there *should* always be at least one supported mode (xsalsa20_poly1305)
+        # there *should* always be at least one supported mode
         modes: List[SupportedModes] = [
             mode for mode in data["modes"] if mode in self._connection.supported_modes
         ]

disnake/types/voice.py

@@ -7,7 +7,10 @@
 from .member import MemberWithUser
 from .snowflake import Snowflake
 
-SupportedModes = Literal["xsalsa20_poly1305_lite", "xsalsa20_poly1305_suffix", "xsalsa20_poly1305"]
+SupportedModes = Literal[
+    # "aead_aes256_gcm_rtpsize",  # supported in libsodium, but not exposed by pynacl
+    "aead_xchacha20_poly1305_rtpsize",
+]
 
 
 class _VoiceState(TypedDict):

disnake/voice_client.py

@@ -228,11 +228,7 @@ def __init__(self, client: Client, channel: abc.Connectable) -> None:
         self.ws: DiscordVoiceWebSocket = MISSING
 
     warn_nacl = not has_nacl
-    supported_modes: Tuple[SupportedModes, ...] = (
-        "xsalsa20_poly1305_lite",
-        "xsalsa20_poly1305_suffix",
-        "xsalsa20_poly1305",
-    )
+    supported_modes: Tuple[SupportedModes, ...] = ("aead_xchacha20_poly1305_rtpsize",)
 
     @property
     def guild(self) -> Guild:
@@ -512,36 +508,35 @@ def _get_voice_packet(self, data):
         header = bytearray(12)
 
         # Formulate rtp header
-        header[0] = 0x80
-        header[1] = 0x78
+        header[0] = 0x80  # version = 2
+        header[1] = 0x78  # payload type = 120 (opus)
         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, f"_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
+    def _get_nonce(self, pad: int):
+        # returns (nonce, padded_nonce).
+        # n.b. all currently implemented modes use the same nonce size (192 bits / 24 bytes)
+        nonce = struct.pack(">I", self._lite_nonce)
 
-        return header + box.encrypt(bytes(data), bytes(nonce)).ciphertext
+        self._lite_nonce += 1
+        if self._lite_nonce > 4294967295:
+            self._lite_nonce = 0
 
-    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 (nonce, nonce.ljust(pad, b"\0"))
 
-        return header + box.encrypt(bytes(data), nonce).ciphertext + nonce
+    def _encrypt_aead_xchacha20_poly1305_rtpsize(self, header: bytes, data) -> bytes:
+        box = nacl.secret.Aead(bytes(self.secret_key))
+        nonce, padded_nonce = self._get_nonce(nacl.secret.Aead.NONCE_SIZE)
 
-    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]
+        return (
+            header
+            + box.encrypt(bytes(data), aad=bytes(header), nonce=padded_nonce).ciphertext
+            + nonce
+        )
 
     def play(
         self, source: AudioSource, *, after: Optional[Callable[[Optional[Exception]], Any]] = None

pyproject.toml

@@ -49,7 +49,7 @@ speed = [
     'cchardet; python_version < "3.10"',
 ]
 voice = [
-    "PyNaCl>=1.3.0,<1.6",
+    "PyNaCl>=1.5.0,<1.6",
 ]
 docs = [
     "sphinx==7.0.1",