refactor auth_util

bumper/db.py

@@ -49,7 +49,7 @@ def user_get(userid):
     return users.get(User.userid == userid)
 
 
-def user_by_deviceid(deviceid):
+def user_by_deviceid(deviceid: str):
     users = db_get().table("users")
     User = Query()
     return users.get(User.devices.any([deviceid]))
@@ -166,7 +166,7 @@ def user_revoke_expired_tokens(userid):
                 tokens.remove(doc_ids=[i.doc_id])
 
 
-def user_revoke_token(userid, token):
+def user_revoke_token(userid: str, token: str):
     opendb = db_get()
     with opendb:
         tokens = opendb.table("tokens")
@@ -287,7 +287,7 @@ def loginByItToken(authcode):
     return {}
 
 
-def check_token(uid, token):
+def check_token(uid: str, token: str) -> bool:
     bumperlog.debug(f"Checking for token: {token}")
     tokens = db_get().table("tokens")
     tmpauth = tokens.get(

bumper/web/auth_util.py

@@ -1,8 +1,12 @@
 """Auth util module."""
 import json
+import logging
 import uuid
 
 from aiohttp import web
+from aiohttp.web_exceptions import HTTPInternalServerError
+from aiohttp.web_request import Request
+from aiohttp.web_response import Response
 
 from bumper import db, use_auth
 from bumper.db import (
@@ -16,7 +20,6 @@ from bumper.db import (
     user_get,
     user_get_token,
     user_revoke_expired_tokens,
-    user_revoke_token,
 )
 from bumper.models import (
     API_ERRORS,
@@ -27,6 +30,7 @@ from bumper.models import (
     EcoVacsHome_Login,
 )
 from bumper.util import get_current_time_as_millis, get_logger
+from bumper.web.plugins import get_success_response
 
 _logger = get_logger("confserver")
 
@@ -63,7 +67,7 @@ async def login(request):
             ):  # Performing basic "auth" using devid, super insecure
                 user = user_by_deviceid(user_devid)
                 if "checkLogin" in request.path:
-                    check_token(
+                    _check_token(
                         apptype, countrycode, user, request.query["accessToken"]
                     )
                 else:
@@ -119,51 +123,33 @@ async def login(request):
         _logger.exception(f"{e}")
 
 
-async def get_authcode(request):
+async def get_authcode(request: Request) -> Response:
     try:
-        apptype = request.match_info.get("apptype", "")
+        user_devid = request.match_info.get("devid", None)  # Ecovacs
-        user_devid = request.match_info.get("devid", "")  # Ecovacs
+        if not user_devid:
-        if user_devid == "":
             user_devid = request.query["deviceId"]  # Ecovacs Home
 
-        if not user_devid == "":
+        if user_devid:
             user = user_by_deviceid(user_devid)
-            token = ""
             if user:
                 if "accessToken" in request.query:
                     token = user_get_token(user["userid"], request.query["accessToken"])
                     if token:
-                    authcode = ""
+                        if "authcode" in token:
-                    if not "authcode" in token:
+                            authcode = token["authcode"]
+                        else:
                             authcode = generate_authcode(
                                 user,
                                 request.match_info.get("country", "us"),
                                 request.query["accessToken"],
                             )
-                    else:
+
-                        authcode = token["authcode"]
+                        data = {
-                    if "global" in apptype:
-                        body = {
-                            "code": RETURN_API_SUCCESS,
-                            "data": {
                             "authCode": authcode,
                             "ecovacsUid": request.query["uid"],
-                            },
-                            "msg": "操作成功",
-                            "success": True,
-                            "time": get_current_time_as_millis(),
                         }
-                    else:
+
-                        body = {
+                        return get_success_response(data)
-                            "code": RETURN_API_SUCCESS,
-                            "data": {
-                                "authCode": authcode,
-                                "ecovacsUid": request.query["uid"],
-                            },
-                            "msg": "操作成功",
-                            "time": get_current_time_as_millis(),
-                        }
-                    return web.json_response(body)
 
         body = {
             "code": ERR_TOKEN_INVALID,
@@ -174,11 +160,13 @@ async def get_authcode(request):
 
         return web.json_response(body)
 
-    except Exception as e:
+    except Exception:  # pylint: disable=broad-except
-        _logger.exception(f"{e}")
+        logging.error("Unexpected exception occurred", exc_info=True)
+
+    raise HTTPInternalServerError
 
 
-def check_token(apptype, countrycode, user, token):
+def _check_token(apptype, countrycode, user, token):
     try:
         if db.check_token(user["userid"], token):
 
@@ -272,7 +260,7 @@ def _auth_any(devid, apptype, country, request):
                 _logger.error(f"No DID for bot: {bot}")
 
         if "checkLogin" in request.path:  # If request was to check a token do so
-            checkToken = check_token(
+            checkToken = _check_token(
                 apptype, countrycode, tmpuser, request.query["accessToken"]
             )
             isGood = json.loads(checkToken.text)
@@ -300,95 +288,3 @@ def _auth_any(devid, apptype, country, request):
 
     except Exception as e:
         _logger.exception(f"{e}")
-
-
-def get_user_account_info(request):
-    try:
-        user_devid = request.match_info.get("devid", "")
-        countrycode = request.match_info.get("country", "us")
-        apptype = request.match_info.get("apptype", "")
-        user = user_by_deviceid(user_devid)
-
-        if "global_" in apptype:  # EcoVacs Home
-            login_details = EcoVacsHome_Login()
-            login_details.ucUid = "fuid_{}".format(user["userid"])
-            login_details.loginName = "fusername_{}".format(user["userid"])
-            login_details.mobile = None
-        else:
-            login_details = EcoVacs_Login()
-
-        login_details.uid = "fuid_{}".format(user["userid"])
-        login_details.username = "fusername_{}".format(user["userid"])
-        login_details.country = countrycode
-        login_details.email = "null@null.com"
-
-        body = {
-            "code": RETURN_API_SUCCESS,
-            "data": {
-                "email": login_details.email,
-                "hasMobile": "N",
-                "hasPassword": "Y",
-                "uid": login_details.uid,
-                "userName": login_details.username,
-                "obfuscatedMobile": None,
-                "mobile": None,
-                "loginName": login_details.loginName,
-            },
-            "msg": "操作成功",
-            "time": get_current_time_as_millis(),
-        }
-
-        # Example body
-        # {
-        # "code": "0000",
-        # "data": {
-        #     "email": "user@gmail.com",
-        #     "hasMobile": "N",
-        #     "hasPassword": "Y",
-        #     "headIco": "",
-        #     "loginName": "user@gmail.com",
-        #     "mobile": null,
-        #     "mobileAreaNo": null,
-        #     "nickname": "",
-        #     "obfuscatedMobile": null,
-        #     "thirdLoginInfoList": [
-        #     {
-        #         "accountType": "WeChat",
-        #         "hasBind": "N"
-        #     }
-        #     ],
-        #     "uid": "20180719212155_*****",
-        #     "userName": "EAY*****"
-        # },
-        # "msg": "操作成功",
-        # "success": true,
-        # "time": 1578203898343
-        # }
-
-        return web.json_response(body)
-
-    except Exception as e:
-        _logger.exception(f"{e}")
-
-
-async def logout(request):
-    try:
-        user_devid = request.match_info.get("devid", "")
-        if not user_devid == "":
-            user = user_by_deviceid(user_devid)
-            if user:
-                if db.check_token(user["userid"], request.query["accessToken"]):
-                    # Deactivate old tokens and authcodes
-                    user_revoke_token(user["userid"], request.query["accessToken"])
-
-        body = {
-            "code": RETURN_API_SUCCESS,
-            "data": None,
-            "msg": "操作成功",
-            "time": get_current_time_as_millis(),
-        }
-
-        return web.json_response(body)
-
-    except Exception as e:
-        _logger.exception(f"{e}")

bumper/web/plugins/v1/private/user.py

@@ -1,11 +1,14 @@
 """User plugin module."""
+import logging
 from typing import Iterable
 
 from aiohttp import web
+from aiohttp.web_exceptions import HTTPInternalServerError
 from aiohttp.web_request import Request
 from aiohttp.web_response import Response
 from aiohttp.web_routedef import AbstractRouteDef
 
+from bumper.db import check_token, user_by_deviceid, user_revoke_token
 from bumper.web import auth_util
 
 from ... import WebserverPlugin, get_success_response
@@ -37,7 +40,7 @@ class UserPlugin(WebserverPlugin):
             web.route(
                 "*",
                 f"{BASE_URL}user/logout",
-                auth_util.logout,
+                _logout,
             ),
             web.route(
                 "*",
@@ -52,7 +55,7 @@ class UserPlugin(WebserverPlugin):
             web.route(
                 "*",
                 f"{BASE_URL}user/getUserAccountInfo",
-                auth_util.get_user_account_info,
+                _get_user_account_info,
             ),
             web.route(
                 "*",
@@ -77,6 +80,74 @@ class UserPlugin(WebserverPlugin):
         ]
 
 
+async def _logout(request: Request) -> Response:
+    try:
+        user_device_id = request.match_info.get("devid", None)
+        if user_device_id:
+            user = user_by_deviceid(user_device_id)
+            if user:
+                if check_token(user["userid"], request.query["accessToken"]):
+                    # Deactivate old tokens and authcodes
+                    user_revoke_token(user["userid"], request.query["accessToken"])
+
+        return get_success_response(None)
+
+    except Exception:  # pylint: disable=broad-except
+        logging.error("Unexpected exception occurred", exc_info=True)
+
+    raise HTTPInternalServerError
+
+
+async def _get_user_account_info(request: Request) -> Response:
+    try:
+        user_devid = request.match_info.get("devid", "")
+        user = user_by_deviceid(user_devid)
+        username = f"fusername_{user['userid']}"
+        return get_success_response(
+            {
+                "email": "null@null.com",
+                "hasMobile": "N",
+                "hasPassword": "Y",
+                "uid": f"fuid_{user['userid']}",
+                "userName": username,
+                "obfuscatedMobile": None,
+                "mobile": None,
+                "loginName": username,
+            }
+        )
+
+        # Example body
+        # {
+        # "code": "0000",
+        # "data": {
+        #     "email": "user@gmail.com",
+        #     "hasMobile": "N",
+        #     "hasPassword": "Y",
+        #     "headIco": "",
+        #     "loginName": "user@gmail.com",
+        #     "mobile": null,
+        #     "mobileAreaNo": null,
+        #     "nickname": "",
+        #     "obfuscatedMobile": null,
+        #     "thirdLoginInfoList": [
+        #     {
+        #         "accountType": "WeChat",
+        #         "hasBind": "N"
+        #     }
+        #     ],
+        #     "uid": "20180719212155_*****",
+        #     "userName": "EAY*****"
+        # },
+        # "msg": "操作成功",
+        # "success": true,
+        # "time": 1578203898343
+        # }
+    except Exception:  # pylint: disable=broad-except
+        logging.error("Unexpected exception occurred", exc_info=True)
+
+    raise HTTPInternalServerError
+
+
 async def _handle_check_agreement(request: Request) -> Response:
     apptype = request.match_info.get("apptype", "")
     data = []