Indicate if a prometheus export was done logged in or not

mu5001tool/__main__.py

@@ -2,12 +2,14 @@ from . import Mu5001Tool
 from .prometheus_exporter import prometheus_exporter
 
 import argparse
+from pathlib import Path
 from pprint import pprint
 
 parser = argparse.ArgumentParser(prog="mu5001tool")
 
 parser.add_argument("--stok", dest="stok", help="Initial session token to use for commands")
 parser.add_argument("--password", dest="password", help="Password for authentication against the device")
+parser.add_argument("--password-file", dest="password_file", type=Path, help="Password for authentication against the device, passed as path to file")
 
 subparsers = parser.add_subparsers()
 
@@ -28,11 +30,12 @@ def run_status(m):
 sp_status = subparsers.add_parser("status", help="General modem status information")
 sp_status.set_defaults(func=run_status)
 
-def run_prometheus_exporter(m):
-    prometheus_exporter(m)
+def run_prometheus_exporter(m, listen_port):
+    prometheus_exporter(m, listen_port)
 
 sp_prometheus_exporter = subparsers.add_parser("prometheus-exporter", help="Serve metrics as prometheus exporter")
 sp_prometheus_exporter.set_defaults(func=run_prometheus_exporter)
+sp_prometheus_exporter.add_argument("--listen-port", dest="listen_port", type=int, default=9242, help="Port for service webserver")
 
 def main():
 
@@ -47,6 +50,7 @@ def main():
     function_arguments.pop("func")
     function_arguments.pop("stok")
     function_arguments.pop("password")
+    function_arguments.pop("password_file")
 
     m = Mu5001Tool()
 
@@ -55,6 +59,12 @@ def main():
 
     if args.password is not None:
         m.set_password(args.password)
+
+    if args.password_file is not None:
+        password = args.password_file.read_text().strip()
+        m.set_password(password)
+
+    if args.password is not None or args.password_file is not None:
         m.login()
 
     args.func(m=m, **function_arguments)

mu5001tool/api.py

@@ -3,9 +3,22 @@ import hashlib
 import requests
 import urllib
 
+class Mu5001ToolException(Exception):
+    pass
+
+class FailedToLogin(Mu5001ToolException):
+    pass
+
+class InvalidPassword(FailedToLogin):
+    pass
+
+class AccountLocked(FailedToLogin):
+    pass
+
 class Mu5001Tool:
     def __init__(self, host="http://192.168.0.1"):
         self._password = None
+        self._password_invalid = False
 
         self.host = host
         self.session = requests.Session()
@@ -49,6 +62,54 @@ class Mu5001Tool:
 
         return login_sha256(login_sha256(password) + ld)
 
+    def is_login_possible(self):
+        data = self.get_cmd_process({
+            "multi_data": "1",
+            "cmd": "login_lock_time,psw_fail_num_str,loginfo",
+        })
+
+        if data.get("loginfo") == "ok":
+            # Already logged in
+            return False
+
+        # How many password attempts left before locking
+        psw_fail_num_str = data.get("psw_fail_num_str")
+
+        if psw_fail_num_str == "":
+            psw_fail_num_str = None
+        else:
+            psw_fail_num_str = int(psw_fail_num_str)
+
+        if psw_fail_num_str > 0:
+            return True
+
+        # How long login is blocked
+        login_lock_time = data.get("login_lock_time")
+
+        if login_lock_time == "":
+            login_lock_time = None
+        else:
+            login_lock_time = int(login_lock_time)
+
+        # Login is blocked
+        return False
+
+
+    def login_if_possible(self):
+        if self._password is None:
+            return
+
+        if self._password_invalid:
+            return
+
+        if not self.is_login_possible():
+            return
+
+        try:
+            self.login()
+        except InvalidPassword as e:
+            self._password_invalid = True
+
     def login(self, password=None):
         if password is None:
             password = self._password
@@ -56,11 +117,22 @@ class Mu5001Tool:
         if password is None:
             raise ValueError("No password provided")
 
-        return self.set_cmd_process({
+        data = self.set_cmd_process({
             "goformId": "LOGIN",
             "password": self.hashed_login_password(password),
         })
 
+        result = data.get("result")
+
+        if result == "3":
+            raise InvalidPassword("Invalid password")
+
+        if result == "1":
+            raise AccountLocked("Account currently locked")
+
+        if result != "0":
+            raise FailedToLogin("Failed to log in")
+
     def logoff(self):
         return self.set_cmd_process({"goformId": "LOGOFF"})
 

mu5001tool/prometheus_exporter.py

@@ -1,6 +1,8 @@
 from . import Mu5001Tool
 from http.server import BaseHTTPRequestHandler, HTTPServer, HTTPStatus
+from pprint import pprint
 import socket
+import time
 import traceback
 
 class HTTPServerV6(HTTPServer):
@@ -42,7 +44,6 @@ def make_prometheus_exporter_request_handler(m):
 
         def export(self):
             cmds_as_metric_value = [
-                    "adjfsdfj",
                 "battery_charging",
                 "battery_pers",
                 "battery_temp",
@@ -89,6 +90,7 @@ def make_prometheus_exporter_request_handler(m):
                 "battery_charg_type",
                 "data_volume_limit_unit",
                 "dial_mode",
+                "loginfo",
                 "mode_main_state",
                 "modem_main_state",
                 "network_provider",
@@ -119,10 +121,16 @@ def make_prometheus_exporter_request_handler(m):
 
             out = []
 
+            if data.get("loginfo") == "ok":
+                out.append("mu5001tool_logged_in 1")
+            else:
+                out.append("mu5001tool_logged_in 0")
+                m.login_if_possible()
+
             for cmd in cmds_as_metric_value:
                 d = data.get(cmd)
 
-                if d is None:
+                if d is None or d == "":
                     continue
 
                 try:
@@ -149,7 +157,7 @@ def make_prometheus_exporter_request_handler(m):
 
     return PrometheusExporterRequestHandler
 
-def prometheus_exporter(m):
-    with HTTPServerV6(("::1", 9242), make_prometheus_exporter_request_handler(m)) as httpd:
+def prometheus_exporter(m, listen_port):
+    with HTTPServerV6(("::1", listen_port), make_prometheus_exporter_request_handler(m)) as httpd:
         print("Starting prometheus exporter on http://[{}]:{}".format(*httpd.socket.getsockname()[:2]))
         httpd.serve_forever()