Tests: Add crypto related fixtures / boilerplate test & class

Author: tuxuser

tests/conftest.py

@@ -1,8 +1,10 @@
-from typing import Dict
+from typing import Dict, Tuple
 import os
 import pytest
 from binascii import unhexlify
 
+from xcloud.protocol import srtp_crypto
+
 @pytest.fixture(scope='session')
 def test_data() -> Dict[str, bytes]:
     data = {}
@@ -38,4 +40,24 @@ def teredo_packet() -> bytes:
     return unhexlify(
         '6000000000003b1520010000338c24f41c38f3fdd2f3c93d20010000'
         '338c24f4043b30e3d2f3c93d01049eb8960803080000c0a889db0c02'
-    )
+    )
+
+@pytest.fixture(scope='session')
+def session_id() -> str:
+    return 'ED309CA5-F87C-439D-A429-63F417B552FA'
+
+@pytest.fixture(scope='session')
+def ice_credentials_client() -> Tuple[str, str]:
+    return ('m99KewV+44E=', 'AneALie0L4P2tpvbh76nremwgQrT12/R3UYTG5VmUJ8=')
+
+@pytest.fixture(scope='session')
+def ice_credentials_host() -> Tuple[str, str]:
+    return ('5yUsZtOzQ+w=', 'bWpvx/cXTk3/IeadJHO4T19W/OZopsbn0MwTAZqZu8w=')
+
+@pytest.fixture(scope='session')
+def srtp_key() -> str:
+    return 'RdHzuLLVGuO1aHILIEVJ1UzR7RWVioepmpy+9SRf'
+
+@pytest.fixture(scope='session')
+def crypto_context(srtp_key: str) -> srtp_crypto.MsSrtpCrypto:
+    return srtp_crypto.MsSrtpCrypto(srtp_key)

tests/data/rtp_connection_probing.bin

@@ -0,0 +1,8 @@
+from xcloud.protocol import srtp_crypto
+
+def test_decrypt(test_data: dict, crypto_context: srtp_crypto.MsSrtpCrypto):
+    rtp_packet_raw = test_data['rtp_connection_probing.bin']
+    plaintext = crypto_context.decrypt_raw(rtp_packet_raw)
+
+    print(plaintext)
+    assert plaintext is not None

tests/test_crypto.py

@@ -0,0 +1,21 @@
+import base64
+from aiortc.rtp import RtpPacket
+
+class MsSrtpCrypto:
+    def __init__(self, master_key: str):
+        try:
+            master_key = base64.b64decode(master_key)
+        except Exception:
+            raise ValueError('Master key is not base64-decodable')
+        
+        self.master_key = master_key
+    
+    def decrypt(self, rtp_data: RtpPacket) -> RtpPacket:
+        raise NotImplementedError('Decryption not implemented')
+
+    def decrypt_raw(self, data: bytes) -> RtpPacket:
+        packet = RtpPacket.parse(data)
+        return self.decrypt(packet)
+    
+    def encrypt(self, rtp_data: RtpPacket) -> RtpPacket:
+        raise NotImplementedError('Encryption not implemented')