From 98b7cb080bccf186231d949ba9ebb73bf873a9c6 Mon Sep 17 00:00:00 2001 From: Armor00 <2654988228@qq.com> Date: Tue, 14 Jan 2025 15:12:43 +0800 Subject: [PATCH] update --- run_service_backend.py | 6 +- src/SKland_Auto_Sign/SecuritySm.py | 317 ++++++++++++++++++++++ src/SKland_Auto_Sign/auto_sign.py | 189 ------------- src/SKland_Auto_Sign/new_auto_sign.py | 375 ++++++++++++++++++++++++++ src/ark_helper_backend/backend.py | 36 +-- src/util/mail.py | 3 +- src/util/utils.py | 13 + 7 files changed, 723 insertions(+), 216 deletions(-) create mode 100644 src/SKland_Auto_Sign/SecuritySm.py delete mode 100644 src/SKland_Auto_Sign/auto_sign.py create mode 100644 src/SKland_Auto_Sign/new_auto_sign.py diff --git a/run_service_backend.py b/run_service_backend.py index eee5d93..2d9ae71 100644 --- a/run_service_backend.py +++ b/run_service_backend.py @@ -11,6 +11,7 @@ from flask_cors import CORS from src.My_Logger.project_logger import LoggerClass from src.ark_helper_backend.backend import ArkHelperBackend from src.util.mail import send_batch_email +from src.util.utils import random_time app = Flask(__name__) CORS(app, supports_credentials=True) @@ -114,7 +115,7 @@ def skland_sign_in_thread(): break time.sleep(20) now = datetime.datetime.now() - if now.strftime('%H:%M') == '08:00' and last_run != now.date(): + if now.strftime('%H:%M') == random_time(480, 720) and last_run != now.date(): ark_back.skland_sign_new() last_run = now.date() except Exception as e: @@ -126,10 +127,11 @@ def skland_sign_in_thread(): if __name__ == '__main__': + send_batch_email(['2654988228@qq.com'], 'test mail', 'test111') print('start') logger = LoggerClass().get_log(logging.INFO) ark_back = ArkHelperBackend(logger=logger) - # threading.Thread(target=skland_sign_in_thread).start() + threading.Thread(target=skland_sign_in_thread).start() threading.Thread(target=data_retrieve_thread).start() app.run(host="192.168.195.194", port=59001) diff --git a/src/SKland_Auto_Sign/SecuritySm.py b/src/SKland_Auto_Sign/SecuritySm.py new file mode 100644 index 0000000..8c45df8 --- /dev/null +++ b/src/SKland_Auto_Sign/SecuritySm.py @@ -0,0 +1,317 @@ +import base64 +import gzip +import hashlib +# 数美加密方法类 +import json +import time +import uuid + +import requests +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.ciphers.algorithms import AES +from cryptography.hazmat.decrepit.ciphers.algorithms import TripleDES +from cryptography.hazmat.primitives.ciphers.base import Cipher +from cryptography.hazmat.primitives.ciphers.modes import CBC, ECB + +# 查询dId请求头 +devices_info_url = "https://fp-it.portal101.cn/deviceprofile/v4" + +# 数美配置 +SM_CONFIG = { + "organization": "UWXspnCCJN4sfYlNfqps", + "appId": "default", + "publicKey": "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCmxMNr7n8ZeT0tE1R9j/mPixoinPkeM+k4VGIn/s0k7N5rJAfnZ0eMER+QhwFvshzo0LNmeUkpR8uIlU/GEVr8mN28sKmwd2gpygqj0ePnBmOW4v0ZVwbSYK+izkhVFk2V/doLoMbWy6b+UnA8mkjvg0iYWRByfRsK2gdl7llqCwIDAQAB", + "protocol": "https", + "apiHost": "fp-it.portal101.cn" +} + +PK = serialization.load_der_public_key(base64.b64decode(SM_CONFIG['publicKey'])) + +DES_RULE = { + "appId": { + "cipher": "DES", + "is_encrypt": 1, + "key": "uy7mzc4h", + "obfuscated_name": "xx" + }, + "box": { + "is_encrypt": 0, + "obfuscated_name": "jf" + }, + "canvas": { + "cipher": "DES", + "is_encrypt": 1, + "key": "snrn887t", + "obfuscated_name": "yk" + }, + "clientSize": { + "cipher": "DES", + "is_encrypt": 1, + "key": "cpmjjgsu", + "obfuscated_name": "zx" + }, + "organization": { + "cipher": "DES", + "is_encrypt": 1, + "key": "78moqjfc", + "obfuscated_name": "dp" + }, + "os": { + "cipher": "DES", + "is_encrypt": 1, + "key": "je6vk6t4", + "obfuscated_name": "pj" + }, + "platform": { + "cipher": "DES", + "is_encrypt": 1, + "key": "pakxhcd2", + "obfuscated_name": "gm" + }, + "plugins": { + "cipher": "DES", + "is_encrypt": 1, + "key": "v51m3pzl", + "obfuscated_name": "kq" + }, + "pmf": { + "cipher": "DES", + "is_encrypt": 1, + "key": "2mdeslu3", + "obfuscated_name": "vw" + }, + "protocol": { + "is_encrypt": 0, + "obfuscated_name": "protocol" + }, + "referer": { + "cipher": "DES", + "is_encrypt": 1, + "key": "y7bmrjlc", + "obfuscated_name": "ab" + }, + "res": { + "cipher": "DES", + "is_encrypt": 1, + "key": "whxqm2a7", + "obfuscated_name": "hf" + }, + "rtype": { + "cipher": "DES", + "is_encrypt": 1, + "key": "x8o2h2bl", + "obfuscated_name": "lo" + }, + "sdkver": { + "cipher": "DES", + "is_encrypt": 1, + "key": "9q3dcxp2", + "obfuscated_name": "sc" + }, + "status": { + "cipher": "DES", + "is_encrypt": 1, + "key": "2jbrxxw4", + "obfuscated_name": "an" + }, + "subVersion": { + "cipher": "DES", + "is_encrypt": 1, + "key": "eo3i2puh", + "obfuscated_name": "ns" + }, + "svm": { + "cipher": "DES", + "is_encrypt": 1, + "key": "fzj3kaeh", + "obfuscated_name": "qr" + }, + "time": { + "cipher": "DES", + "is_encrypt": 1, + "key": "q2t3odsk", + "obfuscated_name": "nb" + }, + "timezone": { + "cipher": "DES", + "is_encrypt": 1, + "key": "1uv05lj5", + "obfuscated_name": "as" + }, + "tn": { + "cipher": "DES", + "is_encrypt": 1, + "key": "x9nzj1bp", + "obfuscated_name": "py" + }, + "trees": { + "cipher": "DES", + "is_encrypt": 1, + "key": "acfs0xo4", + "obfuscated_name": "pi" + }, + "ua": { + "cipher": "DES", + "is_encrypt": 1, + "key": "k92crp1t", + "obfuscated_name": "bj" + }, + "url": { + "cipher": "DES", + "is_encrypt": 1, + "key": "y95hjkoo", + "obfuscated_name": "cf" + }, + "version": { + "is_encrypt": 0, + "obfuscated_name": "version" + }, + "vpw": { + "cipher": "DES", + "is_encrypt": 1, + "key": "r9924ab5", + "obfuscated_name": "ca" + } +} + +BROWSER_ENV = { + 'plugins': 'MicrosoftEdgePDFPluginPortableDocumentFormatinternal-pdf-viewer1,MicrosoftEdgePDFViewermhjfbmdgcfjbbpaeojofohoefgiehjai1', + 'ua': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0', + 'canvas': '259ffe69', # 基于浏览器的canvas获得的值，不知道复用行不行 + 'timezone': -480, # 时区，应该是固定值吧 + 'platform': 'Win32', + 'url': 'https://www.skland.com/', # 固定值 + 'referer': '', + 'res': '1920_1080_24_1.25', # 屏幕宽度_高度_色深_window.devicePixelRatio + 'clientSize': '0_0_1080_1920_1920_1080_1920_1080', + 'status': '0011', # 不知道在干啥 +} + + +# // 将浏览器环境对象的key全部排序，然后对其所有的值及其子对象的值加入数字并字符串相加。若值为数字，则乘以10000(0x2710)再将其转成字符串存入数组,最后再做md5,存入tn变量（tn变量要做加密） +# //把这个对象用加密规则进行加密，然后对结果做GZIP压缩（结果是对象，应该有序列化），最后做AES加密（加密细节目前不清除），密钥为变量priId +# //加密规则：新对象的key使用相对应加解密规则的obfuscated_name值，value为字符串化后进行进行DES加密，再进行btoa加密 + +# 通过测试 +def _DES(o: dict): + result = {} + for i in o.keys(): + if i in DES_RULE.keys(): + rule = DES_RULE[i] + res = o[i] + if rule['is_encrypt'] == 1: + c = Cipher(TripleDES(rule['key'].encode('utf-8')), ECB()) + data = str(res).encode('utf-8') + # 补足字节 + data += b'\x00' * 8 + res = base64.b64encode(c.encryptor().update(data)).decode('utf-8') + result[rule['obfuscated_name']] = res + else: + result[i] = o[i] + return result + + +# 通过测试 +def _AES(v: bytes, k: bytes): + iv = '0102030405060708' + key = AES(k) + c = Cipher(key, CBC(iv.encode('utf-8'))) + c.encryptor() + # 填充明文 + v += b'\x00' + while len(v) % 16 != 0: + v += b'\x00' + return c.encryptor().update(v).hex() + + +def GZIP(o: dict): + # 这个压缩结果似乎和前台不太一样,不清楚是否会影响 + json_str = json.dumps(o, ensure_ascii=False) + stream = gzip.compress(json_str.encode('utf-8'), 2, mtime=0) + return base64.b64encode(stream) + + +# 获得tn的值,后续做DES加密用 +# 通过测试 +def get_tn(o: dict): + sorted_keys = sorted(o.keys()) + + result_list = [] + + for i in sorted_keys: + v = o[i] + if isinstance(v, (int, float)): + v = str(v * 10000) + elif isinstance(v, dict): + v = get_tn(v) + result_list.append(v) + return ''.join(result_list) + + +def get_smid(): + t = time.localtime() + _time = '{}{:0>2d}{:0>2d}{:0>2d}{:0>2d}{:0>2d}'.format(t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, + t.tm_sec) + uid = str(uuid.uuid4()) + v = _time + hashlib.md5(uid.encode('utf-8')).hexdigest() + '00' + smsk_web = hashlib.md5(('smsk_web_' + v).encode('utf-8')).hexdigest()[0:14] + return v + smsk_web + '0' + + +def get_d_id(): + # storageName = '.thumbcache_' + md5(SM_CONFIG['organization']) // 用于从本地存储获得值 + # uid = uuid() + # priId=md5(uid)[0:16] + # ep=rsa(uid,publicKey) + # SMID = localStorage.get(storageName);// 获得本地存储存的值 + # _0x30b2eb为递归md5 + + uid = str(uuid.uuid4()).encode('utf-8') + priId = hashlib.md5(uid).hexdigest()[0:16] + # ep不一定对，先走走看 + ep = PK.encrypt(uid, padding.PKCS1v15()) + ep = base64.b64encode(ep).decode('utf-8') + + browser = BROWSER_ENV.copy() + current_time = int(time.time() * 1000) + browser.update({ + 'vpw': str(uuid.uuid4()), + 'svm': current_time, + 'trees': str(uuid.uuid4()), + 'pmf': current_time + }) + + des_target = { + **browser, + 'protocol': 102, + 'organization': SM_CONFIG['organization'], + 'appId': SM_CONFIG['appId'], + 'os': 'web', + 'version': '3.0.0', + 'sdkver': '3.0.0', + 'box': '', # 似乎是个SMID，但是第一次的时候是空,不过不影响结果 + 'rtype': 'all', + 'smid': get_smid(), + 'subVersion': '1.0.0', + 'time': 0 + } + des_target['tn'] = hashlib.md5(get_tn(des_target).encode()).hexdigest() + + des_result = _AES(GZIP(_DES(des_target)), priId.encode('utf-8')) + + response = requests.post(devices_info_url, json={ + 'appId': 'default', + 'compress': 2, + 'data': des_result, + 'encode': 5, + 'ep': ep, + 'organization': SM_CONFIG['organization'], + 'os': 'web' # 固定值 + }) + + resp = response.json() + if resp['code'] != 1100: + raise Exception("did计算失败，请联系作者") + # 开头必须是B + return 'B' + resp['detail']['deviceId'] \ No newline at end of file diff --git a/src/SKland_Auto_Sign/auto_sign.py b/src/SKland_Auto_Sign/auto_sign.py deleted file mode 100644 index ee95335..0000000 --- a/src/SKland_Auto_Sign/auto_sign.py +++ /dev/null @@ -1,189 +0,0 @@ -import hashlib -import hmac -import json -import os.path -import time -import logging -import requests - -from urllib import parse - -app_code = '4ca99fa6b56cc2ba' -token_env = os.environ.get('TOKEN') -header = { - 'cred': '', - 'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0', - 'Accept-Encoding': 'gzip', - 'Connection': 'close' -} -header_login = { - 'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0', - 'Accept-Encoding': 'gzip', - 'Connection': 'close' -} - -# 签名请求头一定要这个顺序，否则失败 -# timestamp是必填的,其它三个随便填,不要为none即可 -header_for_sign = { - 'platform': '', - 'timestamp': '', - 'dId': '', - 'vName': '' -} - -# 签到url -sign_url = "https://zonai.skland.com/api/v1/game/attendance" -# 绑定的角色url -binding_url = "https://zonai.skland.com/api/v1/game/player/binding" -# 使用token获得认证代码 -grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant" -# 使用认证代码获得cred -cred_code_url = "https://zonai.skland.com/api/v1/user/auth/generate_cred_by_code" - - -def generate_signature(token: str, path, body_or_query): - """ - 获得签名头 - 接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON() - 将此字符串做HMAC加密，算法为SHA-256，密钥token为请求cred接口会返回的一个token值 - 再将加密后的字符串做MD5即得到sign - :param token: 拿cred时候的token - :param path: 请求路径（不包括网址） - :param body_or_query: 如果是GET，则是它的query。POST则为它的body - :return: 计算完毕的sign - """ - # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2 - t = str(int(time.time()) - 2) - token = token.encode('utf-8') - header_ca = json.loads(json.dumps(header_for_sign)) - header_ca['timestamp'] = t - header_ca_str = json.dumps(header_ca, separators=(',', ':')) - s = path + body_or_query + t + header_ca_str - hex_s = hmac.new(token, s.encode('utf-8'), hashlib.sha256).hexdigest() - md5 = hashlib.md5(hex_s.encode('utf-8')).hexdigest().encode('utf-8').decode('utf-8') - return md5, header_ca - - -def get_sign_header(url: str, method, body, old_header, local_s_token): - h = json.loads(json.dumps(old_header)) - p = parse.urlparse(url) - if method.lower() == 'get': - h['sign'], header_ca = generate_signature(local_s_token, p.path, p.query) - else: - h['sign'], header_ca = generate_signature(local_s_token, p.path, json.dumps(body)) - for i in header_ca: - h[i] = header_ca[i] - return h - - -def get_cred_by_token(token): - grant_code = get_grant_code(token) - return get_cred(grant_code) - - -def get_grant_code(token): - response = requests.post(grant_code_url, json={ - 'appCode': app_code, - 'token': token, - 'type': 0 - }, headers=header_login) - resp = response.json() - if response.status_code != 200: - raise Exception(f'获得认证代码失败：{resp}') - if resp.get('status') != 0: - raise Exception(f'获得认证代码失败：{resp["msg"]}') - return resp['data']['code'] - - -def get_cred(grant): - cred_code_url = 'https://zonai.skland.com/api/v1/user/auth/generate_cred_by_code' - resp = requests.post(cred_code_url, json={ - 'code': grant, - 'kind': 1 - }, headers=header_login) - resp = requests.post(cred_code_url, json={ - 'code': grant, - 'kind': 1 - }, headers=header_login).json() - if resp['code'] != 0: - raise Exception(f'获得cred失败：{resp["message"]}') - return resp['data'] - - -def get_binding_list(s_token): - error_message, error_code = 'Success', 0 - result_list = [] - header_dict = get_sign_header(binding_url, 'get', None, header, s_token) - resp = requests.get(binding_url, headers=header_dict).json() - - if resp['code'] != 0: - error_message = f"请求角色列表出现问题：{resp['message']}" - error_code = -1 - return error_message, error_code, [] - for i in resp['data']['list']: - if i.get('appCode') != 'arknights': - continue - result_list.extend(i.get('bindingList')) - return error_message, error_code, result_list - - -def single_sign(cred_resp): - s_token = cred_resp['token'] - header['cred'] = cred_resp['cred'] - msg, code, characters = get_binding_list(s_token) - if code != 0 or not characters: - return msg, code - - content = characters[0] - body = {'gameId': 1, 'uid': content.get('uid')} - - resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, header,s_token), json=body).json() - if resp['code'] != 0: - msg = f'角色{content.get("nickName")}({content.get("channelName")})签到失败了！原因：{resp.get("message")}' - code = -1 - return msg, code - awards = resp['data']['awards'] - for j in awards: - res = j['resource'] - msg = f'角色{content.get("nickName")}({content.get("channelName")})签到成功，获得{res["name"]}×{j.get("count") or 1}' - code = 0 - return msg, code - - -def new_start(): - result_list = [] - token_list = [ - 'fqyRkfJrLlkuax8SzFffxc0W', - 'i//Ozb2bFsXjxF9ZKgAgVi6K' - ] - token_dict = { - 'i//Ozb2bFsXjxF9ZKgAgVi6K': {'cred': 'ytheE5nQW7AU13UJfn3sq4hdvokojwGk', 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '316019'}, - 'fqyRkfJrLlkuax8SzFffxc0W': {'cred': 'tn0EZDtUrSMu3tXh4wK2VJPionqQnMj4', 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '131957'} - } - for i in token_list: - try: - credit_dict = get_cred_by_token(i) - msg, code = single_sign(credit_dict) - except Exception as ex: - print(f'签到失败，原因：{str(ex)}') - print("签到完成！") - - -def new_start_neo(): - token_dict = { - 'i//Ozb2bFsXjxF9ZKgAgVi6K': {'cred': 'ytheE5nQW7AU13UJfn3sq4hdvokojwGk', - 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '316019'}, - 'fqyRkfJrLlkuax8SzFffxc0W': {'cred': 'tn0EZDtUrSMu3tXh4wK2VJPionqQnMj4', - 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '131957'} - } - for token, credit_dict in token_dict.items(): - try: - msg, code = single_sign(credit_dict) - print(f'{msg}: {code}') - except Exception as ex: - print(f'签到失败，原因：{str(ex)}') - print("签到完成！") - - -if __name__ == '__main__': - new_start() diff --git a/src/SKland_Auto_Sign/new_auto_sign.py b/src/SKland_Auto_Sign/new_auto_sign.py new file mode 100644 index 0000000..609dcd9 --- /dev/null +++ b/src/SKland_Auto_Sign/new_auto_sign.py @@ -0,0 +1,375 @@ +import hashlib +import hmac +import json +import logging +import os.path +import threading +import time +from datetime import date +from getpass import getpass +from urllib import parse + +import requests + +from src.SKland_Auto_Sign.SecuritySm import get_d_id + +token_save_name = 'TOKEN.txt' +app_code = '4ca99fa6b56cc2ba' +token_env = os.environ.get('TOKEN') +# 现在想做什么？ +current_type = os.environ.get('SKYLAND_TYPE') + +http_local = threading.local() +header = { + 'cred': '', + 'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0', + 'Accept-Encoding': 'gzip', + 'Connection': 'close' +} +header_login = { + 'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0', + 'Accept-Encoding': 'gzip', + 'Connection': 'close', + 'dId': get_d_id() +} + +# 签名请求头一定要这个顺序，否则失败 +# timestamp是必填的,其它三个随便填,不要为none即可 +header_for_sign = { + 'platform': '', + 'timestamp': '', + 'dId': '', + 'vName': '' +} + +# 签到url +sign_url = "https://zonai.skland.com/api/v1/game/attendance" +# 绑定的角色url +binding_url = "https://zonai.skland.com/api/v1/game/player/binding" +# 验证码url +login_code_url = "https://as.hypergryph.com/general/v1/send_phone_code" +# 验证码登录 +token_phone_code_url = "https://as.hypergryph.com/user/auth/v2/token_by_phone_code" +# 密码登录 +token_password_url = "https://as.hypergryph.com/user/auth/v1/token_by_phone_password" +# 使用token获得认证代码 +grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant" +# 使用认证代码获得cred +cred_code_url = "https://zonai.skland.com/web/v1/user/auth/generate_cred_by_code" + + +def config_logger(): + current_date = date.today().strftime('%Y-%m-%d') + if not os.path.exists('logs'): + os.mkdir('logs') + logger = logging.getLogger() + + file_handler = logging.FileHandler(f'./logs/{current_date}.log', encoding='utf-8') + logger.addHandler(file_handler) + logging.getLogger().setLevel(logging.DEBUG) + file_handler.setLevel(logging.INFO) + formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') + file_handler.setFormatter(formatter) + + def filter_code(text): + filter_key = ['code', 'cred', 'token'] + try: + j = json.loads(text) + if not j.get('data'): + return text + data = j['data'] + for i in filter_key: + if i in data: + data[i] = '*****' + return json.dumps(j, ensure_ascii=False) + except: + return text + + _get = requests.get + _post = requests.post + + def get(*args, **kwargs): + response = _get(*args, **kwargs) + logger.info(f'GET {args[0]} - {response.status_code} - {filter_code(response.text)}') + return response + + def post(*args, **kwargs): + response = _post(*args, **kwargs) + logger.info(f'POST {args[0]} - {response.status_code} - {filter_code(response.text)}') + return response + + # 替换 requests 中的方法 + requests.get = get + requests.post = post + + +def generate_signature(token: str, path, body_or_query): + """ + 获得签名头 + 接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON() + 将此字符串做HMAC加密，算法为SHA-256，密钥token为请求cred接口会返回的一个token值 + 再将加密后的字符串做MD5即得到sign + :param token: 拿cred时候的token + :param path: 请求路径（不包括网址） + :param body_or_query: 如果是GET，则是它的query。POST则为它的body + :return: 计算完毕的sign + """ + # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2 + t = str(int(time.time()) - 2) + token = token.encode('utf-8') + header_ca = json.loads(json.dumps(header_for_sign)) + header_ca['timestamp'] = t + header_ca_str = json.dumps(header_ca, separators=(',', ':')) + s = path + body_or_query + t + header_ca_str + hex_s = hmac.new(token, s.encode('utf-8'), hashlib.sha256).hexdigest() + md5 = hashlib.md5(hex_s.encode('utf-8')).hexdigest().encode('utf-8').decode('utf-8') + # logging.info(f'算出签名: {md5}') + return md5, header_ca + + +def get_sign_header(url: str, method, body, h): + p = parse.urlparse(url) + if method.lower() == 'get': + h['sign'], header_ca = generate_signature(http_local.token, p.path, p.query) + else: + h['sign'], header_ca = generate_signature(http_local.token, p.path, json.dumps(body)) + for i in header_ca: + h[i] = header_ca[i] + return h + + +def login_by_code(): + phone = input('请输入手机号码：') + resp = requests.post(login_code_url, json={'phone': phone, 'type': 2}, headers=header_login).json() + if resp.get("status") != 0: + raise Exception(f"发送手机验证码出现错误：{resp['msg']}") + code = input("请输入手机验证码：") + r = requests.post(token_phone_code_url, json={"phone": phone, "code": code}, headers=header_login).json() + return get_token(r) + + +def login_by_token(): + token_code = input("请输入（登录森空岛电脑官网后请访问这个网址：https://web-api.skland.com/account/info/hg）:") + return parse_user_token(token_code) + + +def parse_user_token(t): + try: + t = json.loads(t) + return t['data']['content'] + except: + pass + return t + + +def login_by_password(): + phone = input('请输入手机号码：') + password = getpass('请输入密码(不会显示在屏幕上面)：') + r = requests.post(token_password_url, json={"phone": phone, "password": password}, headers=header_login).json() + return get_token(r) + + +def get_cred_by_token(token): + grant_code = get_grant_code(token) + return get_cred(grant_code) + + +def get_token(resp): + if resp.get('status') != 0: + raise Exception(f'获得token失败：{resp["msg"]}') + return resp['data']['token'] + + +def get_grant_code(token): + response = requests.post(grant_code_url, json={ + 'appCode': app_code, + 'token': token, + 'type': 0 + }, headers=header_login) + resp = response.json() + if response.status_code != 200: + raise Exception(f'获得认证代码失败：{resp}') + if resp.get('status') != 0: + raise Exception(f'获得认证代码失败：{resp["msg"]}') + return resp['data']['code'] + + +def get_cred(grant): + resp = requests.post(cred_code_url, json={ + 'code': grant, + 'kind': 1 + }, headers=header_login).json() + if resp['code'] != 0: + raise Exception(f'获得cred失败：{resp["message"]}') + return resp['data'] + + +def get_binding_list(): + v = [] + resp = requests.get(binding_url, headers=get_sign_header(binding_url, 'get', None, http_local.header)).json() + + if resp['code'] != 0: + print(f"请求角色列表出现问题：{resp['message']}") + if resp.get('message') == '用户未登录': + print(f'用户登录可能失效了，请重新运行此程序！') + os.remove(token_save_name) + return [] + for i in resp['data']['list']: + if i.get('appCode') != 'arknights': + continue + v.extend(i.get('bindingList')) + return v + + +def list_awards(game_id, uid): + resp = requests.get(sign_url, headers=http_local.header, params={'gameId': game_id, 'uid': uid}).json() + print(resp) + + +def do_sign(cred_resp): + http_local.token = cred_resp['token'] + http_local.header = header.copy() + http_local.header['cred'] = cred_resp['cred'] + characters = get_binding_list() + + for i in characters: + body = { + 'gameId': 1, + 'uid': i.get('uid') + } + # list_awards(1, i.get('uid')) + resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header), + json=body).json() + if resp['code'] != 0: + print(f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}') + continue + awards = resp['data']['awards'] + for j in awards: + res = j['resource'] + print( + f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}' + ) + + +def single_sign(cred_resp): + http_local.token = cred_resp['token'] + http_local.header = header.copy() + http_local.header['cred'] = cred_resp['cred'] + characters = get_binding_list() + + code, msg = -1, 'Default Error' + for i in characters: + body = { + 'gameId': 1, + 'uid': i.get('uid') + } + # list_awards(1, i.get('uid')) + resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header), + json=body).json() + if resp['code'] != 0: + msg = f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}' + code = -1 + continue + awards = resp['data']['awards'] + for j in awards: + res = j['resource'] + msg = f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}' + code = 0 + return msg, code + + +def save(token): + with open(token_save_name, 'w') as f: + f.write(token) + print( + f'您的鹰角网络通行证保存在{token_save_name}, 打开这个可以把它复制到云函数服务器上执行!\n双击添加账号即可再次添加账号') + + +def read(path): + if not os.path.exists(token_save_name): + return [] + v = [] + with open(path, 'r', encoding='utf-8') as f: + for i in f.readlines(): + i = i.strip() + i and i not in v and v.append(i) + return v + + +def read_from_env(): + v = [] + token_list = token_env.split(',') + for i in token_list: + i = i.strip() + if i and i not in v: + v.append(parse_user_token(i)) + print(f'从环境变量中读取到{len(v)}个token...') + return v + + +def init_token(): + if token_env: + print('使用环境变量里面的token') + # 对于github action,不需要存储token,因为token在环境变量里 + return read_from_env() + tokens = [] + tokens.extend(read(token_save_name)) + add_account = current_type == 'add_account' + if add_account: + print('！！！您启用了添加账号模式，将不会签到！！！') + if len(tokens) == 0 or add_account: + tokens.append(input_for_token()) + save('\n'.join(tokens)) + return [] if add_account else tokens + + +def input_for_token(): + print("请输入你需要做什么：") + print("1.使用用户名密码登录（非常推荐）") + print("2.使用手机验证码登录（非常推荐，但可能因为人机验证失败）") + print("3.手动输入鹰角网络通行证账号登录(推荐)") + mode = input('请输入（1，2，3）：') + if mode == '' or mode == '1': + token = login_by_password() + elif mode == '2': + token = login_by_code() + elif mode == '3': + token = login_by_token() + else: + exit(-1) + return token + + +def start(): + # token = init_token() + token = [] + data_dict = { + "user_dict": { + "18021026530": {"password": "Mrfz1790990971", "status": True}, + "13693680360": {"password": "Tanhuobin2022", "status": True} + } + } + for user, content in data_dict['user_dict'].items(): + r = requests.post(token_password_url, json={"phone": user, "password": content['password']}, headers=header_login).json() + token.append(get_token(r)) + for i in token: + try: + do_sign(get_cred_by_token(i)) + except Exception as ex: + print(f'签到失败，原因：{str(ex)}') + logging.error('', exc_info=ex) + print("签到完成！") + + +if __name__ == '__main__': + print('本项目源代码仓库：https://github.com/xxyz30/skyland-auto-sign(已被github官方封禁)') + print('https://gitee.com/FancyCabbage/skyland-auto-sign') + config_logger() + + # logging.info('=========starting==========') + + start_time = time.time() + start() + end_time = time.time() + # logging.info(f'complete with {(end_time - start_time) * 1000} ms') + # logging.info('===========ending============') \ No newline at end of file diff --git a/src/ark_helper_backend/backend.py b/src/ark_helper_backend/backend.py index ba1704c..077b083 100644 --- a/src/ark_helper_backend/backend.py +++ b/src/ark_helper_backend/backend.py @@ -4,12 +4,13 @@ import logging import copy import requests import time +import random from pathlib import Path from src.util.utils import hg_get_new_user_token, load_user_config, validate_token from src.My_Logger.project_logger import LoggerClass -from src.SKland_Auto_Sign.auto_sign import get_cred_by_token, single_sign +from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_sign PROJECT_PATH = Path(__file__).parent.parent.parent.absolute() DATA_PATH = os.path.join(PROJECT_PATH, 'data', 'ArkHelperData') @@ -136,29 +137,16 @@ class ArkHelperBackend: self.single_user_login(phone) self.save_data(payment_status=False, gacha_status=False) - def skland_sign(self): - token_list = [content['token'] for _, content in self.user_data_dict.items()] - for token in token_list: - try: - credit_dict = get_cred_by_token(token) - msg, code = single_sign(credit_dict) - self.logger.info(f'{msg}: {code}') - except Exception as ex: - self.logger.error(f'签到失败，原因：{str(ex)}') - def skland_sign_new(self): - token_dict = { - 'i//Ozb2bFsXjxF9ZKgAgVi6K': {'cred': 'ytheE5nQW7AU13UJfn3sq4hdvokojwGk', - 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '316019'}, - 'fqyRkfJrLlkuax8SzFffxc0W': {'cred': 'tn0EZDtUrSMu3tXh4wK2VJPionqQnMj4', - 'token': 'dbfa23414a942e7609ea4008eeeb7fb0', 'userId': '131957'} - } - for token, credit_dict in token_dict.items(): - try: - msg, code = single_sign(credit_dict) - self.logger.info(f'{msg}: {code}') - except Exception as ex: - self.logger.error(f'签到失败，原因：{str(ex)}') + for user, content in self.user_data_dict.items(): + token = content.get('token') + if token: + try: + msg, code = single_sign(get_cred_by_token(token)) + self.logger.info(f'{msg}: {code}') + except Exception as ex: + self.logger.error(f'签到失败，原因：{str(ex)}') + time.sleep(random.randint(100, 200)) def update_gacha_data(self): for phone in self.user_data_dict: @@ -217,4 +205,4 @@ class ArkHelperBackend: if __name__ == '__main__': ak_back = ArkHelperBackend() # ak_back.update_payment_data() - ak_back.skland_sign() + ak_back.skland_sign_new() diff --git a/src/util/mail.py b/src/util/mail.py index 7debc46..c7f2378 100644 --- a/src/util/mail.py +++ b/src/util/mail.py @@ -16,7 +16,8 @@ def send_batch_email(receiver_email_list, subject, content, file_path=None): """ # sender config sender_email = "ArmorServer@outlook.com" - password = "ServerOnly2024!" + # password = "ServerOnly2024!" + password = 'fczojhcahxwonosh' # mail content message = MIMEMultipart() diff --git a/src/util/utils.py b/src/util/utils.py index cf0f7c9..9363c67 100644 --- a/src/util/utils.py +++ b/src/util/utils.py @@ -1,6 +1,7 @@ import requests import json import os +import random from pathlib import Path @@ -62,5 +63,17 @@ def validate_token(token): return True +def random_time(start_time=480, end_time=720): + # 生成从8*60(480)到12*60(720)之间的随机分钟数 + random_minutes = random.randint(start_time, end_time) + + # 将分钟数转换成小时和分钟 + hour = random_minutes // 60 + minute = random_minutes % 60 + + # 返回格式化的时间 + return "{:02}:{:02}".format(hour, minute) + + if __name__ == '__main__': print(load_user_config())