Mirrors/mealie
1
0
Fork 0
mirror of https://github.com/mealie-recipes/mealie.git synced 2026-01-29 12:03:11 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat: Login with OAuth via OpenID Connect (OIDC) (#3280)

Browse Source 
* initial oidc implementation

* add dynamic scheme

* e2e test setup

* add caching

* fix

* try this

* add libldap-2.5 to runtime dependencies (#2849)

* New translations en-us.json (Norwegian) (#2851)

* New Crowdin updates (#2855)

* New translations en-us.json (Italian)

* New translations en-us.json (Norwegian)

* New translations en-us.json (Portuguese)

* fix

* remove cache

* cache yarn deps

* cache docker image

* cleanup action

* lint

* fix tests

* remove not needed variables

* run code gen

* fix tests

* add docs

* move code into custom scheme

* remove unneeded type

* fix oidc admin

* add more tests

* add better spacing on login page

* create auth providers

* clean up testing stuff

* type fixes

* add OIDC auth method to postgres enum

* add option to bypass login screen and go directly to iDP

* remove check so we can fallback to another auth method oauth fails

* Add provider name to be shown at the login screen

* add new properties to admin about api

* fix spec

* add a prompt to change auth method when changing password

* Create new auth section. Add more info on auth methods

* update docs

* run ruff

* update docs

* format

* docs gen

* formatting

* initialize logger in class

* mypy type fixes

* docs gen

* add models to get proper fields in docs and fix serialization

* validate id token before using it

* only request a mealie token on initial callback

* remove unused method

* fix unit tests

* docs gen

* check for valid idToken before getting token

* add iss to mealie token

* check to see if we already have a mealie token before getting one

* fix lock file

* update authlib

* update lock file

* add remember me environment variable

* add user group setting to allow only certain groups to log in

---------

Co-authored-by: Carter Mintey <cmintey8@gmail.com>
Co-authored-by: Carter <35710697+cmintey@users.noreply.github.com>
This commit is contained in:
Hayden
2024-03-10 13:51:36 -05:00
committed by GitHub
parent bea1a592d7
commit 5f6844eceb
53 changed files with 1533 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
mealie/core/dependencies/dependencies.py
View File

@@ -10,6 +10,7 @@ from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm.session import Session
from mealie.core import root_logger
from mealie.core.config import get_app_dirs, get_app_settings
from mealie.db.db_setup import generate_session
from mealie.repos.all_repositories import get_repositories
@@ -21,6 +22,13 @@ oauth2_scheme_soft_fail = OAuth2PasswordBearer(tokenUrl="/api/auth/token", auto_
ALGORITHM = "HS256"
app_dirs = get_app_dirs()
settings = get_app_settings()
logger = root_logger.get_logger("dependencies")
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
async def is_logged_in(token: str = Depends(oauth2_scheme_soft_fail), session=Depends(generate_session)) -> bool:
@@ -76,13 +84,10 @@ async def try_get_current_user(
async def get_current_user(
request: Request, token: str | None = Depends(oauth2_scheme_soft_fail), session=Depends(generate_session)
request: Request,
token: str | None = Depends(oauth2_scheme_soft_fail),
session=Depends(generate_session),
) -> PrivateUser:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if token is None and "mealie.access_token" in request.cookies:
# Try extract from cookie
token = request.cookies.get("mealie.access_token", "")
@@ -117,12 +122,6 @@ async def get_current_user(
async def get_integration_id(token: str = Depends(oauth2_scheme)) -> str:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
decoded_token = jwt.decode(token, settings.SECRET, algorithms=[ALGORITHM])
return decoded_token.get("integration_id", DEFAULT_INTEGRATION_ID)

3
mealie/core/exceptions.py
View File

@@ -40,3 +40,6 @@ def mealie_registered_exceptions(t: Translator) -> dict:
NoEntryFound: t.t("exceptions.no-entry-found"),
IntegrityError: t.t("exceptions.integrity-error"),
}
class UserLockedOut(Exception): ...

149
mealie/core/security/ldap.py
View File

@@ -1,149 +0,0 @@
import ldap
from ldap.ldapobject import LDAPObject
from mealie.core import root_logger
from mealie.core.config import get_app_settings
from mealie.db.models.users.users import AuthMethod
from mealie.repos.repository_factory import AllRepositories
from mealie.schema.user.user import PrivateUser
logger = root_logger.get_logger("security.ldap")
def search_user(conn: LDAPObject, username: str) -> list[tuple[str, dict[str, list[bytes]]]] | None:
"""
Searches for a user by LDAP_ID_ATTRIBUTE, LDAP_MAIL_ATTRIBUTE, and the provided LDAP_USER_FILTER.
If none or multiple users are found, return False
"""
settings = get_app_settings()
user_filter = ""
if settings.LDAP_USER_FILTER:
# fill in the template provided by the user to maintain backwards compatibility
user_filter = settings.LDAP_USER_FILTER.format(
id_attribute=settings.LDAP_ID_ATTRIBUTE, mail_attribute=settings.LDAP_MAIL_ATTRIBUTE, input=username
)
# Don't assume the provided search filter has (|({id_attribute}={input})({mail_attribute}={input}))
search_filter = (
f"(&(|({settings.LDAP_ID_ATTRIBUTE}={username})({settings.LDAP_MAIL_ATTRIBUTE}={username})){user_filter})"
)
user_entry: list[tuple[str, dict[str, list[bytes]]]] | None = None
try:
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")
if not user_entry:
conn.unbind_s()
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}")
conn.unbind_s()
return None
return user_entry
def get_user(db: AllRepositories, username: str, password: str) -> PrivateUser | bool:
"""Given a username and password, tries to authenticate by BINDing to an
LDAP server
If the BIND succeeds, it will either create a new user of that username on
the server or return an existing one.
Returns False on failure.
"""
settings = get_app_settings()
if settings.LDAP_TLS_INSECURE:
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
conn = ldap.initialize(settings.LDAP_SERVER_URL)
conn.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
conn.set_option(ldap.OPT_REFERRALS, 0)
if settings.LDAP_TLS_CACERTFILE:
conn.set_option(ldap.OPT_X_TLS_CACERTFILE, settings.LDAP_TLS_CACERTFILE)
conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
if settings.LDAP_ENABLE_STARTTLS:
conn.start_tls_s()
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")
conn.unbind_s()
return False
user_entry = search_user(conn, username)
if not user_entry:
return False
user_dn, user_attr = user_entry[0]
# Check the credentials of the user
try:
logger.debug(f"[LDAP] Attempting to bind with '{user_dn}' using the provided password")
conn.simple_bind_s(user_dn, password)
except (ldap.INVALID_CREDENTIALS, ldap.NO_SUCH_OBJECT):
logger.error("[LDAP] Bind failed")
conn.unbind_s()
return False
# Check for existing user
user = db.users.get_one(username, "email", any_case=True)
if not user:
user = db.users.get_one(username, "username", any_case=True)
if user is None:
logger.debug("[LDAP] User is not in Mealie. Creating a new account")
attribute_keys = {
settings.LDAP_ID_ATTRIBUTE: "username",
settings.LDAP_NAME_ATTRIBUTE: "name",
settings.LDAP_MAIL_ATTRIBUTE: "mail",
}
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(
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}")
conn.unbind_s()
return False
attributes[attribute_key] = user_attr[attribute_key][0].decode("utf-8")
user = db.users.create(
{
"username": attributes[settings.LDAP_ID_ATTRIBUTE],
"password": "LDAP",
"full_name": attributes[settings.LDAP_NAME_ATTRIBUTE],
"email": attributes[settings.LDAP_MAIL_ATTRIBUTE],
"admin": False,
"auth_method": AuthMethod.LDAP,
},
)
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")
user.admin = should_be_admin
db.users.update(user.id, user)
conn.unbind_s()
return user

4
mealie/core/security/providers/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from .auth_provider import *
from .credentials_provider import *
from .ldap_provider import *
from .openid_provider import *

71
mealie/core/security/providers/auth_provider.py Normal file
View File

@@ -0,0 +1,71 @@
import abc
from datetime import datetime, timedelta, timezone
from typing import Generic, TypeVar
from jose import jwt
from sqlalchemy.orm.session import Session
from mealie.core.config import get_app_settings
from mealie.repos.all_repositories import get_repositories
from mealie.schema.user.user import PrivateUser
ALGORITHM = "HS256"
ISS = "mealie"
remember_me_duration = timedelta(days=14)
T = TypeVar("T")
class AuthProvider(Generic[T], metaclass=abc.ABCMeta):
"""Base Authentication Provider interface"""
def __init__(self, session: Session, data: T) -> None:
self.session = session
self.data = data
self.user: PrivateUser | None = None
self.__has_tried_user = False
@classmethod
def __subclasshook__(cls, __subclass: type) -> bool:
return hasattr(__subclass, "authenticate") and callable(__subclass.authenticate)
def get_access_token(self, user: PrivateUser, remember_me=False) -> tuple[str, timedelta]:
settings = get_app_settings()
duration = timedelta(hours=settings.TOKEN_TIME)
if remember_me and remember_me_duration > duration:
duration = remember_me_duration
return AuthProvider.create_access_token({"sub": str(user.id)}, duration)
@staticmethod
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> tuple[str, timedelta]:
settings = get_app_settings()
to_encode = data.copy()
expires_delta = expires_delta or timedelta(hours=settings.TOKEN_TIME)
expire = datetime.now(timezone.utc) + expires_delta
to_encode["exp"] = expire
to_encode["iss"] = ISS
return (jwt.encode(to_encode, settings.SECRET, algorithm=ALGORITHM), expires_delta)
def try_get_user(self, username: str) -> PrivateUser | None:
"""Try to get a user from the database, first trying username, then trying email"""
if self.__has_tried_user:
return self.user
db = get_repositories(self.session)
user = user = db.users.get_one(username, "username", any_case=True)
if not user:
user = db.users.get_one(username, "email", any_case=True)
self.user = user
return user
@abc.abstractmethod
async def authenticate(self) -> tuple[str, timedelta] | None:
"""Attempt to authenticate a user"""
raise NotImplementedError

57
mealie/core/security/providers/credentials_provider.py Normal file
View File

@@ -0,0 +1,57 @@
from datetime import timedelta
from sqlalchemy.orm.session import Session
from mealie.core import root_logger
from mealie.core.config import get_app_settings
from mealie.core.exceptions import UserLockedOut
from mealie.core.security.hasher import get_hasher
from mealie.core.security.providers.auth_provider import AuthProvider
from mealie.repos.all_repositories import get_repositories
from mealie.schema.user.auth import CredentialsRequest
from mealie.services.user_services.user_service import UserService
class CredentialsProvider(AuthProvider[CredentialsRequest]):
"""Authentication provider that authenticates a user the database using username/password combination"""
_logger = root_logger.get_logger("credentials_provider")
def __init__(self, session: Session, data: CredentialsRequest) -> None:
super().__init__(session, data)
async def authenticate(self) -> tuple[str, timedelta] | None:
"""Attempt to authenticate a user given a username and password"""
settings = get_app_settings()
db = get_repositories(self.session)
user = self.try_get_user(self.data.username)
if not user:
# To prevent user enumeration we perform the verify_password computation to ensure
# server side time is relatively constant and not vulnerable to timing attacks.
CredentialsProvider.verify_password(
"abc123cba321", "$2b$12$JdHtJOlkPFwyxdjdygEzPOtYmdQF5/R5tHxw5Tq8pxjubyLqdIX5i"
)
return None
if user.login_attemps >= settings.SECURITY_MAX_LOGIN_ATTEMPTS or user.is_locked:
raise UserLockedOut()
if not CredentialsProvider.verify_password(self.data.password, user.password):
user.login_attemps += 1
db.users.update(user.id, user)
if user.login_attemps >= settings.SECURITY_MAX_LOGIN_ATTEMPTS:
user_service = UserService(db)
user_service.lock_user(user)
return None
user.login_attemps = 0
user = db.users.update(user.id, user)
return self.get_access_token(user, self.data.remember_me) # type: ignore
@staticmethod
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Compares a plain string to a hashed password"""
return get_hasher().verify(plain_password, hashed_password)

178
mealie/core/security/providers/ldap_provider.py Normal file
View File

@@ -0,0 +1,178 @@
from datetime import timedelta
import ldap
from ldap.ldapobject import LDAPObject
from sqlalchemy.orm.session import Session
from mealie.core import root_logger
from mealie.core.config import get_app_settings
from mealie.core.security.providers.credentials_provider import CredentialsProvider
from mealie.db.models.users.users import AuthMethod
from mealie.repos.all_repositories import get_repositories
from mealie.schema.user.auth import CredentialsRequest
from mealie.schema.user.user import PrivateUser
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, data: CredentialsRequest) -> None:
super().__init__(session, data)
self.conn = None
async def authenticate(self) -> tuple[str, timedelta] | None:
"""Attempt to authenticate a user given a username and password"""
user = self.try_get_user(self.data.username)
if not user or user.password == "LDAP" or user.auth_method == AuthMethod.LDAP:
user = self.get_user()
if user:
return self.get_access_token(user, self.data.remember_me)
return await super().authenticate()
def search_user(self, conn: LDAPObject) -> list[tuple[str, dict[str, list[bytes]]]] | None:
"""
Searches for a user by LDAP_ID_ATTRIBUTE, LDAP_MAIL_ATTRIBUTE, and the provided LDAP_USER_FILTER.
If none or multiple users are found, return False
"""
if not self.data:
return None
settings = get_app_settings()
user_filter = ""
if settings.LDAP_USER_FILTER:
# fill in the template provided by the user to maintain backwards compatibility
user_filter = settings.LDAP_USER_FILTER.format(
id_attribute=settings.LDAP_ID_ATTRIBUTE,
mail_attribute=settings.LDAP_MAIL_ATTRIBUTE,
input=self.data.username,
)
# Don't assume the provided search filter has (|({id_attribute}={input})({mail_attribute}={input}))
search_filter = "(&(|({id_attribute}={input})({mail_attribute}={input})){filter})".format(
id_attribute=settings.LDAP_ID_ATTRIBUTE,
mail_attribute=settings.LDAP_MAIL_ATTRIBUTE,
input=self.data.username,
filter=user_filter,
)
user_entry: list[tuple[str, dict[str, list[bytes]]]] | None = None
try:
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:
self._logger.error("[LDAP] Bad user search filter")
if not user_entry:
conn.unbind_s()
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:
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
return user_entry
def get_user(self) -> PrivateUser | None:
"""Given a username and password, tries to authenticate by BINDing to an
LDAP server
If the BIND succeeds, it will either create a new user of that username on
the server or return an existing one.
Returns False on failure.
"""
settings = get_app_settings()
db = get_repositories(self.session)
if not self.data:
return None
data = self.data
if settings.LDAP_TLS_INSECURE:
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
conn = ldap.initialize(settings.LDAP_SERVER_URL)
conn.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
conn.set_option(ldap.OPT_REFERRALS, 0)
if settings.LDAP_TLS_CACERTFILE:
conn.set_option(ldap.OPT_X_TLS_CACERTFILE, settings.LDAP_TLS_CACERTFILE)
conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
if settings.LDAP_ENABLE_STARTTLS:
conn.start_tls_s()
try:
conn.simple_bind_s(settings.LDAP_QUERY_BIND, settings.LDAP_QUERY_PASSWORD)
except (ldap.INVALID_CREDENTIALS, ldap.NO_SUCH_OBJECT):
self._logger.error("[LDAP] Unable to bind to with provided user/password")
conn.unbind_s()
return None
user_entry = self.search_user(conn)
if not user_entry:
return None
user_dn, user_attr = user_entry[0]
# Check the credentials of the user
try:
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):
self._logger.error("[LDAP] Bind failed")
conn.unbind_s()
return None
user = self.try_get_user(data.username)
if user is None:
self._logger.debug("[LDAP] User is not in Mealie. Creating a new account")
attribute_keys = {
settings.LDAP_ID_ATTRIBUTE: "username",
settings.LDAP_NAME_ATTRIBUTE: "name",
settings.LDAP_MAIL_ATTRIBUTE: "mail",
}
attributes = {}
for attribute_key, attribute_name in attribute_keys.items():
if attribute_key not in user_attr or len(user_attr[attribute_key]) == 0:
self._logger.error(
f"[LDAP] Unable to create user due to missing '{attribute_name}' ('{attribute_key}') attribute"
)
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")
user = db.users.create(
{
"username": attributes[settings.LDAP_ID_ATTRIBUTE],
"password": "LDAP",
"full_name": attributes[settings.LDAP_NAME_ATTRIBUTE],
"email": attributes[settings.LDAP_MAIL_ATTRIBUTE],
"admin": False,
"auth_method": AuthMethod.LDAP,
},
)
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:
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)
conn.unbind_s()
return user

127
mealie/core/security/providers/openid_provider.py Normal file
View File

@@ -0,0 +1,127 @@
from datetime import timedelta
from functools import lru_cache
import requests
from authlib.jose import JsonWebKey, JsonWebToken, JWTClaims, KeySet
from authlib.jose.errors import ExpiredTokenError
from authlib.oidc.core import CodeIDToken
from sqlalchemy.orm.session import Session
from mealie.core import root_logger
from mealie.core.config import get_app_settings
from mealie.core.security.providers.auth_provider import AuthProvider
from mealie.db.models.users.users import AuthMethod
from mealie.repos.all_repositories import get_repositories
from mealie.schema.user.auth import OIDCRequest
class OpenIDProvider(AuthProvider[OIDCRequest]):
"""Authentication provider that authenticates a user using a token from OIDC ID token"""
_logger = root_logger.get_logger("openid_provider")
def __init__(self, session: Session, data: OIDCRequest) -> None:
super().__init__(session, data)
async def authenticate(self) -> tuple[str, timedelta] | None:
"""Attempt to authenticate a user given a username and password"""
claims = self.get_claims()
if not claims:
return None
settings = get_app_settings()
repos = get_repositories(self.session)
user = self.try_get_user(claims.get("email"))
group_claim = claims.get("groups", [])
is_admin = settings.OIDC_ADMIN_GROUP in group_claim if settings.OIDC_ADMIN_GROUP else False
is_valid_user = settings.OIDC_USER_GROUP in group_claim if settings.OIDC_USER_GROUP else True
if not is_valid_user:
self._logger.debug(
"[OIDC] User does not have the required group. Found: %s - Required: %s",
group_claim,
settings.OIDC_USER_GROUP,
)
return None
if not user:
if not settings.OIDC_SIGNUP_ENABLED:
self._logger.debug("[OIDC] No user found. Not creating a new user - new user creation is disabled.")
return None
self._logger.debug("[OIDC] No user found. Creating new OIDC user.")
user = repos.users.create(
{
"username": claims.get("preferred_username"),
"password": "OIDC",
"full_name": claims.get("name"),
"email": claims.get("email"),
"admin": is_admin,
"auth_method": AuthMethod.OIDC,
}
)
self.session.commit()
return self.get_access_token(user, settings.OIDC_REMEMBER_ME) # type: ignore
if user:
if user.admin != is_admin:
self._logger.debug(f"[OIDC] {'Setting' if is_admin else 'Removing'} user as admin")
user.admin = is_admin
repos.users.update(user.id, user)
return self.get_access_token(user, settings.OIDC_REMEMBER_ME)
self._logger.info("[OIDC] Found user but their AuthMethod does not match OIDC")
return None
def get_claims(self) -> JWTClaims | None:
"""Get the claims from the ID token and check if the required claims are present"""
required_claims = {"preferred_username", "name", "email"}
jwks = OpenIDProvider.get_jwks()
if not jwks:
return None
claims = JsonWebToken(["RS256"]).decode(s=self.data.id_token, key=jwks, claims_cls=CodeIDToken)
try:
claims.validate()
except ExpiredTokenError as e:
self._logger.debug(f"[OIDC] {e.error}: {e.description}")
return None
if not claims:
self._logger.warning("[OIDC] Claims not found")
return None
if not required_claims.issubset(claims.keys()):
self._logger.error(
f"[OIDC] Required claims not present. Expected: {required_claims} Actual: {claims.keys()}"
)
return None
return claims
@lru_cache
@staticmethod
def get_jwks() -> KeySet | None:
"""Get the key set from the open id configuration"""
settings = get_app_settings()
if not (settings.OIDC_READY and settings.OIDC_CONFIGURATION_URL):
return None
configuration = None
with requests.get(settings.OIDC_CONFIGURATION_URL, timeout=5) as config_response:
config_response.raise_for_status()
configuration = config_response.json()
if not configuration:
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:
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:
response.raise_for_status()
return JsonWebKey.import_key_set(response.json())

62
mealie/core/security/security.py
View File

@@ -2,23 +2,35 @@ import secrets
from datetime import datetime, timedelta, timezone
from pathlib import Path
from fastapi import Request
from jose import jwt
from sqlalchemy.orm.session import Session
from mealie.core import root_logger
from mealie.core.config import get_app_settings
from mealie.core.security import ldap
from mealie.core.security.hasher import get_hasher
from mealie.db.models.users.users import AuthMethod
from mealie.repos.all_repositories import get_repositories
from mealie.schema.user import PrivateUser
from mealie.services.user_services.user_service import UserService
from mealie.core.security.providers.auth_provider import AuthProvider
from mealie.core.security.providers.credentials_provider import CredentialsProvider
from mealie.core.security.providers.ldap_provider import LDAPProvider
from mealie.core.security.providers.openid_provider import OpenIDProvider
from mealie.schema.user.auth import CredentialsRequest, CredentialsRequestForm, OIDCRequest
ALGORITHM = "HS256"
logger = root_logger.get_logger("security")
class UserLockedOut(Exception): ...
def get_auth_provider(session: Session, request: Request, data: CredentialsRequestForm) -> AuthProvider:
settings = get_app_settings()
if request.cookies.get("mealie.auth.strategy") == "oidc":
return OpenIDProvider(session, OIDCRequest(id_token=request.cookies.get("mealie.auth._id_token.oidc")))
credentials_request = CredentialsRequest(**data.__dict__)
if settings.LDAP_ENABLED:
return LDAPProvider(session, credentials_request)
return CredentialsProvider(session, credentials_request)
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
@@ -43,44 +55,6 @@ def create_recipe_slug_token(file_path: str | Path) -> str:
return create_access_token(token_data, expires_delta=timedelta(minutes=30))
def authenticate_user(session, email: str, password: str) -> PrivateUser | bool:
settings = get_app_settings()
db = get_repositories(session)
user = db.users.get_one(email, "email", any_case=True)
if not user:
user = db.users.get_one(email, "username", any_case=True)
if settings.LDAP_AUTH_ENABLED and (not user or user.password == "LDAP" or user.auth_method == AuthMethod.LDAP):
return ldap.get_user(db, email, password)
if not user:
# To prevent user enumeration we perform the verify_password computation to ensure
# server side time is relatively constant and not vulnerable to timing attacks.
verify_password("abc123cba321", "$2b$12$JdHtJOlkPFwyxdjdygEzPOtYmdQF5/R5tHxw5Tq8pxjubyLqdIX5i")
return False
if user.login_attemps >= settings.SECURITY_MAX_LOGIN_ATTEMPTS or user.is_locked:
raise UserLockedOut()
elif not verify_password(password, user.password):
user.login_attemps += 1
db.users.update(user.id, user)
if user.login_attemps >= settings.SECURITY_MAX_LOGIN_ATTEMPTS:
user_service = UserService(db)
user_service.lock_user(user)
return False
user.login_attemps = 0
return db.users.update(user.id, user)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Compares a plain string to a hashed password"""
return get_hasher().verify(plain_password, hashed_password)
def hash_password(password: str) -> str:
"""Takes in a raw password and hashes it. Used prior to saving a new password to the database."""
return get_hasher().hash(password)

20
mealie/core/settings/settings.py
View File

@@ -171,6 +171,26 @@ class AppSettings(BaseSettings):
not_none = None not in required
return self.LDAP_AUTH_ENABLED and not_none
# ===============================================
# OIDC Configuration
OIDC_AUTH_ENABLED: bool = False
OIDC_CLIENT_ID: str | None = None
OIDC_CONFIGURATION_URL: str | None = None
OIDC_SIGNUP_ENABLED: bool = True
OIDC_USER_GROUP: str | None = None
OIDC_ADMIN_GROUP: str | None = None
OIDC_AUTO_REDIRECT: bool = False
OIDC_PROVIDER_NAME: str = "OAuth"
OIDC_REMEMBER_ME: bool = False
@property
def OIDC_READY(self) -> bool:
"""Validates OIDC settings are all set"""
required = {self.OIDC_CLIENT_ID, self.OIDC_CONFIGURATION_URL}
not_none = None not in required
return self.OIDC_AUTH_ENABLED and not_none
# ===============================================
# Testing Config