AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import logging
|
|
import socket
|
|
import asyncio
|
|
import re
|
|
from typing import Optional, Tuple
|
|
import aiohttp
|
|
import ssl
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class NetworkUtils:
|
|
"""网络工具类"""
|
|
|
|
@staticmethod
|
|
async def check_port_available(host: str, port: int) -> bool:
|
|
"""检查端口是否可用"""
|
|
try:
|
|
# 尝试创建socket连接
|
|
reader, writer = await asyncio.open_connection(host, port)
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
logger.debug(f"端口 {host}:{port} 已被占用")
|
|
return False
|
|
except (ConnectionRefusedError, asyncio.TimeoutError):
|
|
logger.debug(f"端口 {host}:{port} 可用")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"检查端口可用性时出错 {host}:{port}: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
async def find_available_port(host: str = "localhost", start_port: int = 8000,
|
|
max_attempts: int = 100) -> Optional[int]:
|
|
"""查找可用端口"""
|
|
try:
|
|
for port in range(start_port, start_port + max_attempts):
|
|
if await NetworkUtils.check_port_available(host, port):
|
|
logger.debug(f"找到可用端口: {host}:{port}")
|
|
return port
|
|
logger.warning(f"在范围 {start_port}-{start_port + max_attempts} 内未找到可用端口")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"查找可用端口时出错: {str(e)}", exc_info=True)
|
|
return None
|
|
|
|
@staticmethod
|
|
async def http_request(url: str, method: str = "GET", headers: dict = None,
|
|
data: dict = None, timeout: int = 30) -> Tuple[bool, dict]:
|
|
"""发送HTTP请求"""
|
|
try:
|
|
logger.debug(f"发送HTTP请求: {method} {url}")
|
|
|
|
timeout_obj = aiohttp.ClientTimeout(total=timeout)
|
|
|
|
async with aiohttp.ClientSession(timeout=timeout_obj) as session:
|
|
async with session.request(method, url, headers=headers, json=data) as response:
|
|
response_data = await response.text()
|
|
|
|
result = {
|
|
"status": response.status,
|
|
"headers": dict(response.headers),
|
|
"data": response_data,
|
|
"url": str(response.url)
|
|
}
|
|
|
|
logger.debug(f"HTTP请求完成: {method} {url} -> 状态 {response.status}")
|
|
return True, result
|
|
|
|
except asyncio.TimeoutError:
|
|
logger.error(f"HTTP请求超时: {method} {url}")
|
|
return False, {"error": "请求超时"}
|
|
except aiohttp.ClientError as e:
|
|
logger.error(f"HTTP客户端错误: {method} {url} -> {str(e)}")
|
|
return False, {"error": str(e)}
|
|
except Exception as e:
|
|
logger.error(f"HTTP请求时出错: {method} {url} -> {str(e)}", exc_info=True)
|
|
return False, {"error": str(e)}
|
|
|
|
@staticmethod
|
|
def get_local_ip() -> str:
|
|
"""获取本地IP地址"""
|
|
try:
|
|
# 创建一个socket连接来获取本地IP
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
|
|
s.connect(("8.8.8.8", 80))
|
|
local_ip = s.getsockname()[0]
|
|
|
|
logger.debug(f"获取本地IP: {local_ip}")
|
|
return local_ip
|
|
except Exception as e:
|
|
logger.error(f"获取本地IP时出错: {str(e)}", exc_info=True)
|
|
return "127.0.0.1"
|
|
|
|
@staticmethod
|
|
def is_valid_hostname(hostname: str) -> bool:
|
|
"""验证主机名格式"""
|
|
try:
|
|
if len(hostname) > 255:
|
|
return False
|
|
|
|
if hostname[-1] == ".":
|
|
hostname = hostname[:-1]
|
|
|
|
allowed = re.compile(r"(?!-)[A-Z\d-]{1,63}(?<!-)$", re.IGNORECASE)
|
|
return all(allowed.match(x) for x in hostname.split("."))
|
|
except Exception as e:
|
|
logger.error(f"验证主机名时出错: {str(e)}", exc_info=True)
|
|
return False |