Modify the WSGI environ based on the various ``Forwarded`` headers before calling the wrapped application. Store the original environ values in ``werkzeug.proxy_fix.orig_{key}``.
(
self, environ: WSGIEnvironment, start_response: StartResponse
)
| 124 | return None |
| 125 | |
| 126 | def __call__( |
| 127 | self, environ: WSGIEnvironment, start_response: StartResponse |
| 128 | ) -> t.Iterable[bytes]: |
| 129 | """Modify the WSGI environ based on the various ``Forwarded`` |
| 130 | headers before calling the wrapped application. Store the |
| 131 | original environ values in ``werkzeug.proxy_fix.orig_{key}``. |
| 132 | """ |
| 133 | environ_get = environ.get |
| 134 | orig_remote_addr = environ_get("REMOTE_ADDR") |
| 135 | orig_wsgi_url_scheme = environ_get("wsgi.url_scheme") |
| 136 | orig_http_host = environ_get("HTTP_HOST") |
| 137 | environ.update( |
| 138 | { |
| 139 | "werkzeug.proxy_fix.orig": { |
| 140 | "REMOTE_ADDR": orig_remote_addr, |
| 141 | "wsgi.url_scheme": orig_wsgi_url_scheme, |
| 142 | "HTTP_HOST": orig_http_host, |
| 143 | "SERVER_NAME": environ_get("SERVER_NAME"), |
| 144 | "SERVER_PORT": environ_get("SERVER_PORT"), |
| 145 | "SCRIPT_NAME": environ_get("SCRIPT_NAME"), |
| 146 | } |
| 147 | } |
| 148 | ) |
| 149 | |
| 150 | x_for = self._get_real_value(self.x_for, environ_get("HTTP_X_FORWARDED_FOR")) |
| 151 | if x_for: |
| 152 | environ["REMOTE_ADDR"] = x_for |
| 153 | |
| 154 | x_proto = self._get_real_value( |
| 155 | self.x_proto, environ_get("HTTP_X_FORWARDED_PROTO") |
| 156 | ) |
| 157 | if x_proto: |
| 158 | environ["wsgi.url_scheme"] = x_proto |
| 159 | |
| 160 | x_host = self._get_real_value(self.x_host, environ_get("HTTP_X_FORWARDED_HOST")) |
| 161 | if x_host: |
| 162 | environ["HTTP_HOST"] = environ["SERVER_NAME"] = x_host |
| 163 | # "]" to check for IPv6 address without port |
| 164 | if ":" in x_host and not x_host.endswith("]"): |
| 165 | environ["SERVER_NAME"], environ["SERVER_PORT"] = x_host.rsplit(":", 1) |
| 166 | |
| 167 | x_port = self._get_real_value(self.x_port, environ_get("HTTP_X_FORWARDED_PORT")) |
| 168 | if x_port: |
| 169 | host = environ.get("HTTP_HOST") |
| 170 | if host: |
| 171 | # "]" to check for IPv6 address without port |
| 172 | if ":" in host and not host.endswith("]"): |
| 173 | host = host.rsplit(":", 1)[0] |
| 174 | environ["HTTP_HOST"] = f"{host}:{x_port}" |
| 175 | environ["SERVER_PORT"] = x_port |
| 176 | |
| 177 | x_prefix = self._get_real_value( |
| 178 | self.x_prefix, environ_get("HTTP_X_FORWARDED_PREFIX") |
| 179 | ) |
| 180 | if x_prefix: |
| 181 | environ["SCRIPT_NAME"] = x_prefix |
| 182 | |
| 183 | return self.app(environ, start_response) |
nothing calls this directly
no test coverage detected