c-my/srun-cli-bit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

reformat code

Browse Source
This commit is contained in:
c-my
2022-12-05 16:54:34 +08:00
parent d5847c8d3f
commit df9288a4d1
6 changed files with 92 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.py
+1 -1
View File
@@ -1,4 +1,4 @@
from srunbit.cli import execute
if __name__ == '__main__':
if __name__ == "__main__":
execute()
src/srunbit/encrypt/hash.py
+38 -25
View File
@@ -7,11 +7,11 @@ from typing import Dict, List
def gen_info(data: Dict, token: str) -> str:
json_obj = {
'username': data['username'],
'password': data['password'],
'ip': data['ip'],
'acid': data['ac_id'],
'enc_ver': 'srun_bx1',
"username": data["username"],
"password": data["password"],
"ip": data["ip"],
"acid": data["ac_id"],
"enc_ver": "srun_bx1",
}
json_str = json.dumps(json_obj)
x_encode_res = _x_encode(json_str, token)
@@ -20,22 +20,22 @@ def gen_info(data: Dict, token: str) -> str:
dict_val = "LVoJPiCN2R8G90yg+hmFHuacZ1OWMnrsSTXkYpUq/3dlbfKwv6xztjI7DeBE45QA="
d = dict()
for idx, v in enumerate(dict_key):
d[v] = dict_val[idx:idx + 1]
d[v] = dict_val[idx : idx + 1]
b64_arr = bytearray()
for c in x_encode_res:
b64_arr.append(ord(c))
b64_res = base64.standard_b64encode(b64_arr)
target = ''
target = ""
for s in b64_res:
target += d[chr(s)]
return f"{{SRBX1}}{target}"
def _x_encode(msg: str, key: str) -> str:
if msg == '':
return ''
if msg == "":
return ""
v = _s(msg, True)
k = _s(key, False)
n = len(v) - 1
@@ -72,30 +72,39 @@ def _s(a: str, b: bool) -> List[int]:
c = len(a)
v = []
for i in range(0, c, 4):
tmp = _char_code_at(a, i) | (_char_code_at(a, i + 1) << 8) | (
_char_code_at(a, i + 2) << 16) | _char_code_at(a, i + 3) << 24
tmp = (
_char_code_at(a, i)
| (_char_code_at(a, i + 1) << 8)
| (_char_code_at(a, i + 2) << 16)
| _char_code_at(a, i + 3) << 24
)
v.append(tmp)
if b:
v.append(c)
return v
def _l(a: [int], b: bool) -> str:
def _l(a: List[int], b: bool) -> str:
d = len(a)
c = (d - 1) << 2
if b:
m = a[d - 1]
if m < c - 3 or m > c:
return ''
return ""
c = m
res = []
for s in a:
item = chr(s & 0xff) + chr((s >> 8) & 0xff) + chr((s >> 16) & 0xff) + str(chr((s >> 24) & 0xff))
item = (
chr(s & 0xFF)
+ chr((s >> 8) & 0xFF)
+ chr((s >> 16) & 0xFF)
+ str(chr((s >> 24) & 0xFF))
)
res.append(item)
if b:
return ''.join(res)[0:c]
return "".join(res)[0:c]
else:
return ''.join(res)
return "".join(res)
def _char_code_at(s: str, index: int) -> int:
@@ -105,19 +114,23 @@ def _char_code_at(s: str, index: int) -> int:
def pwd_hmd5(password: str, token: str) -> str:
hm = hmac.new(bytes(token, encoding='utf-8'), bytes(password, encoding='utf-8'), digestmod='MD5')
hm = hmac.new(
bytes(token, encoding="utf-8"),
bytes(password, encoding="utf-8"),
digestmod="MD5",
)
hmd5 = hm.hexdigest()
return f'{{MD5}}{hmd5}'
return f"{{MD5}}{hmd5}"
def checksum(data: Dict, token: str) -> str:
username = data['username']
password = data['password']
acid = data['ac_id']
ip = data['ip']
info = data['info']
str_list = ['', username, password[5:], acid, ip, '200', '1', info]
username = data["username"]
password = data["password"]
acid = data["ac_id"]
ip = data["ip"]
info = data["info"]
str_list = ["", username, password[5:], acid, ip, "200", "1", info]
sum_str = token.join(str_list)
sh = hashlib.sha1()
sh.update(bytes(sum_str, encoding='utf-8'))
sh.update(bytes(sum_str, encoding="utf-8"))
return sh.hexdigest()
src/srunbit/model/account.py
+5 -8
View File
@@ -5,15 +5,12 @@ class Account:
def __init__(self, username, password):
self.username = username
self.password = password
self.access_token = ''
self.acid = ''
self.access_token = ""
self.acid = ""
def to_dict(self):
return {
'username': self.username,
'password': self.password
}
return {"username": self.username, "password": self.password}
@classmethod
def from_dict(self, data: Dict):
return Account(data['username'], data['password'])
def from_dict(cls, data: Dict):
return Account(data["username"], data["password"])
src/srunbit/model/form.py
+11 -11
View File
@@ -3,20 +3,20 @@ from typing import Dict
def login(username: str, password: str, acid: str) -> Dict:
return {
'action': 'login',
'username': username,
'password': password,
'ac_id': acid,
'ip': '',
'info': '',
'chksum': '',
'n': '200',
'type': '1'
"action": "login",
"username": username,
"password": password,
"ac_id": acid,
"ip": "",
"info": "",
"chksum": "",
"n": "200",
"type": "1",
}
def logout(username: str) -> Dict:
return {
'action': 'logout',
'username': username,
"action": "logout",
"username": username,
}
src/srunbit/network/network.py
+32 -32
View File
@@ -28,7 +28,7 @@ class LogoutCode(Enum):
class Network:
def __init__(self):
self.__base_url = 'http://10.0.0.55'
self.__base_url = "http://10.0.0.55"
self.__challenge_url = "/cgi-bin/get_challenge"
self.__portal_url = "/cgi-bin/srun_portal"
self.__successful_url = "/cgi-bin/rad_user_info"
@@ -37,7 +37,7 @@ class Network:
try:
ac_id = self._get_ac_id()[0]
except Exception as e:
logging.debug(f'get ac_id failed: {e}')
logging.debug(f"get ac_id failed: {e}")
return LoginCode.FAILED
username = account.username
password = account.password
@@ -46,35 +46,35 @@ class Network:
try:
challenge = self._get_challenge(username)
except Exception as e:
logging.debug(f'get challenge failed: {e}')
logging.debug(f"get challenge failed: {e}")
return LoginCode.FAILED
token = challenge['challenge']
ip = challenge['client_ip']
form_login['ip'] = ip
form_login['info'] = hash.gen_info(form_login, token)
form_login['password'] = hash.pwd_hmd5('', token)
form_login['chksum'] = hash.checksum(form_login, token)
token = challenge["challenge"]
ip = challenge["client_ip"]
form_login["ip"] = ip
form_login["info"] = hash.gen_info(form_login, token)
form_login["password"] = hash.pwd_hmd5("", token)
form_login["chksum"] = hash.checksum(form_login, token)
try:
json_obj = self._get_json(self.__base_url + self.__portal_url, form_login)
except Exception as e:
logging.debug(f'login failed: {e}')
logging.debug(f"login failed: {e}")
return LoginCode.FAILED
if 'res' not in json_obj:
if "res" not in json_obj:
logging.debug(f'field "res" not in json, got json: {json_obj}')
return LoginCode.FAILED
if json_obj['res'] != 'ok':
error_msg = json_obj['error_msg']
if 'Arrearage users' in error_msg:
if json_obj["res"] != "ok":
error_msg = json_obj["error_msg"]
if "Arrearage users" in error_msg:
return LoginCode.ARREARAGE
elif 'You are already online.' in error_msg:
elif "You are already online." in error_msg:
return LoginCode.ALREADY_ONLINE
elif 'Password is error' in error_msg:
elif "Password is error" in error_msg:
return LoginCode.WRONG_PASSWORD
elif 'User is disabled' in error_msg:
elif "User is disabled" in error_msg:
return LoginCode.USER_DISABLED
else:
logging.debug(f'login res is not ok, got error_msg: {error_msg}')
logging.debug(f"login res is not ok, got error_msg: {error_msg}")
print(json_obj)
return LoginCode.FAILED
return LoginCode.SUCCESSFUL
@@ -84,11 +84,11 @@ class Network:
try:
json_obj = self._get_json(self.__base_url + self.__portal_url, form_logout)
except Exception as e:
logging.debug(f'logout failed: {e}')
logging.debug(f"logout failed: {e}")
return LogoutCode.FAILED
if json_obj['error'] != 'ok':
error_msg = json_obj['error_msg']
if 'You are not online' in error_msg:
if json_obj["error"] != "ok":
error_msg = json_obj["error_msg"]
if "You are not online" in error_msg:
return LogoutCode.ALREADY_OFFLINE
else:
return LogoutCode.FAILED
@@ -98,7 +98,7 @@ class Network:
try:
json_obj = self._get_json(self.__base_url + self.__successful_url)
except Exception as e:
logging.debug(f'get info failed: {e}')
logging.debug(f"get info failed: {e}")
return None
return json_obj
@@ -106,7 +106,7 @@ class Network:
r = requests.get(self.__base_url)
d = urlparse(r.url)
query = parse_qs(d.query)
ac_id = query['ac_id'][0]
ac_id = query["ac_id"][0]
return ac_id
def _get_json(self, url: str, data=None):
@@ -114,24 +114,24 @@ class Network:
data = {}
r = self._request_with_callback(url, data)
body = r.text
s = body.find('(')
e = body.rfind(')')
body = body[s + 1:e]
s = body.find("(")
e = body.rfind(")")
body = body[s + 1 : e]
json_obj = json.loads(body)
return json_obj
def _get_challenge(self, username: str) -> dict:
params = {
'username': username,
'ip': '',
"username": username,
"ip": "",
}
return self._get_json(self.__base_url + self.__challenge_url, params)
def _request_with_callback(self, url: str, params: dict):
params['callback'] = self._gen_callback()
params['_'] = self._gen_callback()
params["callback"] = self._gen_callback()
params["_"] = self._gen_callback()
r = requests.get(url, params=params)
return r
def _gen_callback(self) -> str:
return f'{int(time.time() * 1000)}'
return f"{int(time.time() * 1000)}"
src/srunbit/store/account.py
+5 -5
View File
@@ -8,23 +8,23 @@ from typing import Union
from srunbit.model import Account
home = Path.home()
name = '.bitsrun'
name = ".bitsrun"
path = os.path.join(str(home), name)
def read_account() -> Union[Account, None]:
try:
with open(path, 'rb') as f:
with open(path, "rb") as f:
ss = f.read()
s = base64.b64decode(ss)
return Account.from_dict(json.loads(s))
except Exception as e:
logging.debug(f'failed to read account: {e}')
logging.debug(f"failed to read account: {e}")
return None
def write_account(account: Account):
with open(path, 'wb') as f:
with open(path, "wb") as f:
s = json.dumps(account.to_dict())
ss = base64.b64encode(bytes(s, encoding='utf-8'))
ss = base64.b64encode(bytes(s, encoding="utf-8"))
f.write(ss)