Function process_request

src/websockets/asyncio/server.py:951–995

Perform HTTP Basic Authentication. If it succeeds, set the connection's ``username`` attribute and return :obj:`None`. If it fails, return an HTTP 401 Unauthorized responss.

(
        connection: ServerConnection,
        request: Request,
    )

    assert check_credentials is not None  # help mypy

    async def process_request(
        connection: ServerConnection,
        request: Request,
    ) -> Response | None:
        """
        Perform HTTP Basic Authentication.

        If it succeeds, set the connection's ``username`` attribute and return
        :obj:`None`. If it fails, return an HTTP 401 Unauthorized responss.

        """
        try:
            authorization = request.headers["Authorization"]
        except KeyError:
            response = connection.respond(
                http.HTTPStatus.UNAUTHORIZED,
                "Missing credentials\n",
            )
            response.headers["WWW-Authenticate"] = build_www_authenticate_basic(realm)
            return response

        try:
            username, password = parse_authorization_basic(authorization)
        except InvalidHeader:
            response = connection.respond(
                http.HTTPStatus.UNAUTHORIZED,
                "Unsupported credentials\n",
            )
            response.headers["WWW-Authenticate"] = build_www_authenticate_basic(realm)
            return response

        valid_credentials = check_credentials(username, password)
        if isinstance(valid_credentials, Awaitable):
            valid_credentials = await valid_credentials

        if not valid_credentials:
            response = connection.respond(
                http.HTTPStatus.UNAUTHORIZED,
                "Invalid credentials\n",
            )
            response.headers["WWW-Authenticate"] = build_www_authenticate_basic(realm)
            return response

        connection.username = username
        return None

    return process_request