Nexus Agent
08a0157c95
调度页执行周期可视化/单次执行/分类选机/推送源对齐；登录 IP 门控与无缝导航； 服务器批量后台任务、执行记录、凭据合并、各页激活刷新与错误提示修复。 Co-authored-by: Cursor <cursoragent@cursor.com>
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""Login IP allowlist helpers (shared by auth API precheck and AuthService.login)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
|
|
from server.config import settings
|
|
|
|
_TRUTHY = frozenset({"true", "1", "yes", "on"})
|
|
|
|
|
|
def is_allowlist_enforced() -> bool:
|
|
"""Whether LOGIN_ALLOWLIST_ENABLED is on."""
|
|
val = (getattr(settings, "LOGIN_ALLOWLIST_ENABLED", "false") or "false").lower()
|
|
return val in _TRUTHY
|
|
|
|
|
|
def get_allowed_login_ips() -> list[str]:
|
|
"""Combined subscription + manual IPs from settings."""
|
|
sub_ips_raw = (getattr(settings, "LOGIN_SUBSCRIPTION_IPS", "") or "").strip()
|
|
manual_ips_raw = (getattr(settings, "LOGIN_MANUAL_IPS", "") or "").strip()
|
|
all_ips_raw = ",".join(filter(None, [sub_ips_raw, manual_ips_raw]))
|
|
return [ip.strip() for ip in all_ips_raw.split(",") if ip.strip()]
|
|
|
|
|
|
def ip_in_allowlist(client_ip: str, allowed: list[str]) -> bool:
|
|
"""Check if client_ip matches any entry in the allowlist."""
|
|
try:
|
|
client_addr = ipaddress.ip_address(client_ip)
|
|
for entry in allowed:
|
|
entry = entry.strip()
|
|
try:
|
|
if "/" in entry:
|
|
if client_addr in ipaddress.ip_network(entry, strict=False):
|
|
return True
|
|
elif client_addr == ipaddress.ip_address(entry):
|
|
return True
|
|
except ValueError:
|
|
if client_ip == entry:
|
|
return True
|
|
except ValueError:
|
|
return client_ip in allowed
|
|
return False
|
|
|
|
|
|
def check_login_ip(client_ip: str) -> tuple[bool, bool, str | None]:
|
|
"""Evaluate login IP access.
|
|
|
|
Returns:
|
|
(allowlist_enabled, allowed, message)
|
|
- allowlist off: (False, True, None)
|
|
- allowlist on, empty list: (True, True, None) — nothing to enforce
|
|
- allowlist on, IP ok: (True, True, None)
|
|
- allowlist on, IP blocked: (True, False, message)
|
|
"""
|
|
if not is_allowlist_enforced():
|
|
return False, True, None
|
|
|
|
if not client_ip or client_ip in ("", "unknown"):
|
|
return True, True, None
|
|
|
|
allowed_ips = get_allowed_login_ips()
|
|
if not allowed_ips:
|
|
return True, True, None
|
|
|
|
if ip_in_allowlist(client_ip, allowed_ips):
|
|
return True, True, None
|
|
|
|
return (
|
|
True,
|
|
False,
|
|
f"当前 IP ({client_ip}) 不在登录白名单中，请检查代理或联系管理员",
|
|
)
|
|
|
|
|
|
def is_whitelisted_login_ip(client_ip: str) -> bool:
|
|
"""True when allowlist is enforced, non-empty, and client IP matches."""
|
|
if not is_allowlist_enforced():
|
|
return False
|
|
if not client_ip or client_ip in ("", "unknown"):
|
|
return False
|
|
allowed_ips = get_allowed_login_ips()
|
|
return bool(allowed_ips) and ip_in_allowlist(client_ip, allowed_ips)
|