Маршруты /docs и /openapi.json проверяли только адрес. С выключенным списком адресов это означало открытый доступ к описанию API из интернета. Теперь они отвечают 404, пока docs не включён явно, а включённые требуют адрес из списка и токен - заголовком либо ссылкой /docs?token=ЗНАЧЕНИЕ. Попутно исправлено сравнение токена: compare_digest на строках с не-ASCII бросает TypeError, и токен с кириллицей давал 500 вместо 401. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
"""Проверка доступа: Bearer-токен и список разрешённых адресов."""
|
||
import ipaddress
|
||
import secrets
|
||
from ipaddress import IPv4Network, IPv6Network
|
||
|
||
__all__ = ["check_token", "token_matches", "parse_allowlist", "ip_allowed",
|
||
"client_address"]
|
||
|
||
Network = IPv4Network | IPv6Network
|
||
|
||
|
||
def check_token(header_value: str | None, expected: str) -> bool:
|
||
"""Сверяет заголовок Authorization с настроенным токеном.
|
||
|
||
Пустой токен в конфиге не открывает доступ, а закрывает его полностью -
|
||
иначе забытая настройка молча выставила бы сервис наружу.
|
||
Сравнение идёт в постоянное время, чтобы токен нельзя было подобрать по таймингам.
|
||
"""
|
||
if not header_value:
|
||
return False
|
||
scheme, _, value = header_value.partition(" ")
|
||
if scheme.lower() != "bearer":
|
||
return False
|
||
return token_matches(value, expected)
|
||
|
||
|
||
def token_matches(value: str | None, expected: str) -> bool:
|
||
"""Сверяет голое значение токена, без схемы Bearer.
|
||
|
||
Нужно для ссылок вида ?token=: браузер, открывая страницу по ссылке,
|
||
заголовок Authorization не подставит.
|
||
"""
|
||
if not expected or not value:
|
||
return False
|
||
# Сравниваем байты, а не строки: compare_digest на строках с не-ASCII
|
||
# бросает TypeError, и токен с кириллицей давал бы 500 вместо 401.
|
||
return secrets.compare_digest(value.strip().encode("utf-8"),
|
||
expected.encode("utf-8"))
|
||
|
||
|
||
def parse_allowlist(raw: str) -> list[Network]:
|
||
"""Разбирает список адресов и подсетей через запятую.
|
||
|
||
Некорректные записи пропускаются: опечатка в конфиге не должна ронять сервис.
|
||
"""
|
||
nets: list[Network] = []
|
||
for item in (raw or "").split(","):
|
||
item = item.strip()
|
||
if not item:
|
||
continue
|
||
try:
|
||
nets.append(ipaddress.ip_network(item, strict=False))
|
||
except ValueError:
|
||
continue
|
||
return nets
|
||
|
||
|
||
def ip_allowed(client_ip: str | None, allowlist: list[Network]) -> bool:
|
||
"""Проверяет адрес по списку. Пустой список означает «проверка выключена»."""
|
||
if not allowlist:
|
||
return True
|
||
if not client_ip:
|
||
return False
|
||
try:
|
||
addr = ipaddress.ip_address(client_ip)
|
||
except ValueError:
|
||
return False
|
||
return any(addr in net for net in allowlist)
|
||
|
||
|
||
def client_address(peer_ip: str | None, forwarded: str | None,
|
||
trusted_proxies: list[Network]) -> str | None:
|
||
"""Возвращает адрес клиента с учётом обратного прокси.
|
||
|
||
Заголовку X-Forwarded-For верим только тогда, когда сам запрос пришёл
|
||
от доверенного прокси. Иначе любой желающий подставил бы себе разрешённый
|
||
адрес и обошёл ограничение.
|
||
"""
|
||
if not forwarded or not trusted_proxies or not ip_allowed(peer_ip, trusted_proxies):
|
||
return peer_ip
|
||
first = forwarded.split(",")[0].strip()
|
||
try:
|
||
ipaddress.ip_address(first)
|
||
except ValueError:
|
||
return peer_ip
|
||
return first
|