mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-21 07:02:26 +02:00
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
1298 lines
34 KiB
Python
1298 lines
34 KiB
Python
import copy
|
|
import re
|
|
import json
|
|
import socket
|
|
import os
|
|
import typing
|
|
|
|
import public
|
|
from .regexplib import match_ipv4, match_ipv6, match_safe_path, match_based_host, match_email
|
|
from .exceptions import HintException
|
|
from .structures import aap_t_simple_result
|
|
|
|
|
|
class Param:
|
|
__VALIDATE_OPTS = [
|
|
'>',
|
|
'<',
|
|
'>=',
|
|
'<=',
|
|
'=',
|
|
'in',
|
|
'not in',
|
|
]
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.__nullable: bool = False
|
|
self.__validate_rules: typing.List[_ValidateRule] = []
|
|
self.__filters: typing.List[callable] = []
|
|
|
|
# 验证器 Begin ----->
|
|
|
|
def Require(self):
|
|
"""
|
|
必选参数
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_RequireValidation(self.name))
|
|
return self
|
|
|
|
def Date(self):
|
|
"""
|
|
日期字符串
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_DateValidation(self.name))
|
|
return self
|
|
|
|
def Timestamp(self):
|
|
"""
|
|
Unix时间戳
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_TimestampValidation(self.name))
|
|
return self
|
|
|
|
def Url(self):
|
|
"""
|
|
URL
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_UrlValidation(self.name))
|
|
return self
|
|
|
|
def Ip(self):
|
|
"""
|
|
IP地址
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_IpValidation(self.name))
|
|
return self
|
|
|
|
def Ipv4(self):
|
|
"""
|
|
IPv4地址
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_Ipv4Validation(self.name))
|
|
return self
|
|
|
|
def Ipv6(self):
|
|
"""
|
|
IPv6地址
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_Ipv6Validation(self.name))
|
|
return self
|
|
|
|
def Host(self):
|
|
"""
|
|
主机地址(可以包含端口号)
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_HostValidation(self.name))
|
|
return self
|
|
|
|
def Port(self):
|
|
"""
|
|
端口号
|
|
@return: self
|
|
"""
|
|
return self.Integer('between', [1, 65535])
|
|
|
|
def Json(self):
|
|
"""
|
|
JSON字符串
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_JsonValidation(self.name))
|
|
return self
|
|
|
|
def Array(self):
|
|
"""
|
|
JSON-Array字符串
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_ArrayValidation(self.name))
|
|
return self
|
|
|
|
def Object(self):
|
|
"""
|
|
JSON-Object字符串
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_ObjectValidation(self.name))
|
|
return self
|
|
|
|
def List(self):
|
|
"""
|
|
限制参数数据类型:list
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_ListValidation(self.name))
|
|
return self
|
|
|
|
def Tuple(self):
|
|
"""
|
|
限制参数数据类型:tuple
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_TupleValidation(self.name))
|
|
return self
|
|
|
|
def Dict(self):
|
|
"""
|
|
限制参数数据类型:dict
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_DictValidation(self.name))
|
|
return self
|
|
|
|
def Bool(self):
|
|
"""
|
|
布尔值或boolean字符串 true/false
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_BoolValidation(self.name))
|
|
return self
|
|
|
|
def String(self, opt: typing.Optional[str] = None, length_or_list: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
"""
|
|
字符串
|
|
@param opt: str 运算符
|
|
@param length_or_list: int|list[int|str]|None 字符串长度或字符串集合
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_StringValidation(self.name, opt, length_or_list))
|
|
return self
|
|
|
|
def Number(self, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[int, float, typing.List[typing.Union[int, float]]]] = None):
|
|
"""
|
|
数值
|
|
@param opt: str 运算符
|
|
@param num: int 数值大小
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_NumberValidation(self.name, opt, num))
|
|
return self
|
|
|
|
def Integer(self, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[typing.Union[int, typing.List[int]]]] = None):
|
|
"""
|
|
整数
|
|
@param opt: str 运算符
|
|
@param num: int 数值大小
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_IntegerValidation(self.name, opt, num))
|
|
return self
|
|
|
|
def Float(self, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[int, float, typing.List[typing.Union[int, float]]]] = None):
|
|
"""
|
|
浮点数
|
|
@param opt: str 运算符
|
|
@param num: int 数值大小
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_FloatValidation(self.name, opt, num))
|
|
return self
|
|
|
|
def Alpha(self, opt: typing.Optional[str] = None, length_or_list: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
"""
|
|
纯字母
|
|
@param opt: str 运算符
|
|
@param length_or_list: int|list[int|str]|None 字符串长度或字符串集合
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_AlphaValidation(self.name, opt, length_or_list))
|
|
return self
|
|
|
|
def Alphanum(self, opt: typing.Optional[str] = None, length_or_list: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
"""
|
|
字母+数字
|
|
@param opt: str 运算符
|
|
@param length_or_list: int|list[int|str]|None 字符串长度或字符串集合
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_AlphanumValidation(self.name, opt, length_or_list))
|
|
return self
|
|
|
|
def Mobile(self):
|
|
"""
|
|
(中国)手机号码
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_MobileValidation(self.name))
|
|
return self
|
|
|
|
def Email(self):
|
|
"""
|
|
邮箱地址
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_EmailValidation(self.name))
|
|
return self
|
|
|
|
def Regexp(self, exp: str):
|
|
"""
|
|
正则表达式
|
|
@param exp: str 正则表达式
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_RegexpValidation(self.name, exp))
|
|
return self
|
|
|
|
def File(self):
|
|
"""
|
|
文件上传
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_FileValidation(self.name))
|
|
return self
|
|
|
|
def Size(self, opt: typing.Optional[str] = None, size: typing.Optional[typing.Union[int, typing.List[int]]] = None):
|
|
"""
|
|
上传文件大小
|
|
@param opt: str 运算符
|
|
@param size: int 上传文件大小bytes
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_SizeValidation(self.name, opt, size))
|
|
return self
|
|
|
|
def Mime(self, opt: typing.Optional[str] = None, mime_type: typing.Optional[typing.Union[str, typing.List[str]]] = None):
|
|
"""
|
|
上传文件Mimetype
|
|
@param opt: str 运算符
|
|
@param mime_type: str 上传文件Mimetype
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_MimeValidation(self.name, opt, mime_type))
|
|
return self
|
|
|
|
def Ext(self, opt: typing.Optional[str] = None, ext: typing.Optional[typing.Union[str, typing.List[str]]] = None):
|
|
"""
|
|
上传文件后缀名
|
|
@param opt: str 运算符
|
|
@param ext: str 上传文件后缀名
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_ExtValidation(self.name, opt, ext))
|
|
return self
|
|
|
|
def SafePath(self, force: bool = True):
|
|
"""
|
|
文件路径
|
|
@return: self
|
|
"""
|
|
self.__validate_rules.append(_SafePathValidation(self.name, force))
|
|
return self
|
|
|
|
def Nullable(self):
|
|
self.__nullable = True
|
|
|
|
# <------- 验证器 End
|
|
|
|
# 过滤器 Begin ------>
|
|
|
|
def Trim(self):
|
|
"""
|
|
去除字符串两端空白字符
|
|
@return: self
|
|
"""
|
|
self.__filters.append(lambda x: str(x).strip())
|
|
return self
|
|
|
|
def Xss(self):
|
|
"""
|
|
XSS过滤
|
|
@return: self
|
|
"""
|
|
self.__filters.append(_xssencode)
|
|
return self
|
|
|
|
def Filter(self, f: callable):
|
|
"""
|
|
自定义参数过滤器
|
|
@param f: callable func(x: any) -> any
|
|
@return: self
|
|
"""
|
|
self.__filters.append(f)
|
|
return self
|
|
|
|
# <------- 过滤器 End
|
|
|
|
def do_validate(self, args: dict):
|
|
"""
|
|
执行验证器
|
|
@param args: dict 请求参数列表
|
|
@return: self
|
|
"""
|
|
if self.__nullable:
|
|
args = {k: args[k] for k in filter(lambda k: args[k] is not None, args.keys())}
|
|
|
|
for v in self.__validate_rules:
|
|
v.validate(args)
|
|
|
|
return self
|
|
|
|
def do_filter(self, val, extra_filters: typing.Union[typing.List[callable], typing.Tuple[callable]] = ()) -> any:
|
|
"""
|
|
执行参数过滤器
|
|
@param val: any
|
|
@param extra_filters: list[callable]|tuple[callable]
|
|
@return: any
|
|
"""
|
|
from functools import reduce
|
|
return reduce(lambda x, y: y(x) if x is not None else x, list(extra_filters) + self.__filters, val)
|
|
|
|
|
|
class _ValidateRule:
|
|
def validate(self, args: dict):
|
|
raise NotImplementedError('method validate() not implemented.')
|
|
|
|
|
|
class _RequireValidation(_ValidateRule):
|
|
"""
|
|
必选参数验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} is required'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name in args:
|
|
return
|
|
|
|
if 'FILES' in args and self.name in args['FILES']:
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _DateValidation(_ValidateRule):
|
|
"""
|
|
日期字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid datetime'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if re.match(r'^(?:\d{2}-\d{2}-\d{2}|\d{2}/\d{2}/\d{2})(?: \d{2}:\d{2}(?::\d{2})?)?$',
|
|
str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _TimestampValidation(_ValidateRule):
|
|
"""
|
|
Unix时间戳验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid timestamp'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if re.match(r'^\d{10}$', str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _UrlValidation(_ValidateRule):
|
|
"""
|
|
URL地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid URL'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
regex_obj = re.compile(
|
|
r'^(?:http|ftp)s?://'
|
|
r'(?:(?:[A-Z0-9_](?:[A-Z0-9-_]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-_]{2,}\.?)|'
|
|
r'localhost|'
|
|
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
|
|
r'(?::\d+)?'
|
|
r'(?:/?|[/?]\S+)$', re.IGNORECASE)
|
|
|
|
if regex_obj.match(str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _IpValidation(_ValidateRule):
|
|
"""
|
|
IP地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid IP'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
ipstr = str(args[self.name]).strip()
|
|
|
|
if _is_ipv4(ipstr) or _is_ipv6(ipstr):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _Ipv4Validation(_ValidateRule):
|
|
"""
|
|
IPv4地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid IPv4'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if _is_ipv4(str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _Ipv6Validation(_ValidateRule):
|
|
"""
|
|
IPv6地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid IPv4'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if _is_ipv6(str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _HostValidation(_ValidateRule):
|
|
"""
|
|
主机地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid HOST'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if match_based_host.match(str(args[self.name]).strip()):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _JsonValidation(_ValidateRule):
|
|
"""
|
|
JSON字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid JSON'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
try:
|
|
json.loads(str(args[self.name]).strip())
|
|
return
|
|
except:
|
|
pass
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _ArrayValidation(_ValidateRule):
|
|
"""
|
|
JSON-Array字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid JSON Array'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
try:
|
|
obj = json.loads(str(args[self.name]).strip())
|
|
|
|
if isinstance(obj, list):
|
|
return
|
|
except:
|
|
pass
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _ObjectValidation(_ValidateRule):
|
|
"""
|
|
JSON-Object字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid JSON Object'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
try:
|
|
obj = json.loads(str(args[self.name]).strip())
|
|
|
|
if isinstance(obj, dict):
|
|
return
|
|
except:
|
|
pass
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _BoolValidation(_ValidateRule):
|
|
"""
|
|
bool字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be bool'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
val = args[self.name]
|
|
|
|
if isinstance(val, bool) or re.match(r'^true|false$', str(args[self.name]).strip(), re.IGNORECASE):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _ListValidation(_ValidateRule):
|
|
"""
|
|
list数据类型验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be list'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if isinstance(args[self.name], list):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _TupleValidation(_ValidateRule):
|
|
"""
|
|
tuple数据类型验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be tuple'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if isinstance(args[self.name], tuple):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _DictValidation(_ValidateRule):
|
|
"""
|
|
dict数据类型验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be dict'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if isinstance(args[self.name], dict):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _OperationHelper:
|
|
"""
|
|
运算辅助类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str], operand: typing.Optional[typing.Union[str, int, float, typing.List[typing.Union[str, int, float]]]], data_type):
|
|
self.name: str = name
|
|
self.opt = opt
|
|
self.operand = operand
|
|
self.data_type = data_type
|
|
self._opt_check()
|
|
self._data_type_check()
|
|
|
|
def do(self, val: typing.Union[str, int, float], data_type=None):
|
|
val = str(val).strip()
|
|
|
|
if data_type is not None:
|
|
self.data_type = data_type
|
|
self._data_type_check()
|
|
|
|
if self.opt is None:
|
|
return
|
|
|
|
if self.operand is None:
|
|
return
|
|
|
|
if self.opt == '=':
|
|
self._eq(self._calc_num(val))
|
|
elif self.opt == '>':
|
|
self._gt(self._calc_num(val))
|
|
elif self.opt == '>=':
|
|
self._gte(self._calc_num(val))
|
|
elif self.opt == '<':
|
|
self._lt(self._calc_num(val))
|
|
elif self.opt == '<=':
|
|
self._lte(self._calc_num(val))
|
|
elif self.opt == 'between':
|
|
self._between(self._calc_num(val))
|
|
elif self.opt == 'in':
|
|
self._in(self.data_type(val))
|
|
elif self.opt == 'not in':
|
|
self._not_in(self.data_type(val))
|
|
|
|
def _calc_num(self, val: str) -> typing.Union[int, float]:
|
|
if self.data_type is str:
|
|
return len(val)
|
|
|
|
return self.data_type(val)
|
|
|
|
def _opt_check(self):
|
|
if self.opt is None:
|
|
return
|
|
|
|
self.opt = self.opt.lower()
|
|
|
|
if self.operand is None:
|
|
return
|
|
|
|
if self.opt in ['=', '>', '<', '>=', '<='] and not isinstance(self.operand, int):
|
|
raise HintException('当运算符opt是 \'{}\' 时,运算数只能是int类型或float类型,当前类型 {}'.format(self.opt, type(self.operand)))
|
|
|
|
if self.opt in ['in', 'not in', 'between'] and not isinstance(self.operand, list):
|
|
raise HintException('当运算符opt是 \'{}\' 时,运算数只能是list类型,当前类型 {}'.format(self.opt, type(self.operand)))
|
|
|
|
def _data_type_check(self):
|
|
if self.data_type is str:
|
|
return
|
|
|
|
if self.data_type is int:
|
|
return
|
|
|
|
if self.data_type is float:
|
|
return
|
|
|
|
raise HintException('data_type只能是str、int、float 当前 {}'.format(self.data_type))
|
|
|
|
def _eq(self, num: typing.Union[int, float]):
|
|
if num == self.operand:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must equal {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '', self.operand))
|
|
|
|
def _gt(self, num: typing.Union[int, float]):
|
|
if num > self.operand:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must greater than {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '',
|
|
self.operand))
|
|
|
|
def _gte(self, num: typing.Union[int, float]):
|
|
if num >= self.operand:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must greater than or equal {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '',
|
|
self.operand))
|
|
|
|
def _lt(self, num: typing.Union[int, float]):
|
|
if num < self.operand:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must less than {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '',
|
|
self.operand))
|
|
|
|
def _lte(self, num: typing.Union[int, float]):
|
|
if num <= self.operand:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must less than or equal {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '',
|
|
self.operand))
|
|
|
|
def _between(self, num: typing.Union[int, float]):
|
|
if len(self.operand) != 2:
|
|
raise HintException('当运算符opt是 \'between\' 时,运算数只能是list类型,并且list的长度只能是2,当前list长度 {}'.format(len(self.operand)))
|
|
|
|
if num >= self.operand[0] and num <= self.operand[1]:
|
|
return
|
|
|
|
raise HintException(
|
|
'{}{} must between {} and {}'.format(self.name, ' length' if isinstance(self.data_type, str) else '',
|
|
self.operand[0], self.operand[1]))
|
|
|
|
def _in(self, item: typing.Union[int, float, str]):
|
|
if len(self.operand) < 1:
|
|
raise HintException('当运算符opt是 \'{}\' 时,运算数只能是list类型,并且list的长度必须大于0,当前list长度 0'.format(self.opt))
|
|
|
|
if item in self.operand:
|
|
return
|
|
|
|
raise HintException('{} must in {}'.format(self.name, self.operand))
|
|
|
|
def _not_in(self, item: typing.Union[int, float, str]):
|
|
if len(self.operand) < 1:
|
|
raise HintException('当运算符opt是 \'{}\' 时,运算数只能是list类型,并且list的长度必须大于0,当前list长度 0'.format(self.opt))
|
|
|
|
if item in self.operand:
|
|
raise HintException('{} must not in {}'.format(self.name, self.operand))
|
|
|
|
|
|
class _StringValidation(_ValidateRule):
|
|
"""
|
|
字符串验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, v: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be string'
|
|
self.op = _OperationHelper(name, opt, v, str)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = args[self.name]
|
|
|
|
if isinstance(s, str):
|
|
self.op.do(s)
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _NumberValidation(_ValidateRule):
|
|
"""
|
|
数字验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[int, float, typing.List[typing.Union[int, float]]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be number'
|
|
self.op = _OperationHelper(name, opt, num, float)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
num = args[self.name]
|
|
|
|
if _is_number(num):
|
|
self.op.do(num, _get_number_data_type(num))
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _IntegerValidation(_ValidateRule):
|
|
"""
|
|
整数验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[int, typing.List[int]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be integer'
|
|
self.op = _OperationHelper(name, opt, num, int)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
num = args[self.name]
|
|
|
|
if _is_int(num):
|
|
self.op.do(num)
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _FloatValidation(_ValidateRule):
|
|
"""
|
|
浮点数验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, num: typing.Optional[typing.Union[float, typing.List[float]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be float'
|
|
self.op = _OperationHelper(name, opt, num, float)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
num = args[self.name]
|
|
|
|
if _is_float(num):
|
|
self.op.do(num)
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _AlphaValidation(_ValidateRule):
|
|
"""
|
|
纯字母验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, v: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be alpha'
|
|
self.op = _OperationHelper(name, opt, v, str)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = str(args[self.name]).strip()
|
|
|
|
if re.match(r'^[a-zA-Z]+$', s):
|
|
self.op.do(s)
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _AlphanumValidation(_ValidateRule):
|
|
"""
|
|
字母数字验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, v: typing.Optional[typing.Union[int, typing.List[typing.Union[int, str]]]] = None):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} must be alphanum'
|
|
self.op = _OperationHelper(name, opt, v, str)
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = str(args[self.name]).strip()
|
|
|
|
if re.match(r'^[a-zA-Z0-9]+$', s):
|
|
self.op.do(s)
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _MobileValidation(_ValidateRule):
|
|
"""
|
|
(中国)手机号码验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid mobile'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = str(args[self.name]).strip()
|
|
|
|
if re.match(r'^1[3-9]\d{9}$', s):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _EmailValidation(_ValidateRule):
|
|
"""
|
|
邮箱地址验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid email'
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = str(args[self.name]).strip()
|
|
|
|
if match_email.match(s):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _RegexpValidation(_ValidateRule):
|
|
"""
|
|
正则表达式验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, regexp: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not success verified by regexp'
|
|
self.regexp: str = regexp
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
s = str(args[self.name]).strip()
|
|
|
|
if re.match(self.regexp, s):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _FileValidation(_ValidateRule):
|
|
"""
|
|
文件上传验证类
|
|
"""
|
|
|
|
def __init__(self, name: str):
|
|
self.name: str = name
|
|
self.errmsg: str = '{} not valid file'
|
|
|
|
def validate(self, args: dict):
|
|
if 'FILES' in args and self.name in args['FILES']:
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
class _SizeValidation(_ValidateRule):
|
|
"""
|
|
文件大小验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, size: typing.Optional[typing.Union[int, typing.List[int]]] = None):
|
|
self.name: str = name
|
|
self.op = _OperationHelper(name, opt, size, int)
|
|
|
|
def validate(self, args: dict):
|
|
if 'FILES' not in args or self.name not in args['FILES']:
|
|
return
|
|
|
|
self.op.do(args['FILES'][self.name].content_length)
|
|
|
|
|
|
class _MimeValidation(_ValidateRule):
|
|
"""
|
|
文件mimetype验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, mime_type: typing.Optional[typing.Union[str, typing.List[str]]] = None):
|
|
self.name: str = name
|
|
self.op = _OperationHelper(name, opt, mime_type, str)
|
|
|
|
def validate(self, args: dict):
|
|
if 'FILES' not in args or self.name not in args['FILES']:
|
|
return
|
|
|
|
self.op.do(args['FILES'][self.name].mimetype)
|
|
|
|
|
|
class _ExtValidation(_ValidateRule):
|
|
"""
|
|
文件后缀名验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, opt: typing.Optional[str] = None, ext: typing.Optional[typing.Union[str, typing.List[str]]] = None):
|
|
self.name: str = name
|
|
self.op = _OperationHelper(name, opt, ext, str)
|
|
|
|
def validate(self, args: dict):
|
|
if 'FILES' not in args or self.name not in args['FILES']:
|
|
return
|
|
|
|
f = args['FILES'][self.name]
|
|
|
|
self.op.do(os.path.splitext(f.filename)[-1])
|
|
|
|
|
|
class _SafePathValidation(_ValidateRule):
|
|
"""
|
|
文件路径名验证类
|
|
"""
|
|
|
|
def __init__(self, name: str, force: bool):
|
|
self.name: str = name
|
|
self.errmsg = '{} not safe path'
|
|
self.force = force
|
|
|
|
def validate(self, args: dict):
|
|
if self.name not in args:
|
|
return
|
|
|
|
if _is_safe_path(str(args[self.name]).strip(), self.force):
|
|
return
|
|
|
|
raise HintException(self.errmsg.format(self.name))
|
|
|
|
|
|
def trim_filter() -> callable:
|
|
"""
|
|
获取Trim参数过滤器
|
|
@return: callable
|
|
"""
|
|
return lambda x: str(x).strip() if isinstance(x, str) else x
|
|
|
|
|
|
def xss_filter() -> callable:
|
|
"""
|
|
获取XSS参数过滤器
|
|
@return: callable
|
|
"""
|
|
return _xssencode
|
|
|
|
|
|
def _is_ipv4(ip: str) -> bool:
|
|
'''
|
|
@name 是否是IPV4地址
|
|
@author hwliang
|
|
@param ip<string> IP地址
|
|
@return True/False
|
|
'''
|
|
# 验证基本格式
|
|
if not match_ipv4.match(ip):
|
|
return False
|
|
|
|
# 验证每个段是否在合理范围
|
|
try:
|
|
socket.inet_pton(socket.AF_INET, ip)
|
|
except AttributeError:
|
|
try:
|
|
socket.inet_aton(ip)
|
|
except socket.error:
|
|
return False
|
|
except socket.error:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _is_ipv6(ip: str) -> bool:
|
|
'''
|
|
@name 是否为IPv6地址
|
|
@author hwliang
|
|
@param ip<string> 地址
|
|
@return True/False
|
|
'''
|
|
# 验证基本格式
|
|
if not match_ipv6.match(ip):
|
|
return False
|
|
|
|
# 验证IPv6地址
|
|
try:
|
|
socket.inet_pton(socket.AF_INET6, ip)
|
|
except socket.error:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _xssencode(text: str) -> str:
|
|
"""
|
|
XSS过滤
|
|
@param text: str
|
|
@return bool
|
|
"""
|
|
try:
|
|
from cgi import html
|
|
list = ['`', '~', '&', '#', '/', '*', '$', '@', '<', '>', '\"', '\'', ';', '%', ',', '.', '\\u']
|
|
ret = []
|
|
for i in text:
|
|
if i in list:
|
|
i = ''
|
|
ret.append(i)
|
|
str_convert = ''.join(ret)
|
|
text2 = html.escape(str_convert, quote=True)
|
|
return text2
|
|
except:
|
|
return text.replace('&', '&').replace('"', '"').replace('<', '<').replace('>', '>')
|
|
|
|
|
|
def _is_safe_path(path: str, force: bool=True) -> bool:
|
|
"""
|
|
文件路径过滤
|
|
@param path: str
|
|
@param force: bool
|
|
@return: bool
|
|
"""
|
|
if len(path) > 256:
|
|
return False
|
|
|
|
checks = ['..', './', '\\', '%', '$', '^', '&', '*', '~', '"', "'", ';', '|', '{', '}', '`']
|
|
|
|
for c in checks:
|
|
if path.find(c) > -1:
|
|
return False
|
|
|
|
if force:
|
|
if not match_safe_path.match(path):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def _is_number(s) -> bool:
|
|
"""
|
|
@name 判断输入参数是否一个数字
|
|
@author Zhj<2022-07-18>
|
|
@param s<string|integer|float> 输入参数
|
|
@return bool
|
|
"""
|
|
try:
|
|
float(s)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
|
|
try:
|
|
import unicodedata
|
|
unicodedata.numeric(s)
|
|
return True
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
return False
|
|
|
|
|
|
def _is_int(s) -> bool:
|
|
"""
|
|
判断输入是否是整数
|
|
@param s: any
|
|
@return bool
|
|
"""
|
|
try:
|
|
int(s)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
|
|
return False
|
|
|
|
|
|
def _is_float(s) -> bool:
|
|
"""
|
|
判断输入是否是浮点数
|
|
@param s: any
|
|
@return bool
|
|
"""
|
|
try:
|
|
float(s)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
|
|
return False
|
|
|
|
|
|
def _get_number_data_type(s):
|
|
"""
|
|
获取数字的数据类型
|
|
@param s<string> 输入参数
|
|
@return int|float
|
|
"""
|
|
try:
|
|
int(s)
|
|
return int
|
|
except ValueError:
|
|
pass
|
|
|
|
return float
|
|
|
|
|
|
# 参数验证器
|
|
class Validator:
|
|
def __init__(self, rules: typing.Union[typing.Tuple[Param], typing.List[Param]], raise_exc: bool = True):
|
|
self.__RULES = list(rules)
|
|
self.__RAISE_EXC = raise_exc
|
|
|
|
# 参数格式校验
|
|
def check(self, args: dict) -> aap_t_simple_result:
|
|
try:
|
|
for v in self.__RULES:
|
|
v.do_validate(args)
|
|
except Exception as e:
|
|
if self.__RAISE_EXC:
|
|
raise
|
|
|
|
return aap_t_simple_result(False, str(e))
|
|
|
|
return aap_t_simple_result(True, 'ok')
|
|
|
|
# 参数列表过滤
|
|
def filter(self, args: dict) -> typing.Dict:
|
|
new_args = {}
|
|
|
|
for v in self.__RULES:
|
|
if v.name not in args:
|
|
continue
|
|
|
|
new_args = v.do_filter(args[v.name])
|
|
|
|
return new_args
|