关于Flask接口离谱的Bug、Win上正常、linux上400?

新手上路,请多包涵

先放出原本的代码参考:

import binascii
import random
import string
import time
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from router import app

@app.route('/v3/iParse/<int:fileId>', methods=["GET"])
def iParse(fileId):
    try:
        rUrl = 'https://api.ilanzou.com/unproved/file/redirect'
        rTime = str(int(round(time.time() * 1000)))

        # 构建请求参数
        rParams = {
            "downloadId": aes_ecb_pkcs7_encrypt(f"{fileId}|", 'lanZouY-disk-app'),
            "enable": 1,
            "devType": 3,
            "uuid": generate_random_string(21),
            "timestamp": aes_ecb_pkcs7_encrypt(rTime, 'lanZouY-disk-app'),
            "auth": aes_ecb_pkcs7_encrypt(f"{fileId}|{rTime}", 'lanZouY-disk-app')
        }

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Accept-Encoding': 'gzip, deflate, br',
            'Origin': 'https://www.ilanzou.com',
            'Referer': 'https://www.ilanzou.com/',
            'Sec-Fetch-Dest': 'empty',
            'Sec-Fetch-Mode': 'cors',
            'Sec-Fetch-Site': 'same-site',
            'Connection': 'keep-alive',
            'Cookie': 'down_ip=1'
        }

        response = requests.get(
            url=rUrl,
            headers=headers,
            params=rParams,  # 通过 params 参数传递请求参数
            allow_redirects=False,
            timeout=10
        )

        if response.status_code in (200, 302):
            try:
                if response.status_code == 302:
                    redirect_url = response.headers.get('Location')
                    return {"code": 200, "status": "Parse successful", "url": redirect_url}
                else:
                    # 处理 200 状态码的JSON响应
                    json_response = response.json()
                    if 'url' in json_response:
                        return {"code": 200, "status": "Parse successful", "url": json_response['url']}
                    elif 'data' in json_response and isinstance(json_response['data'], dict):
                        if 'url' in json_response['data']:
                            return {"code": 200, "status": "Parse successful", "url": json_response['data']['url']}
            except Exception as e:
                pass

        return {"code": 400, "status": "Parse failed", "url": None}

    except Exception as e:
        return {"code": 500, "status": f"Server error: {str(e)}", "url": None}


# AES加密
def aes_ecb_pkcs7_encrypt(plaintext, key):
    key_bytes = key.encode('utf-8')
    cipher = AES.new(key_bytes, AES.MODE_ECB)
    padded_plaintext = pad(plaintext.encode('utf-8'), AES.block_size)
    encrypted_bytes = cipher.encrypt(padded_plaintext)
    encrypted_hex = binascii.hexlify(encrypted_bytes).decode('utf-8')
    return encrypted_hex.upper()


# 生成随机字符串
def generate_random_string(length):
    letters = string.ascii_letters + string.digits
    random_str = ''.join(random.choice(letters) for _ in range(length))
    return random_str

这是一个接受参数后发送请求并返回结果的接口,框架是Flask框架,现在遇到了一个非常奇葩的问题,经理一个月的探索和找Bug,依然是成空军。

问题描述:就是发送请求后这里的"timestamp": aes_ecb_pkcs7_encrypt(rTime, 'lanZouY-disk-app'),参数会变成xtamp,为什么?我搜索发现是因为ASCL码的编码导致,我尝试目前网上的各种办法,都一样,目前接口还是返回参数为空。

注意:电脑(Windows)上完全没有问题,就是同一套代码在服务器上就跑步了(我已经通过Docker测试还是一样)。服务器:Linux CentOS 7.6.1810 x86_64(Py3.7.9)

求求救救我吧。

我尝试了网上的几种办法:
1.把timestamp放到第一个参数-》无效
2.前面加&转义-》无效
3.。。。-》无效

阅读 512
avatarAI BotBETA
### 解答

这个问题很可能是由于在不同操作系统中,环境变量或者库的行为差异导致的。特别是在处理字符串和编码时,Linux 和 Windows 之间的行为可能有所不同。

**根本原因**:
在 Linux 环境中,当使用 `AES.new(key_bytes, AES.MODE_ECB)` 创建 AES 加密对象,并且 `key_bytes` 的长度不是 AES 所需的块大小(对于 AES-128 是 16 字节,AES-256 是 32 字节)的倍数时,`Crypto.Cipher.AES` 库可能会表现出不一致的行为。在 Windows 上,可能由于某种原因(比如库的版本、实现细节等),它忽略了长度不匹配的问题,但在 Linux 上则可能引发了问题,导致加密结果不正确,甚至在某些情况下字符串被截断。

**解决方案**:

1. **确保密钥长度正确**:
   确保用于 AES 加密的密钥长度是 AES 加密模式所需的块大小的倍数。对于 AES-128,密钥应该是 16 字节长;对于 AES-256,密钥应该是 32 字节长。如果密钥长度不正确,可以通过截断或填充来修正。

2. **修改密钥**:
   在你的代码中,密钥是 `'lanZouY-disk-app'`,这是一个 14 字节长的字符串。你可以通过添加一个或多个字符来使其长度达到 16 字节(对于 AES-128)。例如,你可以将密钥改为 `'lanZouY-disk-app01'`。

3. **重新测试**:
   在 Linux 环境中重新部署代码,并测试接口是否按预期工作。

4. **避免使用 ECB 模式**:
   ECB 模式(电子密码本模式)是不安全的,因为它容易受到模式分析攻击。如果可能的话,考虑使用更安全的加密模式,如 CBC(密码分组链接模式)或 GCM(伽罗瓦/计数器模式)。

5. **调试和日志**:
   在代码中添加更多的日志记录,以便在出现问题时能够更容易地跟踪和诊断。

通过实施这些更改,你应该能够解决在 Linux 环境中遇到的问题,并使你的 Flask 接口按预期工作。
1 个回答
import binascii
import random
import string
import time
import logging
import requests
from flask import Flask
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad

app = Flask(__name__)

# 设置日志级别
logging.basicConfig(level=logging.DEBUG)

# 强制设置编码环境(避免 Linux 和 Windows 环境差异)
import locale
import sys

locale.setlocale(locale.LC_ALL, 'C.UTF-8')
assert sys.getdefaultencoding() == 'utf-8', "Default encoding is not UTF-8"

@app.route('/v3/iParse/<int:fileId>', methods=["GET"])
def iParse(fileId):
    try:
        rUrl = 'https://api.ilanzou.com/unproved/file/redirect'
        rTime = str(int(round(time.time() * 1000)))

        # 构建请求参数
        rParams = {
            "downloadId": aes_ecb_pkcs7_encrypt(f"{fileId}|", 'lanZouY-disk-app'),
            "enable": 1,
            "devType": 3,
            "uuid": generate_random_string(21),
            "timestamp": aes_ecb_pkcs7_encrypt(rTime, 'lanZouY-disk-app'),
            "auth": aes_ecb_pkcs7_encrypt(f"{fileId}|{rTime}", 'lanZouY-disk-app')
        }

        # 打印调试信息
        logging.debug(f"Generated parameters: {rParams}")

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Accept-Encoding': 'gzip, deflate, br',
            'Origin': 'https://www.ilanzou.com',
            'Referer': 'https://www.ilanzou.com/',
            'Sec-Fetch-Dest': 'empty',
            'Sec-Fetch-Mode': 'cors',
            'Sec-Fetch-Site': 'same-site',
            'Connection': 'keep-alive',
            'Cookie': 'down_ip=1'
        }

        response = requests.get(
            url=rUrl,
            headers=headers,
            params=rParams,
            allow_redirects=False,
            timeout=10
        )

        logging.debug(f"Response status code: {response.status_code}")
        logging.debug(f"Response headers: {response.headers}")
        logging.debug(f"Response content: {response.text}")

        if response.status_code in (200, 302):
            try:
                if response.status_code == 302:
                    redirect_url = response.headers.get('Location')
                    return {"code": 200, "status": "Parse successful", "url": redirect_url}
                else:
                    json_response = response.json()
                    if 'url' in json_response:
                        return {"code": 200, "status": "Parse successful", "url": json_response['url']}
                    elif 'data' in json_response and isinstance(json_response['data'], dict):
                        if 'url' in json_response['data']:
                            return {"code": 200, "status": "Parse successful", "url": json_response['data']['url']}
            except Exception as e:
                logging.error(f"JSON parsing error: {e}")
                pass

        return {"code": 400, "status": "Parse failed", "url": None}

    except Exception as e:
        logging.error(f"Server error: {e}")
        return {"code": 500, "status": f"Server error: {str(e)}", "url": None}


# AES加密函数
def aes_ecb_pkcs7_encrypt(plaintext, key):
    try:
        key_bytes = key.encode('utf-8')  # 确保密钥是 UTF-8 编码
        cipher = AES.new(key_bytes, AES.MODE_ECB)
        padded_plaintext = pad(plaintext.encode('utf-8'), AES.block_size)
        encrypted_bytes = cipher.encrypt(padded_plaintext)
        encrypted_hex = binascii.hexlify(encrypted_bytes).decode('utf-8')
        logging.debug(f"Encrypting plaintext '{plaintext}' with key '{key}': {encrypted_hex.upper()}")
        return encrypted_hex.upper()
    except Exception as e:
        logging.error(f"AES encryption error: {e}")
        raise ValueError(f"Encryption failed: {e}")


# 生成随机字符串
def generate_random_string(length):
    letters = string.ascii_letters + string.digits
    random_str = ''.join(random.choice(letters) for _ in range(length))
    logging.debug(f"Generated random string: {random_str}")
    return random_str


if __name__ == "__main__":
    # 启动服务
    app.run(host="0.0.0.0", port=5000)
  1. 在Linux环境中,需要设置以下环境变量:

    export PYTHONIOENCODING=utf8
    export LANG=C.UTF-8
    export LC_ALL=C.UTF-8
  2. 如果使用Docker,在Dockerfile中添加:

    ENV PYTHONIOENCODING=utf8
    ENV LANG=C.UTF-8
    ENV LC_ALL=C.UTF-8
撰写回答
你尚未登录,登录后可以
  • 和开发者交流问题的细节
  • 关注并接收问题和回答的更新提醒
  • 参与内容的编辑和改进,让解决方法与时俱进
推荐问题
宣传栏