update
This commit is contained in:
@@ -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']
|
||||
@@ -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()
|
||||
@@ -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============')
|
||||
@@ -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()
|
||||
|
||||
+2
-1
@@ -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()
|
||||
|
||||
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user