Secure optional LAN-only self-registration behind trusted proxies
All checks were successful
Deploy production / deploy (push) Successful in 15m8s

This commit is contained in:
admin i 2026-09-16 12:07:00 +03:00
parent 1f35009e09
commit 53cba386e9
5 changed files with 350 additions and 149 deletions

View File

@ -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

View File

@ -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(
"<h1>403 Forbidden</h1>"
"<p>Регистрация доступна только из локальной сети.</p>"
f"<p>Ваш IP: {client_ip}</p>",
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))

View File

@ -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/")

View File

@ -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.

View File

@ -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).