mirror of
https://github.com/django/daphne.git
synced 2025-10-22 11:34:27 +03:00
31bb1bcc23
The headers on my environment aren't bytes, rather str-s, and so getting the host and port from those will result None being passed as a result. Also, since X-Forwarded-For is not to be trusted, and custom nginx configurations can pass a `X-Real-IP` header, add two extra command line parameters to be able to parse custom passed remote IP headers.
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
import importlib
|
|
|
|
from twisted.web.http_headers import Headers
|
|
|
|
|
|
def import_by_path(path):
|
|
"""
|
|
Given a dotted/colon path, like project.module:ClassName.callable,
|
|
returns the object at the end of the path.
|
|
"""
|
|
module_path, object_path = path.split(":", 1)
|
|
target = importlib.import_module(module_path)
|
|
for bit in object_path.split("."):
|
|
target = getattr(target, bit)
|
|
return target
|
|
|
|
|
|
def header_value(headers, header_name) -> str:
|
|
value = headers[header_name]
|
|
if isinstance(value, list):
|
|
value = value[0]
|
|
return value.decode("utf-8") if type(value) is bytes else value
|
|
|
|
|
|
def parse_x_forwarded_for(headers,
|
|
address_header_name="X-Forwarded-For",
|
|
port_header_name="X-Forwarded-Port",
|
|
original=None):
|
|
"""
|
|
Parses an X-Forwarded-For header and returns a host/port pair as a list.
|
|
|
|
@param headers: The twisted-style object containing a request's headers
|
|
@param address_header_name: The name of the expected host header
|
|
@param port_header_name: The name of the expected port header
|
|
@param original: A host/port pair that should be returned if the headers are not in the request
|
|
@return: A list containing a host (string) as the first entry and a port (int) as the second.
|
|
"""
|
|
if not address_header_name:
|
|
return original
|
|
|
|
# Convert twisted-style headers into dicts
|
|
if isinstance(headers, Headers):
|
|
headers = dict(headers.getAllRawHeaders())
|
|
|
|
# Lowercase all header names in the dict
|
|
new_headers = dict()
|
|
for name, values in headers.items():
|
|
name = name.lower()
|
|
name = name if type(name) is bytes else name.encode('utf-8')
|
|
new_headers[name] = values
|
|
headers = new_headers
|
|
|
|
address_header_name = address_header_name.lower().encode('utf-8')
|
|
result = original
|
|
if address_header_name in headers:
|
|
address_value = header_value(headers, address_header_name)
|
|
|
|
if "," in address_value:
|
|
address_value = address_value.split(",")[0].strip()
|
|
|
|
result = [address_value, 0]
|
|
|
|
if port_header_name:
|
|
# We only want to parse the X-Forwarded-Port header if we also parsed the X-Forwarded-For
|
|
# header to avoid inconsistent results.
|
|
port_header_name = port_header_name.lower().encode("utf-8")
|
|
if port_header_name in headers:
|
|
port_value = header_value(headers, port_header_name)
|
|
try:
|
|
result[1] = int(port_value)
|
|
except ValueError:
|
|
pass
|
|
|
|
return result
|