2a69f0193f
* Users are only authenticated against their password on login, and to retrieve a token * Passwords are wiped from the GUI frontend and cookies after login and token retrieval * Tokens are revoked at the end of the session/logout * If the user chooses the "remember me" option, the token is stored in the cookie * Tokens correctly delete themselves on logout * Tokens can expire at user-specified date * Tokens have their last usage time * Tokens can have user defined descriptions * Users can manage login tokens in their account settings
73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
import base64
|
|
from typing import Optional, Tuple
|
|
from szurubooru import model, errors, rest
|
|
from szurubooru.func import auth, users, user_tokens
|
|
from szurubooru.rest.errors import HttpBadRequest
|
|
|
|
|
|
def _authenticate_basic_auth(username: str, password: str) -> model.User:
|
|
''' Try to authenticate user. Throw AuthError for invalid users. '''
|
|
user = users.get_user_by_name(username)
|
|
if not auth.is_valid_password(user, password):
|
|
raise errors.AuthError('Invalid password.')
|
|
return user
|
|
|
|
|
|
def _authenticate_token(
|
|
username: str, token: str) -> Tuple[model.User, model.UserToken]:
|
|
''' Try to authenticate user. Throw AuthError for invalid users. '''
|
|
user = users.get_user_by_name(username)
|
|
user_token = user_tokens.get_by_user_and_token(user, token)
|
|
if not auth.is_valid_token(user_token):
|
|
raise errors.AuthError('Invalid token.')
|
|
return user, user_token
|
|
|
|
|
|
def _get_user(ctx: rest.Context, bump_login: bool) -> Optional[model.User]:
|
|
if not ctx.has_header('Authorization'):
|
|
return None
|
|
|
|
auth_token = None
|
|
|
|
try:
|
|
auth_type, credentials = ctx.get_header('Authorization').split(' ', 1)
|
|
if auth_type.lower() == 'basic':
|
|
username, password = base64.decodebytes(
|
|
credentials.encode('ascii')).decode('utf8').split(':', 1)
|
|
auth_user = _authenticate_basic_auth(username, password)
|
|
elif auth_type.lower() == 'token':
|
|
username, token = base64.decodebytes(
|
|
credentials.encode('ascii')).decode('utf8').split(':', 1)
|
|
auth_user, auth_token = _authenticate_token(username, token)
|
|
else:
|
|
raise HttpBadRequest(
|
|
'ValidationError',
|
|
'Only basic or token HTTP authentication is supported.')
|
|
except ValueError as err:
|
|
msg = (
|
|
'Authorization header values are not properly formed. '
|
|
'Supplied header {0}. Got error: {1}')
|
|
raise HttpBadRequest(
|
|
'ValidationError',
|
|
msg.format(ctx.get_header('Authorization'), str(err)))
|
|
|
|
if bump_login and auth_user.user_id:
|
|
users.bump_user_login_time(auth_user)
|
|
if auth_token is not None:
|
|
user_tokens.bump_usage_time(auth_token)
|
|
ctx.session.commit()
|
|
|
|
return auth_user
|
|
|
|
|
|
def process_request(ctx: rest.Context) -> None:
|
|
''' Bind the user to request. Update last login time if needed. '''
|
|
bump_login = ctx.get_param_as_bool('bump-login', default=False)
|
|
auth_user = _get_user(ctx, bump_login)
|
|
if auth_user:
|
|
ctx.user = auth_user
|
|
|
|
|
|
@rest.middleware.pre_hook
|
|
def process_request_hook(ctx: rest.Context) -> None:
|
|
process_request(ctx)
|