Add support for new enccryption methods

interactions/api/voice/encryption.py

@@ -1,6 +1,7 @@
 import struct
+import secrets
 
-__all__ = ("Encryption",)
+__all__ = ("Decryption", "Encryption")
 
 from abc import ABC, abstractmethod
 
@@ -11,21 +12,33 @@
 except ImportError:
     nacl_imported = False
 
+try:
+    from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305, AESGCM
+
+    cryptography_imported = True
+except ImportError:
+    cryptography_imported = False
+
 
 class Crypt(ABC):
     SUPPORTED = (
+        # todo: need deprecating
         "xsalsa20_poly1305_lite",
         "xsalsa20_poly1305_suffix",
         "xsalsa20_poly1305",
+        "aead_xchacha20_poly1305_rtpsize",
+        "aead_aes256_gcm_rtpsize",
     )
 
     def __init__(self, secret_key) -> None:
-        if not nacl_imported:
+        if not nacl_imported or not cryptography_imported:
             raise RuntimeError("Please install interactions[voice] to use voice components.")
         self.box: secret.SecretBox = secret.SecretBox(bytes(secret_key))
-
         self._xsalsa20_poly1305_lite_nonce: int = 0
 
+        self.xchacha20 = ChaCha20Poly1305(bytes(secret_key))
+        self.aes_gcm = AESGCM(bytes(secret_key))
+
     @abstractmethod
     def xsalsa20_poly1305_lite(self, header: bytes, data) -> bytes:
         raise NotImplementedError
@@ -38,6 +51,14 @@ def xsalsa20_poly1305_suffix(self, header: bytes, data) -> bytes:
     def xsalsa20_poly1305(self, header: bytes, data) -> bytes:
         raise NotImplementedError
 
+    @abstractmethod
+    def aead_xchacha20_poly1305_rtpsize(self, header: bytes, data) -> bytes:
+        raise NotImplementedError
+
+    @abstractmethod
+    def aead_aes256_gcm_rtpsize(self, header: bytes, data) -> bytes:
+        raise NotImplementedError
+
 
 class Encryption(Crypt):
     def encrypt(self, mode: str, header: bytes, data) -> bytes:
@@ -48,6 +69,10 @@ def encrypt(self, mode: str, header: bytes, data) -> bytes:
                 return self.xsalsa20_poly1305_suffix(header, data)
             case "xsalsa20_poly1305":
                 return self.xsalsa20_poly1305(header, data)
+            case "aead_xchacha20_poly1305_rtpsize":
+                return self.aead_xchacha20_poly1305_rtpsize(header, data)
+            case "aead_aes256_gcm_rtpsize":
+                return self.aead_aes256_gcm_rtpsize(header, data)
             case _:
                 raise RuntimeError(f"Unsupported encryption type requested: {mode}")
 
@@ -71,6 +96,22 @@ def xsalsa20_poly1305(self, header: bytes, data) -> bytes:
 
         return header + self.box.encrypt(bytes(data), bytes(nonce)).ciphertext
 
+    def aead_xchacha20_poly1305_rtpsize(self, header: bytes, data) -> bytes:
+        nonce_suffix = secrets.token_bytes(4)
+        nonce = bytearray(24)
+        nonce[:4] = nonce_suffix
+        ciphertext = self.xchacha20.encrypt(bytes(nonce), bytes(data), header)
+
+        return header + ciphertext + nonce_suffix
+
+    def aead_aes256_gcm_rtpsize(self, header: bytes, data) -> bytes:
+        nonce_suffix = secrets.token_bytes(4)
+        nonce = bytearray(12)
+        nonce[:4] = nonce_suffix
+        ciphertext = self.aes_gcm.encrypt(bytes(nonce), bytes(data), header)
+
+        return header + ciphertext + nonce_suffix
+
 
 class Decryption(Crypt):
     def decrypt(self, mode: str, header: bytes, data) -> bytes:
@@ -81,6 +122,10 @@ def decrypt(self, mode: str, header: bytes, data) -> bytes:
                 return self.xsalsa20_poly1305_suffix(header, data)
             case "xsalsa20_poly1305":
                 return self.xsalsa20_poly1305(header, data)
+            case "aead_xchacha20_poly1305_rtpsize":
+                return self.aead_xchacha20_poly1305_rtpsize(header, data)
+            case "aead_aes256_gcm_rtpsize":
+                return self.aead_aes256_gcm_rtpsize(header, data)
             case _:
                 raise RuntimeError(f"Unsupported decryption type requested: {mode}")
 
@@ -98,11 +143,25 @@ def xsalsa20_poly1305_lite(self, header: bytes, data) -> bytes:
 
     def xsalsa20_poly1305_suffix(self, header: bytes, data) -> bytes:
         nonce = data[-24:]
-
         return self.box.decrypt(bytes(data[:-24]), bytes(nonce))
 
     def xsalsa20_poly1305(self, header: bytes, data) -> bytes:
         nonce = bytearray(24)
         nonce[:12] = header
-
         return self.box.decrypt(bytes(data), bytes(nonce))
+
+    def aead_xchacha20_poly1305_rtpsize(self, header: bytes, data) -> bytes:
+        nonce_suffix = data[-4:]
+        ciphertext = data[:-4]
+        nonce = bytearray(24)
+        nonce[:4] = nonce_suffix
+
+        return self.xchacha20.decrypt(bytes(nonce), bytes(ciphertext), header)
+
+    def aead_aes256_gcm_rtpsize(self, header: bytes, data) -> bytes:
+        nonce_suffix = data[-4:]
+        ciphertext = data[:-4]
+        nonce = bytearray(12)
+        nonce[:4] = nonce_suffix
+
+        return self.aes_gcm.decrypt(bytes(nonce), bytes(ciphertext), header)

interactions/api/voice/voice_gateway.py

@@ -197,7 +197,12 @@ async def dispatch_opcode(self, data, op) -> None:
                 self.logger.info(f"Voice connection established; using {data['mode']}")
                 self.selected_mode = data["mode"]
                 self.secret = data["secret_key"]
-                self.encryptor = Encryption(self.secret)
+                try:
+                    self.encryptor = Encryption(self.secret)
+                    self.logger.debug(f"Encryption initialized successfully for mode {data['mode']}")
+                except Exception as e:
+                    self.logger.error(f"Failed to initialize encryption: {e}", exc_info=True)
+                    raise
                 self.ready.set()
                 if self.cond:
                     with self.cond:

pyproject.toml

@@ -36,6 +36,7 @@ pre-commit = { version = "*", optional = true }
 
 [tool.poetry.group.voice.dependencies]
 PyNaCl = "^1.5.0,<1.6"
+cryptography = "46.0.3"
 
 [tool.poetry.group.speedup.dependencies]
 aiodns = "*"