diff --git a/README.md b/README.md index fd18194..be89fdf 100644 --- a/README.md +++ b/README.md @@ -29,10 +29,9 @@ Usage: bitsrun login/logout [OPTIONS] Log into or out of the BIT network. Options: - -u, --username TEXT Username. - -p, --password TEXT Password. + -u, --username TEXT Your username. + -p, --password TEXT Your password. -v, --verbose Verbosely echo API response. - -s, --silent Silent, no output to stdout. --help Show this message and exit. ``` diff --git a/bitsrun/cli.py b/bitsrun/cli.py index f93c2bf..50362ff 100644 --- a/bitsrun/cli.py +++ b/bitsrun/cli.py @@ -1,5 +1,6 @@ import sys from getpass import getpass +from pprint import pprint import click @@ -9,10 +10,9 @@ # A hacky way to specify shared options for multiple click commands: # https://stackoverflow.com/questions/40182157/shared-options-and-flags-between-commands _options = [ - click.option("-u", "--username", help="Username.", required=False), - click.option("-p", "--password", help="Password.", required=False), + click.option("-u", "--username", help="Your username.", required=False), + click.option("-p", "--password", help="Your password.", required=False), click.option("-v", "--verbose", is_flag=True, help="Verbosely echo API response."), - click.option("-s", "--silent", is_flag=True, help="Silent, no output to stdout."), ] @@ -41,55 +41,70 @@ def config_paths(): @cli.command() @add_options(_options) -def login(username, password, verbose, silent): +def login(username, password, verbose): """Log into the BIT network.""" - do_action("login", username, password, verbose, silent) + do_action("login", username, password, verbose) @cli.command() @add_options(_options) -def logout(username, password, verbose, silent): +def logout(username, password, verbose): """Log out of the BIT network.""" - do_action("logout", username, password, verbose, silent) + do_action("logout", username, password, verbose) -def do_action(action, username, password, verbose, silent): - """Log in/out the BIT network.""" +def do_action(action, username, password, verbose): + # Support reading password from stdin when not passed via `--password` if username and not password: password = getpass(prompt="Please enter your password: ") + + # Try to read username and password from args provided. If none, look for config + # files in possible paths. If none, fail and prompt user to provide one. if username and password: user = User(username, password) elif conf := read_config(): - user = User(*conf) + user = User(**conf[0]) + if verbose: + click.echo( + click.style("bitsrun: ", fg="blue") + + "Reading config from " + + click.style(conf[1], fg="yellow", underline=True) + ) else: ctx = click.get_current_context() - ctx.fail("No username/password provided.") + ctx.fail("No username or password provided") try: if action == "login": - res = user.login() - - # Output login result by default if not silent - if not silent: - click.echo(f"{res.get('username')} ({res.get('online_ip')}) logged in") - + resp = user.login() + message = f"{user.username} ({resp['online_ip']}) logged in" elif action == "logout": - res = user.logout() - - # Output logout result by default if not silent - if not silent: - click.echo(f"{res.get('online_ip')} logged out") - + resp = user.logout() + message = f"{resp['online_ip']} logged out" else: # Should not reach here, but just in case - raise ValueError(f"unknown action `{action}`") + raise ValueError(f"Unknown action `{action}`") - # Output direct result of response if verbose + # Output direct result of the API response if verbose if verbose: - click.echo(f"{click.style('info:', fg='blue')} {res}") + click.echo(f"{click.style('bitsrun:', fg='cyan')} Response from API:") + pprint(resp) + + # Handle error from API response. When field `error` is not `ok`, then the + # login/logout action has likely failed. Hints are provided in the `error_msg`. + if resp["error"] != "ok": + raise Exception( + resp["error_msg"] + if resp["error_msg"] + else "Action failed, use --verbose for more info" + ) + + # Print success message + click.echo(f"{click.style('bitsrun:', fg='green')} {message}") except Exception as e: - click.echo(f"{click.style('error:', fg='red')} {e}") + # Exception is caught and printed to stderr + click.echo(f"{click.style('error:', fg='red')} {e}", err=True) # Throw with error code 1 for scripts to pick up error state sys.exit(1) diff --git a/bitsrun/config.py b/bitsrun/config.py index d6865d1..af73ed2 100644 --- a/bitsrun/config.py +++ b/bitsrun/config.py @@ -2,7 +2,7 @@ from os import getenv from pathlib import Path from sys import platform -from typing import Optional, Tuple +from typing import Optional, Tuple, TypedDict from platformdirs import site_config_path, user_config_path @@ -49,7 +49,12 @@ def get_config_paths() -> map: return map(lambda path: path / "bit-user.json", paths) -def read_config() -> Optional[Tuple[str, str]]: +class ConfigType(TypedDict): + username: str + password: str + + +def read_config() -> Optional[Tuple[ConfigType, str]]: """Read config from the first available config file with name `bit-user.json`. The config file should be a JSON file with the following structure: @@ -59,7 +64,7 @@ def read_config() -> Optional[Tuple[str, str]]: ``` Returns: - A tuple of (username, password) if the config file is found. + A tuple of (config, path to config file) if the config file is found. """ paths = get_config_paths() @@ -67,7 +72,7 @@ def read_config() -> Optional[Tuple[str, str]]: try: with open(path) as f: data = json.loads(f.read()) - return data["username"], data["password"] + return data, str(path) except Exception: continue return None diff --git a/bitsrun/user.py b/bitsrun/user.py index 0c2fc9d..e5c4faf 100644 --- a/bitsrun/user.py +++ b/bitsrun/user.py @@ -2,15 +2,15 @@ import json from enum import Enum from hashlib import sha1 -from typing import Dict, Optional, Union +from typing import Dict, Literal, Optional, TypedDict, Union from requests import Session from bitsrun.utils import fkbase64, parse_homepage, xencode -API_BASE = "http://10.0.0.55" -TYPE_CONST = 1 -N_CONST = 200 +_API_BASE = "http://10.0.0.55" +_TYPE_CONST = 1 +_N_CONST = 200 class Action(Enum): @@ -18,15 +18,26 @@ class Action(Enum): LOGOUT = "logout" +class UserResponseType(TypedDict): + client_ip: str + online_ip: str + # Field `error` is also `login_error` when logout action fails + error: Union[Literal["login_error"], Literal["ok"]] + error_msg: str + res: Union[Literal["login_error"], Literal["ok"]] + # Field `username` is not present on login fails and all logout scenarios + username: Optional[str] + + class User: def __init__(self, username: str, password: str): self.username = username self.password = password - self.ip, self.acid = parse_homepage(api_base=API_BASE) + self.ip, self.acid = parse_homepage(api_base=_API_BASE) self.session = Session() - def login(self) -> Dict[str, Union[str, int]]: + def login(self) -> UserResponseType: logged_in_user = self._user_validate() # Raise exception if device is already logged in @@ -35,7 +46,7 @@ def login(self) -> Dict[str, Union[str, int]]: return self._do_action(Action.LOGIN) - def logout(self) -> Dict[str, Union[str, int]]: + def logout(self) -> UserResponseType: logged_in_user = self._user_validate() # Raise exception if device is not logged in @@ -44,9 +55,9 @@ def logout(self) -> Dict[str, Union[str, int]]: return self._do_action(Action.LOGOUT) - def _do_action(self, action: Action) -> Dict[str, Union[str, int]]: + def _do_action(self, action: Action) -> UserResponseType: params = self._make_params(action) - response = self.session.get(API_BASE + "/cgi-bin/srun_portal", params=params) + response = self.session.get(_API_BASE + "/cgi-bin/srun_portal", params=params) return json.loads(response.text[6:-1]) def _get_user_info(self) -> Optional[str]: @@ -56,7 +67,7 @@ def _get_user_info(self) -> Optional[str]: The username of the current logged in user if exists. """ - resp = self.session.get(API_BASE + "/cgi-bin/rad_user_info") + resp = self.session.get(_API_BASE + "/cgi-bin/rad_user_info") data = resp.text if data == "not_online_error": @@ -88,7 +99,7 @@ def _user_validate(self) -> Optional[str]: def _get_token(self) -> str: params = {"callback": "jsonp", "username": self.username, "ip": self.ip} - response = self.session.get(API_BASE + "/cgi-bin/get_challenge", params=params) + response = self.session.get(_API_BASE + "/cgi-bin/get_challenge", params=params) result = json.loads(response.text[6:-1]) return result["challenge"] @@ -101,8 +112,8 @@ def _make_params(self, action: Action) -> Dict[str, Union[int, str]]: "action": action.value, "ac_id": self.acid, "ip": self.ip, - "type": TYPE_CONST, - "n": N_CONST, + "type": _TYPE_CONST, + "n": _N_CONST, } data = { @@ -123,8 +134,8 @@ def _make_params(self, action: Action) -> Dict[str, Union[int, str]]: hmd5, self.acid, self.ip, - N_CONST, - TYPE_CONST, + _N_CONST, + _TYPE_CONST, info, ).encode() ).hexdigest() diff --git a/pyproject.toml b/pyproject.toml index efa985e..d60e2ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "bitsrun" -version = "3.2.4" +version = "3.3.0" description = "A headless login / logout script for 10.0.0.55" authors = ["spencerwooo "] license = "MIT"