Repository: pyca/pyopenssl
Pull request #1279 (2024-01-10 16:47:56)

Add Connection.get_selected_srtp_profile (#1278)
If an SRTP profile was negotiated as part of the handshake, make it possible to retrieve the name of the profile. This is needed to determine which profiles were offered using `Context.set_tlsext_use_srtp` was actually selected.


Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index d504499..408e5c2 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -35,6 +35,8 @@ Changes:
   to the now deprecated ``OpenSSL.crypto.CRL`` arguments.
 - Fixed ``test_set_default_verify_paths`` test so that it is skipped if no
   network connection is available.
+- Added ``OpenSSL.SSL.Connection.get_selected_srtp_profile`` to determine which SRTP profile was negotiated.
+  `#1279 <https://github.com/pyca/pyopenssl/pull/1279>`_.
 
 23.2.0 (2023-05-30)
 -------------------
diff --git a/src/OpenSSL/SSL.py b/src/OpenSSL/SSL.py
index ae07442..4db5240 100644
--- a/src/OpenSSL/SSL.py
+++ b/src/OpenSSL/SSL.py
@@ -2858,6 +2858,19 @@ class Connection:
 
         return _ffi.buffer(data[0], data_len[0])[:]
 
+    def get_selected_srtp_profile(self):
+        """
+        Get the SRTP protocol which was negotiated.
+
+        :returns: A bytestring of the SRTP profile name. If no profile has been
+            negotiated yet, returns an empty bytestring.
+        """
+        profile = _lib.SSL_get_selected_srtp_profile(self._ssl)
+        if not profile:
+            return b""
+
+        return _ffi.string(profile.name)
+
     def request_ocsp(self):
         """
         Called to request that the server sends stapled OCSP data, if
diff --git a/src/OpenSSL/crypto.py b/src/OpenSSL/crypto.py
index fb466ae..1707488 100644
--- a/src/OpenSSL/crypto.py
+++ b/src/OpenSSL/crypto.py
@@ -3162,7 +3162,7 @@ def sign(pkey: PKey, data: Union[str, bytes], digest: str) -> bytes:
 utils.deprecated(
     sign,
     __name__,
-    "sign() is deprecated. Use the equivilant APIs in cryptography.",
+    "sign() is deprecated. Use the equivalent APIs in cryptography.",
     DeprecationWarning,
     name="sign",
 )
@@ -3209,7 +3209,7 @@ def verify(
 utils.deprecated(
     verify,
     __name__,
-    "verify() is deprecated. Use the equivilant APIs in cryptography.",
+    "verify() is deprecated. Use the equivalent APIs in cryptography.",
     DeprecationWarning,
     name="verify",
 )


Test patch:
diff --git a/tests/test_crypto.py b/tests/test_crypto.py
index 6351f0d..3d1bf82 100644
--- a/tests/test_crypto.py
+++ b/tests/test_crypto.py
@@ -770,7 +770,7 @@ def x509_data():
     """
     # Basic setup stuff to generate a certificate
     pkey = PKey()
-    pkey.generate_key(TYPE_RSA, 512)
+    pkey.generate_key(TYPE_RSA, 2048)
     req = X509Req()
     req.set_pubkey(pkey)
     # Authority good you have.
@@ -1123,7 +1123,7 @@ class TestPKey:
         `PKey.generate_key` generates an RSA key when passed `TYPE_RSA` as a
         type and a reasonable number of bits.
         """
-        bits = 512
+        bits = 2048
         key = PKey()
         key.generate_key(TYPE_RSA, bits)
         assert key.type() == TYPE_RSA
@@ -1152,7 +1152,7 @@ class TestPKey:
         generate new keys.
         """
         key = PKey()
-        for type, bits in [(TYPE_RSA, 512), (TYPE_DSA, 576)]:
+        for type, bits in [(TYPE_RSA, 2048), (TYPE_DSA, 576)]:
             key.generate_key(type, bits)
             assert key.type() == type
             assert key.bits() == bits
@@ -1173,7 +1173,7 @@ class TestPKey:
         """
         # A trick to get a public-only key
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         cert = X509()
         cert.set_pubkey(key)
         pub = cert.get_pubkey()
@@ -1487,7 +1487,7 @@ class _PKeyInteractionTestsMixin:
         """
         request = self.signable()
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         request.set_pubkey(key)
         pub = request.get_pubkey()
         with pytest.raises(ValueError):
@@ -1500,7 +1500,7 @@ class _PKeyInteractionTestsMixin:
         """
         request = self.signable()
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         with pytest.raises(ValueError):
             request.sign(key, BAD_DIGEST)
 
@@ -1512,7 +1512,7 @@ class _PKeyInteractionTestsMixin:
         """
         request = self.signable()
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         request.set_pubkey(key)
         request.sign(key, GOOD_DIGEST)
         # If the type has a verify method, cover that too.
@@ -1521,7 +1521,7 @@ class _PKeyInteractionTestsMixin:
             assert request.verify(pub)
             # Make another key that won't verify.
             key = PKey()
-            key.generate_key(TYPE_RSA, 512)
+            key.generate_key(TYPE_RSA, 2048)
             with pytest.raises(Error):
                 request.verify(key)
 
@@ -2773,7 +2773,7 @@ class TestFunction:
         argument but no `passphrase` argument.
         """
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         with pytest.raises(TypeError):
             dump_privatekey(FILETYPE_PEM, key, cipher=GOOD_CIPHER)
 
@@ -2797,7 +2797,7 @@ class TestFunction:
         cipher name.
         """
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         with pytest.raises(ValueError):
             dump_privatekey(FILETYPE_PEM, key, BAD_CIPHER, "passphrase")
 
@@ -2807,7 +2807,7 @@ class TestFunction:
         is neither a `str` nor a callable.
         """
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         with pytest.raises(TypeError):
             dump_privatekey(FILETYPE_PEM, key, GOOD_CIPHER, object())
 
@@ -2817,7 +2817,7 @@ class TestFunction:
         filetype.
         """
         key = PKey()
-        key.generate_key(TYPE_RSA, 512)
+        key.generate_key(TYPE_RSA, 2048)
         with pytest.raises(ValueError):
             dump_privatekey(100, key)
 
diff --git a/tests/test_ssl.py b/tests/test_ssl.py
index d667eec..ca5bf83 100644
--- a/tests/test_ssl.py
+++ b/tests/test_ssl.py
@@ -4380,7 +4380,7 @@ class TestDTLS:
     # Arbitrary number larger than any conceivable handshake volley.
     LARGE_BUFFER = 65536
 
-    def test_it_works_at_all(self):
+    def _test_handshake_and_data(self, srtp_profile):
         s_ctx = Context(DTLS_METHOD)
 
         def generate_cookie(ssl):
@@ -4394,11 +4394,15 @@ class TestDTLS:
         s_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
         s_ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
         s_ctx.set_options(OP_NO_QUERY_MTU)
+        if srtp_profile is not None:
+            s_ctx.set_tlsext_use_srtp(srtp_profile)
         s = Connection(s_ctx)
         s.set_accept_state()
 
         c_ctx = Context(DTLS_METHOD)
         c_ctx.set_options(OP_NO_QUERY_MTU)
+        if srtp_profile is not None:
+            c_ctx.set_tlsext_use_srtp(srtp_profile)
         c = Connection(c_ctx)
         c.set_connect_state()
 
@@ -4480,6 +4484,14 @@ class TestDTLS:
         pump()
         assert s.read(100) == b"goodbye"
 
+        # Check whether SRTP was negotiated
+        if srtp_profile is not None:
+            assert s.get_selected_srtp_profile() == srtp_profile
+            assert c.get_selected_srtp_profile() == srtp_profile
+        else:
+            assert s.get_selected_srtp_profile() == b""
+            assert c.get_selected_srtp_profile() == b""
+
         # Check that the MTU set/query functions are doing *something*
         c.set_ciphertext_mtu(1000)
         try:
@@ -4492,6 +4504,12 @@ class TestDTLS:
         except NotImplementedError:  # OpenSSL 1.1.0 and earlier
             pass
 
+    def test_it_works_at_all(self):
+        self._test_handshake_and_data(srtp_profile=None)
+
+    def test_it_works_with_srtp(self):
+        self._test_handshake_and_data(srtp_profile=b"SRTP_AES128_CM_SHA1_80")
+
     def test_timeout(self, monkeypatch):
         c_ctx = Context(DTLS_METHOD)
         c = Connection(c_ctx)