Move off deprecated pyOpenSSL functionality (#7968)

* migrate off pyOpenSSL's crypto module

* `asyncio.iscoroutinefunction` -> `inspect.iscoroutinefunction`

mitmproxy/addons/tlsconfig.py

@@ -11,7 +11,6 @@ from typing import TypedDict
 from aioquic.h3.connection import H3_ALPN
 from aioquic.tls import CipherSuite
 from cryptography import x509
-from OpenSSL import crypto
 from OpenSSL import SSL
 
 from mitmproxy import certs
@@ -240,7 +239,7 @@ class TlsConfig:
             min_version=net_tls.Version[ctx.options.tls_version_client_min],
             max_version=net_tls.Version[ctx.options.tls_version_client_max],
             cipher_list=tuple(cipher_list),
-            ecdh_curve=ctx.options.tls_ecdh_curve_client,
+            ecdh_curve=net_tls.get_curve(ctx.options.tls_ecdh_curve_client),
             chain_file=entry.chain_file,
             request_client_cert=ctx.options.request_client_cert,
             alpn_select_callback=alpn_select_callback,
@@ -335,7 +334,7 @@ class TlsConfig:
             min_version=net_tls.Version[ctx.options.tls_version_server_min],
             max_version=net_tls.Version[ctx.options.tls_version_server_max],
             cipher_list=tuple(cipher_list),
-            ecdh_curve=ctx.options.tls_ecdh_curve_server,
+            ecdh_curve=net_tls.get_curve(ctx.options.tls_ecdh_curve_server),
             verify=verify,
             ca_path=ctx.options.ssl_verify_upstream_trusted_confdir,
             ca_pemfile=ctx.options.ssl_verify_upstream_trusted_ca,
@@ -521,13 +520,10 @@ class TlsConfig:
                 ctx.options.tls_ecdh_curve_client,
                 ctx.options.tls_ecdh_curve_server,
             ]:
-                if ecdh_curve is not None:
-                    try:
-                        crypto.get_elliptic_curve(ecdh_curve)
-                    except Exception as e:
-                        raise exceptions.OptionsError(
-                            f"Invalid ECDH curve: {ecdh_curve!r}"
-                        ) from e
+                if ecdh_curve is not None and ecdh_curve not in net_tls.EC_CURVES:
+                    raise exceptions.OptionsError(
+                        f"Invalid ECDH curve: {ecdh_curve!r}. Valid curves are: {', '.join(net_tls.EC_CURVES)}"
+                    )
 
         if "tls_version_client_min" in updated:
             self._warn_unsupported_version("tls_version_client_min", True)

mitmproxy/net/tls.py

@@ -1,5 +1,6 @@
 import os
 import threading
+import typing
 from collections.abc import Callable
 from collections.abc import Iterable
 from enum import Enum
@@ -10,9 +11,12 @@ from typing import Any
 from typing import BinaryIO
 
 import certifi
-from OpenSSL import crypto
+import OpenSSL
+from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurve
+from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurveOID
+from cryptography.hazmat.primitives.asymmetric.ec import get_curve_for_oid
+from cryptography.x509 import ObjectIdentifier
 from OpenSSL import SSL
-from OpenSSL.crypto import X509
 
 from mitmproxy import certs
 
@@ -86,6 +90,27 @@ def is_supported_version(version: Version):
         return False
 
 
+EC_CURVES: dict[str, EllipticCurve] = {}
+for oid in EllipticCurveOID.__dict__.values():
+    if isinstance(oid, ObjectIdentifier):
+        curve = get_curve_for_oid(oid)()
+        EC_CURVES[curve.name] = curve
+
+
+@typing.overload
+def get_curve(name: str) -> EllipticCurve: ...
+
+
+@typing.overload
+def get_curve(name: None) -> None: ...
+
+
+def get_curve(name: str | None) -> EllipticCurve | None:
+    if name is None:
+        return None
+    return EC_CURVES[name]
+
+
 class MasterSecretLogger:
     def __init__(self, filename: Path):
         self.filename = filename.expanduser()
@@ -127,7 +152,7 @@ def _create_ssl_context(
     min_version: Version,
     max_version: Version,
     cipher_list: Iterable[str] | None,
-    ecdh_curve: str | None,
+    ecdh_curve: EllipticCurve | None,
 ) -> SSL.Context:
     context = SSL.Context(method.value)
 
@@ -145,7 +170,7 @@ def _create_ssl_context(
     # ECDHE for Key exchange
     if ecdh_curve is not None:
         try:
-            context.set_tmp_ecdh(crypto.get_elliptic_curve(ecdh_curve))
+            context.set_tmp_ecdh(ecdh_curve)
         except ValueError as e:
             raise RuntimeError(f"Elliptic curve specification error: {e}") from e
 
@@ -170,7 +195,7 @@ def create_proxy_server_context(
     min_version: Version,
     max_version: Version,
     cipher_list: tuple[str, ...] | None,
-    ecdh_curve: str | None,
+    ecdh_curve: EllipticCurve | None,
     verify: Verify,
     ca_path: str | None,
     ca_pemfile: str | None,
@@ -219,7 +244,7 @@ def create_client_proxy_context(
     min_version: Version,
     max_version: Version,
     cipher_list: tuple[str, ...] | None,
-    ecdh_curve: str | None,
+    ecdh_curve: EllipticCurve | None,
     chain_file: Path | None,
     alpn_select_callback: Callable[[SSL.Connection, list[bytes]], Any] | None,
     request_client_cert: bool,
@@ -269,7 +294,7 @@ def create_client_proxy_context(
 
 def accept_all(
     conn_: SSL.Connection,
-    x509: X509,
+    x509: OpenSSL.crypto.X509,
     errno: int,
     err_depth: int,
     is_cert_verified: int,

mitmproxy/utils/signals.py

@@ -67,7 +67,7 @@ class _SignalMixin:
 
 class _SyncSignal(Generic[P], _SignalMixin):
     def connect(self, receiver: Callable[P, None]) -> None:
-        assert not asyncio.iscoroutinefunction(receiver)
+        assert not inspect.iscoroutinefunction(receiver)
         super().connect(receiver)
 
     def disconnect(self, receiver: Callable[P, None]) -> None:

test/mitmproxy/net/test_tls.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 import pytest
+from cryptography.hazmat.primitives.asymmetric import ec
 from OpenSSL import SSL
 
 from mitmproxy import certs
@@ -101,3 +102,7 @@ def test_is_dtls_record_magic():
     assert not tls.starts_like_dtls_record(bytes.fromhex("160300"))
     assert not tls.starts_like_dtls_record(bytes.fromhex("160304"))
     assert not tls.starts_like_dtls_record(bytes.fromhex("150301"))
+
+
+def test_get_curve():
+    assert isinstance(tls.get_curve("secp256r1"), ec.SECP256R1)