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)
207 lines
7.5 KiB
Python
207 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import logging
|
|
import re
|
|
import os
|
|
from typing import Any, List, Optional, Callable, Dict
|
|
from urllib.parse import urlparse
|
|
import ipaddress
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ValidationUtils:
|
|
"""验证工具类"""
|
|
|
|
@staticmethod
|
|
def is_valid_email(email: str) -> bool:
|
|
"""验证邮箱格式"""
|
|
try:
|
|
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
|
|
result = bool(re.match(pattern, email))
|
|
logger.debug(f"邮箱验证: {email} -> {result}")
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"验证邮箱时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def is_valid_url(url: str) -> bool:
|
|
"""验证URL格式"""
|
|
try:
|
|
result = urlparse(url)
|
|
is_valid = all([result.scheme, result.netloc])
|
|
logger.debug(f"URL验证: {url} -> {is_valid}")
|
|
return is_valid
|
|
except Exception as e:
|
|
logger.error(f"验证URL时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def is_valid_ip(ip: str) -> bool:
|
|
"""验证IP地址格式"""
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
logger.debug(f"IP地址验证: {ip} -> True")
|
|
return True
|
|
except ValueError:
|
|
logger.debug(f"IP地址验证: {ip} -> False")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"验证IP地址时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def is_valid_port(port: int) -> bool:
|
|
"""验证端口号"""
|
|
try:
|
|
is_valid = 1 <= port <= 65535
|
|
logger.debug(f"端口验证: {port} -> {is_valid}")
|
|
return is_valid
|
|
except Exception as e:
|
|
logger.error(f"验证端口时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def validate_string(value: Any, min_length: int = 0, max_length: int = None,
|
|
pattern: str = None) -> bool:
|
|
"""验证字符串"""
|
|
try:
|
|
if not isinstance(value, str):
|
|
logger.debug(f"字符串验证失败: 不是字符串类型")
|
|
return False
|
|
|
|
if len(value) < min_length:
|
|
logger.debug(f"字符串验证失败: 长度小于 {min_length}")
|
|
return False
|
|
|
|
if max_length and len(value) > max_length:
|
|
logger.debug(f"字符串验证失败: 长度大于 {max_length}")
|
|
return False
|
|
|
|
if pattern and not re.match(pattern, value):
|
|
logger.debug(f"字符串验证失败: 不匹配模式 {pattern}")
|
|
return False
|
|
|
|
logger.debug(f"字符串验证通过: 长度 {len(value)}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证字符串时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def validate_number(value: Any, min_value: float = None, max_value: float = None) -> bool:
|
|
"""验证数字"""
|
|
try:
|
|
if not isinstance(value, (int, float)):
|
|
# 尝试转换
|
|
try:
|
|
value = float(value)
|
|
except (ValueError, TypeError):
|
|
logger.debug(f"数字验证失败: 无法转换为数字")
|
|
return False
|
|
|
|
if min_value is not None and value < min_value:
|
|
logger.debug(f"数字验证失败: 值小于 {min_value}")
|
|
return False
|
|
|
|
if max_value is not None and value > max_value:
|
|
logger.debug(f"数字验证失败: 值大于 {max_value}")
|
|
return False
|
|
|
|
logger.debug(f"数字验证通过: {value}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证数字时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def validate_list(value: Any, min_length: int = 0, max_length: int = None,
|
|
item_validator: Callable = None) -> bool:
|
|
"""验证列表"""
|
|
try:
|
|
if not isinstance(value, list):
|
|
logger.debug(f"列表验证失败: 不是列表类型")
|
|
return False
|
|
|
|
if len(value) < min_length:
|
|
logger.debug(f"列表验证失败: 长度小于 {min_length}")
|
|
return False
|
|
|
|
if max_length and len(value) > max_length:
|
|
logger.debug(f"列表验证失败: 长度大于 {max_length}")
|
|
return False
|
|
|
|
if item_validator:
|
|
for i, item in enumerate(value):
|
|
if not item_validator(item):
|
|
logger.debug(f"列表验证失败: 第 {i} 项验证失败")
|
|
return False
|
|
|
|
logger.debug(f"列表验证通过: 长度 {len(value)}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证列表时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def validate_dict(value: Any, required_keys: List[str] = None,
|
|
key_validators: Dict[str, Callable] = None) -> bool:
|
|
"""验证字典"""
|
|
try:
|
|
if not isinstance(value, dict):
|
|
logger.debug(f"字典验证失败: 不是字典类型")
|
|
return False
|
|
|
|
# 检查必需键
|
|
if required_keys:
|
|
for key in required_keys:
|
|
if key not in value:
|
|
logger.debug(f"字典验证失败: 缺少必需键 {key}")
|
|
return False
|
|
|
|
# 检查键值验证器
|
|
if key_validators:
|
|
for key, validator in key_validators.items():
|
|
if key in value and not validator(value[key]):
|
|
logger.debug(f"字典验证失败: 键 {key} 的值验证失败")
|
|
return False
|
|
|
|
logger.debug(f"字典验证通过: 键数 {len(value)}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证字典时出错: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def validate_file_path(file_path: str, check_exists: bool = True,
|
|
check_readable: bool = False, check_writable: bool = False) -> bool:
|
|
"""验证文件路径"""
|
|
try:
|
|
from pathlib import Path
|
|
|
|
path = Path(file_path)
|
|
|
|
if check_exists and not path.exists():
|
|
logger.debug(f"文件路径验证失败: 文件不存在")
|
|
return False
|
|
|
|
if check_readable and not os.access(path, os.R_OK):
|
|
logger.debug(f"文件路径验证失败: 文件不可读")
|
|
return False
|
|
|
|
if check_writable and not os.access(path, os.W_OK):
|
|
logger.debug(f"文件路径验证失败: 文件不可写")
|
|
return False
|
|
|
|
logger.debug(f"文件路径验证通过: {file_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"验证文件路径时出错: {str(e)}", exc_info=True)
|
|
return False
|