diff --git a/src/main.py b/src/main.py index c11af5e..c5ee0b5 100644 --- a/src/main.py +++ b/src/main.py @@ -1,4 +1,4 @@ from srunbit.cli import execute -if __name__ == '__main__': +if __name__ == "__main__": execute() diff --git a/src/srunbit/encrypt/hash.py b/src/srunbit/encrypt/hash.py index e7c8cd9..43fa2db 100644 --- a/src/srunbit/encrypt/hash.py +++ b/src/srunbit/encrypt/hash.py @@ -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() diff --git a/src/srunbit/model/account.py b/src/srunbit/model/account.py index c92b343..ce4ab99 100644 --- a/src/srunbit/model/account.py +++ b/src/srunbit/model/account.py @@ -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"]) diff --git a/src/srunbit/model/form.py b/src/srunbit/model/form.py index 9445aec..14e36f8 100644 --- a/src/srunbit/model/form.py +++ b/src/srunbit/model/form.py @@ -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, } diff --git a/src/srunbit/network/network.py b/src/srunbit/network/network.py index 83e1dc3..a3c4469 100644 --- a/src/srunbit/network/network.py +++ b/src/srunbit/network/network.py @@ -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)}" diff --git a/src/srunbit/store/account.py b/src/srunbit/store/account.py index a420567..421deb3 100644 --- a/src/srunbit/store/account.py +++ b/src/srunbit/store/account.py @@ -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)