Игорь Надеин
53cba386e9
All checks were successful
Deploy production / deploy (push) Successful in 15m8s
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Fail-closed registration policy for the CORP LAN and its trusted proxies."""
|
|
|
|
import ipaddress
|
|
|
|
from django.conf import settings
|
|
from rest_framework.permissions import BasePermission
|
|
|
|
|
|
def _address(value):
|
|
# Reject scoped IPv6, ports, empty strings and non-address header values.
|
|
if not isinstance(value, str) or not value or "%" in value:
|
|
raise ValueError("Invalid client address")
|
|
address = ipaddress.ip_address(value)
|
|
if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
|
|
return address.ipv4_mapped
|
|
return address
|
|
|
|
|
|
def _networks(values):
|
|
if not isinstance(values, (list, tuple)):
|
|
raise ValueError("Expected an explicit list of networks")
|
|
return tuple(ipaddress.ip_network(value, strict=True) for value in values)
|
|
|
|
|
|
def get_client_ip(request) -> str:
|
|
"""Walk XFF from the socket peer backwards, stopping at the first untrusted IP.
|
|
|
|
Only explicitly configured proxies may supply forwarding information. Each
|
|
proxy must append its actual peer (not overwrite it with a client header).
|
|
Never fall back to a proxy's own LAN address when the chain is missing/bad.
|
|
X-Real-IP, Forwarded and arbitrary custom headers are deliberately ignored.
|
|
"""
|
|
try:
|
|
peer = _address(request.META.get("REMOTE_ADDR", ""))
|
|
trusted = _networks(getattr(settings, "REGISTRATION_TRUSTED_PROXIES", ()))
|
|
if not any(peer in network for network in trusted):
|
|
return str(peer)
|
|
|
|
forwarded = request.META.get("HTTP_X_FORWARDED_FOR", "")
|
|
if not isinstance(forwarded, str) or len(forwarded) > 2048:
|
|
return ""
|
|
hops = forwarded.split(",")
|
|
if len(hops) > 16:
|
|
return ""
|
|
for value in reversed(hops):
|
|
peer = _address(value.strip())
|
|
if not any(peer in network for network in trusted):
|
|
return str(peer)
|
|
return "" # Only proxies, no independently identified client.
|
|
except (ValueError, TypeError):
|
|
return ""
|
|
|
|
|
|
def is_ip_allowed(ip_address: str, allowed_networks=None) -> bool:
|
|
"""Only explicitly allowed client networks; no '*' or private-IP fallback."""
|
|
try:
|
|
address = _address(ip_address)
|
|
networks = _networks(
|
|
getattr(settings, "REGISTRATION_ALLOWED_IPS", ())
|
|
if allowed_networks is None
|
|
else allowed_networks
|
|
)
|
|
excluded = _networks(getattr(settings, "REGISTRATION_DENIED_IPS", ()))
|
|
trusted = _networks(getattr(settings, "REGISTRATION_TRUSTED_PROXIES", ()))
|
|
if any(address in network for network in excluded + trusted):
|
|
return False
|
|
return any(address in network for network in networks)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
class RegistrationFromLAN(BasePermission):
|
|
"""Applied only to self-registration, before validation, mail or LDAP writes."""
|
|
|
|
message = "Регистрация доступна только из локальной сети компании."
|
|
code = "registration_lan_only"
|
|
|
|
def has_permission(self, request, view):
|
|
if not getattr(settings, "REGISTRATION_LAN_ONLY_ENABLED", False):
|
|
return True
|
|
return is_ip_allowed(get_client_ip(request))
|