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