run linting

Author: eirikhex

tests/conftest.py

@@ -113,24 +113,20 @@ def tls_ca_ssl_context(tls_certificate_authority: trustme.CA) -> ssl.SSLContext:
     tls_certificate_authority.configure_trust(ssl_ctx)
     return ssl_ctx
 
+
 @pytest.fixture
-def tls_client_ssl_context(tls_certificate_authority: trustme.CA, tls_client_certificate: trustme.LeafCert) -> ssl.SSLContext:
+def tls_client_ssl_context(
+    tls_certificate_authority: trustme.CA, tls_client_certificate: trustme.LeafCert
+) -> ssl.SSLContext:
     ssl_ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
     tls_certificate_authority.configure_trust(ssl_ctx)
 
     # Load the client certificate chain into the SSL context
     with tls_client_certificate.private_key_and_cert_chain_pem.tempfile() as client_cert_pem:
         ssl_ctx.load_cert_chain(certfile=client_cert_pem)
-    
 
     return ssl_ctx
 
-@pytest.fixture
-def tls_client_certificate_pem_path(tls_client_certificate: trustme.LeafCert):
-    private_key_and_cert_chain = tls_client_certificate.private_key_and_cert_chain_pem
-    with private_key_and_cert_chain.tempfile() as client_cert_pem:
-        yield client_cert_pem
-
 
 @pytest.fixture(scope="package")
 def reload_directory_structure(tmp_path_factory: pytest.TempPathFactory):
@@ -283,8 +279,3 @@ def ws_protocol_cls(request: pytest.FixtureRequest):
 )
 def http_protocol_cls(request: pytest.FixtureRequest):
     return import_from_string(request.param)
-@pytest.fixture
-def tls_client_certificate_pem_path(tls_client_certificate: trustme.LeafCert):
-    private_key_and_cert_chain = tls_client_certificate.private_key_and_cert_chain_pem
-    with private_key_and_cert_chain.tempfile() as client_cert_pem:
-        yield client_cert_pem

tests/test_ssl.py

@@ -2,7 +2,6 @@
 
 import httpx
 import pytest
-
 from cryptography import x509
 
 from tests.utils import run_server
@@ -46,8 +45,6 @@ async def test_run(
     ],
     indirect=["tls_client_certificate"],
 )
-
-
 @pytest.mark.anyio
 async def test_run_httptools_client_cert(
     tls_client_ssl_context,
@@ -59,11 +56,13 @@ async def test_run_httptools_client_cert(
     async def app(scope, receive, send):
         assert scope["type"] == "http"
         assert len(scope["extensions"]["tls"]["client_cert_chain"]) >= 1
-        cert = x509.load_pem_x509_certificate(scope["extensions"]["tls"]["client_cert_chain"][0].encode('utf-8'))
+        cert = x509.load_pem_x509_certificate(scope["extensions"]["tls"]["client_cert_chain"][0].encode("utf-8"))
         assert cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value == expected_common_name
-        cipher_suites = [cipher['name'] for cipher in ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER).get_ciphers()]
+        cipher_suites = [cipher["name"] for cipher in ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER).get_ciphers()]
         assert scope["extensions"]["tls"]["cipher_suite"] in cipher_suites
-        assert (scope["extensions"]["tls"]["tls_version"].startswith("TLSv") or scope["extensions"]["tls"]["tls_version"].startswith("SSLv"))
+        assert scope["extensions"]["tls"]["tls_version"].startswith("TLSv") or scope["extensions"]["tls"][
+            "tls_version"
+        ].startswith("SSLv")
 
         await send({"type": "http.response.start", "status": 204, "headers": []})
         await send({"type": "http.response.body", "body": b"", "more_body": False})

uvicorn/protocols/http/h11_impl.py

@@ -27,13 +27,15 @@
     service_unavailable,
 )
 from uvicorn.protocols.utils import (
+    TLSInfo,
     get_client_addr,
     get_local_addr,
     get_path_with_query_string,
     get_remote_addr,
     get_tls_info,
     is_ssl,
 )
+from uvicorn.server import ServerState
 
 
 def _get_status_phrase(status_code: int) -> bytes:
@@ -89,7 +91,7 @@ def __init__(
         self.server: tuple[str, int] | None = None
         self.client: tuple[str, int] | None = None
         self.scheme: Literal["http", "https"] | None = None
-        self.tls: dict[object, object] = {}
+        self.tls: TLSInfo = TLSInfo()
 
         # Per-request state
         self.scope: HTTPScope = None  # type: ignore[assignment]
@@ -231,8 +233,8 @@ def handle_events(self) -> None:
                 }
 
                 if self.config.is_ssl:
-                    self.scope["extensions"]["tls"] = self.tls
-                    
+                    self.scope["extensions"]["tls"] = cast(dict[object, object], self.tls)
+
                 if self._should_upgrade():
                     self.handle_websocket_upgrade(event)
                     return

uvicorn/protocols/http/httptools_impl.py

@@ -33,13 +33,15 @@
     service_unavailable,
 )
 from uvicorn.protocols.utils import (
+    TLSInfo,
     get_client_addr,
     get_local_addr,
     get_path_with_query_string,
     get_remote_addr,
     get_tls_info,
     is_ssl,
 )
+from uvicorn.server import ServerState
 
 HEADER_RE = re.compile(b'[\x00-\x1f\x7f()<>@,;:[]={} \t\\"]')
 HEADER_VALUE_RE = re.compile(b"[\x00-\x08\x0a-\x1f\x7f]")
@@ -103,7 +105,7 @@ def __init__(
         self.client: tuple[str, int] | None = None
         self.scheme: Literal["http", "https"] | None = None
         self.pipeline: deque[tuple[RequestResponseCycle, ASGI3Application]] = deque()
-        self.tls: dict[object, object] = {}
+        self.tls: TLSInfo = TLSInfo()
 
         # Per-request state
         self.scope: HTTPScope = None  # type: ignore[assignment]
@@ -124,7 +126,7 @@ def connection_made(  # type: ignore[override]
         self.scheme = "https" if is_ssl(transport) else "http"
 
         if self.config.is_ssl:
-            self.tls = get_tls_info(transport,self.config)
+            self.tls = get_tls_info(transport, self.config)
 
         if self.logger.level <= TRACE_LOG_LEVEL:
             prefix = "%s:%d - " % self.client if self.client else ""
@@ -250,11 +252,11 @@ def on_message_begin(self) -> None:
             "root_path": self.root_path,
             "headers": self.headers,
             "state": self.app_state.copy(),
-                "extensions": {},
+            "extensions": {},
         }
 
         if self.config.is_ssl:
-            self.scope["extensions"]["tls"] = self.tls
+            self.scope["extensions"]["tls"] = cast(dict[object, object], self.tls)
 
     # Parser callbacks
     def on_url(self, url: bytes) -> None:

uvicorn/protocols/utils.py

@@ -3,7 +3,7 @@
 import asyncio
 import ssl
 import urllib.parse
-import sys
+from typing import TypedDict
 
 from uvicorn._types import WWWScope
 from uvicorn.config import Config
@@ -59,16 +59,22 @@ def get_path_with_query_string(scope: WWWScope) -> str:
     return path_with_query_string
 
 
+class TLSInfo(TypedDict, total=False):
+    server_cert: str | None
+    client_cert_chain: list[str]
+    tls_version: str | None
+    cipher_suite: str | None
 
-def get_tls_info(transport: asyncio.Transport, server_config: Config) -> dict[object, object]:
+
+def get_tls_info(transport: asyncio.Transport, server_config: Config) -> TLSInfo:
     ###
     # server_cert: Unable to set from transport information, need to set from server_config
-    # client_cert_chain: 
+    # client_cert_chain:
     # tls_version:
     # cipher_suite:
     ###
 
-    ssl_info: dict[object, object] = {
+    ssl_info: TLSInfo = {
         "server_cert": None,
         "client_cert_chain": [],
         "tls_version": None,
@@ -79,12 +85,16 @@ def get_tls_info(transport: asyncio.Transport, server_config: Config) -> dict[ob
 
     ssl_object = transport.get_extra_info("ssl_object")
     if ssl_object is not None:
-        client_chain = ssl_object.get_verified_chain() if hasattr(ssl_object, "get_verified_chain") else [ssl_object.getpeercert(binary_form=True)]
+        client_chain = (
+            ssl_object.get_verified_chain()
+            if hasattr(ssl_object, "get_verified_chain")
+            else [ssl_object.getpeercert(binary_form=True)]
+        )
         for cert in client_chain:
             if cert is not None:
                 ssl_info["client_cert_chain"].append(ssl.DER_cert_to_PEM_cert(cert))
-            
+
         ssl_info["tls_version"] = ssl_object.version()
         ssl_info["cipher_suite"] = ssl_object.cipher()[0] if ssl_object.cipher() else None
 
-    return ssl_info
+    return ssl_info