Version 0.47.0 (#2937)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

.github/dependabot.yml

@@ -4,6 +4,8 @@ updates:
     directory: "/"
     schedule:
       interval: "monthly"
+    cooldown:
+      default-days: 7
     groups:
       python-packages:
         patterns:
@@ -12,6 +14,8 @@ updates:
     directory: "/"
     schedule:
       interval: monthly
+    cooldown:
+      default-days: 7
     groups:
       github-actions:
         patterns:

.github/workflows/benchmark.yml

@@ -1,4 +1,3 @@
----
 name: CodSpeed
 
 on:
@@ -17,18 +16,21 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: "actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd" # v6.0.2
+        with:
+          persist-credentials: false
 
       - name: Install uv
-        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
         with:
           python-version: "3.13"
+          enable-cache: true
 
       - name: Install dependencies
         run: scripts/install
         shell: bash
 
       - name: Run the benchmarks
-        uses: CodSpeedHQ/action@281164b0f014a4e7badd2c02cecad9b595b70537 # v4
+        uses: CodSpeedHQ/action@db35df748deb45fdef0960669f57d627c1956c30 # v4
         with:
           mode: instrumentation
           run: uv run pytest tests/benchmarks/ --codspeed -n 0

.github/workflows/main.yml

@@ -1,4 +1,3 @@
----
 name: Test Suite
 
 on:
@@ -13,6 +12,9 @@ jobs:
     runs-on: "${{ matrix.os }}"
     timeout-minutes: 10
 
+    permissions:
+      contents: read
+
     strategy:
       fail-fast: false
       matrix:
@@ -21,9 +23,11 @@ jobs:
 
     steps:
       - uses: "actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd" # v6.0.2
+        with:
+          persist-credentials: false
 
       - name: Install uv
-        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
         with:
           python-version: ${{ matrix.python-version }}
           enable-cache: ${{ matrix.os != 'windows-latest' }}
@@ -48,11 +52,62 @@ jobs:
         run: scripts/coverage
         shell: bash
 
+  docs-preview:
+    if: github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+      pull-requests: write
+
+    environment:
+      name: cloudflare
+      url: ${{ steps.deploy.outputs.deployment-url }}
+
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+        with:
+          python-version: "3.12"
+          enable-cache: false
+
+      - name: Install dependencies
+        run: scripts/install
+
+      - name: Build docs
+        run: uv run mkdocs build
+
+      - name: Deploy preview to Cloudflare Pages
+        id: deploy
+        uses: cloudflare/wrangler-action@da0e0dfe58b7a431659754fdf3f186c529afbe65 # v3.14.1
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          command: >
+            pages deploy ./site
+            --project-name uvicorn
+            --commit-hash ${{ github.event.pull_request.head.sha }}
+            --branch ${{ github.head_ref }}
+
+      - name: Comment preview URL on PR
+        uses: thollander/actions-comment-pull-request@24bffb9b452ba05a4f3f77933840a6a841d1b32b # v3.0.1
+        with:
+          message: |
+            :book: Docs preview: ${{ steps.deploy.outputs.deployment-url }}
+          comment-tag: docs-preview
+
   # https://github.com/marketplace/actions/alls-green#why
   check:
     if: always()
     needs: [tests]
     runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+
     steps:
       - name: Decide whether the needed jobs succeeded or failed
         uses: re-actors/alls-green@05ac9388f0aebcb5727afa17fcccfecd6f8ec5fe # v1.2.2

.github/workflows/publish.yml

@@ -10,14 +10,19 @@ jobs:
   build:
     runs-on: ubuntu-latest
 
+    permissions:
+      contents: read
+
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
 
       - name: Install uv
-        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
         with:
           python-version: "3.11"
-          enable-cache: true
+          enable-cache: false
 
       - name: Install dependencies
         run: scripts/install
@@ -51,7 +56,7 @@ jobs:
 
     steps:
       - name: Download artifacts
-        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
         with:
           name: package-distributions
           path: dist/
@@ -67,9 +72,11 @@ jobs:
       contents: write
 
     steps:
-      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      # `mkdocs gh-deploy` pushes the built docs to `gh-pages`, so this job needs
+      # a real checkout with the authenticated origin remote preserved.
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2  # zizmor: ignore[artipacked]
       - name: Download artifacts
-        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
         with:
           name: documentation
           path: site/
@@ -80,10 +87,10 @@ jobs:
           git config user.email 41898282+github-actions[bot]@users.noreply.github.com
 
       - name: Install uv
-        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
         with:
           python-version: "3.12"
-          enable-cache: true
+          enable-cache: false
 
       - name: Install dependencies
         run: scripts/install
@@ -95,14 +102,19 @@ jobs:
     runs-on: ubuntu-latest
     needs: build
 
+    permissions:
+      contents: read
+
     environment:
       name: cloudflare
       url: https://uvicorn.dev
 
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
       - name: Download artifacts
-        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
         with:
           name: documentation
           path: site/

.github/workflows/zizmor.yml

@@ -0,0 +1,25 @@
+name: GitHub Actions Security Analysis
+
+on:
+  push:
+    branches: ["main"]
+  pull_request:
+    branches: ["**"]
+
+permissions: {}
+
+jobs:
+  zizmor:
+    runs-on: ubuntu-latest
+
+    permissions:
+      security-events: write # Required for upload-sarif (used by zizmor-action) to upload SARIF files.
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Run zizmor 🌈
+        uses: zizmorcore/zizmor-action@71321a20a9ded102f6e9ce5718a2fcec2c4f70d8 # v0.5.2

docs/concepts/logging.md

@@ -0,0 +1,319 @@
+Uvicorn uses Python's built-in [`logging`](https://docs.python.org/3/library/logging.html)
+module, and provides three loggers out of the box:
+
+| Logger name      | Purpose                                            |
+|------------------|----------------------------------------------------|
+| `uvicorn`        | Parent logger (rarely used directly)               |
+| `uvicorn.error`  | Server-level messages (startup, shutdown, errors)   |
+| `uvicorn.access` | Per-request access log lines                        |
+
+!!! note
+    Despite its name, `uvicorn.error` is **not** limited to error messages.
+    It is the general-purpose server logger, similar to how Gunicorn names its
+    main logger. See [#562](https://github.com/encode/uvicorn/issues/562) for
+    background.
+
+## Default Configuration
+
+By default, Uvicorn applies the following
+[`dictConfig()`](https://docs.python.org/3/library/logging.config.html#logging.config.dictConfig)
+configuration:
+
+```python
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "default": {
+            "()": "uvicorn.logging.DefaultFormatter",
+            "fmt": "%(levelprefix)s %(message)s",
+            "use_colors": None,
+        },
+        "access": {
+            "()": "uvicorn.logging.AccessFormatter",
+            "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
+        },
+    },
+    "handlers": {
+        "default": {
+            "formatter": "default",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stderr",
+        },
+        "access": {
+            "formatter": "access",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stdout",
+        },
+    },
+    "loggers": {
+        "uvicorn": {"handlers": ["default"], "level": "INFO", "propagate": False},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+    },
+}
+```
+
+## Custom Logging Configuration
+
+You can supply a custom logging configuration file with the `--log-config`
+option (or `log_config` when calling `uvicorn.run()`).
+
+Uvicorn supports three file formats:
+
+| Extension      | Loader                       | Notes                                       |
+|----------------|------------------------------|---------------------------------------------|
+| `.json`        | `logging.config.dictConfig`  | Standard JSON `dictConfig` schema.          |
+| `.yaml`/`.yml` | `logging.config.dictConfig`  | Requires **PyYAML** (`uvicorn[standard]`).  |
+| Any other      | `logging.config.fileConfig`  | Classic INI-style format.                   |
+
+### YAML Example
+
+Create a file named `log_config.yaml`:
+
+```yaml
+version: 1
+disable_existing_loggers: false
+formatters:
+  default:
+    "()": uvicorn.logging.DefaultFormatter
+    fmt: "%(asctime)s - %(levelprefix)s %(message)s"
+    datefmt: "%Y-%m-%d %H:%M:%S"
+    use_colors: null
+  access:
+    "()": uvicorn.logging.AccessFormatter
+    fmt: '%(asctime)s - %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'
+    datefmt: "%Y-%m-%d %H:%M:%S"
+handlers:
+  default:
+    formatter: default
+    class: logging.StreamHandler
+    stream: ext://sys.stderr
+  access:
+    formatter: access
+    class: logging.StreamHandler
+    stream: ext://sys.stdout
+loggers:
+  uvicorn:
+    handlers:
+      - default
+    level: INFO
+    propagate: false
+  uvicorn.error:
+    level: INFO
+  uvicorn.access:
+    handlers:
+      - access
+    level: INFO
+    propagate: false
+```
+
+Then pass it to Uvicorn:
+
+=== "CLI"
+
+    ```bash
+    uvicorn main:app --log-config log_config.yaml
+    ```
+
+=== "Programmatic"
+
+    ```python
+    uvicorn.run("main:app", log_config="log_config.yaml")
+    ```
+
+### JSON Example
+
+Create a file named `log_config.json`:
+
+```json
+{
+  "version": 1,
+  "disable_existing_loggers": false,
+  "formatters": {
+    "default": {
+      "()": "uvicorn.logging.DefaultFormatter",
+      "fmt": "%(asctime)s - %(levelprefix)s %(message)s",
+      "datefmt": "%Y-%m-%d %H:%M:%S",
+      "use_colors": null
+    },
+    "access": {
+      "()": "uvicorn.logging.AccessFormatter",
+      "fmt": "%(asctime)s - %(levelprefix)s %(client_addr)s - \"%(request_line)s\" %(status_code)s",
+      "datefmt": "%Y-%m-%d %H:%M:%S"
+    }
+  },
+  "handlers": {
+    "default": {
+      "formatter": "default",
+      "class": "logging.StreamHandler",
+      "stream": "ext://sys.stderr"
+    },
+    "access": {
+      "formatter": "access",
+      "class": "logging.StreamHandler",
+      "stream": "ext://sys.stdout"
+    }
+  },
+  "loggers": {
+    "uvicorn": {
+      "handlers": ["default"],
+      "level": "INFO",
+      "propagate": false
+    },
+    "uvicorn.error": {
+      "level": "INFO"
+    },
+    "uvicorn.access": {
+      "handlers": ["access"],
+      "level": "INFO",
+      "propagate": false
+    }
+  }
+}
+```
+
+### Programmatic `dictConfig`
+
+You can also pass a dictionary directly when running programmatically:
+
+```python
+import uvicorn
+
+log_config = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "default": {
+            "()": "uvicorn.logging.DefaultFormatter",
+            "fmt": "%(asctime)s - %(levelprefix)s %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+        "access": {
+            "()": "uvicorn.logging.AccessFormatter",
+            "fmt": '%(asctime)s - %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+    },
+    "handlers": {
+        "default": {
+            "formatter": "default",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stderr",
+        },
+        "access": {
+            "formatter": "access",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stdout",
+        },
+    },
+    "loggers": {
+        "uvicorn": {"handlers": ["default"], "level": "INFO", "propagate": False},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+    },
+}
+
+uvicorn.run("main:app", log_config=log_config)
+```
+
+## Common Recipes
+
+### Writing Logs to a File
+
+To write Uvicorn's server logs to a file in addition to the console, add a `FileHandler` to the `uvicorn` logger:
+
+```yaml
+version: 1
+disable_existing_loggers: false
+formatters:
+  default:
+    "()": uvicorn.logging.DefaultFormatter
+    fmt: "%(asctime)s - %(levelprefix)s %(message)s"
+    datefmt: "%Y-%m-%d %H:%M:%S"
+    use_colors: false
+  access:
+    "()": uvicorn.logging.AccessFormatter
+    fmt: '%(asctime)s - %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'
+    datefmt: "%Y-%m-%d %H:%M:%S"
+handlers:
+  default:
+    formatter: default
+    class: logging.StreamHandler
+    stream: ext://sys.stderr
+  access:
+    formatter: access
+    class: logging.StreamHandler
+    stream: ext://sys.stdout
+  file:
+    formatter: default
+    class: logging.FileHandler
+    filename: uvicorn.log
+loggers:
+  uvicorn:
+    handlers:
+      - default
+      - file
+    level: INFO
+    propagate: false
+  uvicorn.error:
+    level: INFO
+  uvicorn.access:
+    handlers:
+      - access
+    level: INFO
+    propagate: false
+```
+
+In this example, `uvicorn.access` still writes to stdout only. To write access
+logs to the file as well, add `file` to the `uvicorn.access.handlers` list.
+
+### Disabling Access Logs
+
+Use the `--no-access-log` CLI flag, or set `access_log=False` programmatically.
+This removes all handlers from `uvicorn.access` without affecting the
+`uvicorn.error` logger.
+
+### Disabling Colors
+
+Pass `--no-use-colors` on the command line, or set `use_colors=False`
+programmatically. When using a custom `--log-config`, set `use_colors: false`
+on each formatter that extends `uvicorn.logging.ColourizedFormatter`.
+
+### Using a Standard Formatter
+
+If you do not need Uvicorn's colorized output, you can use the standard
+`logging.Formatter` instead:
+
+```yaml
+version: 1
+disable_existing_loggers: false
+formatters:
+  default:
+    format: "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+    datefmt: "%Y-%m-%d %H:%M:%S"
+handlers:
+  default:
+    formatter: default
+    class: logging.StreamHandler
+    stream: ext://sys.stderr
+loggers:
+  uvicorn:
+    handlers:
+      - default
+    level: INFO
+    propagate: false
+  uvicorn.error:
+    level: INFO
+  uvicorn.access:
+    handlers:
+      - default
+    level: INFO
+    propagate: false
+```
+
+!!! warning
+    When using a standard `logging.Formatter` for the access logger, the
+    `%(client_addr)s`, `%(request_line)s`, and `%(status_code)s` placeholders
+    are **not** available. The access log line will be formatted using only the
+    standard `%(message)s` field.

docs/css/extra.css

@@ -0,0 +1,57 @@
+.md-nav__sponsors {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  gap: 0.5rem;
+  margin: 1.2rem 0.4rem 0.6rem;
+  padding: 0.9rem 0.6rem 0.8rem;
+  background-color: color-mix(in srgb, var(--md-primary-fg-color) 8%, transparent);
+  border-radius: 0.4rem;
+}
+
+.md-nav__sponsors-title {
+  margin: 0 0 0.1rem;
+  font-size: 0.6rem;
+  font-weight: 700;
+  text-transform: uppercase;
+  letter-spacing: 0.05em;
+  color: var(--md-default-fg-color--light);
+}
+
+.md-nav__sponsor {
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  width: 100%;
+  padding: 0.25rem;
+  border-radius: 0.2rem;
+  transition: opacity 0.15s;
+}
+
+.md-nav__sponsor:hover {
+  opacity: 0.75;
+}
+
+.md-nav__sponsor img {
+  max-width: 100%;
+  max-height: 1.6rem;
+  object-fit: contain;
+}
+
+.md-nav__sponsor-cta {
+  display: inline-block;
+  margin-top: 0.15rem;
+  padding: 0.25rem 0.6rem;
+  font-size: 0.65rem;
+  font-weight: 600;
+  color: var(--md-primary-bg-color);
+  background-color: var(--md-primary-fg-color);
+  border-radius: 0.2rem;
+  text-decoration: none;
+  transition: opacity 0.15s;
+}
+
+.md-nav__sponsor-cta:hover {
+  opacity: 0.85;
+  color: var(--md-primary-bg-color);
+}

docs/deployment/index.md

@@ -82,7 +82,7 @@ The default process manager monitors the status of child processes and automatic
 
 You can also manage child processes by sending specific signals to the main process. (Not supported on Windows.)
 
-- `SIGHUP`: Work processeses are graceful restarted one after another. If you update the code, the new worker process will use the new code.
+- `SIGHUP`: Work processes are graceful restarted one after another. If you update the code, the new worker process will use the new code.
 - `SIGTTIN`: Increase the number of worker processes by one.
 - `SIGTTOU`: Decrease the number of worker processes by one.
 
@@ -225,6 +225,36 @@ It's also possible to use certificates with uvicorn's worker for gunicorn.
 $ gunicorn --keyfile=./key.pem --certfile=./cert.pem -k uvicorn.workers.UvicornWorker main:app
 ```
 
+### Customizing the SSL context
+
+For TLS scenarios that the `--ssl-*` flags don't cover (e.g., mutual TLS, custom `SSLContext.options`, bumping `minimum_version`, loading certificates from memory), pass an `ssl_context_factory` to `uvicorn.run()` or `Config`.
+
+The factory receives the `Config` instance and a `default_ssl_context_factory` callable that builds the standard context from the `ssl_*` settings on `Config`. Use it to start from uvicorn's default and mutate it, or ignore it and build your own context from scratch - the `ssl_*` settings are only consumed by the default factory, so if you don't call it they're effectively unused.
+
+```python
+import ssl
+from collections.abc import Callable
+
+import uvicorn
+from uvicorn.config import Config
+
+
+def ssl_context_factory(config: Config, default_ssl_context_factory: Callable[[], ssl.SSLContext]) -> ssl.SSLContext:
+    context = default_ssl_context_factory()
+    context.minimum_version = ssl.TLSVersion.TLSv1_3
+    return context
+
+
+uvicorn.run(
+    "main:app",
+    ssl_keyfile="key.pem",
+    ssl_certfile="cert.pem",
+    ssl_context_factory=ssl_context_factory,
+)
+```
+
+The factory is called inside each worker process, so it works with `--reload` and `--workers > 1`. The factory itself must be picklable in those modes (a top-level function is fine; lambdas and local closures are not). The `ssl_*` settings on `Config` are only consumed by `default_ssl_context_factory()`; if you build the context yourself without calling it, those settings are ignored.
+
 ## Proxies and Forwarded Headers
 
 When running an application behind one or more proxies, certain information about the request is lost.

docs/index.md

@@ -44,6 +44,18 @@ and means we're now able to start building a common set of tooling usable across
 
 Uvicorn currently supports **HTTP/1.1** and **WebSockets**.
 
+## Sponsorship
+
+Help us keep Uvicorn maintained and sustainable by [becoming a sponsor](https://github.com/sponsors/Kludex).
+
+**Current sponsors:**
+
+<div style="display: flex; flex-wrap: wrap; gap: 2rem; align-items: center; margin: 1rem 0;">
+    <a href="https://fastapi.tiangolo.com">
+        <img src="img/fastapi-logo.png" alt="FastAPI" style="height: 80px;">
+    </a>
+</div>
+
 ## Quickstart
 
 **Uvicorn** is available on [PyPI](https://pypi.org/project/uvicorn/) so installation is as simple as:

docs/overrides/partials/nav.html

@@ -44,4 +44,15 @@
        {{ item.render(nav_item, path, 1) }}
      {% endfor %}
    </ul>
+
+   <!-- Sponsors -->
+   <div class="md-nav__sponsors">
+     <p class="md-nav__sponsors-title">Sponsors</p>
+     <a href="https://fastapi.tiangolo.com" title="FastAPI" class="md-nav__sponsor">
+       <img src="{{ 'img/fastapi-logo.png' | url }}" alt="FastAPI">
+     </a>
+     <a href="https://github.com/sponsors/Kludex" class="md-nav__sponsor-cta">
+       Become a sponsor! ❤️
+     </a>
+   </div>
  </nav>

docs/release-notes.md

@@ -2,6 +2,76 @@
 toc_depth: 2
 ---
 
+## 0.47.0 (May 14, 2026)
+
+### Added
+
+* Add `ssl_context_factory` for custom `SSLContext` configuration (#2920)
+
+### Changed
+
+* Eagerly import the ASGI app in the parent process (#2919)
+
+### Fixed
+
+* Treat `fd=0` as a valid file descriptor with reload/workers (#2927)
+
+## 0.46.0 (April 23, 2026)
+
+### Added
+
+* Support `ws_max_size` in `wsproto` implementation (#2915)
+* Support `ws_ping_interval` and `ws_ping_timeout` in `wsproto` implementation (#2916)
+
+### Changed
+
+* Use `bytearray` for incoming WebSocket message buffer in `websockets-sansio` (#2917)
+
+## 0.45.0 (April 21, 2026)
+
+### Added
+
+* Add `--reset-contextvars` flag to isolate ASGI request context (#2912)
+* Accept `os.PathLike` for `log_config` (#2905)
+* Accept `log_level` strings case-insensitively (#2907)
+
+### Changed
+
+* Revert "Emit `http.disconnect` on server shutdown for streaming responses" (#2913)
+* Revert "Explicitly start ASGI run with empty context" (#2911)
+
+### Fixed
+
+* Preserve forwarded client ports in proxy headers middleware (#2903)
+* Raise helpful `ImportError` when PyYAML is missing for YAML log config (#2906)
+
+## 0.44.0 (April 6, 2026)
+
+### Added
+
+* Implement websocket keepalive pings for websockets-sansio (#2888)
+
+## 0.43.0 (April 3, 2026)
+
+You can quit Uvicorn now. We heard you, @pamelafox - all 47 of your Ctrl+C's (thanks for flagging it, and thanks to @tiangolo for the fix 🙏). [See the tweet](https://x.com/pamelafox/status/2039097686155227623).
+
+### Changed
+
+* Emit `http.disconnect` ASGI `receive()` event on server shutting down for streaming responses (#2829)
+* Use native `context` parameter for `create_task` on Python 3.11+ (#2859)
+* Drop cast in ASGI types (#2875)
+
+## 0.42.0 (March 16, 2026)
+
+### Changed
+
+* Use `bytearray` for request body accumulation to avoid O(n^2) allocation on fragmented bodies (#2845)
+
+### Fixed
+
+* Escape brackets and backslash in httptools `HEADER_RE` regex (#2824)
+* Fix multiple issues in websockets sans-io implementation (#2825)
+
 ## 0.41.0 (February 16, 2026)
 
 ### Added

docs/settings.md

@@ -39,6 +39,7 @@ uvicorn itself.
 * `APP` - The ASGI application to run, in the format `"<module>:<attribute>"`.
 * `--factory` - Treat `APP` as an application factory, i.e. a `() -> <ASGI app>` callable.
 * `--app-dir <path>` - Look for APP in the specified directory by adding it to the PYTHONPATH. **Default:** *Current working directory*.
+* `--reset-contextvars` - Run each ASGI request in a fresh `contextvars.Context`. Workaround for a [context leak in asyncio](https://github.com/python/cpython/issues/140947); only relevant when using the `asyncio` event loop (uvloop is not affected). Enabling this hides any context set in the lifespan or by external instrumentation from ASGI handlers. **Default:** *False*.
 
 ## Socket Binding
 
@@ -93,10 +94,10 @@ Using Uvicorn with watchfiles will enable the following options (which are other
 * `--loop <str>` - Set the event loop implementation. The uvloop implementation provides greater performance, but is not compatible with Windows or PyPy. **Options:** *'auto', 'asyncio', 'uvloop'.* **Default:** *'auto'*.
 * `--http <str>` - Set the HTTP protocol implementation. The httptools implementation provides greater performance, but it not compatible with PyPy. **Options:** *'auto', 'h11', 'httptools'.* **Default:** *'auto'*.
 * `--ws <str>` - Set the WebSockets protocol implementation. Either of the `websockets` and `wsproto` packages are supported. There are two versions of `websockets` supported: `websockets` and `websockets-sansio`. Use `'none'` to ignore all websocket requests. **Options:** *'auto', 'none', 'websockets', 'websockets-sansio', 'wsproto'.* **Default:** *'auto'*.
-* `--ws-max-size <int>` - Set the WebSockets max message size, in bytes. Only available with the `websockets` protocol. **Default:** *16777216* (16 MB).
+* `--ws-max-size <int>` - Set the WebSockets max message size, in bytes. **Default:** *16777216* (16 MB).
 * `--ws-max-queue <int>` - Set the maximum length of the WebSocket incoming message queue. Only available with the `websockets` protocol. **Default:** *32*.
-* `--ws-ping-interval <float>` - Set the WebSockets ping interval, in seconds. Only available with the `websockets` protocol. **Default:** *20.0*.
-* `--ws-ping-timeout <float>` - Set the WebSockets ping timeout, in seconds. Only available with the `websockets` protocol. **Default:** *20.0*.
+* `--ws-ping-interval <float>` - Set the WebSockets ping interval, in seconds. **Default:** *20.0*.
+* `--ws-ping-timeout <float>` - Set the WebSockets ping timeout, in seconds. **Default:** *20.0*.
 * `--ws-per-message-deflate <bool>` - Enable/disable WebSocket per-message-deflate compression. Only available with the `websockets` protocol. **Default:** *True*.
 * `--lifespan <str>` - Set the Lifespan protocol implementation. **Options:** *'auto', 'on', 'off'.* **Default:** *'auto'*.
 * `--h11-max-incomplete-event-size <int>` - Set the maximum number of bytes to buffer of an incomplete event. Only available for `h11` HTTP protocol implementation. **Default:** *16384* (16 KB).
@@ -137,6 +138,8 @@ The [SSL context](https://docs.python.org/3/library/ssl.html#ssl.SSLContext) can
 
 To understand more about the SSL context options, please refer to the [Python documentation](https://docs.python.org/3/library/ssl.html).
 
+For advanced TLS scenarios that the flags above don't cover (e.g., mutual TLS, certificate pinning, custom `SSLContext.options`), pass an `ssl_context_factory` to `uvicorn.run()` or `Config`. See [Running with HTTPS](deployment/index.md#customizing-the-ssl-context) for details.
+
 ## Resource Limits
 
 * `--limit-concurrency <int>` - Maximum number of concurrent connections or tasks to allow, before issuing HTTP 503 responses. Useful for ensuring known memory usage patterns even under over-resourced loads.

docs/sponsorship.md

@@ -1,184 +0,0 @@
-# ✨ Sponsor Starlette & Uvicorn ✨
-
-Thank you for your interest in sponsoring Starlette and Uvicorn! ❤️
-
-Your support *directly* contributes to the ongoing development, maintenance, and long-term sustainability of both projects.
-
-<div style="display: flex; justify-content: center; gap: 4rem; margin: 2rem 0; text-align: center;">
-    <div style="padding: 1rem;">
-        <h3 style="color: #6e5494; font-size: 2em; margin-bottom: 0.5rem;">67M+</h3>
-        <p>Starlette Downloads/Month</p>
-    </div>
-    <div style="padding: 1rem;">
-        <h3 style="color: #6e5494; font-size: 2em; margin-bottom: 0.5rem;">57M+</h3>
-        <p>Uvicorn Downloads/Month</p>
-    </div>
-    <div style="padding: 1rem;">
-        <h3 style="color: #6e5494; font-size: 2em; margin-bottom: 0.5rem;">19K+</h3>
-        <p>Combined GitHub Stars</p>
-    </div>
-</div>
-
-## Why Sponsor?
-
-While Starlette and Uvicorn are part of the [Encode](https://github.com/encode) organization,
-they have been primarily maintained by [**Marcelo Trylesinski (Kludex)**](https://github.com/Kludex)
-for the past several years. His dedication and consistent work have been instrumental in keeping
-these projects robust, secure, and up-to-date.
-
-This sponsorship page was created to give the community an opportunity to support Marcelo's continued
-efforts in maintaining and improving both projects. Your sponsorship directly enables him to
-dedicate more time and resources to maintaining and improving these essential tools:
-
-- [x] **Active Development:** Developing new features, enhancing existing ones, and
-  keeping both projects aligned with the latest developments in the Python and ASGI ecosystems. 💻
-- [x] **Community Support:** Providing better support, addressing user issues,
-  and cultivating a welcoming environment for contributors. 🤝
-- [x] **Long-Term Stability:** Ensuring the long-term viability of both projects through strategic
-  planning and addressing technical debt. 🌳
-- [x] **Bug Fixes & Maintenance:** Providing prompt attention to bug reports and
-  general maintenance to keep the projects reliable. 🔨
-- [x] **Security:** Ensuring robust security practices, conducting regular security audits, and
-  promptly addressing vulnerabilities to protect millions of production deployments. 🔒
-- [x] **Documentation:** Creating comprehensive guides, tutorials, and examples to help users of all skill levels. 📖
-
-## How Sponsorship Works
-
-We currently manage sponsorships *exclusively* through **GitHub Sponsors**. This platform integrates seamlessly with the GitHub ecosystem, making it easy for organizations to contribute.
-
-<div style="text-align: center; padding: 2rem; margin: 2rem 0; background: linear-gradient(135deg, #6e5494, #24292e); border-radius: 10px; color: white;">
-    <h2 style="color: white; margin-bottom: 1rem;">🌟 Become a Sponsor Today! 🌟</h2>
-    <p style="margin-bottom: 1.5rem; font-size: 1.1em;">Your support helps keep Starlette and Uvicorn growing stronger!</p>
-    <a href="https://github.com/sponsors/Kludex"
-       style="display: inline-block; padding: 1rem 2rem; background-color: #238636; color: white; text-decoration: none; border-radius: 6px; font-size: 1.2em; font-weight: bold; transition: all 0.3s ease-in-out;"
-       onmouseover="this.style.backgroundColor='#2ea043';this.style.transform='translateY(-2px)'"
-       onmouseout="this.style.backgroundColor='#238636';this.style.transform='translateY(0)'">
-        ❤️ Sponsor on GitHub
-    </a>
-</div>
-
-## Sponsorship Tiers 🎁
-
-<div style="display: grid; grid-template-columns: repeat(auto-fit, minmax(250px, 1fr)); gap: 1.5rem; margin: 2rem 0;">
-    <div style="padding: 1.5rem; border: 1px solid #e1e4e8; border-radius: 6px; background: #fff; display: flex; flex-direction: column;">
-        <h3 style="color: #cd7f32;">🥉 Bronze Sponsor</h3>
-        <div style="font-size: 1.5em; margin: 1rem 0;">$100<span style="font-size: 0.6em;">/month</span></div>
-        <ul style="list-style: none; padding: 0; margin-bottom: 1rem; min-height: 90px;">
-            <li>✓ Company name on Sponsors page</li>
-            <li>✓ Small logo with link</li>
-            <li>✓ Our eternal gratitude</li>
-        </ul>
-        <div style="text-align: center; margin-top: auto;">
-            <a href="https://github.com/sponsors/Kludex" style="display: inline-block; padding: 0.5rem 1rem; background-color: #cd7f32; color: white; text-decoration: none; border-radius: 6px; font-weight: bold; transition: opacity 0.2s;" onmouseover="this.style.opacity='0.8'" onmouseout="this.style.opacity='1'">
-                Become a Bronze Sponsor
-            </a>
-        </div>
-    </div>
-    <div style="padding: 1.5rem; border: 1px solid #e1e4e8; border-radius: 6px; background: #fff; display: flex; flex-direction: column;">
-        <h3 style="color: #c0c0c0;">🥈 Silver Sponsor</h3>
-        <div style="font-size: 1.5em; margin: 1rem 0;">$250<span style="font-size: 0.6em;">/month</span></div>
-        <ul style="list-style: none; padding: 0; margin-bottom: 1rem; min-height: 90px;">
-            <li>✓ All Bronze benefits</li>
-            <li>✓ Medium-sized logo</li>
-            <li>✓ Release notes mention</li>
-        </ul>
-        <div style="text-align: center; margin-top: auto;">
-            <a href="https://github.com/sponsors/Kludex" style="display: inline-block; padding: 0.5rem 1rem; background-color: #c0c0c0; color: white; text-decoration: none; border-radius: 6px; font-weight: bold; transition: opacity 0.2s;" onmouseover="this.style.opacity='0.8'" onmouseout="this.style.opacity='1'">
-                Become a Silver Sponsor
-            </a>
-        </div>
-    </div>
-    <div style="padding: 1.5rem; border: 1px solid #e1e4e8; border-radius: 6px; background: #fff; position: relative; overflow: hidden; display: flex; flex-direction: column;">
-        <div style="position: absolute; top: 10px; right: -25px; background: #238636; color: white; padding: 5px 30px; transform: rotate(45deg);">
-            Popular
-        </div>
-        <h3 style="color: #ffd700;">🥇 Gold Sponsor</h3>
-        <div style="font-size: 1.5em; margin: 1rem 0;">$500<span style="font-size: 0.6em;">/month</span></div>
-        <ul style="list-style: none; padding: 0; margin-bottom: 1rem; min-height: 90px;">
-            <li>✓ All Silver benefits</li>
-            <li>✓ Large logo on main pages</li>
-            <li>✓ Priority support</li>
-        </ul>
-        <div style="text-align: center; margin-top: auto;">
-            <a href="https://github.com/sponsors/Kludex" style="display: inline-block; padding: 0.5rem 1rem; background-color: #ffd700; color: black; text-decoration: none; border-radius: 6px; font-weight: bold; transition: opacity 0.2s;" onmouseover="this.style.opacity='0.8'" onmouseout="this.style.opacity='1'">
-                Become a Gold Sponsor
-            </a>
-        </div>
-    </div>
-</div>
-
-<div style="text-align: center; margin: 2rem 0;">
-    <h3>🤝 Custom Sponsor</h3>
-    <p>Looking for something different? <a href="mailto:marcelotryle@gmail.com">Contact us</a> to discuss custom sponsorship options!</p>
-</div>
-
-## Current Sponsors
-
-**Thank you to our generous sponsors!** 🙏
-
-<div style="display: flex; flex-direction: column; gap: 3rem; margin: 2rem 0;">
-    <div>
-        <h3 style="text-align: center; color: #ffd700; margin-bottom: 1.5rem;">🏆 Gold Sponsors</h3>
-        <div style="display: flex; flex-wrap: wrap; justify-content: center; gap: 2rem; align-items: center;">
-            <a href="https://fastapi.tiangolo.com" style="text-decoration: none;">
-                <div style="width: 200px; background: #f6f8fa; border-radius: 8px; padding: 1rem; text-align: center;">
-                    <div style="height: 100px; display: flex; align-items: center; justify-content: center; margin-bottom: 0.75rem;">
-                        <img src="https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png" alt="FastAPI" style="max-width: 100%; max-height: 100%; object-fit: contain;">
-                    </div>
-                    <p style="margin: 0; color: #57606a; font-size: 0.9em;">Modern, fast web framework for building APIs with Python 3.8+</p>
-                </div>
-            </a>
-        </div>
-    </div>
-
-    <div>
-        <h3 style="text-align: center; color: #c0c0c0; margin-bottom: 1.5rem;">🥈 Silver Sponsors</h3>
-        <div style="display: flex; flex-wrap: wrap; justify-content: center; gap: 2rem; align-items: center;">
-            <!-- Add Silver Sponsors here -->
-        </div>
-    </div>
-
-    <div>
-        <h3 style="text-align: center; color: #cd7f32; margin-bottom: 1.5rem;">🥉 Bronze Sponsors</h3>
-        <div style="display: flex; flex-wrap: wrap; justify-content: center; gap: 2rem; align-items: center;">
-            <!-- Add Bronze Sponsors here -->
-        </div>
-    </div>
-</div>
-
-## Alternative Sponsorship Platforms
-
-<div style="background: #f6f8fa; padding: 1.5rem; border-radius: 8px; margin: 2rem 0;">
-    <h3>📢 We Want Your Input!</h3>
-    <p>We are currently evaluating whether to expand our sponsorship options beyond GitHub Sponsors. If your company would be interested in sponsoring Starlette and Uvicorn but prefers to use a different platform (e.g., Open Collective, direct invoicing), please let us know!</p>
-    <p>Your feedback is invaluable in helping us make sponsorship as accessible as possible. Share your thoughts by:</p>
-    <ul>
-        <li>Opening a discussion on our <a href="https://github.com/Kludex/starlette/discussions">GitHub repository</a></li>
-        <li>Contacting us directly at <a href="mailto:marcelotryle@gmail.com">marcelotryle@gmail.com</a></li>
-    </ul>
-</div>
-
-<a id="acknowledgments"></a>
-
-## Community & Future Plans 🌟
-
-We want to express our deepest gratitude to all the contributors who have helped shape Starlette and
-Uvicorn over the years. These projects wouldn't be what they are today without the incredible work of
-every single contributor.
-
-Special thanks to some of our most impactful contributors:
-
-- **Tom Christie** ([@tomchristie](https://github.com/tomchristie)) - The original creator of Starlette and Uvicorn.
-- **Adrian Garcia Badaracco** ([@adriangb](https://github.com/adriangb)) - Major contributor to Starlette.
-- **Thomas Grainger** ([@graingert](https://github.com/graingert)) - Major contributor to AnyIO, and significant contributions to Starlette and Uvicorn.
-- **Alex Grönholm** ([@agronholm](https://github.com/agronholm)) - Creator of AnyIO.
-- **Florimond Manca** ([@florimondmanca](https://github.com/florimondmanca)) - Important contributions to Starlette and Uvicorn.
-
-If you want your name removed from the list above, or if I forgot a significant contributor, please let me know.
-You can view all contributors on GitHub:
-[Starlette Contributors](https://github.com/Kludex/starlette/graphs/contributors) / [Uvicorn Contributors](https://github.com/Kludex/uvicorn/graphs/contributors).
-
-While the current sponsorship program directly supports Marcelo's maintenance work, we are exploring ways
-to distribute funding to other key contributors in the future. This initiative is still in early planning
-stages, as we want to ensure a fair and sustainable model that recognizes the valuable contributions of
-our community members.

mkdocs.yml

@@ -54,6 +54,7 @@ nav:
   - Concepts:
       - ASGI: concepts/asgi.md
       - Lifespan: concepts/lifespan.md
+      - Logging: concepts/logging.md
       - WebSockets: concepts/websockets.md
       - Event Loop: concepts/event-loop.md
   - Deployment:
@@ -61,7 +62,6 @@ nav:
       - Docker: deployment/docker.md
   - Release Notes: release-notes.md
   - Contributing: contributing.md
-  - Sponsorship: sponsorship.md
 
 extra:
   analytics:
@@ -79,6 +79,9 @@ extra:
     - icon: fontawesome/solid/globe
       link: https://fastapiexpert.com
 
+extra_css:
+  - css/extra.css
+
 markdown_extensions:
   - attr_list
   - admonition

pyproject.toml

@@ -53,7 +53,7 @@ dev = [
     # We add uvicorn[standard] so `uv sync` considers the extras.
     "uvicorn[standard]",
     "ruff==0.15.1",
-    "pytest==9.0.2",
+    "pytest==9.0.3",
     "pytest-mock==3.15.1",
     "pytest-xdist[psutil]==3.8.0",
     "pytest-codspeed>=4.1.1",
@@ -82,7 +82,8 @@ docs = [
 
 [tool.uv]
 default-groups = ["dev", "docs"]
-required-version = ">=0.8.6"
+required-version = ">=0.9.17"
+exclude-newer = "7 days"
 
 [project.scripts]
 uvicorn = "uvicorn.main:main"

tests/benchmarks/http.py

@@ -7,7 +7,6 @@ from typing import TYPE_CHECKING, Any, TypeAlias
 from uvicorn._types import ASGIApplication, Scope
 from uvicorn.config import Config
 from uvicorn.lifespan.off import LifespanOff
-from uvicorn.lifespan.on import LifespanOn
 from uvicorn.protocols.http.h11_impl import H11Protocol
 from uvicorn.server import ServerState
 
@@ -158,16 +157,17 @@ class MockProtocol(asyncio.Protocol):
     scope: Scope
 
 
+def make_config(app: ASGIApplication, **kwargs: Any) -> Config:
+    return Config(app=app, **kwargs)
+
+
 def get_connected_protocol(
-    app: ASGIApplication,
+    config: Config,
     http_protocol_cls: type[HTTPProtocol],
-    lifespan: LifespanOff | LifespanOn | None = None,
-    **kwargs: Any,
 ) -> MockProtocol:
     loop = MockLoop()
     transport = MockTransport()
-    config = Config(app=app, **kwargs)
-    lifespan = lifespan or LifespanOff(config)
+    lifespan = LifespanOff(config)
     server_state = ServerState()
     protocol = http_protocol_cls(config=config, server_state=server_state, app_state=lifespan.state, _loop=loop)  # type: ignore
     protocol.connection_made(transport)  # type: ignore[arg-type]

tests/benchmarks/test_http.py

@@ -15,6 +15,7 @@ from tests.benchmarks.http import (
     SIMPLE_POST_REQUEST,
     START_POST_REQUEST,
     get_connected_protocol,
+    make_config,
 )
 from tests.response import Response
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
@@ -28,6 +29,9 @@ _plain_text_app = Response("Hello, world", media_type="text/plain")
 _no_content_app = Response(b"", status_code=204)
 _chunked_app = Response(b"Hello, world!", status_code=200, headers={"transfer-encoding": "chunked"})
 
+_plain_text_config = make_config(_plain_text_app)
+_chunked_config = make_config(_chunked_app)
+
 
 async def _body_echo_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
     body = b""
@@ -41,26 +45,29 @@ async def _body_echo_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGIS
     await send({"type": "http.response.body", "body": body})
 
 
+_body_echo_config = make_config(_body_echo_app)
+
+
 async def test_bench_simple_get(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(SIMPLE_GET_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_simple_post(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(SIMPLE_POST_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_large_post(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(LARGE_POST_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_pipelined_requests(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(SIMPLE_GET_REQUEST * 3)
     await protocol.loop.run_one()
     await protocol.loop.run_one()
@@ -68,7 +75,7 @@ async def test_bench_pipelined_requests(http_protocol_cls: type[HTTPProtocol]) -
 
 
 async def test_bench_keepalive_reuse(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(SIMPLE_GET_REQUEST)
     await protocol.loop.run_one()
     protocol.data_received(SIMPLE_GET_REQUEST)
@@ -76,25 +83,25 @@ async def test_bench_keepalive_reuse(http_protocol_cls: type[HTTPProtocol]) -> N
 
 
 async def test_bench_chunked_response(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_chunked_app, http_protocol_cls)
+    protocol = get_connected_protocol(_chunked_config, http_protocol_cls)
     protocol.data_received(SIMPLE_GET_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_http10(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(HTTP10_GET_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_connection_close(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(CONNECTION_CLOSE_REQUEST)
     await protocol.loop.run_one()
 
 
 async def test_bench_fragmented_body(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_plain_text_app, http_protocol_cls)
+    protocol = get_connected_protocol(_plain_text_config, http_protocol_cls)
     protocol.data_received(FRAGMENTED_POST_HEADERS)
     for chunk in FRAGMENTED_BODY_CHUNKS:
         protocol.data_received(chunk)
@@ -102,7 +109,7 @@ async def test_bench_fragmented_body(http_protocol_cls: type[HTTPProtocol]) -> N
 
 
 async def test_bench_post_body_receive(http_protocol_cls: type[HTTPProtocol]) -> None:
-    protocol = get_connected_protocol(_body_echo_app, http_protocol_cls)
+    protocol = get_connected_protocol(_body_echo_config, http_protocol_cls)
     protocol.data_received(START_POST_REQUEST)
     protocol.data_received(FINISH_POST_REQUEST)
     await protocol.loop.run_one()

tests/benchmarks/test_ws.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import importlib.util
+from typing import TYPE_CHECKING
+
+import pytest
+
+from tests.benchmarks.http import make_config
+from tests.benchmarks.ws import WS_UPGRADE, get_connected_ws_protocol
+from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
+
+if TYPE_CHECKING:
+    from tests.benchmarks.ws import WSProtocolClass
+
+pytestmark = [pytest.mark.anyio, pytest.mark.benchmark]
+
+
+@pytest.fixture(
+    params=[
+        pytest.param(
+            "wsproto",
+            marks=pytest.mark.skipif(not importlib.util.find_spec("wsproto"), reason="wsproto not installed."),
+            id="wsproto",
+        ),
+        pytest.param("websockets-sansio", id="websockets-sansio"),
+    ]
+)
+def ws_cls(request: pytest.FixtureRequest) -> WSProtocolClass:
+    if request.param == "wsproto":
+        from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
+
+        return WSProtocol
+    from uvicorn.protocols.websockets.websockets_sansio_impl import WebSocketsSansIOProtocol
+
+    return WebSocketsSansIOProtocol
+
+
+async def _ws_accept_close_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+    await receive()
+    await send({"type": "websocket.accept"})
+    await send({"type": "websocket.close", "code": 1000})
+
+
+async def _ws_send_text_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+    await receive()
+    await send({"type": "websocket.accept"})
+    await send({"type": "websocket.send", "text": "Hello, world!"})
+    await send({"type": "websocket.close", "code": 1000})
+
+
+_ws_accept_close_config = make_config(_ws_accept_close_app, access_log=False)
+_ws_send_text_config = make_config(_ws_send_text_app, access_log=False)
+
+
+async def test_bench_ws_handshake(ws_cls: WSProtocolClass) -> None:
+    protocol = get_connected_ws_protocol(_ws_accept_close_config, ws_cls)
+    protocol.data_received(WS_UPGRADE)
+    await protocol.loop.run_one()
+
+
+async def test_bench_ws_send_text(ws_cls: WSProtocolClass) -> None:
+    protocol = get_connected_ws_protocol(_ws_send_text_config, ws_cls)
+    protocol.data_received(WS_UPGRADE)
+    await protocol.loop.run_one()

tests/benchmarks/ws.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, TypeAlias
+
+from tests.benchmarks.http import MockLoop, MockTransport
+from uvicorn.config import Config
+from uvicorn.lifespan.off import LifespanOff
+from uvicorn.server import ServerState
+
+if TYPE_CHECKING:
+    from uvicorn.protocols.websockets.websockets_sansio_impl import WebSocketsSansIOProtocol
+    from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
+
+    WSProtocolClass: TypeAlias = type[WSProtocol] | type[WebSocketsSansIOProtocol]
+
+WS_UPGRADE = (
+    b"GET / HTTP/1.1\r\n"
+    b"Host: example.org\r\n"
+    b"Upgrade: websocket\r\n"
+    b"Connection: Upgrade\r\n"
+    b"Sec-WebSocket-Key: YmVuY2htYXJra2V5MTIzNA==\r\n"
+    b"Sec-WebSocket-Version: 13\r\n"
+    b"\r\n"
+)
+
+# Masked text frame: "Hello, world!" (13 bytes) with zero mask key
+WS_TEXT_FRAME = b"\x81\x8d\x00\x00\x00\x00Hello, world!"
+
+# Masked close frame: code 1000 with zero mask key
+WS_CLOSE_FRAME = b"\x88\x82\x00\x00\x00\x00\x03\xe8"
+
+
+def get_connected_ws_protocol(config: Config, ws_protocol_cls: WSProtocolClass) -> Any:
+    loop = MockLoop()
+    transport = MockTransport()
+    lifespan = LifespanOff(config)
+    server_state = ServerState()
+    protocol = ws_protocol_cls(config=config, server_state=server_state, app_state=lifespan.state, _loop=loop)  # type: ignore[arg-type]
+    protocol.connection_made(transport)  # type: ignore[arg-type]
+    return protocol

tests/middleware/test_proxy_headers.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
+import contextlib
+import ipaddress
 from typing import TYPE_CHECKING
 
 import httpx
-import httpx._transports.asgi
 import pytest
 import websockets.client
 
@@ -30,6 +31,9 @@ async def default_app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISend
         client_addr = "NONE"  # pragma: no cover
     else:
         host, port = client
+        with contextlib.suppress(ValueError):
+            if ipaddress.ip_address(host).version == 6:
+                host = f"[{host}]"
         client_addr = f"{host}:{port}"
 
     response = Response(f"{scheme}://{client_addr}", media_type="text/plain")
@@ -426,6 +430,31 @@ async def test_proxy_headers_multiple_proxies(trusted_hosts: str | list[str], ex
     assert response.text == expected
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("trusted_hosts", "expected"),
+    [
+        # always trust
+        ("*", "https://1.2.3.4:1234"),
+        # all proxies are trusted
+        (["127.0.0.1", "2001:db8::1", "192.168.0.2"], "https://1.2.3.4:1234"),
+        # should set first untrusted as remote address
+        (["192.168.0.2", "127.0.0.1"], "https://[2001:db8::1]:8080"),
+        # Mixed literals and networks
+        (["127.0.0.1", "2001:db8::/32", "192.168.0.2"], "https://1.2.3.4:1234"),
+    ],
+)
+async def test_proxy_headers_multiple_proxies_with_ports(trusted_hosts: str | list[str], expected: str) -> None:
+    async with make_httpx_client(trusted_hosts) as client:
+        headers = {
+            X_FORWARDED_FOR: "1.2.3.4:1234, [2001:db8::1]:8080, 192.168.0.2:9000",
+            X_FORWARDED_PROTO: "https",
+        }
+        response = await client.get("/", headers=headers)
+    assert response.status_code == 200
+    assert response.text == expected
+
+
 @pytest.mark.anyio
 async def test_proxy_headers_invalid_x_forwarded_for() -> None:
     async with make_httpx_client("*") as client:
@@ -441,6 +470,38 @@ async def test_proxy_headers_invalid_x_forwarded_for() -> None:
     assert response.text == "https://1.2.3.4:0"
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("forwarded_for", "expected"),
+    [
+        # IPv4 without port
+        ("1.2.3.4", "https://1.2.3.4:0"),
+        # IPv4 with port
+        ("1.2.3.4:1234", "https://1.2.3.4:1234"),
+        # Bracketed IPv6 with port
+        ("[2001:db8::1]:443", "https://[2001:db8::1]:443"),
+        # Bracketed IPv6 without port
+        ("[2001:db8::1]", "https://[2001:db8::1]:0"),
+        # Bare IPv6 without port
+        ("2001:db8::1", "https://[2001:db8::1]:0"),
+        # Invalid IPv4 port falls back to the original host value
+        ("1.2.3.4:notaport", "https://1.2.3.4:notaport:0"),
+        # Invalid bracketed IPv6 port keeps the host and drops the port
+        ("[2001:db8::1]:notaport", "https://[2001:db8::1]:0"),
+        # Trailing data after a bracketed IPv6 host is left untouched
+        ("[2001:db8::1]extra", "https://[2001:db8::1]extra:0"),
+        # Malformed bracket is left untouched
+        ("[2001:db8::1", "https://[2001:db8::1:0"),
+    ],
+)
+async def test_proxy_headers_x_forwarded_for_port_shapes(forwarded_for: str, expected: str) -> None:
+    async with make_httpx_client("*") as client:
+        headers = {X_FORWARDED_FOR: forwarded_for, X_FORWARDED_PROTO: "https"}
+        response = await client.get("/", headers=headers)
+    assert response.status_code == 200
+    assert response.text == expected
+
+
 @pytest.mark.anyio
 @pytest.mark.parametrize(
     "forwarded_proto,expected",

tests/protocols/test_websocket.py

@@ -10,6 +10,7 @@ import websockets
 import websockets.client
 import websockets.exceptions
 from websockets.extensions.permessage_deflate import ClientPerMessageDeflateFactory
+from websockets.frames import Opcode
 from websockets.typing import Subprotocol
 
 from tests.response import Response
@@ -27,6 +28,7 @@ from uvicorn._types import (
 )
 from uvicorn.config import Config
 from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
+from uvicorn.protocols.websockets.websockets_sansio_impl import WebSocketsSansIOProtocol
 
 try:
     from uvicorn.protocols.websockets.wsproto_impl import WSProtocol as _WSProtocol
@@ -42,6 +44,7 @@ if TYPE_CHECKING:
 
     HTTPProtocol: TypeAlias = "type[H11Protocol | HttpToolsProtocol]"
     WSProtocol: TypeAlias = "type[_WSProtocol | WebSocketProtocol]"
+    KeepaliveWSProtocol: TypeAlias = "type[_WSProtocol | WebSocketsSansIOProtocol]"
 
 pytestmark = pytest.mark.anyio
 
@@ -750,6 +753,61 @@ async def test_send_binary_data_to_server_bigger_than_default_on_websockets(
                 assert ws.close_code == expected_result
 
 
+async def test_fragmented_message_exceeding_max_size(
+    ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
+):
+    """Stream non-FIN fragments past `ws_max_size` - the server must close with 1009."""
+
+    class App(WebSocketResponse):
+        async def websocket_connect(self, message: WebSocketConnectEvent):
+            await self.send({"type": "websocket.accept"})
+
+    config = Config(
+        app=App, ws=ws_protocol_cls, http=http_protocol_cls, lifespan="off", ws_max_size=2048, port=unused_tcp_port
+    )
+    async with run_server(config):
+        async with websockets.connect(f"ws://127.0.0.1:{unused_tcp_port}") as ws:
+            payload = b"A" * 1024
+            with pytest.raises(websockets.exceptions.ConnectionClosed) as exc_info:
+                await ws.write_frame(False, Opcode.BINARY, payload)
+                for _ in range(63):  # 64 KiB total, well past 2 KiB budget
+                    await ws.write_frame(False, Opcode.CONT, payload)
+                await ws.recv()
+    assert exc_info.value.rcvd is not None
+    assert exc_info.value.rcvd.code == 1009
+
+
+async def test_fragmented_message_reassembly(
+    ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
+):
+    """Server reassembles a fragmented message and delivers it to the app intact."""
+
+    received: list[bytes] = []
+
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
+        assert scope["type"] == "websocket"
+        connect = await receive()
+        assert connect["type"] == "websocket.connect"
+        await send({"type": "websocket.accept"})
+        message = await receive()
+        assert message["type"] == "websocket.receive"
+        payload = message.get("bytes")
+        assert payload is not None
+        received.append(payload)
+        await send({"type": "websocket.close"})
+
+    config = Config(app=app, ws=ws_protocol_cls, http=http_protocol_cls, lifespan="off", port=unused_tcp_port)
+    async with run_server(config):
+        async with websockets.connect(f"ws://127.0.0.1:{unused_tcp_port}") as ws:
+            payload = b"A" * 512
+            await ws.write_frame(False, Opcode.BINARY, payload)
+            for _ in range(4):
+                await ws.write_frame(False, Opcode.CONT, payload)
+            await ws.write_frame(True, Opcode.CONT, payload)
+
+    assert received == [b"A" * 512 * 6]
+
+
 async def test_server_reject_connection(
     ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
 ):
@@ -883,6 +941,8 @@ async def test_server_reject_connection_with_invalid_status(
         response = await wsresponse(url)
         assert response.status_code == 500
         assert response.content == b"Internal Server Error"
+        assert response.headers["content-length"] == "21"
+        assert response.headers["connection"] == "close"
 
     config = Config(app=app, ws=ws_protocol_cls, http=http_protocol_cls, lifespan="off", port=unused_tcp_port)
     async with run_server(config):
@@ -1200,3 +1260,118 @@ async def test_lifespan_state(ws_protocol_cls: WSProtocol, http_protocol_cls: HT
         assert is_open
 
     assert expected_states == actual_states
+
+
+@pytest.fixture(
+    params=[
+        pytest.param(
+            "uvicorn.protocols.websockets.wsproto_impl:WSProtocol",
+            marks=skip_if_no_wsproto,
+            id="wsproto",
+        ),
+        pytest.param(
+            "uvicorn.protocols.websockets.websockets_sansio_impl:WebSocketsSansIOProtocol", id="websockets-sansio"
+        ),
+    ]
+)
+def keepalive_ws_protocol_cls(request: pytest.FixtureRequest):
+    from uvicorn.importer import import_from_string
+
+    return import_from_string(request.param)
+
+
+async def test_server_keepalive_ping_pong(
+    keepalive_ws_protocol_cls: KeepaliveWSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
+):
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
+        while True:
+            message = await receive()
+            if message["type"] == "websocket.connect":
+                await send({"type": "websocket.accept"})
+            elif message["type"] == "websocket.disconnect":
+                break
+
+    config = Config(
+        app=app,
+        ws=keepalive_ws_protocol_cls,
+        http=http_protocol_cls,
+        lifespan="off",
+        ws_ping_interval=0.1,
+        ws_ping_timeout=5.0,
+        port=unused_tcp_port,
+    )
+    async with run_server(config) as server:
+        # The websockets client auto-responds to ping frames, keeping the connection alive.
+        async with websockets.connect(f"ws://127.0.0.1:{unused_tcp_port}", ping_interval=None):
+            protocol = list(server.server_state.connections)[0]
+            assert isinstance(protocol, (_WSProtocol, WebSocketsSansIOProtocol))
+
+            # Wait until the server sends at least one keepalive ping, then
+            # sleep past the timeout window and ensure the connection stays open.
+            # This verifies that the client answered the ping without depending
+            # on clock granularity for the measured RTT.
+            async def ping_sent() -> None:
+                while protocol.ping_sent_at == 0.0:
+                    await asyncio.sleep(0.05)
+
+            await asyncio.wait_for(ping_sent(), timeout=5.0)
+            await asyncio.sleep(0.2)
+            assert not protocol.transport.is_closing()
+
+
+async def test_server_keepalive_ping_timeout(
+    keepalive_ws_protocol_cls: KeepaliveWSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
+):
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
+        while True:
+            message = await receive()
+            if message["type"] == "websocket.connect":
+                await send({"type": "websocket.accept"})
+            elif message["type"] == "websocket.disconnect":
+                break
+
+    config = Config(
+        app=app,
+        ws=keepalive_ws_protocol_cls,
+        http=http_protocol_cls,
+        lifespan="off",
+        ws_ping_interval=0.1,
+        ws_ping_timeout=0.1,
+        log_level="trace",
+        port=unused_tcp_port,
+    )
+    async with run_server(config):
+        async with websockets.connect(f"ws://127.0.0.1:{unused_tcp_port}", ping_interval=None) as websocket:
+            # Swallow outgoing pong frames so the server's ping never gets ack'd.
+            websocket.transport.write = lambda data: None  # type: ignore[method-assign]
+            with pytest.raises(websockets.exceptions.ConnectionClosedError) as exc_info:
+                await asyncio.wait_for(websocket.recv(), timeout=1)
+            assert exc_info.value.rcvd is not None
+            assert exc_info.value.rcvd.code == 1011
+            assert exc_info.value.rcvd.reason == "keepalive ping timeout"
+
+
+async def test_server_keepalive_disabled(
+    keepalive_ws_protocol_cls: KeepaliveWSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int
+):
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
+        while True:
+            message = await receive()
+            if message["type"] == "websocket.connect":
+                await send({"type": "websocket.accept"})
+            elif message["type"] == "websocket.disconnect":
+                break
+
+    config = Config(
+        app=app,
+        ws=keepalive_ws_protocol_cls,
+        http=http_protocol_cls,
+        lifespan="off",
+        ws_ping_interval=None,
+        port=unused_tcp_port,
+    )
+    async with run_server(config) as server:
+        async with websockets.connect(f"ws://127.0.0.1:{unused_tcp_port}", ping_interval=None):
+            protocol = list(server.server_state.connections)[0]
+            assert isinstance(protocol, (_WSProtocol, WebSocketsSansIOProtocol))
+            assert protocol.ping_timer is None

tests/test_config.py

@@ -366,6 +366,35 @@ def test_log_config_yaml(
     mocked_logging_config_module.dictConfig.assert_called_once_with(logging_config)
 
 
+def test_log_config_yaml_missing_pyyaml(mocked_logging_config_module: MagicMock, mocker: MockerFixture) -> None:
+    """
+    Test that a helpful error is raised when PyYAML is not installed.
+    """
+    mocker.patch.dict(sys.modules, {"yaml": None})
+    with pytest.raises(ImportError, match=r"Install the PyYAML package or uvicorn\[standard\]"):
+        Config(app=asgi_app, log_config="log_config.yaml")
+
+
+def test_log_config_pathlike(
+    mocked_logging_config_module: MagicMock,
+    logging_config: dict[str, Any],
+    json_logging_config: str,
+    mocker: MockerFixture,
+    tmp_path: Path,
+) -> None:
+    """
+    Test that one can pass a `os.PathLike` (e.g. `pathlib.Path`) as the log config path.
+    """
+    path = tmp_path / "log_config.json"
+    mocked_open = mocker.patch("uvicorn.config.open", mocker.mock_open(read_data=json_logging_config))
+
+    config = Config(app=asgi_app, log_config=path)
+    config.load()
+
+    mocked_open.assert_called_once_with(os.fspath(path))
+    mocked_logging_config_module.dictConfig.assert_called_once_with(logging_config)
+
+
 @pytest.mark.parametrize("config_file", ["log_config.ini", configparser.ConfigParser(), io.StringIO()])
 def test_log_config_file(
     mocked_logging_config_module: MagicMock,
@@ -462,6 +491,13 @@ def test_config_log_effective_level(log_level: int, uvicorn_logger_level: int) -
     assert logging.getLogger("uvicorn.asgi").getEffectiveLevel() == effective_level
 
 
+@pytest.mark.parametrize("log_level", ["INFO", "Info", "info"])
+def test_config_log_level_case_insensitive(log_level: str) -> None:
+    config = Config(app=asgi_app, log_level=log_level)
+    config.load()
+    assert logging.getLogger("uvicorn.error").level == logging.INFO
+
+
 def test_ws_max_size() -> None:
     config = Config(app=asgi_app, ws_max_size=1000)
     config.load()
@@ -517,6 +553,37 @@ def test_bind_fd_works_with_reload_or_workers(reload: bool, workers: int):  # pr
     fdsock.close()
 
 
+@pytest.fixture
+def stdin_socket() -> Iterator[socket.socket]:  # pragma: py-win32
+    with closing(socket.socket(socket.AF_INET)) as sock:
+        sock.bind(("127.0.0.1", 0))
+        saved_stdin = os.dup(0)
+        os.dup2(sock.fileno(), 0)
+        try:
+            yield sock
+        finally:
+            os.dup2(saved_stdin, 0)
+            os.close(saved_stdin)
+
+
+@pytest.mark.parametrize(
+    "reload, workers",
+    [
+        (True, 1),
+        (False, 2),
+    ],
+    ids=["--reload=True --workers=1", "--reload=False --workers=2"],
+)
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+def test_bind_stdin_works_with_reload_or_workers(
+    reload: bool, workers: int, stdin_socket: socket.socket
+):  # pragma: py-win32
+    config = Config(app=asgi_app, fd=0, reload=reload, workers=workers)
+    config.load()
+    with closing(config.bind_socket()) as sock:
+        assert sock.getsockname() == stdin_socket.getsockname()
+
+
 @pytest.mark.parametrize(
     "reload, workers, expected",
     [

tests/test_main.py

@@ -1,7 +1,9 @@
 import importlib
 import inspect
 import socket
+import sys
 from logging import WARNING
+from pathlib import Path
 
 import httpx
 import pytest
@@ -12,6 +14,7 @@ from uvicorn import Server
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.config import Config
 from uvicorn.main import run
+from uvicorn.supervisors import Multiprocess
 
 pytestmark = pytest.mark.anyio
 
@@ -85,6 +88,61 @@ def test_run_invalid_app_config_combination(caplog: pytest.LogCaptureFixture) ->
     )
 
 
+def test_run_fails_fast_in_parent_on_bad_app_path(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Bad app path with `--workers > 1` exits in the parent.
+
+    Regression for https://github.com/encode/uvicorn/discussions/2440: without
+    parent-side validation the supervisor restarts dying workers forever.
+    """
+
+    def fail(*args: object, **kwargs: object) -> None:  # pragma: no cover
+        pytest.fail("parent reached supervisor; should have exited on bad app path")
+
+    monkeypatch.setattr(Config, "bind_socket", fail)
+    monkeypatch.setattr(Multiprocess, "run", fail)
+
+    with pytest.raises(SystemExit) as exit_exception:
+        run("tests.test_main:nonexistent_attr", workers=2)
+    assert exit_exception.value.code == 1
+    assert any("Error loading ASGI app" in record.message for record in caplog.records)
+
+
+def test_run_imports_app_before_starting_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """`uvicorn.run()` imports the app before `Server.run` opens the event loop.
+
+    Regression for https://github.com/encode/uvicorn/issues/941: an app whose
+    module body calls `asyncio.run(...)` crashes with "loop already running"
+    if Uvicorn imports it inside the server's event loop. The parent must
+    import the app synchronously, before `Server.run` enters `asyncio.run`.
+    """
+    module = tmp_path / "eager_async_app.py"
+    module.write_text(
+        "import asyncio\n"
+        "async def _build():\n"
+        "    async def app(scope, receive, send):\n"
+        "        pass\n"
+        "    return app\n"
+        "app = asyncio.run(_build())\n"
+    )
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    imported_before_server_run: list[bool] = []
+
+    def tracking_run(self: Server, sockets: object = None) -> None:
+        imported_before_server_run.append("eager_async_app" in sys.modules)
+        self.started = True
+
+    monkeypatch.setattr(Server, "run", tracking_run)
+
+    # The import side effect (`eager_async_app` lands in `sys.modules`) must
+    # happen before `Server.run`, which is where the event loop opens.
+    run("eager_async_app:app")
+
+    assert imported_before_server_run == [True]
+
+
 def test_run_startup_failure(caplog: pytest.LogCaptureFixture) -> None:
     async def app(scope, receive, send):
         assert scope["type"] == "lifespan"

tests/test_server.py

@@ -15,12 +15,12 @@ import pytest
 
 from tests.protocols.test_http import SIMPLE_GET_REQUEST
 from tests.utils import run_server
-from uvicorn import Server
 from uvicorn._types import ASGIApplication, ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.config import Config
 from uvicorn.protocols.http.flow_control import HIGH_WATER_LIMIT
 from uvicorn.protocols.http.h11_impl import H11Protocol
 from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
+from uvicorn.server import Server
 
 pytestmark = pytest.mark.anyio
 
@@ -142,23 +142,25 @@ async def test_limit_max_requests_jitter(
     config = Config(
         app=app, limit_max_requests=1, limit_max_requests_jitter=2, port=unused_tcp_port, http=http_protocol_cls
     )
-    server = Server(config=config)
-    limit = server.limit_max_requests
-    assert limit is not None
-    assert 1 <= limit <= 3
-    task = asyncio.create_task(server.serve())
-    while not server.started:
-        await asyncio.sleep(0.01)
-    async with httpx.AsyncClient() as client:
-        for _ in range(limit + 1):
-            await client.get(f"http://127.0.0.1:{unused_tcp_port}")
-    await task
+    async with run_server(config) as server:
+        limit = server.limit_max_requests
+        assert limit is not None
+        assert 1 <= limit <= 3
+        async with httpx.AsyncClient() as client:
+            tasks = [client.get(f"http://127.0.0.1:{unused_tcp_port}") for _ in range(limit + 1)]
+            await asyncio.gather(*tasks)
     assert f"Maximum request limit of {limit} exceeded. Terminating process." in caplog.text
 
 
 @contextlib.asynccontextmanager
-async def server(*, app: ASGIApplication, port: int, http_protocol_cls: type[H11Protocol | HttpToolsProtocol]):
-    config = Config(app=app, port=port, loop="asyncio", http=http_protocol_cls)
+async def _raw_server(
+    *,
+    app: ASGIApplication,
+    port: int,
+    http_protocol_cls: type[H11Protocol | HttpToolsProtocol],
+    reset_contextvars: bool = False,
+):
+    config = Config(app=app, port=port, loop="asyncio", http=http_protocol_cls, reset_contextvars=reset_contextvars)
     server = Server(config=config)
     task = asyncio.create_task(server.serve())
 
@@ -186,10 +188,36 @@ async def server(*, app: ASGIApplication, port: int, http_protocol_cls: type[H11
         await task
 
 
-async def test_no_contextvars_pollution_asyncio(
+async def test_contextvars_preserved_by_default(
     http_protocol_cls: type[H11Protocol | HttpToolsProtocol], unused_tcp_port: int
 ):
-    """Non-regression test for https://github.com/encode/uvicorn/issues/2167."""
+    """By default, context set outside the ASGI task is visible inside it."""
+    ctx: contextvars.ContextVar[str] = contextvars.ContextVar("ctx")
+    ctx.set("outer-value")
+
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
+        assert scope["type"] == "http"
+        while True:
+            message = await receive()
+            assert message["type"] == "http.request"
+            if not message["more_body"]:
+                break
+        body = json.dumps({"ctx": ctx.get("MISSING")}).encode("utf-8")
+        headers = [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode("utf-8"))]
+        await send({"type": "http.response.start", "status": 200, "headers": headers})
+        await send({"type": "http.response.body", "body": body})
+
+    async with _raw_server(app=app, http_protocol_cls=http_protocol_cls, port=unused_tcp_port) as extract_json_body:
+        assert await extract_json_body(SIMPLE_GET_REQUEST) == {"ctx": "outer-value"}
+
+
+async def test_reset_contextvars_asyncio(
+    http_protocol_cls: type[H11Protocol | HttpToolsProtocol], unused_tcp_port: int
+):
+    """With reset_contextvars=True, each ASGI run starts with a fresh context.
+
+    Non-regression test for https://github.com/encode/uvicorn/issues/2167.
+    """
     default_contextvars = {c.name for c in contextvars.copy_context().keys()}
     ctx: contextvars.ContextVar[str] = contextvars.ContextVar("ctx")
 
@@ -209,14 +237,13 @@ async def test_no_contextvars_pollution_asyncio(
             if not message["more_body"]:
                 break
 
-        # return the initial context for empty assertion
         body = json.dumps(initial_context).encode("utf-8")
         headers = [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode("utf-8"))]
         await send({"type": "http.response.start", "status": 200, "headers": headers})
         await send({"type": "http.response.body", "body": body})
 
-    # body has to be larger than HIGH_WATER_LIMIT to trigger a reading pause on the main thread
-    # and a resumption inside the ASGI task
+    # body larger than HIGH_WATER_LIMIT forces a reading pause on the main thread
+    # and a resumption inside the ASGI task, which is where the original pollution showed up.
     large_body = b"a" * (HIGH_WATER_LIMIT + 1)
     large_request = b"\r\n".join(
         [
@@ -229,6 +256,8 @@ async def test_no_contextvars_pollution_asyncio(
         ]
     )
 
-    async with server(app=app, http_protocol_cls=http_protocol_cls, port=unused_tcp_port) as extract_json_body:
+    async with _raw_server(
+        app=app, http_protocol_cls=http_protocol_cls, port=unused_tcp_port, reset_contextvars=True
+    ) as extract_json_body:
         assert await extract_json_body(large_request) == {}
         assert await extract_json_body(SIMPLE_GET_REQUEST) == {}

tests/test_ssl.py

@@ -1,9 +1,17 @@
+from __future__ import annotations
+
+import ssl
+from collections.abc import Callable
+from typing import TypeAlias
+
 import httpx
 import pytest
 
 from tests.utils import run_server
 from uvicorn.config import Config
 
+DefaultFactory: TypeAlias = Callable[[], ssl.SSLContext]
+
 
 async def app(scope, receive, send):
     assert scope["type"] == "http"
@@ -92,3 +100,108 @@ async def test_run_password(
         async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
             response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
     assert response.status_code == 204
+
+
+@pytest.mark.anyio
+async def test_run_ssl_context_factory_default(
+    tls_ca_ssl_context: ssl.SSLContext,
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+    unused_tcp_port: int,
+) -> None:
+    """A factory that just delegates to the default factory should produce a working server."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return default_ssl_context_factory()
+
+    config = Config(
+        app=app,
+        loop="asyncio",
+        limit_max_requests=1,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_context_factory=ssl_context_factory,
+        port=unused_tcp_port,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
+            response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+@pytest.mark.anyio
+async def test_run_ssl_context_factory_custom(
+    tls_ca_ssl_context: ssl.SSLContext,
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+    unused_tcp_port: int,
+) -> None:
+    """A factory that builds its own SSLContext from scratch should work without ssl_keyfile/ssl_certfile."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.load_cert_chain(tls_certificate_server_cert_path, tls_certificate_private_key_path)
+        return ctx
+
+    config = Config(
+        app=app,
+        loop="asyncio",
+        limit_max_requests=1,
+        ssl_context_factory=ssl_context_factory,
+        port=unused_tcp_port,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
+            response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+def test_ssl_context_factory_mutates_default(
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+) -> None:
+    """The factory can call the default and mutate the result (e.g., bump TLS minimum version)."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        ctx = default_ssl_context_factory()
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        return ctx
+
+    config = Config(
+        app=app,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_context_factory=ssl_context_factory,
+    )
+    config.load()
+    assert config.is_ssl
+    assert isinstance(config.ssl, ssl.SSLContext)
+    assert config.ssl.minimum_version == ssl.TLSVersion.TLSv1_3
+
+
+def test_default_ssl_context_factory_requires_ssl_certfile() -> None:
+    """Calling `default_ssl_context_factory()` without `ssl_certfile` raises a clear error."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return default_ssl_context_factory()
+
+    config = Config(app=app, ssl_context_factory=ssl_context_factory)
+    with pytest.raises(RuntimeError, match="requires `ssl_certfile`"):
+        config.load()
+
+
+def test_ssl_context_factory_must_return_ssl_context() -> None:
+    def bad_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> object:
+        return "not an SSLContext"
+
+    config = Config(app=app, ssl_context_factory=bad_factory)  # type: ignore[arg-type]
+    with pytest.raises(TypeError, match="must return an `ssl.SSLContext`"):
+        config.load()
+
+
+def test_is_ssl_true_when_only_factory_set() -> None:
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)  # pragma: no cover
+
+    config = Config(app=app, ssl_context_factory=ssl_context_factory)
+    assert config.is_ssl is True

uvicorn/__init__.py

@@ -1,5 +1,5 @@
 from uvicorn.config import Config
 from uvicorn.main import Server, main, run
 
-__version__ = "0.41.0"
+__version__ = "0.47.0"
 __all__ = ["main", "run", "Config", "Server"]

uvicorn/config.py

@@ -193,7 +193,7 @@ class Config:
         ws_per_message_deflate: bool = True,
         lifespan: LifespanType = "auto",
         env_file: str | os.PathLike[str] | None = None,
-        log_config: dict[str, Any] | str | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
+        log_config: dict[str, Any] | str | os.PathLike[str] | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
         log_level: str | int | None = None,
         access_log: bool = True,
         use_colors: bool | None = None,
@@ -225,9 +225,11 @@ class Config:
         ssl_cert_reqs: int = ssl.CERT_NONE,
         ssl_ca_certs: str | os.PathLike[str] | None = None,
         ssl_ciphers: str = "TLSv1",
+        ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
         headers: list[tuple[str, str]] | None = None,
         factory: bool = False,
         h11_max_incomplete_event_size: int | None = None,
+        reset_contextvars: bool = False,
     ):
         self.app = app
         self.host = host
@@ -271,10 +273,12 @@ class Config:
         self.ssl_cert_reqs = ssl_cert_reqs
         self.ssl_ca_certs = ssl_ca_certs
         self.ssl_ciphers = ssl_ciphers
+        self.ssl_context_factory = ssl_context_factory
         self.headers: list[tuple[str, str]] = headers or []
         self.encoded_headers: list[tuple[bytes, bytes]] = []
         self.factory = factory
         self.h11_max_incomplete_event_size = h11_max_incomplete_event_size
+        self.reset_contextvars = reset_contextvars
 
         self.loaded = False
         self.configure_logging()
@@ -355,7 +359,7 @@ class Config:
 
     @property
     def is_ssl(self) -> bool:
-        return bool(self.ssl_keyfile or self.ssl_certfile)
+        return bool(self.ssl_keyfile or self.ssl_certfile or self.ssl_context_factory)
 
     @property
     def use_subprocess(self) -> bool:
@@ -365,6 +369,9 @@ class Config:
         logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
 
         if self.log_config is not None:
+            if isinstance(self.log_config, os.PathLike):
+                self.log_config = os.fspath(self.log_config)
+
             if isinstance(self.log_config, dict):
                 if self.use_colors in (True, False):
                     self.log_config["formatters"]["default"]["use_colors"] = self.use_colors
@@ -375,9 +382,12 @@ class Config:
                     loaded_config = json.load(file)
                     logging.config.dictConfig(loaded_config)
             elif isinstance(self.log_config, str) and self.log_config.endswith((".yaml", ".yml")):
-                # Install the PyYAML package or the uvicorn[standard] optional
-                # dependencies to enable this functionality.
-                import yaml
+                try:
+                    import yaml
+                except ImportError as e:
+                    raise ImportError(
+                        "Install the PyYAML package or uvicorn[standard] to use `--log-config` with YAML files."
+                    ) from e
 
                 with open(self.log_config) as file:
                     loaded_config = yaml.safe_load(file)
@@ -389,7 +399,7 @@ class Config:
 
         if self.log_level is not None:
             if isinstance(self.log_level, str):
-                log_level = LOG_LEVELS[self.log_level]
+                log_level = LOG_LEVELS[self.log_level.lower()]
             else:
                 log_level = self.log_level
             logging.getLogger("uvicorn.error").setLevel(log_level)
@@ -399,12 +409,43 @@ class Config:
             logging.getLogger("uvicorn.access").handlers = []
             logging.getLogger("uvicorn.access").propagate = False
 
+    def load_app(self) -> Any:
+        """Import the app and return it. Exits on failure."""
+        try:
+            return import_from_string(self.app)
+        except ImportFromStringError as exc:
+            logger.error("Error loading ASGI app. %s" % exc)
+            sys.exit(1)
+
     def load(self) -> None:
         assert not self.loaded
 
-        if self.is_ssl:
+        if self.ssl_context_factory is not None:
+
+            def default_factory() -> ssl.SSLContext:
+                if not self.ssl_certfile:
+                    raise RuntimeError(
+                        "`default_ssl_context_factory()` requires `ssl_certfile` to be set on `Config`. "
+                        "Either pass `ssl_certfile` (and optionally `ssl_keyfile`) or build the `SSLContext` "
+                        "directly inside `ssl_context_factory` without calling the default factory."
+                    )
+                return create_ssl_context(
+                    keyfile=self.ssl_keyfile,
+                    certfile=self.ssl_certfile,
+                    password=self.ssl_keyfile_password,
+                    ssl_version=self.ssl_version,
+                    cert_reqs=self.ssl_cert_reqs,
+                    ca_certs=self.ssl_ca_certs,
+                    ciphers=self.ssl_ciphers,
+                )
+
+            context = self.ssl_context_factory(self, default_factory)
+            if not isinstance(context, ssl.SSLContext):
+                raise TypeError(f"`ssl_context_factory` must return an `ssl.SSLContext`, got {type(context).__name__}")
+            self.ssl: ssl.SSLContext | None = context
+        elif self.is_ssl:
             assert self.ssl_certfile
-            self.ssl: ssl.SSLContext | None = create_ssl_context(
+            self.ssl = create_ssl_context(
                 keyfile=self.ssl_keyfile,
                 certfile=self.ssl_certfile,
                 password=self.ssl_keyfile_password,
@@ -437,11 +478,7 @@ class Config:
 
         self.lifespan_class = import_from_string(LIFESPAN[self.lifespan])
 
-        try:
-            self.loaded_app = import_from_string(self.app)
-        except ImportFromStringError as exc:
-            logger.error("Error loading ASGI app. %s" % exc)
-            sys.exit(1)
+        self.loaded_app = self.load_app()
 
         try:
             self.loaded_app = self.loaded_app()
@@ -500,7 +537,7 @@ class Config:
 
     def bind_socket(self) -> socket.socket:
         logger_args: list[str | int]
-        if self.uds:  # pragma: py-win32
+        if self.uds is not None:  # pragma: py-win32
             path = self.uds
             sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
             try:
@@ -515,7 +552,7 @@ class Config:
             sock_name_format = "%s"
             color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
             logger_args = [self.uds]
-        elif self.fd:  # pragma: py-win32
+        elif self.fd is not None:  # pragma: py-win32
             sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
             message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
             fd_name_format = "%s"

uvicorn/main.py

@@ -372,6 +372,13 @@ def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> No
     default=None,
     help="For h11, the maximum number of bytes to buffer of an incomplete event.",
 )
+@click.option(
+    "--reset-contextvars",
+    is_flag=True,
+    default=False,
+    help="Run each ASGI request in a fresh contextvars.Context. Hides context set in the lifespan.",
+    show_default=True,
+)
 @click.option(
     "--factory",
     is_flag=True,
@@ -428,6 +435,7 @@ def main(
     use_colors: bool,
     app_dir: str,
     h11_max_incomplete_event_size: int | None,
+    reset_contextvars: bool,
     factory: bool,
 ) -> None:
     run(
@@ -480,6 +488,7 @@ def main(
         factory=factory,
         app_dir=app_dir,
         h11_max_incomplete_event_size=h11_max_incomplete_event_size,
+        reset_contextvars=reset_contextvars,
     )
 
 
@@ -507,7 +516,7 @@ def run(
     reload_delay: float = 0.25,
     workers: int | None = None,
     env_file: str | os.PathLike[str] | None = None,
-    log_config: dict[str, Any] | str | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
+    log_config: dict[str, Any] | str | os.PathLike[str] | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
     log_level: str | int | None = None,
     access_log: bool = True,
     proxy_headers: bool = True,
@@ -529,11 +538,13 @@ def run(
     ssl_cert_reqs: int = ssl.CERT_NONE,
     ssl_ca_certs: str | os.PathLike[str] | None = None,
     ssl_ciphers: str = "TLSv1",
+    ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
     headers: list[tuple[str, str]] | None = None,
     use_colors: bool | None = None,
     app_dir: str | None = None,
     factory: bool = False,
     h11_max_incomplete_event_size: int | None = None,
+    reset_contextvars: bool = False,
 ) -> None:
     if app_dir is not None:
         sys.path.insert(0, app_dir)
@@ -583,18 +594,21 @@ def run(
         ssl_cert_reqs=ssl_cert_reqs,
         ssl_ca_certs=ssl_ca_certs,
         ssl_ciphers=ssl_ciphers,
+        ssl_context_factory=ssl_context_factory,
         headers=headers,
         use_colors=use_colors,
         factory=factory,
         h11_max_incomplete_event_size=h11_max_incomplete_event_size,
+        reset_contextvars=reset_contextvars,
     )
-    server = Server(config=config)
-
     if (config.reload or config.workers > 1) and not isinstance(app, str):
         logger = logging.getLogger("uvicorn.error")
         logger.warning("You must pass the application as an import string to enable 'reload' or 'workers'.")
         sys.exit(1)
 
+    config.load_app()
+    server = Server(config=config)
+
     try:
         if config.should_reload:
             sock = config.bind_socket()
@@ -604,8 +618,8 @@ def run(
             Multiprocess(config, target=server.run, sockets=[sock]).run()
         else:
             server.run()
-    except KeyboardInterrupt:
-        pass  # pragma: full coverage
+    except KeyboardInterrupt:  # pragma: full coverage
+        pass
     finally:
         if config.uds and os.path.exists(config.uds):
             os.remove(config.uds)  # pragma: py-win32

uvicorn/middleware/proxy_headers.py

@@ -45,16 +45,12 @@ class ProxyHeadersMiddleware:
 
             if b"x-forwarded-for" in headers:
                 x_forwarded_for = headers[b"x-forwarded-for"].decode("latin1")
-                host = self.trusted_hosts.get_trusted_client_host(x_forwarded_for)
+                host, port = self.trusted_hosts.get_trusted_client_address(x_forwarded_for)
 
                 if host:
                     # If the x-forwarded-for header is empty then host is an empty string.
                     # Only set the client if we actually got something usable.
                     # See: https://github.com/Kludex/uvicorn/issues/1068
-
-                    # We've lost the connecting client's port information by now,
-                    # so only include the host.
-                    port = 0
                     scope["client"] = (host, port)
 
         return await self.app(scope, receive, send)
@@ -64,6 +60,41 @@ def _parse_raw_hosts(value: str) -> list[str]:
     return [item.strip() for item in value.split(",")]
 
 
+def _parse_host_port(value: str) -> tuple[str, int]:
+    """Parse a forwarded host value into host and optional port.
+
+    Accepts bare IPs, IPv4 `host:port`, and bracketed IPv6 `[host]:port`.
+    Any unrecognized or malformed value is treated conservatively and returned
+    without a port so trust checks do not silently normalize arbitrary input.
+    """
+
+    if value.startswith("["):
+        bracket_end = value.find("]")
+        if bracket_end == -1:
+            return value, 0
+
+        host = value[1:bracket_end]
+        remainder = value[bracket_end + 1 :]
+        if not remainder:
+            return host, 0
+        if not remainder.startswith(":"):
+            return value, 0
+
+        try:
+            return host, int(remainder[1:])
+        except ValueError:
+            return host, 0
+
+    if value.count(":") == 1:
+        host, port = value.rsplit(":", 1)
+        try:
+            return host, int(port)
+        except ValueError:
+            return value, 0
+
+    return value, 0
+
+
 class _TrustedHosts:
     """Container for trusted hosts and networks"""
 
@@ -122,21 +153,22 @@ class _TrustedHosts:
         except ValueError:
             return host in self.trusted_literals
 
-    def get_trusted_client_host(self, x_forwarded_for: str) -> str:
-        """Extract the client host from x_forwarded_for header
+    def get_trusted_client_address(self, x_forwarded_for: str) -> tuple[str, int]:
+        """Extract the client address from x_forwarded_for header.
 
         In general this is the first "untrusted" host in the forwarded for list.
         """
         x_forwarded_for_hosts = _parse_raw_hosts(x_forwarded_for)
 
         if self.always_trust:
-            return x_forwarded_for_hosts[0]
+            return _parse_host_port(x_forwarded_for_hosts[0])
 
         # Note: each proxy appends to the header list so check it in reverse order
-        for host in reversed(x_forwarded_for_hosts):
+        for host_port in reversed(x_forwarded_for_hosts):
+            host, port = _parse_host_port(host_port)
             if host not in self:
-                return host
+                return host, port
 
         # All hosts are trusted meaning that the client was also a trusted proxy
         # See https://github.com/Kludex/uvicorn/issues/1068#issuecomment-855371576
-        return x_forwarded_for_hosts[0]
+        return _parse_host_port(x_forwarded_for_hosts[0])

uvicorn/protocols/http/h11_impl.py

@@ -4,8 +4,9 @@ import asyncio
 import contextvars
 import http
 import logging
+import sys
 from collections.abc import Callable
-from typing import Any, Literal, cast
+from typing import Any, Literal
 from urllib.parse import unquote
 
 import h11
@@ -62,7 +63,6 @@ class H11Protocol(asyncio.Protocol):
         )
         self.ws_protocol_class = config.ws_protocol_class
         self.root_path = config.root_path
-        self.root_path_encoded = config.root_path.encode("ascii")
         self.limit_concurrency = config.limit_concurrency
         self.app_state = app_state
 
@@ -200,7 +200,7 @@ class H11Protocol(asyncio.Protocol):
                 raw_path, _, query_string = event.target.partition(b"?")
                 path = unquote(raw_path.decode("ascii"))
                 full_path = self.root_path + path
-                full_raw_path = self.root_path_encoded + raw_path
+                full_raw_path = self.root_path.encode("ascii") + raw_path
                 self.scope = {
                     "type": "http",
                     "asgi": {"version": self.config.asgi_version, "spec_version": "2.3"},
@@ -250,12 +250,16 @@ class H11Protocol(asyncio.Protocol):
                     message_event=asyncio.Event(),
                     on_response=self.on_response_complete,
                 )
-                # For the asyncio loop, we need to explicitly start with an empty context
-                # as it can be polluted from previous ASGI runs.
-                # See https://github.com/python/cpython/issues/140947 for details.
-                task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
-                # TODO: Replace the line above with the line below for Python >= 3.11
-                # task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+                if self.config.reset_contextvars:
+                    # Opt-in workaround for https://github.com/python/cpython/issues/140947:
+                    # asyncio can leak context vars between tasks. Hides context set in the
+                    # lifespan or by external instrumentation.
+                    if sys.version_info >= (3, 11):  # pragma: py-lt-311
+                        task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
+                    else:  # pragma: py-gte-311
+                        task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
+                else:
+                    task = self.loop.create_task(self.cycle.run_asgi(app))
                 task.add_done_callback(self.tasks.discard)
                 self.tasks.add(task)
 
@@ -453,8 +457,6 @@ class RequestResponseCycle:
 
     # ASGI interface
     async def send(self, message: ASGISendEvent) -> None:
-        message_type = message["type"]
-
         if self.flow.write_paused and not self.disconnected:
             await self.flow.drain()  # pragma: full coverage
 
@@ -463,10 +465,8 @@ class RequestResponseCycle:
 
         if not self.response_started:
             # Sending response status line and headers
-            if message_type != "http.response.start":
-                msg = "Expected ASGI message 'http.response.start', but got '%s'."
-                raise RuntimeError(msg % message_type)
-            message = cast("HTTPResponseStartEvent", message)
+            if message["type"] != "http.response.start":
+                raise RuntimeError(f"Expected ASGI message 'http.response.start', but got '{message['type']}'.")
 
             self.response_started = True
             self.waiting_for_100_continue = False
@@ -495,10 +495,8 @@ class RequestResponseCycle:
 
         elif not self.response_complete:
             # Sending response body
-            if message_type != "http.response.body":
-                msg = "Expected ASGI message 'http.response.body', but got '%s'."
-                raise RuntimeError(msg % message_type)
-            message = cast("HTTPResponseBodyEvent", message)
+            if message["type"] != "http.response.body":
+                raise RuntimeError(f"Expected ASGI message 'http.response.body', but got '{message['type']}'.")
 
             body = message.get("body", b"")
             more_body = message.get("more_body", False)
@@ -517,8 +515,7 @@ class RequestResponseCycle:
 
         else:
             # Response already sent
-            msg = "Unexpected ASGI message '%s' sent, after response already completed."
-            raise RuntimeError(msg % message_type)
+            raise RuntimeError(f"Unexpected ASGI message '{message['type']}' sent, after response already completed.")
 
         if self.response_complete:
             if self.conn.our_state is h11.MUST_CLOSE or not self.keep_alive:
@@ -542,10 +539,6 @@ class RequestResponseCycle:
         if self.disconnected or self.response_complete:
             return {"type": "http.disconnect"}
 
-        message: HTTPRequestEvent = {
-            "type": "http.request",
-            "body": bytes(self.body),
-            "more_body": self.more_body,
-        }
+        message: HTTPRequestEvent = {"type": "http.request", "body": bytes(self.body), "more_body": self.more_body}
         self.body = bytearray()
         return message

uvicorn/protocols/http/httptools_impl.py

@@ -5,11 +5,12 @@ import contextvars
 import http
 import logging
 import re
+import sys
 import urllib
 from asyncio.events import TimerHandle
 from collections import deque
 from collections.abc import Callable
-from typing import Any, Literal, cast
+from typing import Any, Literal
 
 import httptools
 
@@ -18,7 +19,6 @@ from uvicorn._types import (
     ASGIReceiveEvent,
     ASGISendEvent,
     HTTPRequestEvent,
-    HTTPResponseStartEvent,
     HTTPScope,
 )
 from uvicorn.config import Config
@@ -70,7 +70,6 @@ class HttpToolsProtocol(asyncio.Protocol):
 
         self.ws_protocol_class = config.ws_protocol_class
         self.root_path = config.root_path
-        self.root_path_encoded = config.root_path.encode("ascii")
         self.limit_concurrency = config.limit_concurrency
         self.app_state = app_state
 
@@ -259,7 +258,7 @@ class HttpToolsProtocol(asyncio.Protocol):
         if "%" in path:
             path = urllib.parse.unquote(path)
         full_path = self.root_path + path
-        full_raw_path = self.root_path_encoded + raw_path
+        full_raw_path = self.root_path.encode("ascii") + raw_path
         self.scope["path"] = full_path
         self.scope["raw_path"] = full_raw_path
         self.scope["query_string"] = parsed_url.query or b""
@@ -290,19 +289,26 @@ class HttpToolsProtocol(asyncio.Protocol):
         )
         if existing_cycle is None or existing_cycle.response_complete:
             # Standard case - start processing the request.
-            # For the asyncio loop, we need to explicitly start with an empty context
-            # as it can be polluted from previous ASGI runs.
-            # See https://github.com/python/cpython/issues/140947 for details.
-            task = contextvars.Context().run(self.loop.create_task, self.cycle.run_asgi(app))
-            # TODO: Replace the line above with the line below for Python >= 3.11
-            # task = self.loop.create_task(self.cycle.run_asgi(app), context=contextvars.Context())
-            task.add_done_callback(self.tasks.discard)
-            self.tasks.add(task)
+            self._start_asgi_task(self.cycle, app)
         else:
             # Pipelined HTTP requests need to be queued up.
             self.flow.pause_reading()
             self.pipeline.appendleft((self.cycle, app))
 
+    def _start_asgi_task(self, cycle: RequestResponseCycle, app: ASGI3Application) -> None:
+        if self.config.reset_contextvars:
+            # Opt-in workaround for https://github.com/python/cpython/issues/140947:
+            # asyncio can leak context vars between tasks. Hides context set in the
+            # lifespan or by external instrumentation.
+            if sys.version_info >= (3, 11):  # pragma: py-lt-311
+                task = self.loop.create_task(cycle.run_asgi(app), context=contextvars.Context())
+            else:  # pragma: py-gte-311
+                task = contextvars.Context().run(self.loop.create_task, cycle.run_asgi(app))
+        else:
+            task = self.loop.create_task(cycle.run_asgi(app))
+        task.add_done_callback(self.tasks.discard)
+        self.tasks.add(task)
+
     def on_body(self, body: bytes) -> None:
         if (self.parser.should_upgrade() and self._should_upgrade()) or self.cycle.response_complete:
             return
@@ -333,9 +339,7 @@ class HttpToolsProtocol(asyncio.Protocol):
         # Keep-Alive timeout instead.
         if self.pipeline:
             cycle, app = self.pipeline.pop()
-            task = self.loop.create_task(cycle.run_asgi(app))
-            task.add_done_callback(self.tasks.discard)
-            self.tasks.add(task)
+            self._start_asgi_task(cycle, app)
         else:
             self.timeout_keep_alive_task = self.loop.call_later(
                 self.timeout_keep_alive, self.timeout_keep_alive_handler
@@ -456,8 +460,6 @@ class RequestResponseCycle:
 
     # ASGI interface
     async def send(self, message: ASGISendEvent) -> None:
-        message_type = message["type"]
-
         if self.flow.write_paused and not self.disconnected:
             await self.flow.drain()  # pragma: full coverage
 
@@ -466,10 +468,8 @@ class RequestResponseCycle:
 
         if not self.response_started:
             # Sending response status line and headers
-            if message_type != "http.response.start":
-                msg = "Expected ASGI message 'http.response.start', but got '%s'."
-                raise RuntimeError(msg % message_type)
-            message = cast("HTTPResponseStartEvent", message)
+            if message["type"] != "http.response.start":
+                raise RuntimeError(f"Expected ASGI message 'http.response.start', but got '{message['type']}'.")
 
             self.response_started = True
             self.waiting_for_100_continue = False
@@ -520,11 +520,10 @@ class RequestResponseCycle:
 
         elif not self.response_complete:
             # Sending response body
-            if message_type != "http.response.body":
-                msg = "Expected ASGI message 'http.response.body', but got '%s'."
-                raise RuntimeError(msg % message_type)
+            if message["type"] != "http.response.body":
+                raise RuntimeError(f"Expected ASGI message 'http.response.body', but got '{message['type']}'.")
 
-            body = cast(bytes, message.get("body", b""))
+            body = message.get("body", b"")
             more_body = message.get("more_body", False)
 
             # Write response body
@@ -558,8 +557,7 @@ class RequestResponseCycle:
 
         else:
             # Response already sent
-            msg = "Unexpected ASGI message '%s' sent, after response already completed."
-            raise RuntimeError(msg % message_type)
+            raise RuntimeError(f"Unexpected ASGI message '{message['type']}' sent, after response already completed.")
 
     async def receive(self) -> ASGIReceiveEvent:
         if self.waiting_for_100_continue and not self.transport.is_closing():

uvicorn/protocols/websockets/websockets_impl.py

@@ -20,15 +20,10 @@ from websockets.typing import Subprotocol
 from uvicorn._types import (
     ASGI3Application,
     ASGISendEvent,
-    WebSocketAcceptEvent,
-    WebSocketCloseEvent,
     WebSocketConnectEvent,
     WebSocketDisconnectEvent,
     WebSocketReceiveEvent,
-    WebSocketResponseBodyEvent,
-    WebSocketResponseStartEvent,
     WebSocketScope,
-    WebSocketSendEvent,
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
@@ -262,11 +257,8 @@ class WebSocketProtocol(WebSocketServerProtocol):
         self.transport.close()
 
     async def asgi_send(self, message: ASGISendEvent) -> None:
-        message_type = message["type"]
-
         if not self.handshake_started_event.is_set():
-            if message_type == "websocket.accept":
-                message = cast("WebSocketAcceptEvent", message)
+            if message["type"] == "websocket.accept":
                 self.logger.info(
                     '%s - "WebSocket %s" [accepted]',
                     get_client_addr(self.scope),
@@ -283,8 +275,7 @@ class WebSocketProtocol(WebSocketServerProtocol):
                     )
                 self.handshake_started_event.set()
 
-            elif message_type == "websocket.close":
-                message = cast("WebSocketCloseEvent", message)
+            elif message["type"] == "websocket.close":
                 self.logger.info(
                     '%s - "WebSocket %s" 403',
                     get_client_addr(self.scope),
@@ -294,8 +285,7 @@ class WebSocketProtocol(WebSocketServerProtocol):
                 self.handshake_started_event.set()
                 self.closed_event.set()
 
-            elif message_type == "websocket.http.response.start":
-                message = cast("WebSocketResponseStartEvent", message)
+            elif message["type"] == "websocket.http.response.start":
                 self.logger.info(
                     '%s - "WebSocket %s" %d',
                     get_client_addr(self.scope),
@@ -311,50 +301,48 @@ class WebSocketProtocol(WebSocketServerProtocol):
                 self.handshake_started_event.set()
 
             else:
-                msg = (
+                raise RuntimeError(
                     "Expected ASGI message 'websocket.accept', 'websocket.close', "
-                    "or 'websocket.http.response.start' but got '%s'."
+                    f"or 'websocket.http.response.start' but got '{message['type']}'."
                 )
-                raise RuntimeError(msg % message_type)
 
         elif not self.closed_event.is_set() and self.initial_response is None:
             await self.handshake_completed_event.wait()
 
             try:
-                if message_type == "websocket.send":
-                    message = cast("WebSocketSendEvent", message)
+                if message["type"] == "websocket.send":
                     bytes_data = message.get("bytes")
                     text_data = message.get("text")
                     data = text_data if bytes_data is None else bytes_data
                     await self.send(data)  # type: ignore[arg-type]
 
-                elif message_type == "websocket.close":
-                    message = cast("WebSocketCloseEvent", message)
+                elif message["type"] == "websocket.close":
                     code = message.get("code", 1000)
                     reason = message.get("reason", "") or ""
                     await self.close(code, reason)
                     self.closed_event.set()
 
                 else:
-                    msg = "Expected ASGI message 'websocket.send' or 'websocket.close', but got '%s'."
-                    raise RuntimeError(msg % message_type)
+                    raise RuntimeError(
+                        f"Expected ASGI message 'websocket.send' or 'websocket.close', but got '{message['type']}'."
+                    )
             except ConnectionClosed as exc:
                 raise ClientDisconnected from exc
 
         elif self.initial_response is not None:
-            if message_type == "websocket.http.response.body":
-                message = cast("WebSocketResponseBodyEvent", message)
+            if message["type"] == "websocket.http.response.body":
                 body = self.initial_response[2] + message["body"]
                 self.initial_response = self.initial_response[:2] + (body,)
                 if not message.get("more_body", False):
                     self.closed_event.set()
             else:
-                msg = "Expected ASGI message 'websocket.http.response.body' but got '%s'."
-                raise RuntimeError(msg % message_type)
+                raise RuntimeError(f"Expected ASGI message 'websocket.http.response.body' but got '{message['type']}'.")
 
         else:
-            msg = "Unexpected ASGI message '%s', after sending 'websocket.close' or response already completed."
-            raise RuntimeError(msg % message_type)
+            raise RuntimeError(
+                f"Unexpected ASGI message '{message['type']}', after sending 'websocket.close' "
+                "or response already completed."
+            )
 
     async def asgi_receive(self) -> WebSocketDisconnectEvent | WebSocketConnectEvent | WebSocketReceiveEvent:
         if not self.connect_sent:

uvicorn/protocols/websockets/websockets_sansio_impl.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import random
+import struct
 import sys
+from asyncio import TimerHandle
 from asyncio.transports import BaseTransport, Transport
 from http import HTTPStatus
 from typing import Any, Literal, cast
@@ -17,12 +20,7 @@ from websockets.server import ServerProtocol
 from uvicorn._types import (
     ASGIReceiveEvent,
     ASGISendEvent,
-    WebSocketAcceptEvent,
-    WebSocketCloseEvent,
-    WebSocketResponseBodyEvent,
-    WebSocketResponseStartEvent,
     WebSocketScope,
-    WebSocketSendEvent,
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
@@ -97,8 +95,17 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
         self.writable = asyncio.Event()
         self.writable.set()
 
+        # Keepalive state
+        self.ping_interval = config.ws_ping_interval
+        self.ping_timeout = config.ws_ping_timeout
+        self.ping_timer: TimerHandle | None = None
+        self.pong_timer: TimerHandle | None = None
+        self.pending_ping_payload: bytes | None = None
+        self.ping_sent_at: float = 0.0
+        self.last_ping_rtt: float = 0.0
+
         # Buffers
-        self.bytes = b""
+        self.bytes = bytearray()
 
     def connection_made(self, transport: BaseTransport) -> None:
         """Called when a connection is made."""
@@ -114,6 +121,7 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
             self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket connection made", prefix)
 
     def connection_lost(self, exc: Exception | None) -> None:
+        self.stop_keepalive()
         code = 1005 if self.handshake_complete else 1006
         self.queue.put_nowait({"type": "websocket.disconnect", "code": code})
         self.connections.remove(self)
@@ -130,6 +138,7 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
         pass
 
     def shutdown(self) -> None:
+        self.stop_keepalive()
         if self.handshake_complete:
             self.queue.put_nowait({"type": "websocket.disconnect", "code": 1012})
             self.conn.send_close(1012)
@@ -160,7 +169,7 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                 elif event.opcode == Opcode.PING:
                     self.handle_ping()
                 elif event.opcode == Opcode.PONG:
-                    pass  # pragma: no cover
+                    self.handle_pong(event)
                 elif event.opcode == Opcode.CLOSE:
                     self.handle_close(event)
                 else:
@@ -207,19 +216,19 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
         task.add_done_callback(self.on_task_complete)
         self.tasks.add(task)
 
-    def handle_cont(self, event: Frame) -> None:  # pragma: no cover
-        self.bytes += event.data
+    def handle_cont(self, event: Frame) -> None:
+        self.bytes.extend(event.data)
         if event.fin:
             self.send_receive_event_to_app()
 
     def handle_text(self, event: Frame) -> None:
-        self.bytes = event.data
+        self.bytes = bytearray(event.data)
         self.curr_msg_data_type: Literal["text", "bytes"] = "text"
         if event.fin:
             self.send_receive_event_to_app()
 
     def handle_bytes(self, event: Frame) -> None:
-        self.bytes = event.data
+        self.bytes = bytearray(event.data)
         self.curr_msg_data_type = "bytes"
         if event.fin:
             self.send_receive_event_to_app()
@@ -234,7 +243,7 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                 self.handle_parser_exception()
                 return
         else:
-            self.queue.put_nowait({"type": "websocket.receive", "bytes": self.bytes})
+            self.queue.put_nowait({"type": "websocket.receive", "bytes": bytes(self.bytes)})
         if not self.read_paused:
             self.read_paused = True
             self.transport.pause_reading()
@@ -243,6 +252,67 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
         output = self.conn.data_to_send()
         self.transport.write(b"".join(output))
 
+    def handle_pong(self, event: Frame) -> None:
+        # Ignore unsolicited pongs and stale pongs whose payload doesn't match the ping currently in flight
+        if self.pending_ping_payload is None or bytes(event.data) != self.pending_ping_payload:
+            return  # pragma: no cover
+
+        self.last_ping_rtt = self.loop.time() - self.ping_sent_at
+        self.pending_ping_payload = None
+        # The peer answered in time; cancel the pong deadline and chain the next ping. This `schedule_ping()` call is
+        # what keeps the keepalive loop running when ping_timeout is set. When ping_timeout is None the next ping is
+        # already scheduled by `send_keepalive_ping`, so we must not schedule a duplicate here.
+        if self.pong_timer is not None:
+            self.pong_timer.cancel()
+            self.pong_timer = None
+            self.schedule_ping()
+
+    def start_keepalive(self) -> None:
+        if self.ping_interval is not None and self.ping_interval > 0:
+            self.schedule_ping()
+
+    def stop_keepalive(self) -> None:
+        if self.ping_timer is not None:
+            self.ping_timer.cancel()
+            self.ping_timer = None
+        if self.pong_timer is not None:  # pragma: no cover
+            self.pong_timer.cancel()
+            self.pong_timer = None
+        self.pending_ping_payload = None
+
+    def schedule_ping(self) -> None:
+        assert self.ping_interval is not None
+        delay = max(0.0, self.ping_interval - self.last_ping_rtt)
+        self.ping_timer = self.loop.call_later(delay, self.send_keepalive_ping)
+
+    def send_keepalive_ping(self) -> None:
+        self.ping_timer = None
+        if self.close_sent or self.transport.is_closing():  # pragma: no cover
+            return
+        # Random 4-byte payload identifies this ping; `handle_pong` uses it to ignore stale or unsolicited pongs.
+        # See https://github.com/python-websockets/websockets/blob/4d229bf9f583d593aa103287aee0a77c9fbc3a79/src/websockets/asyncio/connection.py#L624
+        self.pending_ping_payload = struct.pack("!I", random.getrandbits(32))
+        self.ping_sent_at = self.loop.time()
+        self.conn.send_ping(self.pending_ping_payload)
+        self.transport.write(b"".join(self.conn.data_to_send()))
+        if self.ping_timeout is not None:
+            self.pong_timer = self.loop.call_later(self.ping_timeout, self.keepalive_timeout)
+        else:  # pragma: no cover
+            self.schedule_ping()
+
+    def keepalive_timeout(self) -> None:
+        self.pong_timer = None
+        self.pending_ping_payload = None
+        if self.close_sent or self.transport.is_closing():  # pragma: no cover
+            return
+        if self.logger.level <= TRACE_LOG_LEVEL:
+            prefix = "%s:%d - " % self.client if self.client else ""
+            self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket keepalive ping timeout", prefix)
+        self.conn.fail(1011, "keepalive ping timeout")
+        self.transport.write(b"".join(self.conn.data_to_send()))
+        self.close_sent = True
+        self.transport.close()
+
     def handle_close(self, event: Frame) -> None:
         if not self.close_sent and not self.transport.is_closing():
             assert self.conn.close_rcvd is not None
@@ -295,11 +365,8 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
     async def send(self, message: ASGISendEvent) -> None:
         await self.writable.wait()
 
-        message_type = message["type"]
-
         if not self.handshake_complete and self.initial_response is None:
-            if message_type == "websocket.accept":
-                message = cast(WebSocketAcceptEvent, message)
+            if message["type"] == "websocket.accept":
                 self.logger.info(
                     '%s - "WebSocket %s" [accepted]',
                     get_client_addr(self.scope),
@@ -319,9 +386,9 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                     self.conn.send_response(self.response)
                     output = self.conn.data_to_send()
                     self.transport.write(b"".join(output))
+                    self.start_keepalive()
 
-            elif message_type == "websocket.close":
-                message = cast(WebSocketCloseEvent, message)
+            elif message["type"] == "websocket.close":
                 self.queue.put_nowait({"type": "websocket.disconnect", "code": 1006})
                 self.logger.info(
                     '%s - "WebSocket %s" 403',
@@ -335,8 +402,7 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                 self.handshake_complete = True
                 self.transport.write(b"".join(output))
                 self.transport.close()
-            elif message_type == "websocket.http.response.start" and self.initial_response is None:
-                message = cast(WebSocketResponseStartEvent, message)
+            elif message["type"] == "websocket.http.response.start" and self.initial_response is None:
                 if not (100 <= message["status"] < 600):
                     raise RuntimeError("Invalid HTTP status code '%d' in response." % message["status"])
                 self.logger.info(
@@ -351,17 +417,14 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                 ]
                 self.initial_response = (message["status"], headers, b"")
             else:
-                msg = (
+                raise RuntimeError(
                     "Expected ASGI message 'websocket.accept', 'websocket.close' "
-                    "or 'websocket.http.response.start' "
-                    "but got '%s'."
+                    f"or 'websocket.http.response.start' but got '{message['type']}'."
                 )
-                raise RuntimeError(msg % message_type)
 
         elif not self.close_sent and self.initial_response is None:
             try:
-                if message_type == "websocket.send":
-                    message = cast(WebSocketSendEvent, message)
+                if message["type"] == "websocket.send":
                     bytes_data = message.get("bytes")
                     text_data = message.get("text")
                     if bytes_data is not None:
@@ -371,9 +434,8 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                     output = self.conn.data_to_send()
                     self.transport.write(b"".join(output))
 
-                elif message_type == "websocket.close":
+                elif message["type"] == "websocket.close":
                     if not self.transport.is_closing():
-                        message = cast(WebSocketCloseEvent, message)
                         code = message.get("code", 1000)
                         reason = message.get("reason", "") or ""
                         self.queue.put_nowait({"type": "websocket.disconnect", "code": code, "reason": reason})
@@ -383,13 +445,13 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                         self.close_sent = True
                         self.transport.close()
                 else:
-                    msg = "Expected ASGI message 'websocket.send' or 'websocket.close', but got '%s'."
-                    raise RuntimeError(msg % message_type)
+                    raise RuntimeError(
+                        f"Expected ASGI message 'websocket.send' or 'websocket.close', but got '{message['type']}'."
+                    )
             except InvalidState:
                 raise ClientDisconnected()
         elif self.initial_response is not None:
-            if message_type == "websocket.http.response.body":
-                message = cast(WebSocketResponseBodyEvent, message)
+            if message["type"] == "websocket.http.response.body":
                 body = self.initial_response[2] + message["body"]
                 self.initial_response = self.initial_response[:2] + (body,)
                 if not message.get("more_body", False):
@@ -402,12 +464,10 @@ class WebSocketsSansIOProtocol(asyncio.Protocol):
                     self.transport.write(b"".join(output))
                     self.transport.close()
             else:  # pragma: no cover
-                msg = "Expected ASGI message 'websocket.http.response.body' but got '%s'."
-                raise RuntimeError(msg % message_type)
+                raise RuntimeError(f"Expected ASGI message 'websocket.http.response.body' but got '{message['type']}'.")
 
         else:
-            msg = "Unexpected ASGI message '%s', after sending 'websocket.close'."
-            raise RuntimeError(msg % message_type)
+            raise RuntimeError(f"Unexpected ASGI message '{message['type']}', after sending 'websocket.close'.")
 
     async def receive(self) -> ASGIReceiveEvent:
         message = await self.queue.get()

uvicorn/protocols/websockets/wsproto_impl.py

@@ -2,6 +2,10 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import random
+import struct
+from asyncio import TimerHandle
+from io import BytesIO, StringIO
 from typing import Any, Literal, cast
 from urllib.parse import unquote
 
@@ -11,17 +15,7 @@ from wsproto.connection import ConnectionState
 from wsproto.extensions import Extension, PerMessageDeflate
 from wsproto.utilities import LocalProtocolError, RemoteProtocolError
 
-from uvicorn._types import (
-    ASGI3Application,
-    ASGISendEvent,
-    WebSocketAcceptEvent,
-    WebSocketCloseEvent,
-    WebSocketEvent,
-    WebSocketResponseBodyEvent,
-    WebSocketResponseStartEvent,
-    WebSocketScope,
-    WebSocketSendEvent,
-)
+from uvicorn._types import ASGI3Application, ASGISendEvent, WebSocketEvent, WebSocketReceiveEvent, WebSocketScope
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
 from uvicorn.protocols.utils import (
@@ -35,6 +29,36 @@ from uvicorn.protocols.utils import (
 from uvicorn.server import ServerState
 
 
+class FrameTooLargeError(Exception):
+    """Raised when accumulated websocket message bytes exceed `ws_max_size`."""
+
+
+class WebsocketBuffer:
+    def __init__(self, max_length: int) -> None:
+        self.value: BytesIO | StringIO | None = None
+        self.length = 0
+        self.max_length = max_length
+
+    def extend(self, event: events.TextMessage | events.BytesMessage) -> None:
+        if self.value is None:
+            self.value = StringIO() if isinstance(event, events.TextMessage) else BytesIO()
+        self.value.write(event.data)  # type: ignore[arg-type]
+        # `ws_max_size` is a byte budget, so count UTF-8 bytes for text.
+        self.length += len(event.data.encode()) if isinstance(event, events.TextMessage) else len(event.data)
+        if self.length > self.max_length:
+            raise FrameTooLargeError
+
+    def clear(self) -> None:
+        self.value = None
+        self.length = 0
+
+    def to_message(self) -> WebSocketReceiveEvent:
+        if isinstance(self.value, StringIO):
+            return {"type": "websocket.receive", "text": self.value.getvalue()}
+        assert isinstance(self.value, BytesIO)
+        return {"type": "websocket.receive", "bytes": self.value.getvalue()}
+
+
 class WSProtocol(asyncio.Protocol):
     def __init__(
         self,
@@ -78,15 +102,21 @@ class WSProtocol(asyncio.Protocol):
         self.writable = asyncio.Event()
         self.writable.set()
 
-        # Buffers
-        self.bytes = b""
-        self.text = ""
+        # Keepalive state
+        self.ping_interval = config.ws_ping_interval
+        self.ping_timeout = config.ws_ping_timeout
+        self.ping_timer: TimerHandle | None = None
+        self.pong_timer: TimerHandle | None = None
+        self.pending_ping_payload: bytes | None = None
+        self.ping_sent_at: float = 0.0
+        self.last_ping_rtt: float = 0.0
+
+        # Buffer
+        self.buffer = WebsocketBuffer(self.config.ws_max_size)
 
     # Protocol interface
 
-    def connection_made(  # type: ignore[override]
-        self, transport: asyncio.Transport
-    ) -> None:
+    def connection_made(self, transport: asyncio.Transport) -> None:  # type: ignore[override]
         self.connections.add(self)
         self.transport = transport
         self.server = get_local_addr(transport)
@@ -98,6 +128,7 @@ class WSProtocol(asyncio.Protocol):
             self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket connection made", prefix)
 
     def connection_lost(self, exc: Exception | None) -> None:
+        self.stop_keepalive()
         code = 1005 if self.handshake_complete else 1006
         self.queue.put_nowait({"type": "websocket.disconnect", "code": code})
         self.connections.remove(self)
@@ -125,16 +156,18 @@ class WSProtocol(asyncio.Protocol):
 
     def handle_events(self) -> None:
         for event in self.conn.events():
+            if self.close_sent:
+                return
             if isinstance(event, events.Request):
                 self.handle_connect(event)
-            elif isinstance(event, events.TextMessage):
-                self.handle_text(event)
-            elif isinstance(event, events.BytesMessage):
-                self.handle_bytes(event)
+            elif isinstance(event, (events.TextMessage, events.BytesMessage)):
+                self.handle_message(event)
             elif isinstance(event, events.CloseConnection):
                 self.handle_close(event)
             elif isinstance(event, events.Ping):
                 self.handle_ping(event)
+            elif isinstance(event, events.Pong):
+                self.handle_pong(event)
 
     def pause_writing(self) -> None:
         """
@@ -149,6 +182,7 @@ class WSProtocol(asyncio.Protocol):
         self.writable.set()  # pragma: full coverage
 
     def shutdown(self) -> None:
+        self.stop_keepalive()
         if self.handshake_complete:
             self.queue.put_nowait({"type": "websocket.disconnect", "code": 1012})
             output = self.conn.send(wsproto.events.CloseConnection(code=1012))
@@ -190,21 +224,20 @@ class WSProtocol(asyncio.Protocol):
         task.add_done_callback(self.on_task_complete)
         self.tasks.add(task)
 
-    def handle_text(self, event: events.TextMessage) -> None:
-        self.text += event.data
+    def handle_message(self, event: events.TextMessage | events.BytesMessage) -> None:
+        try:
+            self.buffer.extend(event)
+        except FrameTooLargeError:
+            self.close_sent = True
+            reason = f"Message exceeds the maximum size ({self.config.ws_max_size} bytes)"
+            self.queue.put_nowait({"type": "websocket.disconnect", "code": 1009, "reason": reason})
+            if not self.transport.is_closing():
+                self.transport.write(self.conn.send(wsproto.events.CloseConnection(code=1009, reason=reason)))
+                self.transport.close()
+            return
         if event.message_finished:
-            self.queue.put_nowait({"type": "websocket.receive", "text": self.text})
-            self.text = ""
-            if not self.read_paused:
-                self.read_paused = True
-                self.transport.pause_reading()
-
-    def handle_bytes(self, event: events.BytesMessage) -> None:
-        self.bytes += event.data
-        # todo: we may want to guard the size of self.bytes and self.text
-        if event.message_finished:
-            self.queue.put_nowait({"type": "websocket.receive", "bytes": self.bytes})
-            self.bytes = b""
+            self.queue.put_nowait(self.buffer.to_message())
+            self.buffer.clear()
             if not self.read_paused:
                 self.read_paused = True
                 self.transport.pause_reading()
@@ -218,6 +251,65 @@ class WSProtocol(asyncio.Protocol):
     def handle_ping(self, event: events.Ping) -> None:
         self.transport.write(self.conn.send(event.response()))
 
+    def handle_pong(self, event: events.Pong) -> None:
+        # Ignore unsolicited pongs and stale pongs whose payload doesn't match the ping currently in flight.
+        if self.pending_ping_payload is None or bytes(event.payload) != self.pending_ping_payload:
+            return  # pragma: no cover
+
+        self.last_ping_rtt = self.loop.time() - self.ping_sent_at
+        self.pending_ping_payload = None
+        # The peer answered in time; cancel the pong deadline and chain the next ping. This `schedule_ping()` call is
+        # what keeps the keepalive loop running when ping_timeout is set. When ping_timeout is None the next ping is
+        # already scheduled by `send_keepalive_ping`, so we must not schedule a duplicate here.
+        if self.pong_timer is not None:
+            self.pong_timer.cancel()
+            self.pong_timer = None
+            self.schedule_ping()
+
+    def start_keepalive(self) -> None:
+        if self.ping_interval is not None and self.ping_interval > 0:
+            self.schedule_ping()
+
+    def stop_keepalive(self) -> None:
+        if self.ping_timer is not None:
+            self.ping_timer.cancel()
+            self.ping_timer = None
+        if self.pong_timer is not None:  # pragma: no cover
+            self.pong_timer.cancel()
+            self.pong_timer = None
+        self.pending_ping_payload = None
+
+    def schedule_ping(self) -> None:
+        assert self.ping_interval is not None
+        delay = max(0.0, self.ping_interval - self.last_ping_rtt)
+        self.ping_timer = self.loop.call_later(delay, self.send_keepalive_ping)
+
+    def send_keepalive_ping(self) -> None:
+        self.ping_timer = None
+        if self.close_sent or self.transport.is_closing():  # pragma: no cover
+            return
+        # Random 4-byte payload identifies this ping; `handle_pong` uses it to ignore stale or unsolicited pongs.
+        self.pending_ping_payload = struct.pack("!I", random.getrandbits(32))
+        self.ping_sent_at = self.loop.time()
+        self.transport.write(self.conn.send(wsproto.events.Ping(payload=self.pending_ping_payload)))
+        if self.ping_timeout is not None:
+            self.pong_timer = self.loop.call_later(self.ping_timeout, self.keepalive_timeout)
+        else:  # pragma: no cover
+            self.schedule_ping()
+
+    def keepalive_timeout(self) -> None:
+        self.pong_timer = None
+        self.pending_ping_payload = None
+        if self.close_sent or self.transport.is_closing():  # pragma: no cover
+            return
+        if self.logger.level <= TRACE_LOG_LEVEL:
+            prefix = "%s:%d - " % self.client if self.client else ""
+            self.logger.log(TRACE_LOG_LEVEL, "%sWebSocket keepalive ping timeout", prefix)
+        reason = "keepalive ping timeout"
+        self.transport.write(self.conn.send(wsproto.events.CloseConnection(code=1011, reason=reason)))
+        self.close_sent = True
+        self.transport.close()
+
     def send_500_response(self) -> None:
         if self.response_started or self.handshake_complete:
             return  # we cannot send responses anymore
@@ -249,11 +341,8 @@ class WSProtocol(asyncio.Protocol):
     async def send(self, message: ASGISendEvent) -> None:
         await self.writable.wait()
 
-        message_type = message["type"]
-
         if not self.handshake_complete:
-            if message_type == "websocket.accept":
-                message = cast(WebSocketAcceptEvent, message)
+            if message["type"] == "websocket.accept":
                 self.logger.info(
                     '%s - "WebSocket %s" [accepted]',
                     get_client_addr(self.scope),
@@ -274,8 +363,9 @@ class WSProtocol(asyncio.Protocol):
                         )
                     )
                     self.transport.write(output)
+                    self.start_keepalive()
 
-            elif message_type == "websocket.close":
+            elif message["type"] == "websocket.close":
                 self.queue.put_nowait({"type": "websocket.disconnect", "code": 1006})
                 self.logger.info(
                     '%s - "WebSocket %s" 403',
@@ -289,8 +379,7 @@ class WSProtocol(asyncio.Protocol):
                 self.transport.write(output)
                 self.transport.close()
 
-            elif message_type == "websocket.http.response.start":
-                message = cast(WebSocketResponseStartEvent, message)
+            elif message["type"] == "websocket.http.response.start":
                 # ensure status code is in the valid range
                 if not (100 <= message["status"] < 600):
                     msg = "Invalid HTTP status code '%d' in response."
@@ -312,17 +401,14 @@ class WSProtocol(asyncio.Protocol):
                 self.response_started = True
 
             else:
-                msg = (
+                raise RuntimeError(
                     "Expected ASGI message 'websocket.accept', 'websocket.close' "
-                    "or 'websocket.http.response.start' "
-                    "but got '%s'."
+                    f"or 'websocket.http.response.start' but got '{message['type']}'."
                 )
-                raise RuntimeError(msg % message_type)
 
         elif not self.close_sent and not self.response_started:
             try:
-                if message_type == "websocket.send":
-                    message = cast(WebSocketSendEvent, message)
+                if message["type"] == "websocket.send":
                     bytes_data = message.get("bytes")
                     text_data = message.get("text")
                     data = text_data if bytes_data is None else bytes_data
@@ -330,8 +416,7 @@ class WSProtocol(asyncio.Protocol):
                     if not self.transport.is_closing():
                         self.transport.write(output)
 
-                elif message_type == "websocket.close":
-                    message = cast(WebSocketCloseEvent, message)
+                elif message["type"] == "websocket.close":
                     self.close_sent = True
                     code = message.get("code", 1000)
                     reason = message.get("reason", "") or ""
@@ -342,13 +427,13 @@ class WSProtocol(asyncio.Protocol):
                         self.transport.close()
 
                 else:
-                    msg = "Expected ASGI message 'websocket.send' or 'websocket.close', but got '%s'."
-                    raise RuntimeError(msg % message_type)
+                    raise RuntimeError(
+                        f"Expected ASGI message 'websocket.send' or 'websocket.close', but got '{message['type']}'."
+                    )
             except LocalProtocolError as exc:
                 raise ClientDisconnected from exc
         elif self.response_started:
-            if message_type == "websocket.http.response.body":
-                message = cast("WebSocketResponseBodyEvent", message)
+            if message["type"] == "websocket.http.response.body":
                 body_finished = not message.get("more_body", False)
                 reject_data = events.RejectData(data=message["body"], body_finished=body_finished)
                 output = self.conn.send(reject_data)
@@ -360,12 +445,10 @@ class WSProtocol(asyncio.Protocol):
                     self.transport.close()
 
             else:
-                msg = "Expected ASGI message 'websocket.http.response.body' but got '%s'."
-                raise RuntimeError(msg % message_type)
+                raise RuntimeError(f"Expected ASGI message 'websocket.http.response.body' but got '{message['type']}'.")
 
         else:
-            msg = "Unexpected ASGI message '%s', after sending 'websocket.close'."
-            raise RuntimeError(msg % message_type)
+            raise RuntimeError(f"Unexpected ASGI message '{message['type']}', after sending 'websocket.close'.")
 
     async def receive(self) -> WebSocketEvent:
         message = await self.queue.get()