(self, request_headers: Headers)
| 100 | return origin in self.allow_origins |
| 101 | |
| 102 | def preflight_response(self, request_headers: Headers) -> Response: |
| 103 | requested_origin = request_headers["origin"] |
| 104 | requested_method = request_headers["access-control-request-method"] |
| 105 | requested_headers = request_headers.get("access-control-request-headers") |
| 106 | requested_private_network = request_headers.get("access-control-request-private-network") |
| 107 | |
| 108 | headers = dict(self.preflight_headers) |
| 109 | failures: list[str] = [] |
| 110 | |
| 111 | if self.is_allowed_origin(origin=requested_origin): |
| 112 | if self.preflight_explicit_allow_origin: |
| 113 | # The "else" case is already accounted for in self.preflight_headers |
| 114 | # and the value would be "*". |
| 115 | headers["Access-Control-Allow-Origin"] = requested_origin |
| 116 | else: |
| 117 | failures.append("origin") |
| 118 | |
| 119 | if requested_method not in self.allow_methods: |
| 120 | failures.append("method") |
| 121 | |
| 122 | # If we allow all headers, then we have to mirror back any requested |
| 123 | # headers in the response. |
| 124 | if self.allow_all_headers and requested_headers is not None: |
| 125 | headers["Access-Control-Allow-Headers"] = requested_headers |
| 126 | elif requested_headers is not None: |
| 127 | for header in [h.lower() for h in requested_headers.split(",")]: |
| 128 | if header.strip() not in self.allow_headers: |
| 129 | failures.append("headers") |
| 130 | break |
| 131 | |
| 132 | if requested_private_network is not None: |
| 133 | if self.allow_private_network: |
| 134 | headers["Access-Control-Allow-Private-Network"] = "true" |
| 135 | else: |
| 136 | failures.append("private-network") |
| 137 | |
| 138 | # We don't strictly need to use 400 responses here, since its up to |
| 139 | # the browser to enforce the CORS policy, but its more informative |
| 140 | # if we do. |
| 141 | if failures: |
| 142 | failure_text = "Disallowed CORS " + ", ".join(failures) |
| 143 | return PlainTextResponse(failure_text, status_code=400, headers=headers) |
| 144 | |
| 145 | return PlainTextResponse("OK", status_code=200, headers=headers) |
| 146 | |
| 147 | async def simple_response(self, scope: Scope, receive: Receive, send: Send, request_headers: Headers) -> None: |
| 148 | send = functools.partial(self.send, send=send, request_headers=request_headers) |
no test coverage detected