Crypto: Add nonce generation from pysrtp

Author: tuxuser

tests/test_crypto.py

@@ -4,12 +4,11 @@
 
 def test_decrypt(test_data: dict, crypto_context: srtp_crypto.SrtpContext):
     rtp_packet_raw = test_data['rtp_connection_probing.bin']
-    rtp_header, rtp_body = rtp_packet_raw[:12], rtp_packet_raw[12:]
 
-    plaintext = crypto_context.decrypt(rtp_body, aad=rtp_header)
+    plaintext = crypto_context.decrypt_packet(rtp_packet_raw)
     with pytest.raises(Exception):
         # Skip 1 byte of "additional data" to ensure invalid data
-        crypto_context.decrypt(rtp_body, aad=rtp_header[1:])
+        crypto_context.decrypt(rtp_packet_raw[:-1])
 
     assert plaintext is not None
 

xcloud/protocol/srtp_crypto.py

@@ -10,6 +10,8 @@
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
+from . import utils
+
 class TransformDirection(Enum):
     Encrypt = 0
     Decrypt = 1
@@ -110,6 +112,9 @@ def __init__(self, master_keys: SrtpMasterKeys):
         """
         MS-SRTP context
         """
+        self.roc = 0
+        self.seq = 0
+
         self.master_keys = master_keys
         self.session_keys = SrtpContext._derive_session_keys(
             self.master_keys.master_key, self.master_keys.master_salt
@@ -149,19 +154,13 @@ def _derive_single_key(master_key, master_salt, key_index: int = 0, max_bytes: i
         import binascii
         '''SRTP key derivation, https://tools.ietf.org/html/rfc3711#section-4.3'''
 
-        def bytes_to_int(b):
-            return int.from_bytes(b, byteorder='big')
-
-        def int_to_bytes(i, n_bytes):
-            return i.to_bytes(n_bytes, byteorder='big')
-
         assert len(master_key) == 128 // 8
         assert len(master_salt) == 112 // 8
-        salt = bytes_to_int(master_salt)
+        salt = utils.bytes_to_int(master_salt)
 
         DIV = lambda x, y: 0 if y == 0 else x // y
         prng = lambda iv: SrtpContext._crypt_ctr_oneshot(
-            master_key, int_to_bytes(iv, 16), b'\x00' * 16, max_bytes=max_bytes
+            master_key, utils.int_to_bytes(iv, 16), b'\x00' * 16, max_bytes=max_bytes
         )
         r = DIV(pkt_i, key_derivation_rate)  # pkt_i is always 48 bits
         derive_key_from_label = lambda label: prng(
@@ -189,19 +188,26 @@ def _decrypt(ctx: AESGCM, nonce: bytes, data: bytes, aad: bytes) -> bytes:
     def _encrypt(ctx: AESGCM, nonce: bytes, data: bytes, aad: bytes) -> bytes:
         return ctx.encrypt(nonce, data, aad)
 
-    def _get_transformed_nonce(self, transform_direction: TransformDirection) -> bytes:
-        # Skip first 2 bytes of Nonce key
-        nonce = bytearray(self.session_keys.salt_key[2:])
-        # TODO: Implement transform logic
-        # FIXME: Just tranforming the Nonce to a known value for
-        #        our single test packet
-        nonce[-1] += 1
-        return nonce
-
-    def decrypt(self, data: bytes, aad: bytes) -> bytes:
-        nonce = self._get_transformed_nonce(TransformDirection.Decrypt)
-        return SrtpContext._decrypt(self.decryptor_ctx, nonce, data, aad)
+    @staticmethod
+    def packet_index(roc, seq):
+        return seq + (roc << 16)
+
+    @staticmethod
+    def _calc_iv(salt, ssrc, pkt_i):
+        salt = utils.bytes_to_int(salt)
+        iv = ((ssrc << (48)) + pkt_i) ^ salt
+        return utils.int_to_bytes(iv, 12)
 
-    def encrypt(self, data: bytes, aad: bytes) -> RtpPacket:
-        nonce = self._get_transformed_nonce(TransformDirection.Encrypt)
-        return SrtpContext._encrypt(self.decryptor_ctx, nonce, data, aad)
+    def decrypt_packet(self, rtp_packet: bytes) -> RtpPacket:
+        rtp_header = rtp_packet[:12]
+        parsed = RtpPacket.parse(rtp_packet)
+        
+        if parsed.sequence_number < self.seq:
+            self.roc += 1
+        self.seq = parsed.sequence_number
+        pkt_i = SrtpContext.packet_index(self.roc, self.seq)
+        iv = SrtpContext._calc_iv(self.session_keys.salt_key[2:], parsed.ssrc, pkt_i)
+
+        decrypted_payload = SrtpContext._decrypt(self.decryptor_ctx, iv, parsed.payload, rtp_header)
+        parsed.payload = decrypted_payload
+        return parsed

xcloud/protocol/utils.py

@@ -0,0 +1,5 @@
+def bytes_to_int(b):
+    return int.from_bytes(b, byteorder='big')
+
+def int_to_bytes(i, n_bytes):
+    return i.to_bytes(n_bytes, byteorder='big')

xcloud/scripts/pcap_reader.py

@@ -23,7 +23,8 @@ class XcloudPcapParser:
     def __init__(self, srtp_key: Optional[str]):
         self.crypto: Optional[srtp_crypto.SrtpContext] = None
         if srtp_key:
-            self.crypto = srtp_crypto.SrtpContext.from_base64(srtp_key)
+            self.outgoing_crypto = srtp_crypto.SrtpContext.from_base64(srtp_key)
+            self.incoming_crypto = srtp_crypto.SrtpContext.from_base64(srtp_key)
 
     @property
     def PACKET_TYPES(self):
@@ -33,35 +34,30 @@ def PACKET_TYPES(self):
             (teredo.TeredoPacket.parse, self.get_info_teredo)
         ]
 
-    def get_info_stun(self, stun: stun.Message) -> None:
+    def get_info_stun(self, stun: stun.Message, is_client: bool) -> None:
         return f'STUN: {stun}'
 
-    def brute_force_nonce(self, nonce_orig: bytes) -> Generator:
-        for byte1 in range(0, 0xFF):
-            for byte2 in range(0, 0xFF):
-                nonce_transform = b''.join([nonce_orig[:5], struct.pack('!B', byte1), nonce_orig[6:11], struct.pack('!B', byte2)])
-                yield nonce_transform
-
-    def get_info_rtp(self, rtp: rtp.RtpPacket) -> None:
+    def get_info_rtp(self, rtp: rtp.RtpPacket, is_client: bool) -> None:
         try:
             payload_name = packets.PayloadType(rtp.payload_type)
         except:
             payload_name = '<UNKNOWN>'
 
-        info_str = f'RTP: {payload_name.name} {rtp} SSRC={rtp.ssrc}'
-        if self.crypto:
-            rtp_packet_serialized = rtp.serialize()
-            rtp_header, rtp_data = rtp_packet_serialized[:12], rtp_packet_serialized[12:]
-            nonce_orig = self.crypto.session_keys.nonce_key[2:]
-            for nonce_transformed in self.brute_force_nonce(nonce_orig):
-                try:
-                    decrypted = self.crypto._decrypt(self.crypto.decryptor_ctx, nonce_transformed, rtp_data, rtp_header)
-                    info_str += "\n" + hexdump(decrypted, result='return') + "\n"
-                except Exception:
-                    pass
+        direction = 'OUT -> ' if is_client else '<- IN '
+        info_str = f'{direction} RTP: {payload_name.name} {rtp} SSRC={rtp.ssrc}'
+        if self.incoming_crypto and self.outgoing_crypto:
+            rtp_packet = rtp.serialize()
+            try:
+                if is_client:
+                    rtp_decrypted = self.outgoing_crypto.decrypt_packet(rtp_packet)
+                else:
+                    rtp_decrypted = self.incoming_crypto.decrypt_packet(rtp_packet)
+                info_str += "\n" + hexdump(rtp_decrypted.payload, result='return') + "\n"
+            except Exception:
+                info_str += "\n DECRYPTION FAILED \n"
         return info_str
 
-    def get_info_teredo(self, teredo: teredo.TeredoPacket) -> None:
+    def get_info_teredo(self, teredo: teredo.TeredoPacket, is_client: bool) -> None:
         info = f'TEREDO: {teredo}'
         if teredo.ipv6.next_header != ipv6.NO_NEXT_HEADER:
             data = teredo.ipv6.data
@@ -77,7 +73,8 @@ def get_info_general(self, packet: Any) -> Optional[str]:
             for cls, info_func in self.PACKET_TYPES:
                 try:
                     instance = cls(data)
-                    info = info_func(instance)
+                    is_client = (packet.dport == 54881)
+                    info = info_func(instance, is_client)
                     return info
                 except:
                     pass
@@ -96,16 +93,15 @@ def packet_filter(self, filepath):
                     continue
 
                 ip = eth.data
-                subpacket = ip.data
-                if not isinstance(subpacket, dpkt.udp.UDP):
+                if not isinstance(ip.data, dpkt.udp.UDP):
                     continue
 
-                yield(subpacket, ts)
+                yield(ip, ts)
 
 
     def parse_file(self, pcap_filepath: str) -> None:
         for packet, timestamp in self.packet_filter(pcap_filepath):
-            info = self.get_info_general(packet)
+            info = self.get_info_general(packet.data)
             if info:
                 print(info)