507 lines
18 KiB
Python
507 lines
18 KiB
Python
from tools import *
|
|
from Utils.Config import *
|
|
from Utils.Lib import *
|
|
from Utils.Secrets import get as _secret_get
|
|
|
|
|
|
APIS=flask.Blueprint("api", __name__, subdomain="api", static_folder = JoinPath(FlaskTempLate, "API"), static_url_path="/")
|
|
|
|
TOKEN = _secret_get('wecom', 'TOKEN', fallback='')
|
|
EncodingAESKey = _secret_get('wecom', 'ENCODING_AES_KEY', fallback='')
|
|
CorpID = _secret_get('wecom', 'CORP_ID', fallback='')
|
|
|
|
|
|
class WXBizMsgCrypt:
|
|
def __init__(self, sToken, sEncodingAESKey, sCorpId):
|
|
self.token = sToken
|
|
self.corpId = sCorpId
|
|
# 提取aeskey
|
|
aes_key = base64.b64decode(sEncodingAESKey + '=')
|
|
self.key = aes_key
|
|
self.mode = AES.MODE_CBC
|
|
# 设置加解密模式为AES的CBC模式
|
|
|
|
def VerifyURL(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr):
|
|
# 验证URL函数，返回解密后的明文
|
|
try:
|
|
# 计算签名
|
|
signature = self._getSHA1(self.token, sTimeStamp, sNonce, sEchoStr)
|
|
if signature != sMsgSignature:
|
|
return (40001, "签名验证错误")
|
|
|
|
# 解密
|
|
ret, sReplyEchoStr = self._decrypt(sEchoStr)
|
|
return (0, sReplyEchoStr)
|
|
except Exception as e:
|
|
return (-40003, "解密失败")
|
|
|
|
def DecryptMsg(self, sMsgSignature, sTimeStamp, sNonce, sPostData):
|
|
# 解密函数，用于解密POST请求的数据
|
|
try:
|
|
# 解析XML
|
|
xml_tree = ET.fromstring(sPostData)
|
|
encrypt = xml_tree.find("Encrypt").text
|
|
|
|
# 计算签名
|
|
signature = self._getSHA1(self.token, sTimeStamp, sNonce, encrypt)
|
|
if signature != sMsgSignature:
|
|
return (40001, "签名验证错误")
|
|
|
|
# 解密
|
|
ret, content = self._decrypt(encrypt)
|
|
return (0, content)
|
|
except Exception as e:
|
|
return (-40003, "解密失败")
|
|
|
|
def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):
|
|
# 加密函数，用于加密响应消息
|
|
if timestamp is None:
|
|
timestamp = str(int(time.time()))
|
|
|
|
# 加密
|
|
ret, encrypt = self._encrypt(sReplyMsg)
|
|
if ret != 0:
|
|
return (ret, None)
|
|
|
|
# 生成签名
|
|
signature = self._getSHA1(self.token, timestamp, sNonce, encrypt)
|
|
|
|
# 生成XML
|
|
xml = self._generate_response_xml(encrypt, signature, timestamp, sNonce)
|
|
return (0, xml)
|
|
|
|
def _getSHA1(self, token, timestamp, nonce, encrypt):
|
|
# 生成签名
|
|
try:
|
|
sortlist = [token, timestamp, nonce, encrypt]
|
|
sortlist.sort()
|
|
sha = hashlib.sha1()
|
|
sha.update(''.join(sortlist).encode())
|
|
return sha.hexdigest()
|
|
except Exception as e:
|
|
return None
|
|
|
|
def _encrypt(self, text):
|
|
# 加密消息
|
|
try:
|
|
# 生成16位随机字符串
|
|
random_str = ''.join(random.sample('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', 16))
|
|
text = text.encode('utf-8')
|
|
# 计算网络字节序
|
|
msg_len = self._length(text)
|
|
# 拼接
|
|
raw = random_str + msg_len + text + self.corpId.encode('utf-8')
|
|
# 填充
|
|
padding = len(raw) % 32
|
|
padnum = 32 - padding if padding != 0 else 32
|
|
raw = raw + bytes([padnum]) * padnum
|
|
# 加密
|
|
cipher = AES.new(self.key, self.mode, self.key[:16])
|
|
encrypted = cipher.encrypt(raw)
|
|
# 返回Base64编码
|
|
return (0, base64.b64encode(encrypted).decode('utf-8'))
|
|
except Exception as e:
|
|
return (-40003, None)
|
|
|
|
def _decrypt(self, text):
|
|
# 解密消息
|
|
try:
|
|
# Base64解码
|
|
text = base64.b64decode(text)
|
|
# 解密
|
|
cipher = AES.new(self.key, self.mode, self.key[:16])
|
|
decrypted = cipher.decrypt(text)
|
|
# 去除填充
|
|
padnum = decrypted[-1]
|
|
if padnum < 1 or padnum > 32:
|
|
padnum = 0
|
|
result = decrypted[:-padnum]
|
|
# 提取内容
|
|
content = result[16:]
|
|
msglen = int.from_bytes(content[:4], 'big')
|
|
message = content[4:4+msglen].decode('utf-8')
|
|
from_corpid = content[4+msglen:].decode('utf-8')
|
|
# 验证CorpID
|
|
if from_corpid != self.corpId:
|
|
return (-40005, None)
|
|
return (0, message)
|
|
except Exception as e:
|
|
return (-40003, None)
|
|
|
|
def _length(self, text):
|
|
# 将消息长度转换为4字节的网络字节序
|
|
l = len(text)
|
|
return l.to_bytes(4, 'big')
|
|
|
|
def _generate_response_xml(self, encrypt, signature, timestamp, nonce):
|
|
# 生成响应XML
|
|
xml = f"""<xml>
|
|
<Encrypt><![CDATA[{encrypt}]]></Encrypt>
|
|
<MsgSignature><![CDATA[{signature}]]></MsgSignature>
|
|
<TimeStamp>{timestamp}</TimeStamp>
|
|
<Nonce><![CDATA[{nonce}]]></Nonce>
|
|
</xml>"""
|
|
return xml
|
|
|
|
# 初始化加密类实例
|
|
wxcpt = WXBizMsgCrypt(TOKEN, EncodingAESKey, CorpID)
|
|
|
|
# 企业微信消息接收接口
|
|
@APIS.route('/', methods=['GET', 'POST'])
|
|
def APISIndex():
|
|
msg_signature = flask.request.args.get('msg_signature', '')
|
|
timestamp = flask.request.args.get('timestamp', '')
|
|
nonce = flask.request.args.get('nonce', '')
|
|
|
|
if not msg_signature or not timestamp or not nonce:
|
|
return flask.render_template("/API/index.html")
|
|
|
|
if flask.request.method == 'GET':
|
|
# 处理URL验证请求
|
|
echostr = flask.request.args.get('echostr', '')
|
|
ret, sEchoStr = wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
|
|
if ret != 0:
|
|
return flask.Response("URL验证失败", status=400)
|
|
return flask.Response(sEchoStr)
|
|
else:
|
|
# 处理消息接收请求
|
|
post_data = flask.request.get_data(as_text=True)
|
|
ret, message = wxcpt.DecryptMsg(msg_signature, timestamp, nonce, post_data)
|
|
if ret != 0:
|
|
return flask.Response("消息解密失败", status=400)
|
|
|
|
print("INFO: ", message)
|
|
|
|
# 解析消息内容
|
|
try:
|
|
msg_xml = ET.fromstring(message)
|
|
msg_type = msg_xml.find("MsgType").text
|
|
# 这里可以根据消息类型进行不同的处理
|
|
# 例如：文本消息、图片消息、事件等
|
|
|
|
# 示例：简单回复文本消息
|
|
if msg_type == "text":
|
|
content = msg_xml.find("Content").text
|
|
# 构建回复消息
|
|
reply_content = f"收到消息：{content}"
|
|
reply_xml = f"""<xml>
|
|
<ToUserName><![CDATA[{msg_xml.find("FromUserName").text}]]></ToUserName>
|
|
<FromUserName><![CDATA[{msg_xml.find("ToUserName").text}]]></FromUserName>
|
|
<CreateTime>{int(time.time())}</CreateTime>
|
|
<MsgType><![CDATA[text]]></MsgType>
|
|
<Content><![CDATA[{reply_content}]]></Content>
|
|
</xml>"""
|
|
|
|
# 加密回复消息
|
|
sNonce = ''.join(random.sample('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', 16))
|
|
ret, encrypted_xml = wxcpt.EncryptMsg(reply_xml, sNonce)
|
|
if ret != 0:
|
|
return flask.Response("消息加密失败", status=500)
|
|
return flask.Response(encrypted_xml)
|
|
else:
|
|
# 其他类型消息默认回复空串
|
|
return flask.Response("")
|
|
except Exception as e:
|
|
# 异常处理，返回空串表示成功接收但不回复
|
|
return flask.Response("")
|
|
|
|
|
|
|
|
|
|
@APIS.route('/ipinfo.api', methods=['GET'])
|
|
def APISIPINFOAPI():
|
|
remote_addr = flask.request.remote_addr
|
|
x_forwarded_for = flask.request.headers.get('X-Forwarded-For', None)
|
|
x_real_ip = flask.request.headers.get('X-Real-IP', None)
|
|
cf_connecting_ip = flask.request.headers.get('CF-Connecting-IP', None)
|
|
headers = dict(flask.request.headers)
|
|
request_info = {
|
|
'method': flask.request.method,
|
|
'url': flask.request.url,
|
|
'base_url': flask.request.base_url,
|
|
'full_path': flask.request.full_path,
|
|
'path': flask.request.path,
|
|
'query_string': flask.request.query_string.decode('utf-8'),
|
|
'is_secure': flask.request.is_secure,
|
|
'is_json': flask.request.is_json,
|
|
'content_type': flask.request.content_type,
|
|
'content_length': flask.request.content_length
|
|
}
|
|
ip_info = {
|
|
'ip': {
|
|
'remote_addr': remote_addr,
|
|
'x_forwarded_for': x_forwarded_for,
|
|
'x_real_ip': x_real_ip,
|
|
'cloudflare_ip': cf_connecting_ip,
|
|
'determined_ip': x_forwarded_for.split(',')[0].strip() if x_forwarded_for else remote_addr
|
|
},
|
|
'network': {
|
|
'user_agent': flask.request.user_agent.string,
|
|
'referrer': flask.request.referrer,
|
|
'language': flask.request.accept_languages.to_header(),
|
|
'encoding': flask.request.accept_encodings.to_header(),
|
|
'charset': flask.request.accept_charsets.to_header(),
|
|
'mimetypes': flask.request.accept_mimetypes.to_header()
|
|
},
|
|
'request': request_info,
|
|
'headers': headers
|
|
}
|
|
return flask.jsonify(ip_info)
|
|
|
|
@APIS.route("/mcstatus.api")
|
|
def APISMCStatusAPI():
|
|
Host = flask.request.json.get('Host') if flask.request.method == 'POST' else flask.request.args.get('Host')
|
|
Port = flask.request.json.get('Port') if flask.request.method == 'POST' else flask.request.args.get('Port')
|
|
if not Host: return flask.jsonify({"Error": "Host Required"}), 400
|
|
try:
|
|
# Try Java Edition
|
|
JavaServerInstance = JavaServer.lookup(f"{Host}:{Port}") if Port else JavaServer.lookup(Host)
|
|
ServerStatus = JavaServerInstance.status()
|
|
return flask.jsonify({
|
|
"Type": "Java",
|
|
"Host": Host,
|
|
"Port": Port or 25565,
|
|
"Version": ServerStatus.version.name,
|
|
"Players": f"{ServerStatus.players.online}/{ServerStatus.players.max}",
|
|
"Ping": ServerStatus.latency,
|
|
"Motd": str(ServerStatus.description)
|
|
})
|
|
except:
|
|
try:
|
|
# Try Bedrock Edition
|
|
BedrockPort = Port or 19132
|
|
SocketTest = socket.create_connection((Host, BedrockPort), timeout=1)
|
|
SocketTest.close()
|
|
BedrockServerInstance = BedrockServer.lookup(f"{Host}:{BedrockPort}")
|
|
BedrockStatus = BedrockServerInstance.status()
|
|
return flask.jsonify({
|
|
"Type": "Bedrock",
|
|
"Host": Host,
|
|
"Port": BedrockPort,
|
|
"Version": BedrockStatus.version.version,
|
|
"Players": f"{BedrockStatus.players_online}/{BedrockStatus.players_max}",
|
|
"Ping": BedrockStatus.latency,
|
|
"Motd": BedrockStatus.motd
|
|
})
|
|
except Exception as Error:
|
|
return flask.jsonify({"Error": str(Error)}), 404
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 扩展支持的媒体类型及其对应的MIME类型
|
|
MEDIA_EXTENSIONS = {
|
|
'.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg',
|
|
'.png': 'image/png',
|
|
'.gif': 'image/gif',
|
|
'.bmp': 'image/bmp',
|
|
'.mp4': 'video/mp4',
|
|
'.ogg': 'audio/ogg',
|
|
'.mp3': 'audio/mpeg',
|
|
'.wav': 'audio/wav'
|
|
}
|
|
def GRIFZ(ZIP_FILE_PATH: str, ZIP_PASSWORD: str, IN: str = ""):
|
|
try:
|
|
with zipfile.ZipFile(ZIP_FILE_PATH, 'r') as zf:
|
|
all_files = zf.namelist()
|
|
|
|
# 筛选媒体文件，并检查是否在指定文件夹下
|
|
media_files = []
|
|
for f in all_files:
|
|
# 检查文件扩展名是否为支持的媒体类型
|
|
if os.path.splitext(f.lower())[1] in MEDIA_EXTENSIONS:
|
|
# 如果IN不为空，则只保留指定文件夹下的文件
|
|
if IN:
|
|
# 使用os.path.normpath统一路径格式，确保跨平台兼容性
|
|
normalized_path = os.path.normpath(f)
|
|
normalized_in = os.path.normpath(IN)
|
|
|
|
# 检查文件是否在IN指定的文件夹下
|
|
if normalized_path.startswith(normalized_in):
|
|
media_files.append(f)
|
|
else:
|
|
media_files.append(f)
|
|
if not media_files:
|
|
return None, "NO SUPPORTED MEDIA FILES FOUND"
|
|
|
|
random_media = random.choice(media_files)
|
|
with zf.open(random_media, pwd=ZIP_PASSWORD.encode('utf-8')) as media_file:
|
|
media_data = io.BytesIO(media_file.read())
|
|
ext = os.path.splitext(random_media)[1].lower()
|
|
return media_data, ext
|
|
|
|
except Exception as e:
|
|
return None, f"ERROR: {str(e)}"
|
|
|
|
def ResizeImage(image_data, size=(25, 25)):
|
|
try:
|
|
img = Image.open(image_data)
|
|
if img.mode in ('RGBA', 'LA'):
|
|
background = Image.new(img.mode[:-1], img.size, (255, 255, 255))
|
|
background.paste(img, img.split()[-1])
|
|
img = background
|
|
resized_img = img.resize(size, Image.Resampling.LANCZOS)
|
|
output = io.BytesIO()
|
|
format = img.format or 'JPEG'
|
|
resized_img.save(output, format=format)
|
|
output.seek(0)
|
|
return output
|
|
except Exception as e:
|
|
return None, f"压缩失败: {str(e)}"
|
|
|
|
@APIS.route("/RandomFurryPornographyPicture.api")
|
|
def APISRandomFurryPornographyPictureAPI():
|
|
PATH=os.path.join(TempLate, "FTP/yellow.zip")
|
|
APIKEY = flask.request.args.get("APIKEY")
|
|
if APIKEY == "2525": PATH = os.path.join(TempLate, "FTP/YELLOW_RESIZE.zip")
|
|
img_data, ext = GRIFZ(PATH, APIKEY)
|
|
if img_data is None: return flask.jsonify({"Code": 1, "Status": 1, "Error": ext}), 500
|
|
mime_type = f'image/{ext[1:]}' if ext != '.jpg' else 'image/jpeg'
|
|
if APIKEY == "2525": img_data = ResizeImage(img_data)
|
|
return flask.send_file(
|
|
img_data,
|
|
mimetype=mime_type,
|
|
as_attachment=False
|
|
)
|
|
|
|
def _RandomMediaFromZip(zip_path, folder_prefix):
|
|
PATH = os.path.join(TempLate, zip_path)
|
|
APIKEY = flask.request.args.get("APIKEY")
|
|
in_folder = flask.request.args.get("IN", "")
|
|
data, ext = GRIFZ(PATH, APIKEY, folder_prefix + in_folder)
|
|
if data is None:
|
|
return flask.jsonify({"Code": 1, "Status": 1, "Error": ext}), 500
|
|
mime_type = MEDIA_EXTENSIONS.get(ext, 'application/octet-stream')
|
|
data.seek(0)
|
|
if ext == '.mp4':
|
|
file_size = data.getbuffer().nbytes
|
|
range_header = flask.request.headers.get('Range', None)
|
|
if range_header:
|
|
start, end = range_header.replace('bytes=', '').split('-')
|
|
start = int(start)
|
|
end = int(end) if end else file_size - 1
|
|
data.seek(start)
|
|
chunk = data.read(end - start + 1)
|
|
response = flask.Response(
|
|
chunk,
|
|
status=206,
|
|
mimetype=mime_type,
|
|
headers={
|
|
'Content-Range': f'bytes {start}-{end}/{file_size}',
|
|
'Accept-Ranges': 'bytes',
|
|
'Content-Length': str(end - start + 1),
|
|
'Content-Disposition': 'inline; filename="random.mp4"'
|
|
}
|
|
)
|
|
else:
|
|
response = flask.Response(
|
|
flask.stream_with_context(data),
|
|
mimetype=mime_type,
|
|
headers={
|
|
'Accept-Ranges': 'bytes',
|
|
'Content-Length': str(file_size),
|
|
'Content-Disposition': 'inline; filename="random.mp4"'
|
|
}
|
|
)
|
|
return response
|
|
else:
|
|
return flask.send_file(
|
|
data,
|
|
mimetype=mime_type,
|
|
as_attachment=False,
|
|
download_name=f"random{ext}"
|
|
)
|
|
|
|
@APIS.route("/RandomEMSTC.api")
|
|
def APISRandomEMSTCAPI():
|
|
return _RandomMediaFromZip("FTP/EMSTC.zip", "EMSTC-main/")
|
|
|
|
@APIS.route("/RandomFurryMEME.api")
|
|
def APISRandomFurryMEMEAPI():
|
|
return _RandomMediaFromZip("FTP/Furry.zip", "Furry/")
|
|
|
|
@APIS.route("/RandomProgrammerMEME.api")
|
|
def APISRandomProgrammerMEMEAPI():
|
|
return _RandomMediaFromZip("FTP/Programmer.zip", "Programmer/")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@APIS.route("/Asset/Download/<sha1>")
|
|
def APIAssetDownload(sha1):
|
|
asset = GERPFileAsset.query.filter(GERPFileAsset.FileSHA1 == sha1).first()
|
|
if not asset:
|
|
return {"code": 404, "msg": "文件不存在"}, 404
|
|
|
|
RESOURCE_ROOT = os.path.join(os.path.abspath(os.path.join(os.getcwd(), "resource")))
|
|
file_path = os.path.join(RESOURCE_ROOT, sha1[0:2], sha1[2:4], sha1[4:6], sha1)
|
|
if not os.path.exists(file_path):
|
|
return {"code": 404, "msg": "文件存储不存在"}, 404
|
|
|
|
download = flask.request.args.get("download", "0") == "1"
|
|
if download:
|
|
return flask.send_file(file_path, mimetype=asset.MimeType or "application/octet-stream", as_attachment=True, download_name=asset.FileName)
|
|
|
|
response = flask.send_file(file_path, mimetype=asset.MimeType or "application/octet-stream")
|
|
response.headers["Cache-Control"] = "public, max-age=31536000"
|
|
response.headers["ETag"] = sha1
|
|
return response
|
|
|
|
|
|
|
|
|
|
|
|
@APIS.errorhandler(404)
|
|
def APISNotFound(e):
|
|
return "API 404", 404
|
|
@APIS.errorhandler(500)
|
|
def APISInternalServerError(e):
|
|
return "API 500", 500
|
|
@APIS.errorhandler(503)
|
|
def APISServiceTemporarilyUnavailable(e):
|
|
return "API 503", 503
|
|
@APIS.errorhandler(flask_wtf.csrf.CSRFError)
|
|
def APISCSRError(e):
|
|
return "CSRF 503", 503
|