Codex Release Bot
140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
"""Resolve public/private IPs for server alert Telegram messages."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger("nexus.server_alert_ips")
|
||
|
||
|
||
def is_public_ip(value: str | None) -> bool:
|
||
text = (value or "").strip()
|
||
if not text:
|
||
return False
|
||
try:
|
||
addr = ipaddress.ip_address(text)
|
||
except ValueError:
|
||
return False
|
||
return not (
|
||
addr.is_private
|
||
or addr.is_loopback
|
||
or addr.is_link_local
|
||
or addr.is_reserved
|
||
)
|
||
|
||
|
||
def is_private_ip(value: str | None) -> bool:
|
||
text = (value or "").strip()
|
||
if not text:
|
||
return False
|
||
try:
|
||
addr = ipaddress.ip_address(text)
|
||
except ValueError:
|
||
return False
|
||
return addr.is_private or addr.is_loopback or addr.is_link_local
|
||
|
||
|
||
def classify_domain_ip(domain: str | None) -> tuple[str | None, str | None]:
|
||
"""Return (public_ip, private_ip) when domain is a literal IP."""
|
||
text = (domain or "").strip()
|
||
if not text:
|
||
return None, None
|
||
try:
|
||
ipaddress.ip_address(text)
|
||
except ValueError:
|
||
return None, None
|
||
if is_public_ip(text):
|
||
return text, None
|
||
if is_private_ip(text):
|
||
return None, text
|
||
return None, None
|
||
|
||
|
||
def merge_alert_ips(
|
||
*,
|
||
domain: str | None = None,
|
||
extra_attrs: dict[str, Any] | None = None,
|
||
system_info: dict[str, Any] | None = None,
|
||
) -> tuple[str | None, str | None]:
|
||
"""Pick public/private IPs for Telegram display.
|
||
|
||
外网 SSOT：servers.domain 为公网 IP 时优先（运维登记连接地址，如 120.79.11.13）。
|
||
内网：Agent 上报 private_ip 优先，否则 domain 为私网 IP 时使用。
|
||
Agent ipify 仅作 domain/extra 均无公网时的兜底。
|
||
"""
|
||
public_ip: str | None = None
|
||
private_ip: str | None = None
|
||
|
||
info = system_info if isinstance(system_info, dict) else {}
|
||
attrs = extra_attrs if isinstance(extra_attrs, dict) else {}
|
||
dom_pub, dom_priv = classify_domain_ip(domain)
|
||
|
||
if dom_pub:
|
||
public_ip = dom_pub
|
||
|
||
if not public_ip:
|
||
for key in ("public_ip", "wan_ip", "external_ip"):
|
||
raw = attrs.get(key)
|
||
if raw and is_public_ip(str(raw)):
|
||
public_ip = str(raw).strip()
|
||
break
|
||
|
||
if not public_ip:
|
||
for key in ("public_ip", "wan_ip", "external_ip"):
|
||
raw = info.get(key)
|
||
if raw and is_public_ip(str(raw)):
|
||
public_ip = str(raw).strip()
|
||
break
|
||
|
||
for key in ("private_ip", "lan_ip", "internal_ip"):
|
||
raw = info.get(key)
|
||
if raw and is_private_ip(str(raw)):
|
||
private_ip = str(raw).strip()
|
||
break
|
||
|
||
if not private_ip and dom_priv:
|
||
private_ip = dom_priv
|
||
|
||
return public_ip, private_ip
|
||
|
||
|
||
def format_alert_ip_line(public_ip: str | None, private_ip: str | None) -> str:
|
||
pub = (public_ip or "").strip() or "—"
|
||
priv = (private_ip or "").strip() or "—"
|
||
return f"IP地址：{pub}(外) {priv}(内)"
|
||
|
||
|
||
async def resolve_alert_ips_for_server(server_id: int) -> tuple[str | None, str | None]:
|
||
"""Load server row + Redis heartbeat system_info for Telegram IP lines."""
|
||
from server.infrastructure.database.server_repo import ServerRepositoryImpl
|
||
from server.infrastructure.database.session import AsyncSessionLocal
|
||
from server.infrastructure.redis.client import get_redis
|
||
|
||
domain: str | None = None
|
||
extra_attrs: dict[str, Any] | None = None
|
||
system_info: dict[str, Any] | None = None
|
||
|
||
try:
|
||
async with AsyncSessionLocal() as session:
|
||
server = await ServerRepositoryImpl(session).get_by_id(server_id)
|
||
if server:
|
||
domain = server.domain
|
||
extra_attrs = server.extra_attrs if isinstance(server.extra_attrs, dict) else None
|
||
except Exception as e:
|
||
logger.warning("resolve_alert_ips DB read failed server_id=%s: %s", server_id, e)
|
||
|
||
try:
|
||
redis = get_redis()
|
||
raw = await redis.hget(f"heartbeat:{server_id}", "system_info")
|
||
if raw:
|
||
parsed = json.loads(raw)
|
||
if isinstance(parsed, dict):
|
||
system_info = parsed
|
||
except Exception as e:
|
||
logger.debug("resolve_alert_ips Redis read failed server_id=%s: %s", server_id, e)
|
||
|
||
return merge_alert_ips(domain=domain, extra_attrs=extra_attrs, system_info=system_info)
|