initialize logger in class

mealie/core/security/providers/ldap_provider.py

@@ -12,12 +12,12 @@
 from mealie.repos.all_repositories import get_repositories
 from mealie.schema.user.user import PrivateUser
 
-logger = root_logger.get_logger("ldap_provider")
-
 
 class LDAPProvider(CredentialsProvider):
     """Authentication provider that authenticats a user against an LDAP server using username/password combination"""
 
+    _logger = root_logger.get_logger("ldap_provider")
+
     def __init__(self, session: Session, request: Request) -> None:
         super().__init__(session, request)
         self.conn = None
@@ -60,27 +60,27 @@ def search_user(self, conn: LDAPObject) -> list[tuple[str, dict[str, list[bytes]
 
         user_entry: list[tuple[str, dict[str, list[bytes]]]] | None = None
         try:
-            logger.debug(f"[LDAP] Starting search with filter: {search_filter}")
+            self._logger.debug(f"[LDAP] Starting search with filter: {search_filter}")
             user_entry = conn.search_s(
                 settings.LDAP_BASE_DN,
                 ldap.SCOPE_SUBTREE,
                 search_filter,
                 [settings.LDAP_ID_ATTRIBUTE, settings.LDAP_NAME_ATTRIBUTE, settings.LDAP_MAIL_ATTRIBUTE],
             )
         except ldap.FILTER_ERROR:
-            logger.error("[LDAP] Bad user search filter")
+            self._logger.error("[LDAP] Bad user search filter")
 
         if not user_entry:
             conn.unbind_s()
-            logger.error("[LDAP] No user was found with the provided user filter")
+            self._logger.error("[LDAP] No user was found with the provided user filter")
             return None
 
         # we only want the entries that have a dn
         user_entry = [(dn, attr) for dn, attr in user_entry if dn]
 
         if len(user_entry) > 1:
-            logger.warning("[LDAP] Multiple users found with the provided user filter")
-            logger.debug(f"[LDAP] The following entries were returned: {user_entry}")
+            self._logger.warning("[LDAP] Multiple users found with the provided user filter")
+            self._logger.debug(f"[LDAP] The following entries were returned: {user_entry}")
             conn.unbind_s()
             return None
 
@@ -118,7 +118,7 @@ def get_user(self) -> PrivateUser | None:
         try:
             conn.simple_bind_s(settings.LDAP_QUERY_BIND, settings.LDAP_QUERY_PASSWORD)
         except (ldap.INVALID_CREDENTIALS, ldap.NO_SUCH_OBJECT):
-            logger.error("[LDAP] Unable to bind to with provided user/password")
+            self._logger.error("[LDAP] Unable to bind to with provided user/password")
             conn.unbind_s()
             return None
 
@@ -129,17 +129,17 @@ def get_user(self) -> PrivateUser | None:
 
         # Check the credentials of the user
         try:
-            logger.debug(f"[LDAP] Attempting to bind with '{user_dn}' using the provided password")
+            self._logger.debug(f"[LDAP] Attempting to bind with '{user_dn}' using the provided password")
             conn.simple_bind_s(user_dn, data.password)
         except (ldap.INVALID_CREDENTIALS, ldap.NO_SUCH_OBJECT):
-            logger.error("[LDAP] Bind failed")
+            self._logger.error("[LDAP] Bind failed")
             conn.unbind_s()
             return None
 
         user = self.try_get_user(data.username)
 
         if user is None:
-            logger.debug("[LDAP] User is not in Mealie. Creating a new account")
+            self._logger.debug("[LDAP] User is not in Mealie. Creating a new account")
 
             attribute_keys = {
                 settings.LDAP_ID_ATTRIBUTE: "username",
@@ -149,10 +149,10 @@ def get_user(self) -> PrivateUser | None:
             attributes = {}
             for attribute_key, attribute_name in attribute_keys.items():
                 if attribute_key not in user_attr or len(user_attr[attribute_key]) == 0:
-                    logger.error(
+                    self._logger.error(
                         f"[LDAP] Unable to create user due to missing '{attribute_name}' ('{attribute_key}') attribute"
                     )
-                    logger.debug(f"[LDAP] User has the following attributes: {user_attr}")
+                    self._logger.debug(f"[LDAP] User has the following attributes: {user_attr}")
                     conn.unbind_s()
                     return None
                 attributes[attribute_key] = user_attr[attribute_key][0].decode("utf-8")
@@ -171,7 +171,7 @@ def get_user(self) -> PrivateUser | None:
         if settings.LDAP_ADMIN_FILTER:
             should_be_admin = len(conn.search_s(user_dn, ldap.SCOPE_BASE, settings.LDAP_ADMIN_FILTER, [])) > 0
             if user.admin != should_be_admin:
-                logger.debug(f"[LDAP] {'Setting' if should_be_admin else 'Removing'} user as admin")
+                self._logger.debug(f"[LDAP] {'Setting' if should_be_admin else 'Removing'} user as admin")
                 user.admin = should_be_admin
                 db.users.update(user.id, user)
 

mealie/core/security/providers/openid_provider.py

@@ -10,12 +10,12 @@
 from mealie.db.models.users.users import AuthMethod
 from mealie.repos.all_repositories import get_repositories
 
-logger = root_logger.get_logger("openid_provider")
-
 
 class OpenIDProvider(AuthProvider):
     """Authentication provider that authenticates a user using a token from OIDC ID token"""
 
+    _logger = root_logger.get_logger("openid_provider")
+
     async def authenticate(self) -> tuple[str, timedelta] | None:
         """Attempt to authenticate a user given a username and password"""
 
@@ -31,10 +31,10 @@ async def authenticate(self) -> tuple[str, timedelta] | None:
 
         if not user:
             if not settings.OIDC_SIGNUP_ENABLED:
-                logger.debug("[OIDC] No user found. Not creating a new user - new user creation is disabled.")
+                self._logger.debug("[OIDC] No user found. Not creating a new user - new user creation is disabled.")
                 return None
 
-            logger.debug("[OIDC] No user found. Creating new OIDC user.")
+            self._logger.debug("[OIDC] No user found. Creating new OIDC user.")
 
             user = repos.users.create(
                 {
@@ -51,12 +51,12 @@ async def authenticate(self) -> tuple[str, timedelta] | None:
 
         if user:
             if user.admin != admin_claim:
-                logger.debug(f"[OIDC] {'Setting' if admin_claim else 'Removing'} user as admin")
+                self._logger.debug(f"[OIDC] {'Setting' if admin_claim else 'Removing'} user as admin")
                 user.admin = admin_claim
                 repos.users.update(user.id, user)
             return self.get_access_token(user)
 
-        logger.info("[OIDC] Found user but their AuthMethod does not match OIDC")
+        self._logger.info("[OIDC] Found user but their AuthMethod does not match OIDC")
         return None
 
     def get_claims(self) -> JWTClaims | None:
@@ -67,10 +67,12 @@ def get_claims(self) -> JWTClaims | None:
             return None
         claims = JsonWebToken(["RS256"]).decode(s=self.request.cookies.get("mealie.auth._id_token.oidc"), key=jwks)
         if not claims:
-            logger.warning("[OIDC] Claims not found")
+            self._logger.warning("[OIDC] Claims not found")
             return None
         if not required_claims.issubset(claims.keys()):
-            logger.error(f"[OIDC] Required claims not present. Expected: {required_claims} Actual: {claims.keys()}")
+            self._logger.error(
+                f"[OIDC] Required claims not present. Expected: {required_claims} Actual: {claims.keys()}"
+            )
             return None
         return claims
 
@@ -88,12 +90,12 @@ def get_jwks() -> KeySet | None:
             configuration = config_response.json()
 
         if not configuration:
-            logger.warning("[OIDC] Unable to fetch configuration from the OIDC_CONFIGURATION_URL")
+            OpenIDProvider._logger.warning("[OIDC] Unable to fetch configuration from the OIDC_CONFIGURATION_URL")
             return None
 
         jwks_uri = configuration.get("jwks_uri", None)
         if not jwks_uri:
-            logger.warning("[OIDC] Unable to find the jwks_uri from the OIDC_CONFIGURATION_URL")
+            OpenIDProvider._logger.warning("[OIDC] Unable to find the jwks_uri from the OIDC_CONFIGURATION_URL")
             return None
 
         with requests.get(jwks_uri, timeout=5) as response:

mealie/core/settings/settings.py

@@ -174,10 +174,10 @@ def LDAP_ENABLED(self) -> bool:
     # ===============================================
     # OIDC Configuration
     OIDC_AUTH_ENABLED: bool = False
-    OIDC_CLIENT_ID: NoneStr
-    OIDC_CONFIGURATION_URL: NoneStr
+    OIDC_CLIENT_ID: str | None = None
+    OIDC_CONFIGURATION_URL: str | None = None
     OIDC_SIGNUP_ENABLED: bool = True
-    OIDC_ADMIN_GROUP: NoneStr
+    OIDC_ADMIN_GROUP: str | None = None
     OIDC_AUTO_REDIRECT: bool = False
     OIDC_PROVIDER_NAME: str = "OAuth"