app: replace dict with UserInfo throughout Moonraker

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>

moonraker/common.py

@@ -224,7 +224,7 @@ class APIDefinition:
         request_type: RequestType,
         transport: Optional[APITransport] = None,
         ip_addr: Optional[IPAddress] = None,
-        user: Optional[Dict[str, Any]] = None
+        user: Optional[UserInfo] = None
     ) -> Coroutine:
         return self.callback(
             WebRequest(self.endpoint, args, request_type, transport, ip_addr, user)
@@ -313,7 +313,7 @@ class APITransport:
         return TransportType.INTERNAL
 
     @property
-    def user_info(self) -> Optional[Dict[str, Any]]:
+    def user_info(self) -> Optional[UserInfo]:
         return None
 
     @property
@@ -350,14 +350,14 @@ class BaseRemoteConnection(APITransport):
             "url": ""
         }
         self._need_auth: bool = False
-        self._user_info: Optional[Dict[str, Any]] = None
+        self._user_info: Optional[UserInfo] = None
 
     @property
-    def user_info(self) -> Optional[Dict[str, Any]]:
+    def user_info(self) -> Optional[UserInfo]:
         return self._user_info
 
     @user_info.setter
-    def user_info(self, uinfo: Dict[str, Any]) -> None:
+    def user_info(self, uinfo: UserInfo) -> None:
         self._user_info = uinfo
         self._need_auth = False
 
@@ -443,7 +443,7 @@ class BaseRemoteConnection(APITransport):
     def on_user_logout(self, user: str) -> bool:
         if self._user_info is None:
             return False
-        if user == self._user_info.get("username", ""):
+        if user == self._user_info.username:
             self._user_info = None
             return True
         return False
@@ -529,7 +529,7 @@ class WebRequest:
         request_type: RequestType = RequestType(0),
         transport: Optional[APITransport] = None,
         ip_addr: Optional[IPAddress] = None,
-        user: Optional[Dict[str, Any]] = None
+        user: Optional[UserInfo] = None
     ) -> None:
         self.endpoint = endpoint
         self.args = args
@@ -561,7 +561,7 @@ class WebRequest:
     def get_ip_address(self) -> Optional[IPAddress]:
         return self.ip_addr
 
-    def get_current_user(self) -> Optional[Dict[str, Any]]:
+    def get_current_user(self) -> Optional[UserInfo]:
         return self.current_user
 
     def _get_converted_arg(self,

moonraker/components/application.py

@@ -56,6 +56,7 @@ if TYPE_CHECKING:
     from ..server import Server
     from ..eventloop import EventLoop
     from ..confighelper import ConfigHelper
+    from ..common import UserInfo
     from .klippy_connection import KlippyConnection as Klippy
     from ..utils import IPAddress
     from .websockets import WebsocketManager, WebSocket
@@ -160,10 +161,10 @@ class PrimaryRouter(MutableRouter):
         else:
             log_method = access_log.error
         request_time = 1000.0 * handler.request.request_time()
-        user = handler.current_user
+        user: Optional[UserInfo] = handler.current_user
         username = "No User"
-        if user is not None and 'username' in user:
-            username = user['username']
+        if user is not None:
+            username = user.username
         log_method(
             f"{status_code} {handler._request_summary()} "
             f"[{username}] {request_time:.2f}ms"
@@ -725,7 +726,7 @@ class RPCHandler(AuthorizedRequestHandler, APITransport):
         return TransportType.HTTP
 
     @property
-    def user_info(self) -> Optional[Dict[str, Any]]:
+    def user_info(self) -> Optional[UserInfo]:
         return self.current_user
 
     @property

moonraker/components/authorization.py

@@ -43,7 +43,7 @@ if TYPE_CHECKING:
     from .ldap import MoonrakerLDAP
     IPAddr = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
     IPNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
-    OneshotToken = Tuple[IPAddr, Optional[Dict[str, Any]], asyncio.Handle]
+    OneshotToken = Tuple[IPAddr, Optional[UserInfo], asyncio.Handle]
 
 # Helpers for base64url encoding and decoding
 def base64url_encode(data: bytes) -> bytes:
@@ -363,7 +363,7 @@ class Authorization:
         user_info = web_request.get_current_user()
         if user_info is None:
             raise self.server.error("No user logged in")
-        username: str = user_info['username']
+        username: str = user_info.username
         if username in RESERVED_USERS:
             raise self.server.error(
                 f"Invalid log out request for user {username}")
@@ -389,9 +389,9 @@ class Authorization:
             sources.append("ldap")
         login_req = self.force_logins and len(self.users) > 1
         request_trusted: Optional[bool] = None
-        user = web_request.current_user
+        user = web_request.get_current_user()
         req_ip = web_request.ip_addr
-        if user is not None and user.get("username") == TRUSTED_USER:
+        if user is not None and user.username == TRUSTED_USER:
             request_trusted = True
         elif req_ip is not None:
             request_trusted = await self._check_authorized_ip(req_ip)
@@ -431,15 +431,15 @@ class Authorization:
             user = web_request.get_current_user()
             if user is None:
                 return {
-                    'username': None,
-                    'source': None,
-                    'created_on': None,
+                    "username": None,
+                    "source": None,
+                    "created_on": None,
                 }
             else:
                 return {
-                    'username': user['username'],
-                    'source': user.get("source", "moonraker"),
-                    'created_on': user.get('created_on')
+                    "username": user.username,
+                    "source": user.source,
+                    "created_on": user.created_on
                 }
         elif req_type == RequestType.POST:
             # Create User
@@ -473,17 +473,17 @@ class Authorization:
         user_info = web_request.get_current_user()
         if user_info is None:
             raise self.server.error("No Current User")
-        username = user_info['username']
-        if user_info.get("source", "moonraker") == "ldap":
+        username = user_info.username
+        if user_info.source == "ldap":
             raise self.server.error(
                 f"Can´t Reset password for ldap user {username}")
         if username in RESERVED_USERS:
             raise self.server.error(
                 f"Invalid Reset Request for user {username}")
-        salt = bytes.fromhex(user_info['salt'])
+        salt = bytes.fromhex(user_info.salt)
         hashed_pass = hashlib.pbkdf2_hmac(
             'sha256', password.encode(), salt, HASH_ITER).hex()
-        if hashed_pass != user_info['password']:
+        if hashed_pass != user_info.password:
             raise self.server.error("Invalid Password")
         new_hashed_pass = hashlib.pbkdf2_hmac(
             'sha256', new_pass.encode(), salt, HASH_ITER).hex()
@@ -557,7 +557,7 @@ class Authorization:
         if jwt_secret_hex is None:
             private_key = Signer()
             jwk_id = base64url_encode(secrets.token_bytes()).decode()
-            user_info.jwt_secret = private_key.hex_seed().decode()
+            user_info.jwt_secret = private_key.hex_seed().decode()  # type: ignore
             user_info.jwk_id = jwk_id
             self.users[username] = user_info
             await self._sync_user(username)
@@ -579,7 +579,7 @@ class Authorization:
                 "authorization:user_created",
                 {'username': username})
         elif conn is not None:
-            conn.user_info = user_info.as_dict()
+            conn.user_info = user_info
         return {
             'username': username,
             'token': token,
@@ -592,10 +592,9 @@ class Authorization:
         username: str = web_request.get_str('username')
         current_user = web_request.get_current_user()
         if current_user is not None:
-            curname = current_user.get('username', None)
-            if curname is not None and curname == username:
-                raise self.server.error(
-                    f"Cannot delete logged in user {curname}")
+            curname = current_user.username
+            if curname == username:
+                raise self.server.error(f"Cannot delete logged in user {curname}")
         if username in RESERVED_USERS:
             raise self.server.error(
                 f"Invalid Request for reserved user {username}")
@@ -655,7 +654,7 @@ class Authorization:
         # verify header
         if header.get('typ') != "JWT" or header.get('alg') != "EdDSA":
             raise self.server.error("Invalid JWT header")
-        jwk_id = header.get('kid')
+        jwk_id: Optional[str] = header.get('kid')
         if jwk_id not in self.public_jwks:
             raise self.server.error("Invalid key ID")
 
@@ -683,7 +682,7 @@ class Authorization:
             raise self.server.error("Unknown user", 401)
         return user_info
 
-    def validate_jwt(self, token: str) -> Dict[str, Any]:
+    def validate_jwt(self, token: str) -> UserInfo:
         try:
             user_info = self.decode_jwt(token)
         except Exception as e:
@@ -692,13 +691,13 @@ class Authorization:
             raise self.server.error(
                 f"Failed to decode JWT: {e}", 401
             ) from e
-        return user_info.as_dict()
+        return user_info
 
-    def validate_api_key(self, api_key: str) -> Dict[str, Any]:
+    def validate_api_key(self, api_key: str) -> UserInfo:
         if not self.enable_api_key:
             raise self.server.error("API Key authentication is disabled", 401)
         if api_key and api_key == self.api_key:
-            return self.users[API_USER].as_dict()
+            return self.users[API_USER]
         raise self.server.error("Invalid API Key", 401)
 
     def _load_private_key(self, secret: str) -> Signer:
@@ -747,10 +746,7 @@ class Authorization:
     def _oneshot_token_expire_handler(self, token):
         self.oneshot_tokens.pop(token, None)
 
-    def get_oneshot_token(self,
-                          ip_addr: IPAddr,
-                          user: Optional[Dict[str, Any]]
-                          ) -> str:
+    def get_oneshot_token(self, ip_addr: IPAddr, user: Optional[UserInfo]) -> str:
         token = base64.b32encode(os.urandom(20)).decode()
         event_loop = self.server.get_event_loop()
         hdl = event_loop.delay_callback(
@@ -825,10 +821,9 @@ class Authorization:
                 return self.trusted_users[ip]["user"]
         return None
 
-    def _check_oneshot_token(self,
-                             token: str,
-                             cur_ip: Optional[IPAddr]
-                             ) -> Optional[Dict[str, Any]]:
+    def _check_oneshot_token(
+        self, token: str, cur_ip: Optional[IPAddr]
+    ) -> Optional[UserInfo]:
         if token in self.oneshot_tokens:
             ip_addr, user, hdl = self.oneshot_tokens.pop(token)
             hdl.cancel()
@@ -847,14 +842,14 @@ class Authorization:
 
     async def authenticate_request(
         self, request: HTTPServerRequest, auth_required: bool = True
-    ) -> Optional[Dict[str, Any]]:
+    ) -> Optional[UserInfo]:
         if request.method == "OPTIONS":
             return None
 
         # Check JSON Web Token
         jwt_user = self._check_json_web_token(request, auth_required)
         if jwt_user is not None:
-            return jwt_user.as_dict()
+            return jwt_user
 
         try:
             ip = ipaddress.ip_address(request.remote_ip)  # type: ignore
@@ -874,7 +869,7 @@ class Authorization:
         if self.enable_api_key:
             key: Optional[str] = request.headers.get("X-Api-Key")
             if key and key == self.api_key:
-                return self.users[API_USER].as_dict()
+                return self.users[API_USER]
 
         # If the force_logins option is enabled and at least one user is created
         # then trusted user authentication is disabled
@@ -887,7 +882,7 @@ class Authorization:
         # then it is acceptable to return None
         trusted_user = await self._check_trusted_connection(ip)
         if trusted_user is not None:
-            return trusted_user.as_dict()
+            return trusted_user
         if not auth_required:
             return None
 

moonraker/components/file_manager/file_manager.py

@@ -43,7 +43,7 @@ from typing import (
 if TYPE_CHECKING:
     from inotify_simple import Event as InotifyEvent
     from ...confighelper import ConfigHelper
-    from ...common import WebRequest
+    from ...common import WebRequest, UserInfo
     from ..klippy_connection import KlippyConnection
     from ..job_queue import JobQueue
     from ..job_state import JobState
@@ -902,7 +902,7 @@ class FileManager:
         started: bool = False
         queued: bool = False
         if upload_info['start_print']:
-            user: Optional[Dict[str, Any]] = upload_info.get("user")
+            user: Optional[UserInfo] = upload_info.get("user")
             if can_start:
                 kapis: APIComp = self.server.lookup_component('klippy_apis')
                 try:

moonraker/components/history.py

@@ -29,7 +29,7 @@ from typing import (
 
 if TYPE_CHECKING:
     from ..confighelper import ConfigHelper
-    from ..common import WebRequest
+    from ..common import WebRequest, UserInfo
     from .database import MoonrakerDatabase as DBComp
     from .job_state import JobState
     from .file_manager.file_manager import FileManager
@@ -372,8 +372,8 @@ class History:
             # `CLEAR_PAUSE/SDCARD_RESET_FILE` workflow
             await self.finish_job("cancelled", prev_stats)
 
-    def _on_job_requested(self, user: Optional[Dict[str, Any]]) -> None:
-        username = (user or {}).get("username", "No User")
+    def _on_job_requested(self, user: Optional[UserInfo]) -> None:
+        username = user.username if user is not None else "No User"
         self.job_user = username
         if self.current_job is not None:
             self.current_job.user = username

moonraker/components/job_queue.py

@@ -21,7 +21,7 @@ from typing import (
 )
 if TYPE_CHECKING:
     from ..confighelper import ConfigHelper
-    from ..common import WebRequest
+    from ..common import WebRequest, UserInfo
     from .klippy_apis import KlippyAPI
     from .file_manager.file_manager import FileManager
 
@@ -168,7 +168,7 @@ class JobQueue:
                         filenames: Union[str, List[str]],
                         check_exists: bool = True,
                         reset: bool = False,
-                        user: Optional[Dict[str, Any]] = None
+                        user: Optional[UserInfo] = None
                         ) -> None:
         async with self.lock:
             # Make sure that the file exists
@@ -324,7 +324,7 @@ class JobQueue:
         await self.pause_queue()
 
 class QueuedJob:
-    def __init__(self, filename: str, user: Optional[Dict[str, Any]] = None) -> None:
+    def __init__(self, filename: str, user: Optional[UserInfo] = None) -> None:
         self.filename = filename
         self.job_id = f"{id(self):016X}"
         self.time_added = time.time()
@@ -334,7 +334,7 @@ class QueuedJob:
         return self.filename
 
     @property
-    def user(self) -> Optional[Dict[str, Any]]:
+    def user(self) -> Optional[UserInfo]:
         return self._user
 
     def as_dict(self, cur_time: float) -> Dict[str, Any]:

moonraker/components/klippy_apis.py

@@ -24,6 +24,7 @@ from typing import (
 )
 if TYPE_CHECKING:
     from ..confighelper import ConfigHelper
+    from ..common import UserInfo
     from .klippy_connection import KlippyConnection as Klippy
     Subscription = Dict[str, Optional[List[Any]]]
     SubCallback = Callable[[Dict[str, Dict[str, Any]], float], Optional[Coroutine]]
@@ -127,7 +128,7 @@ class KlippyAPI(APITransport):
         self,
         filename: str,
         wait_klippy_started: bool = False,
-        user: Optional[Dict[str, Any]] = None
+        user: Optional[UserInfo] = None
     ) -> str:
         # WARNING: Do not call this method from within the following
         # event handlers when "wait_klippy_started" is set to True:

moonraker/components/simplyprint.py

@@ -17,7 +17,7 @@ import logging.handlers
 import tempfile
 from queue import SimpleQueue
 from ..loghelper import LocalQueueHandler
-from ..common import APITransport, JobEvent, KlippyState
+from ..common import APITransport, JobEvent, KlippyState, UserInfo
 from ..utils import json_wrapper as jsonw
 
 from typing import (
@@ -1493,6 +1493,7 @@ class PrintHandler:
         self.download_progress: int = -1
         self.pending_file: str = ""
         self.last_started: str = ""
+        self.sp_user = UserInfo("SimplyPrint", "")
 
     def download_file(self, url: str, start: bool):
         coro = self._download_sp_file(url, start)
@@ -1598,7 +1599,7 @@ class PrintHandler:
         kapi: KlippyAPI = self.server.lookup_component("klippy_apis")
         data = {"state": "started"}
         try:
-            await kapi.start_print(pending, user={"username": "SimplyPrint"})
+            await kapi.start_print(pending, user=self.sp_user)
         except Exception:
             logging.exception("Print Failed to start")
             data["state"] = "error"