"""路由、白名单和转发器共用的网络基础函数。""" import functools import ipaddress import socket import threading import time def ttl_cache(seconds: int, maxsize: int = 16384): """带 TTL 的小型线程安全缓存,并保留 ``cache_clear`` 兼容接口。""" def decorator(function): store = {} lock = threading.Lock() @functools.wraps(function) def wrapper(*args, **kwargs): key = (args, tuple(sorted(kwargs.items()))) now = time.time() with lock: hit = store.get(key) if hit and now - hit[0] < seconds: return hit[1] value = function(*args, **kwargs) with lock: store[key] = (now, value) if len(store) > maxsize: cutoff = now - seconds for cached_key, (timestamp, _) in list(store.items()): if timestamp < cutoff: del store[cached_key] if len(store) > maxsize: store.clear() return value def cache_clear(): with lock: store.clear() wrapper.cache_clear = cache_clear return wrapper return decorator def is_private_ipv4_int(value: int) -> bool: """整数形式 IPv4 是否属于常见私网、回环或链路本地段。""" return ( (value & 0xFF000000) == 0x0A000000 or (value & 0xFFF00000) == 0xAC100000 or (value & 0xFFFF0000) == 0xC0A80000 or (value & 0xFF000000) == 0x7F000000 or (value & 0xFFFF0000) == 0xA9FE0000 ) def resolve_host_ips(host: str): """解析主机名的 IPv4/IPv6 集合;字面量 IP 返回自身,失败返回空集合。""" try: return {ipaddress.ip_address(host)} except ValueError: pass try: return { ipaddress.ip_address(address[4][0]) for address in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM) } except Exception: return set() def is_private_host(host: str) -> bool: """目标是否命中私网、回环或 IPv6 ULA/链路本地地址。""" for address in resolve_host_ips(host): if address.version == 4 and is_private_ipv4_int(int(address)): return True if address.version == 6 and ( address.is_loopback or address.is_link_local or address in ipaddress.ip_network("fc00::/7") ): return True return False