mirror of
https://github.com/xiaoyaDev/xiaoya-alist.git
synced 2026-10-16 21:50:48 +08:00
feat: encapsulation decryption code
This commit is contained in:
DDSRem
@@ -13,54 +13,79 @@ from Crypto.Util.Padding import unpad
|
||||
|
||||
|
||||
app = Flask(__name__)
|
||||
headers = {
|
||||
"token": "6733b42e28cdba32",
|
||||
'User-Agent': 'Mozilla/5.0 (Linux; U; Android 9; zh-cn; SM-S908E Build/TP1A.220624.014) AppleWebKit/533.1 (KHTML, like Gecko) Mobile Safari/533.1', # noqa: E501
|
||||
'Host': 'api.extscreen.com'
|
||||
}
|
||||
|
||||
|
||||
def h(char_array, modifier):
|
||||
unique_chars = list(dict.fromkeys(char_array))
|
||||
numeric_modifier = int(modifier[7:])
|
||||
transformed_string = "".join([chr(abs(ord(c) - (numeric_modifier % 127) - 1) + 33 if abs(ord(c) - (numeric_modifier % 127) - 1) < 33 else # noqa: E501
|
||||
abs(ord(c) - (numeric_modifier % 127) - 1)) for c in unique_chars])
|
||||
return transformed_string
|
||||
class AliyunPanTvToken:
|
||||
"""
|
||||
阿里云盘 TV Token 解密刷新模块
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.timestamp = str(requests.get("http://api.extscreen.com/timestamp", timeout=10).json()["data"]["timestamp"])
|
||||
self.unique_id = uuid.uuid4().hex
|
||||
self.wifimac = str(random.randint(10**11, 10**12 - 1))
|
||||
self.headers = {
|
||||
"token": "6733b42e28cdba32",
|
||||
"User-Agent": "Mozilla/5.0 (Linux; U; Android 9; zh-cn; SM-S908E Build/TP1A.220624.014) AppleWebKit/533.1 (KHTML, like Gecko) Mobile Safari/533.1", # noqa: E501
|
||||
"Host": "api.extscreen.com",
|
||||
}
|
||||
|
||||
def decrypt(ciphertext, iv, t, unique_id, wifimac):
|
||||
try:
|
||||
key = generate_key(t, unique_id, wifimac)
|
||||
cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv = bytes.fromhex(iv))
|
||||
decrypted = unpad(cipher.decrypt(base64.b64decode(ciphertext)), AES.block_size)
|
||||
dec = decrypted.decode('utf-8')
|
||||
return dec
|
||||
except Exception as error:
|
||||
print("Decryption failed", error)
|
||||
raise error
|
||||
def h(self, char_array, modifier):
|
||||
unique_chars = list(dict.fromkeys(char_array))
|
||||
numeric_modifier = int(modifier[7:])
|
||||
transformed_string = "".join(
|
||||
[
|
||||
chr(
|
||||
abs(ord(c) - (numeric_modifier % 127) - 1) + 33
|
||||
if abs(ord(c) - (numeric_modifier % 127) - 1) < 33
|
||||
# noqa: E501
|
||||
else abs(ord(c) - (numeric_modifier % 127) - 1)
|
||||
)
|
||||
for c in unique_chars
|
||||
]
|
||||
)
|
||||
return transformed_string
|
||||
|
||||
def get_params(self):
|
||||
params = {
|
||||
"akv": "2.8.1496",
|
||||
"apv": "1.3.8",
|
||||
"b": "samsung",
|
||||
"d": self.unique_id,
|
||||
"m": "SM-S908E",
|
||||
"mac": "",
|
||||
"n": "SM-S908E",
|
||||
"t": self.timestamp,
|
||||
"wifiMac": self.wifimac,
|
||||
}
|
||||
return params
|
||||
|
||||
def get_params(t, unique_id, wifimac):
|
||||
params = {
|
||||
'akv': '2.8.1496',
|
||||
'apv': '1.3.8',
|
||||
'b': 'samsung',
|
||||
'd': unique_id,
|
||||
'm': 'SM-S908E',
|
||||
'mac': '',
|
||||
'n': 'SM-S908E',
|
||||
't': t,
|
||||
'wifiMac': wifimac,
|
||||
}
|
||||
return params
|
||||
def generate_key(self):
|
||||
params = self.get_params()
|
||||
sorted_keys = sorted(params.keys())
|
||||
concatenated_params = "".join([params[key] for key in sorted_keys if key != "t"])
|
||||
hashed_key = self.h(list(concatenated_params), self.timestamp)
|
||||
return hashlib.md5(hashed_key.encode("utf-8")).hexdigest()
|
||||
|
||||
def decrypt(self, ciphertext, iv):
|
||||
try:
|
||||
key = self.generate_key()
|
||||
cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC, iv=bytes.fromhex(iv))
|
||||
decrypted = unpad(cipher.decrypt(base64.b64decode(ciphertext)), AES.block_size)
|
||||
dec = decrypted.decode("utf-8")
|
||||
return dec
|
||||
except Exception as error:
|
||||
print("Decryption failed %s", error)
|
||||
raise error
|
||||
|
||||
def generate_key(t, unique_id, wifimac):
|
||||
params = get_params(t, unique_id, wifimac)
|
||||
sorted_keys = sorted(params.keys())
|
||||
concatenated_params = "".join([params[key] for key in sorted_keys if key != "t"])
|
||||
hashed_key = h(list(concatenated_params), t)
|
||||
return hashlib.md5(hashed_key.encode('utf-8')).hexdigest()
|
||||
def get_headers(self):
|
||||
return {**self.get_params(), **self.headers}
|
||||
|
||||
def get_token(self, data):
|
||||
token_data = requests.post(
|
||||
"http://api.extscreen.com/aliyundrive/v3/token", data=data, headers=self.get_headers(), timeout=10
|
||||
).json()["data"]
|
||||
return self.decrypt(token_data["ciphertext"], token_data["iv"])
|
||||
|
||||
|
||||
@app.route('/oauth/alipan/token', methods=['POST'])
|
||||
@@ -69,26 +94,11 @@ def oauth_token():
|
||||
refresh_token = data.get('refresh_token', None)
|
||||
if not refresh_token:
|
||||
return Response(json.dumps({"error": "No refresh_token provided"}), status=400, mimetype='application/json')
|
||||
|
||||
req_body = {
|
||||
"refresh_token": refresh_token
|
||||
}
|
||||
|
||||
timestamp = str(requests.get('http://api.extscreen.com/timestamp', timeout=10).json()['data']['timestamp'])
|
||||
unique_id = uuid.uuid4().hex
|
||||
wifimac = str(random.randint(10**11, 10**12 - 1))
|
||||
|
||||
resp = requests.post("http://api.extscreen.com/aliyundrive/v3/token", data=req_body, headers={**get_params(timestamp, unique_id, wifimac), **headers}, timeout=10) # noqa: E501
|
||||
if resp.status_code == 200:
|
||||
resp_data = resp.json()
|
||||
ciphertext = resp_data["data"]["ciphertext"]
|
||||
iv = resp_data["data"]["iv"]
|
||||
|
||||
token_data = decrypt(ciphertext, iv, timestamp, unique_id, wifimac)
|
||||
token = json.loads(token_data)
|
||||
return Response(json.dumps(token), status=200, mimetype='application/json')
|
||||
else:
|
||||
return Response(resp.content, status=resp.status_code, mimetype='application/json')
|
||||
client = AliyunPanTvToken()
|
||||
return Response(json.dumps(json.loads(client.get_token(req_body))), status=200, mimetype='application/json')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -23,146 +23,167 @@ from Crypto.Util.Padding import unpad
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
app = Flask(__name__)
|
||||
timestamp = str(requests.get('http://api.extscreen.com/timestamp', timeout=10).json()['data']['timestamp'])
|
||||
unique_id = uuid.uuid4().hex
|
||||
wifimac = str(random.randint(10**11, 10**12 - 1))
|
||||
headers = {
|
||||
"token": "6733b42e28cdba32",
|
||||
'User-Agent': 'Mozilla/5.0 (Linux; U; Android 9; zh-cn; SM-S908E Build/TP1A.220624.014) AppleWebKit/533.1 (KHTML, like Gecko) Mobile Safari/533.1', # noqa: E501
|
||||
'Host': 'api.extscreen.com'
|
||||
}
|
||||
|
||||
|
||||
def h(char_array, modifier):
|
||||
unique_chars = list(dict.fromkeys(char_array))
|
||||
numeric_modifier = int(modifier[7:])
|
||||
transformed_string = "".join([chr(abs(ord(c) - (numeric_modifier % 127) - 1) + 33 if abs(ord(c) - (numeric_modifier % 127) - 1) < 33 else # noqa: E501
|
||||
abs(ord(c) - (numeric_modifier % 127) - 1)) for c in unique_chars])
|
||||
return transformed_string
|
||||
class AliyunPanTvToken:
|
||||
"""
|
||||
阿里云盘 TV Token 解密刷新模块
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.timestamp = str(requests.get("http://api.extscreen.com/timestamp", timeout=10).json()["data"]["timestamp"])
|
||||
self.unique_id = uuid.uuid4().hex
|
||||
self.wifimac = str(random.randint(10**11, 10**12 - 1))
|
||||
self.headers = {
|
||||
"token": "6733b42e28cdba32",
|
||||
"User-Agent": "Mozilla/5.0 (Linux; U; Android 9; zh-cn; SM-S908E Build/TP1A.220624.014) AppleWebKit/533.1 (KHTML, like Gecko) Mobile Safari/533.1", # noqa: E501
|
||||
"Host": "api.extscreen.com",
|
||||
}
|
||||
|
||||
def h(self, char_array, modifier):
|
||||
unique_chars = list(dict.fromkeys(char_array))
|
||||
numeric_modifier = int(modifier[7:])
|
||||
transformed_string = "".join(
|
||||
[
|
||||
chr(
|
||||
abs(ord(c) - (numeric_modifier % 127) - 1) + 33
|
||||
if abs(ord(c) - (numeric_modifier % 127) - 1) < 33
|
||||
# noqa: E501
|
||||
else abs(ord(c) - (numeric_modifier % 127) - 1)
|
||||
)
|
||||
for c in unique_chars
|
||||
]
|
||||
)
|
||||
return transformed_string
|
||||
|
||||
def get_params(self):
|
||||
params = {
|
||||
"akv": "2.8.1496",
|
||||
"apv": "1.3.8",
|
||||
"b": "samsung",
|
||||
"d": self.unique_id,
|
||||
"m": "SM-S908E",
|
||||
"mac": "",
|
||||
"n": "SM-S908E",
|
||||
"t": self.timestamp,
|
||||
"wifiMac": self.wifimac,
|
||||
}
|
||||
return params
|
||||
|
||||
def generate_key(self):
|
||||
params = self.get_params()
|
||||
sorted_keys = sorted(params.keys())
|
||||
concatenated_params = "".join([params[key] for key in sorted_keys if key != "t"])
|
||||
hashed_key = self.h(list(concatenated_params), self.timestamp)
|
||||
return hashlib.md5(hashed_key.encode("utf-8")).hexdigest()
|
||||
|
||||
def decrypt(self, ciphertext, iv):
|
||||
try:
|
||||
key = self.generate_key()
|
||||
cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC, iv=bytes.fromhex(iv))
|
||||
decrypted = unpad(cipher.decrypt(base64.b64decode(ciphertext)), AES.block_size)
|
||||
dec = decrypted.decode("utf-8")
|
||||
return dec
|
||||
except Exception as error:
|
||||
logging.error("Decryption failed %s", error)
|
||||
raise error
|
||||
|
||||
def get_headers(self):
|
||||
return {**self.get_params(), **self.headers}
|
||||
|
||||
def get_token(self, data):
|
||||
token_data = requests.post(
|
||||
"http://api.extscreen.com/aliyundrive/v3/token", data=data, headers=self.get_headers(), timeout=10
|
||||
).json()["data"]
|
||||
return self.decrypt(token_data["ciphertext"], token_data["iv"])
|
||||
|
||||
def get_qrcode_url(self):
|
||||
data = requests.post(
|
||||
"http://api.extscreen.com/aliyundrive/qrcode",
|
||||
data={
|
||||
"scopes": ",".join(["user:base", "file:all:read", "file:all:write"]),
|
||||
"width": 500,
|
||||
"height": 500,
|
||||
},
|
||||
headers=self.get_headers(),
|
||||
timeout=10,
|
||||
).json()["data"]
|
||||
qr_link = "https://www.aliyundrive.com/o/oauth/authorize?sid=" + data["sid"]
|
||||
return {"qr_link": qr_link, "sid": data["sid"]}
|
||||
|
||||
|
||||
def decrypt(ciphertext, iv, t, _unique_id, _wifimac):
|
||||
try:
|
||||
key = generate_key(t, _unique_id, _wifimac)
|
||||
cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv = bytes.fromhex(iv))
|
||||
decrypted = unpad(cipher.decrypt(base64.b64decode(ciphertext)), AES.block_size)
|
||||
dec = decrypted.decode('utf-8')
|
||||
return dec
|
||||
except Exception as error:
|
||||
logging.error("Decryption failed %s", error)
|
||||
raise error
|
||||
|
||||
|
||||
def get_params(t, _unique_id, _wifimac):
|
||||
params = {
|
||||
'akv': '2.8.1496',
|
||||
'apv': '1.3.8',
|
||||
'b': 'samsung',
|
||||
'd': _unique_id,
|
||||
'm': 'SM-S908E',
|
||||
'mac': '',
|
||||
'n': 'SM-S908E',
|
||||
't': t,
|
||||
'wifiMac': _wifimac,
|
||||
}
|
||||
return params
|
||||
|
||||
|
||||
def generate_key(t, _unique_id, _wifimac):
|
||||
params = get_params(t, _unique_id, _wifimac)
|
||||
sorted_keys = sorted(params.keys())
|
||||
concatenated_params = "".join([params[key] for key in sorted_keys if key != "t"])
|
||||
hashed_key = h(list(concatenated_params), t)
|
||||
return hashlib.md5(hashed_key.encode('utf-8')).hexdigest()
|
||||
|
||||
|
||||
def get_qrcode_url():
|
||||
data = requests.post('http://api.extscreen.com/aliyundrive/qrcode', data={
|
||||
'scopes': ','.join(["user:base", "file:all:read", "file:all:write"]),
|
||||
"width": 500,
|
||||
"height": 500,
|
||||
}, headers={**get_params(timestamp, unique_id, wifimac), **headers}, timeout=10).json()['data']
|
||||
_qr_link = "https://www.aliyundrive.com/o/oauth/authorize?sid=" + data['sid']
|
||||
return {'qr_link': _qr_link, 'sid': data['sid']}
|
||||
|
||||
|
||||
def check_qrcode_status(_sid):
|
||||
status = 'NotLoggedIn'
|
||||
def check_qrcode_status(sid):
|
||||
status = "NotLoggedIn"
|
||||
_auth_code = None
|
||||
while status != 'LoginSuccess':
|
||||
while status != "LoginSuccess":
|
||||
time.sleep(3)
|
||||
result = requests.get(f'https://openapi.alipan.com/oauth/qrcode/{_sid}/status', timeout=10).json()
|
||||
status = result['status']
|
||||
if status == 'LoginSuccess':
|
||||
_auth_code = result['authCode']
|
||||
return {'auth_code': _auth_code}
|
||||
result = requests.get(f"https://openapi.alipan.com/oauth/qrcode/{sid}/status", timeout=10).json()
|
||||
status = result["status"]
|
||||
if status == "LoginSuccess":
|
||||
_auth_code = result["authCode"]
|
||||
return {"auth_code": _auth_code}
|
||||
|
||||
|
||||
def get_token(code):
|
||||
token_data = requests.post('http://api.extscreen.com/aliyundrive/v3/token', data={
|
||||
'code': code,
|
||||
}, headers={**get_params(timestamp, unique_id, wifimac), **headers}, timeout=10).json()['data']
|
||||
ciphertext = token_data['ciphertext']
|
||||
iv = token_data['iv']
|
||||
token_data = decrypt(ciphertext, iv, timestamp, unique_id, wifimac)
|
||||
data = { "code": code }
|
||||
token_data = CLIENT.get_token(data)
|
||||
parsed_json = json.loads(token_data)
|
||||
refresh_token = parsed_json['refresh_token']
|
||||
if sys.platform.startswith('win32'):
|
||||
refresh_token = parsed_json["refresh_token"]
|
||||
if sys.platform.startswith("win32"):
|
||||
file_path = ""
|
||||
else:
|
||||
file_path = "/data/"
|
||||
with open(f"{file_path}myopentoken.txt", "w", encoding='utf-8') as file:
|
||||
with open(f"{file_path}myopentoken.txt", "w", encoding="utf-8") as file:
|
||||
file.write(refresh_token)
|
||||
logging.info('myopentoken.txt 文件更新成功！')
|
||||
with open(f"{file_path}open_tv_token_url.txt", "w", encoding='utf-8') as file:
|
||||
logging.info("myopentoken.txt 文件更新成功！")
|
||||
with open(f"{file_path}open_tv_token_url.txt", "w", encoding="utf-8") as file:
|
||||
file.write("https://alipan-tv-token.pages.dev/refresh")
|
||||
logging.info('open_tv_token_url.txt 文件更新成功！')
|
||||
logging.info("open_tv_token_url.txt 文件更新成功！")
|
||||
|
||||
|
||||
@app.route('/')
|
||||
@app.route("/")
|
||||
def main_page():
|
||||
return render_template('qrcode.html')
|
||||
return render_template("qrcode.html")
|
||||
|
||||
|
||||
@app.route('/get_qrcode', methods=['GET'])
|
||||
@app.route("/get_qrcode", methods=["GET"])
|
||||
def get_qrcode():
|
||||
return jsonify(get_qrcode_url())
|
||||
return jsonify(CLIENT.get_qrcode_url())
|
||||
|
||||
|
||||
@app.route('/check_qrcode/<sid>', methods=['GET'])
|
||||
@app.route("/check_qrcode/<sid>", methods=["GET"])
|
||||
def check_qrcode(sid):
|
||||
return jsonify(check_qrcode_status(sid))
|
||||
|
||||
|
||||
@app.route('/get_tokens', methods=['POST'])
|
||||
@app.route("/get_tokens", methods=["POST"])
|
||||
def get_tokens():
|
||||
_auth_code = request.json.get('auth_code')
|
||||
_auth_code = request.json.get("auth_code")
|
||||
get_token(_auth_code)
|
||||
return jsonify({'status': 'completed'})
|
||||
return jsonify({"status": "completed"})
|
||||
|
||||
|
||||
@app.route('/shutdown_server', methods=['GET'])
|
||||
@app.route("/shutdown_server", methods=["GET"])
|
||||
def shutdown():
|
||||
os._exit(0)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = argparse.ArgumentParser(description='AliyunPan TV Token')
|
||||
parser.add_argument('--qrcode_mode', type=str, required=True, help='扫码模式')
|
||||
CLIENT = AliyunPanTvToken()
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="AliyunPan TV Token")
|
||||
parser.add_argument("--qrcode_mode", type=str, required=True, help="扫码模式")
|
||||
args = parser.parse_args()
|
||||
if args.qrcode_mode == 'web':
|
||||
app.run(host='0.0.0.0', port=34256)
|
||||
elif args.qrcode_mode == 'shell':
|
||||
date = get_qrcode_url()
|
||||
qr_link = date['qr_link']
|
||||
_sid = date['sid']
|
||||
if args.qrcode_mode == "web":
|
||||
app.run(host="0.0.0.0", port=34256)
|
||||
elif args.qrcode_mode == "shell":
|
||||
date = CLIENT.get_qrcode_url()
|
||||
qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_H, box_size=5, border=4)
|
||||
qr.add_data(qr_link)
|
||||
qr.add_data(date["qr_link"])
|
||||
qr.make(fit=True)
|
||||
logging.info('请打开阿里云盘扫描此二维码！')
|
||||
logging.info("请打开阿里云盘扫描此二维码！")
|
||||
qr.print_ascii(invert=True, tty=sys.stdout.isatty())
|
||||
auth_code = check_qrcode_status(_sid)['auth_code']
|
||||
auth_code = check_qrcode_status(date["sid"])["auth_code"]
|
||||
get_token(auth_code)
|
||||
else:
|
||||
logging.error('未知的扫码模式')
|
||||
logging.error("未知的扫码模式")
|
||||
os._exit(1)
|
||||
|
||||
Reference in New Issue
Block a user