diff --git a/backend/api/auth/endpoints/registration.py b/backend/api/auth/endpoints/registration.py index de6239c1..8e1a48b2 100644 --- a/backend/api/auth/endpoints/registration.py +++ b/backend/api/auth/endpoints/registration.py @@ -22,6 +22,7 @@ from api.v1.employees.serializers import ( from api.v1.employees.views._helpers import Employee from api.v1.employees.views.mixins import LdapUserCreationMixin from common.emails import send_templated_mail +from common.ip_restrictions import RegistrationFromLAN from employees.models import Position, Skill from employees.utils import _normalize_phone @@ -31,6 +32,7 @@ logger = logging.getLogger(__name__) class RegisterAPIView(LdapUserCreationMixin, AnonymousAPIView): + permission_classes = [RegistrationFromLAN] throttle_scope = "registration" parser_classes = (JSONParser, FormParser, MultiPartParser) @@ -212,6 +214,9 @@ class RegisterAPIView(LdapUserCreationMixin, AnonymousAPIView): 400: OpenApiResponse( description="Ошибка валидации или занятый email/телефон." ), + 403: OpenApiResponse( + description="Регистрация разрешена только из локальной сети компании." + ), }, ) @transaction.atomic diff --git a/backend/common/ip_restrictions.py b/backend/common/ip_restrictions.py index 23b4a64f..b81fa240 100644 --- a/backend/common/ip_restrictions.py +++ b/backend/common/ip_restrictions.py @@ -1,154 +1,81 @@ -""" -Ограничение доступа к определенным функциям по IP-адресу. -""" +"""Fail-closed registration policy for the CORP LAN and its trusted proxies.""" import ipaddress -from functools import wraps -from typing import List from django.conf import settings -from django.http import HttpResponse +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: - """ - Получает реальный IP-адрес клиента из request. - Учитывает заголовки прокси (X-Forwarded-For, X-Real-IP). - """ - x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") - if x_forwarded_for: - # Берем первый IP из списка (клиентский IP) - ip = x_forwarded_for.split(",")[0].strip() - else: - ip = request.META.get("REMOTE_ADDR", "") - return ip + """Walk XFF from the socket peer backwards, stopping at the first untrusted IP. - -def is_local_ip(ip_address: str) -> bool: - """ - Проверяет, является ли IP-адрес локальным. - - Локальными считаются: - - 127.0.0.0/8 (localhost) - - 10.0.0.0/8 (частная сеть класса A) - - 172.16.0.0/12 (частная сеть класса B) - - 192.168.0.0/16 (частная сеть класса C) - - ::1 (IPv6 localhost) - - fe80::/10 (IPv6 link-local) + 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: - ip = ipaddress.ip_address(ip_address) - return ip.is_private or ip.is_loopback or ip.is_link_local - except ValueError: - # Невалидный IP - считаем небезопасным - return False + 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: List[str] = None) -> bool: - """ - Проверяет, разрешен ли доступ с данного IP-адреса. - - Args: - ip_address: IP-адрес для проверки - allowed_networks: Список разрешенных сетей в формате CIDR - или отдельных IP. Если None, используется - REGISTRATION_ALLOWED_IPS из settings. - - Returns: - True если IP разрешен, False иначе - """ - if allowed_networks is None: - # Проверяем настройки Django - allowed_networks = getattr(settings, "REGISTRATION_ALLOWED_IPS", None) - - # Если настройка не задана, разрешаем только локальные IP - if allowed_networks is None: - return is_local_ip(ip_address) - - # Пустой список = запретить всем - if not allowed_networks: - return False - - # Специальное значение для разрешения всех IP - if allowed_networks == ["*"] or "*" in allowed_networks: - return True - +def is_ip_allowed(ip_address: str, allowed_networks=None) -> bool: + """Only explicitly allowed client networks; no '*' or private-IP fallback.""" try: - ip = ipaddress.ip_address(ip_address) - - for network_str in allowed_networks: - # Проверяем, является ли это сетью (содержит /) или отдельным IP - if "/" in network_str: - network = ipaddress.ip_network(network_str, strict=False) - if ip in network: - return True - else: - # Отдельный IP-адрес - allowed_ip = ipaddress.ip_address(network_str) - if ip == allowed_ip: - return True - - return False - except ValueError: - # Невалидный IP - запрещаем + 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 -def local_ip_required(view_func): - """ - Декоратор для ограничения доступа к view только с локальных IP. +class RegistrationFromLAN(BasePermission): + """Applied only to self-registration, before validation, mail or LDAP writes.""" - Использование: - @local_ip_required - def my_view(request): - ... - """ + message = "Регистрация доступна только из локальной сети компании." + code = "registration_lan_only" - @wraps(view_func) - def wrapper(request, *args, **kwargs): - client_ip = get_client_ip(request) - - if not is_ip_allowed(client_ip): - return HttpResponse( - "

403 Forbidden

" - "

Регистрация доступна только из локальной сети.

" - f"

Ваш IP: {client_ip}

", - status=403, - content_type="text/html; charset=utf-8", - ) - - return view_func(request, *args, **kwargs) - - return wrapper - - -def local_ip_required_api(view_class): - """ - Декоратор для ограничения доступа к API View только с локальных IP. - - Использование: - @local_ip_required_api - class MyAPIView(APIView): - ... - """ - original_dispatch = view_class.dispatch - - def dispatch_wrapper(self, request, *args, **kwargs): - from rest_framework import status as drf_status - from rest_framework.response import Response - - client_ip = get_client_ip(request) - - if not is_ip_allowed(client_ip): - return Response( - { - "detail": "Регистрация доступна только из локальной сети.", - "client_ip": client_ip, - }, - status=drf_status.HTTP_403_FORBIDDEN, - ) - - return original_dispatch(self, request, *args, **kwargs) - - view_class.dispatch = dispatch_wrapper - return view_class + 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)) diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index 74b8808a..2feba9f4 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -237,19 +237,16 @@ PHONENUMBER_DEFAULT_REGION = "RU" # ----------------------------------------------------------------------------- # IP ОГРАНИЧЕНИЯ ДЛЯ РЕГИСТРАЦИИ # ----------------------------------------------------------------------------- -# Список разрешенных IP-адресов или сетей для регистрации (формат CIDR) -# Примеры: -# None - разрешены только локальные IP (по умолчанию) -# ['*'] - разрешены все IP -# ['192.168.1.0/24', '10.0.0.0/8'] - конкретные сети -# ['192.168.1.100', '192.168.1.101'] - конкретные IP -REGISTRATION_ALLOWED_IPS = [ - "127.0.0.0/8", # localhost - "10.0.0.0/8", # приватная сеть класса A - "172.16.0.0/12", # приватная сеть класса B (172.16-31.x.x) - "172.11.0.0/16", # ваша корпоративная сеть - "192.168.0.0/16", # приватная сеть класса C -] +REGISTRATION_LAN_ONLY_ENABLED = ( + os.getenv("REGISTRATION_LAN_ONLY_ENABLED", "false").strip().lower() == "true" +) +# Only the confirmed CORP LAN, not every private network or all of 172.11/16. +REGISTRATION_ALLOWED_IPS = ["172.11.0.0/23"] +# Client -> Synology -> local nginx -> loopback-only ASGI server. +# These peers must append the actual upstream IP to X-Forwarded-For. +REGISTRATION_TRUSTED_PROXIES = ["127.0.0.1/32", "::1/128", "172.11.1.254/32"] +# Do not mistake a NAT gateway (or network/broadcast address) for a LAN user. +REGISTRATION_DENIED_IPS = ["172.11.0.1/32", "172.11.0.0/32", "172.11.1.255/32"] # Медиа файлы MEDIA_ROOT = os.path.join(BASE_DIR, "media/") diff --git a/backend/tests/api/auth/test_registration_network.py b/backend/tests/api/auth/test_registration_network.py new file mode 100644 index 00000000..b49c83c6 --- /dev/null +++ b/backend/tests/api/auth/test_registration_network.py @@ -0,0 +1,215 @@ +import ast +import os +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from django.core import mail +from django.urls import reverse +from rest_framework.test import APIClient + +from common.ip_restrictions import RegistrationFromLAN, get_client_ip, is_ip_allowed +from employees.models import Employee +from tests.api.auth.test_auth_and_registration import register_payload + + +@pytest.fixture(autouse=True) +def registration_policy(settings): + settings.REGISTRATION_LAN_ONLY_ENABLED = True + settings.REGISTRATION_ALLOWED_IPS = ["172.11.0.0/23"] + settings.REGISTRATION_TRUSTED_PROXIES = [ + "127.0.0.1/32", + "::1/128", + "172.11.1.254/32", + ] + settings.REGISTRATION_DENIED_IPS = [ + "172.11.0.1/32", + "172.11.0.0/32", + "172.11.1.255/32", + ] + settings.LDAP_ENABLED = False + settings.LDAP_WRITE_ENABLED = False + + +@pytest.mark.parametrize( + "peer,forwarded,expected", + [ + ("172.11.0.100", "", "172.11.0.100"), + ("203.0.113.7", "172.11.0.100", "203.0.113.7"), + ("127.0.0.1", "172.11.1.128, 172.11.1.254", "172.11.1.128"), + ("172.11.1.254", "172.11.1.128, 172.11.1.254", "172.11.1.128"), + ("127.0.0.1", "172.11.0.100, 203.0.113.7, 172.11.1.254", "203.0.113.7"), + ("127.0.0.1", "bad, 203.0.113.7, 172.11.1.254", "203.0.113.7"), + ("::1", "::ffff:172.11.1.128, 172.11.1.254", "172.11.1.128"), + ("::ffff:203.0.113.7", "172.11.0.100", "203.0.113.7"), + ("127.0.0.1", "2001:db8::7, 172.11.1.254", "2001:db8::7"), + ("127.0.0.1", "172.11.0.100, 172.11.0.1", "172.11.0.1"), + ("127.0.0.1", "", ""), + ("172.11.1.254", "", ""), + ("127.0.0.1", "172.11.1.254, ::1", ""), + ("127.0.0.1", "172.11.0.100, bad, 172.11.1.254", ""), + ("127.0.0.1", "172.11.0.100, , 172.11.1.254", ""), + ("127.0.0.1", "172.11.0.100:1234", ""), + ("127.0.0.1", "fe80::1%eth0", ""), + ("127.0.0.1", "[::1]", ""), + ("127.0.0.1", "172.11.0.100," + "127.0.0.1," * 16, ""), + ("127.0.0.1", " " * 2049 + "172.11.0.100", ""), + ("", "172.11.0.100", ""), + ("bad", "172.11.0.100", ""), + ], +) +def test_trusted_proxy_chain(peer, forwarded, expected): + request = SimpleNamespace( + META={ + "REMOTE_ADDR": peer, + "HTTP_X_FORWARDED_FOR": forwarded, + "HTTP_X_REAL_IP": "172.11.0.100", + "HTTP_FORWARDED": "for=172.11.0.100", + "HTTP_X_ROBOTAIL_CLIENT_IP": "172.11.0.100", + } + ) + assert get_client_ip(request) == expected + + +@pytest.mark.parametrize( + "address", ["172.11.0.2", "172.11.0.255", "172.11.1.0", "172.11.1.253"] +) +def test_confirmed_lan_allowed(address): + assert is_ip_allowed(address) + + +@pytest.mark.parametrize( + "address", + [ + "172.11.2.1", + "172.11.255.1", + "10.0.0.1", + "192.168.0.1", + "172.16.0.1", + "127.0.0.1", + "::1", + "172.11.0.1", + "172.11.1.254", + "172.11.0.0", + "172.11.1.255", + "203.0.113.7", + "2001:db8::7", + "", + "not-an-ip", + ], +) +def test_other_networks_and_infrastructure_denied(address): + assert not is_ip_allowed(address) + + +@pytest.mark.parametrize( + "networks", [None, [], ["*"], ["bad"], ["172.11.0.0/23", "bad"], "172.11.0.0/23"] +) +def test_invalid_allowlist_fails_closed(settings, networks): + settings.REGISTRATION_ALLOWED_IPS = networks + assert not is_ip_allowed("172.11.0.100") + + +@pytest.mark.parametrize( + "value,expected", + [(None, False), ("", False), ("false", False), ("true", True), (" TRUE ", True)], +) +def test_env_switch_defaults_off(monkeypatch, value, expected): + # Evaluate the actual setting expression without reloading Django or .env. + source = Path(__file__).parents[3] / "eusrr_backend" / "settings.py" + tree = ast.parse(source.read_text(encoding="utf-8")) + setting = next( + node + for node in tree.body + if isinstance(node, ast.Assign) + and any( + isinstance(t, ast.Name) and t.id == "REGISTRATION_LAN_ONLY_ENABLED" + for t in node.targets + ) + ) + if value is None: + monkeypatch.delenv("REGISTRATION_LAN_ONLY_ENABLED", raising=False) + else: + monkeypatch.setenv("REGISTRATION_LAN_ONLY_ENABLED", value) + assert ( + eval(compile(ast.Expression(setting.value), str(source), "eval"), {"os": os}) + is expected + ) + + +@pytest.mark.parametrize("route", ["api:register", "api:v1:register"]) +@pytest.mark.parametrize( + "peer,forwarded", + [ + ("203.0.113.7", "172.11.0.100"), + ("127.0.0.1", "172.11.0.100, 203.0.113.7, 172.11.1.254"), + ("172.11.1.254", ""), + ("127.0.0.1", "172.11.0.100, 172.11.0.1, 172.11.1.254"), + ], +) +@pytest.mark.django_db +def test_external_registration_denied_before_side_effects(route, peer, forwarded): + url = reverse(route) + before = Employee.objects.count() + with ( + patch("api.auth.endpoints.registration.RegisterSerializer") as serializer, + patch( + "api.auth.endpoints.registration.RegisterAPIView.create_ldap_user" + ) as ldap, + ): + response = APIClient().post( + url, + register_payload(), + format="json", + REMOTE_ADDR=peer, + HTTP_X_FORWARDED_FOR=forwarded, + HTTP_X_REAL_IP="172.11.0.100", + ) + assert response.status_code == 403 + assert response.data["detail"].code == "registration_lan_only" + serializer.assert_not_called() + ldap.assert_not_called() + assert Employee.objects.count() == before + assert not mail.outbox + + +@pytest.mark.parametrize("route", ["api:register", "api:v1:register"]) +@pytest.mark.parametrize( + "enabled,client", [(True, "172.11.1.128"), (False, "203.0.113.7")] +) +@pytest.mark.django_db +def test_registration_succeeds_in_each_mode(settings, route, enabled, client): + settings.REGISTRATION_LAN_ONLY_ENABLED = enabled + response = APIClient().post( + reverse(route), + register_payload(), + format="json", + REMOTE_ADDR="127.0.0.1", + HTTP_X_FORWARDED_FOR=f"{client}, 172.11.1.254", + ) + assert response.status_code == 201 + assert Employee.objects.filter(email="ivan@example.com").exists() + assert len(mail.outbox) == 1 + + +def test_missing_flag_disables_permission(settings): + del settings.REGISTRATION_LAN_ONLY_ENABLED + assert RegistrationFromLAN().has_permission(SimpleNamespace(META={}), None) + + +@pytest.mark.parametrize( + "route", + [ + "api:token_obtain_pair", + "api:password-reset", + "api:verify-email", + "api:resend-email", + ], +) +@pytest.mark.django_db +def test_other_auth_endpoints_remain_available_externally(route): + response = APIClient(REMOTE_ADDR="203.0.113.7").post( + reverse(route), {}, format="json" + ) + assert response.status_code == 400 # Normal validation, not the LAN restriction. diff --git a/docs/registration-network-policy.md b/docs/registration-network-policy.md new file mode 100644 index 00000000..b1fab664 --- /dev/null +++ b/docs/registration-network-policy.md @@ -0,0 +1,57 @@ +# Регистрация только из локальной сети + +Переключатель в **серверном** backend `.env` (не frontend и не секрет runner): + +```dotenv +REGISTRATION_LAN_ONLY_ENABLED=true +``` + +Если переменной нет, она пустая или равна `false`, ограничение **выключено**. +Включает его только `true` (регистр и пробелы по краям не важны). После изменения +`.env` нужен перезапуск процессов Django или штатный деплой: настройки читаются +при старте. Сам `.env` в Git не добавляется. + +При включении оба адреса `/api/auth/register/` и `/api/v1/auth/register/` +принимают регистрацию только из `172.11.0.0/23`. Снаружи возвращается HTTP 403 +с понятным сообщением до валидации, создания пользователя, отправки письма и LDAP. +Вход, OIDC, восстановление пароля и подтверждение уже отправленного письма этим +ограничением не затрагиваются. Страница формы остаётся доступной; защиту +обеспечивает API, а не скрытие кнопки. + +## Адрес клиента и границы доверия + +Рабочий маршрут: клиент → Synology `172.11.1.254` → nginx CORP → ASGI на loopback. +Оба прокси должны дописывать настоящий IP своего непосредственного клиента +в `X-Forwarded-For`. На CORP это `$proxy_add_x_forwarded_for`. + +Проверка идёт справа налево, пропуская только явно доверенные прокси +(`127.0.0.1`, `::1`, `172.11.1.254`), до первого недоверенного адреса. +Подставленный клиентом префикс не влияет на решение. Другие заголовки адреса +не используются. Если запрос пришёл от доверенного прокси, но адрес клиента +отсутствует, повреждён или состоит только из адресов прокси, доступ запрещён. +Router `172.11.0.1`, прокси и network/broadcast адреса не считаются пользователями. +`172.11.0.0/16` и произвольные частные сети не разрешены. + +Нельзя открывать backend-порты наружу или добавлять всю LAN в доверенные прокси. +Если пограничный NAT скрывает внешний адрес за другим LAN-IP, сначала исправить +передачу исходного адреса/добавить этот шлюз в исключения: по одному IP приложение +не отличит такой запрос от локального. Список прокси не является списком клиентов. + +16 сентября 2026 проверены реальные заголовки через Synology: + +- обычный LAN-запрос: `172.11.1.128, 172.11.1.254`; +- запрос с подставленным XFF: `172.11.0.123, 172.11.1.128, 172.11.1.254`. + +## Проверка + +`pytest tests/api/auth/test_registration_network.py` проверяет оба режима, +обе версии API, успешную регистрацию, отсутствие побочных эффектов при отказе, +подмену заголовков, IPv6, повреждённые цепочки и доступность остальных auth API. +В production проверять без создания учёток: POST пустого JSON `{}` из LAN +должен вернуть 400 (валидация), извне при включённом флаге — 403. + +Отключение: `REGISTRATION_LAN_ONLY_ENABLED=false` и перезапуск Django/деплой. +Это снова разрешит регистрацию извне; другие настройки авторизации не меняются. + +Основа реализации: [DRF permissions](https://www.django-rest-framework.org/api-guide/permissions/) +и [nginx proxy_add_x_forwarded_for](https://nginx.org/en/docs/http/ngx_http_proxy_module.html#var_proxy_add_x_forwarded_for).