Created
December 19, 2018 09:27
-
-
Save simonluijk/a54d35b27439ea2b2a119e6b7c44750c to your computer and use it in GitHub Desktop.
Validates the X_FORWARDED_FOR header in django middleware
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from django.conf import settings | |
from django.core.exceptions import PermissionDenied | |
from netaddr import IPAddress, IPNetwork | |
class ProxyChainMiddleware(object): | |
""" | |
Validates the X_FORWARDED_FOR header against settings.PROXY_CHAIN. Then | |
sets REMOTE_ADDR to the IP address of the first untrusted IP in | |
X_FORWARDED_FOR. | |
Currently it only checks a strict hierarchy of reverse proxies. Could also | |
parse the rest of X_FORWARDED_FOR against a list of trusted ISP proxies. | |
""" | |
def __init__(self): | |
chain = getattr(settings, "PROXY_CHAIN", []) | |
self.proxy_chain = [map(IPNetwork, nwlist) for nwlist in chain] | |
def __is_proxy(self, ip, network_list): | |
return any(map(lambda network: ip in network, network_list)) | |
def process_request(self, request): | |
if settings.DEBUG: | |
# Don't validate proxies in development. | |
return None | |
if len(self.proxy_chain) == 0: | |
# No trusted proxies, leave REMOTE_ADDR as is. | |
return None | |
# Validate REMOTE_ADDR is in first proxy list in proxy chain | |
remote_addr = IPAddress(request.META['REMOTE_ADDR']) | |
if not self.__is_proxy(remote_addr, self.proxy_chain[0]): | |
raise PermissionDenied() | |
try: | |
forwareded = request.META['HTTP_X_FORWARDED_FOR'].split(',') | |
except KeyError: | |
raise PermissionDenied() | |
forwareded = map(IPAddress, | |
map(lambda ip: ip.strip(), reversed(forwareded))) | |
# Request should traverse the whole proxy chain. | |
passed = all( | |
map(lambda args: self.__is_proxy(*args), | |
zip(forwareded, self.proxy_chain[1:]))) | |
if not passed: | |
raise PermissionDenied() | |
try: | |
remote_addr = forwareded[len(self.proxy_chain) - 1] | |
request.META['REMOTE_ADDR'] = str(remote_addr) | |
except IndexError: | |
# Last proxy did not append an IP to X_FORWARDED_FOR. | |
raise PermissionDenied() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment