jumpserver/apps/authentication/mixins.py

457 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

# -*- coding: utf-8 -*-
#
import inspect
from urllib.parse import urlencode
from functools import partial
import time
from django.core.cache import cache
from django.conf import settings
from django.contrib import auth
from django.utils.translation import ugettext as _
from django.contrib.auth import (
BACKEND_SESSION_KEY, _get_backends,
PermissionDenied, user_login_failed, _clean_credentials
)
from django.shortcuts import reverse, redirect
from django.views.generic.edit import FormView
from common.utils import get_object_or_none, get_request_ip, get_logger, bulk_get, FlashMessageUtil
from users.models import User, MFAType
from users.utils import LoginBlockUtil, MFABlockUtils
from . import errors
from .utils import rsa_decrypt, gen_key_pair
from .signals import post_auth_success, post_auth_failed
from .const import RSA_PRIVATE_KEY, RSA_PUBLIC_KEY
logger = get_logger(__name__)
def check_backend_can_auth(username, backend_path, allowed_auth_backends):
if allowed_auth_backends is not None and backend_path not in allowed_auth_backends:
logger.debug('Skip user auth backend: {}, {} not in'.format(
username, backend_path, ','.join(allowed_auth_backends)
)
)
return False
return True
def authenticate(request=None, **credentials):
"""
If the given credentials are valid, return a User object.
"""
username = credentials.get('username')
allowed_auth_backends = User.get_user_allowed_auth_backends(username)
for backend, backend_path in _get_backends(return_tuples=True):
# 预先检查,不浪费认证时间
if not check_backend_can_auth(username, backend_path, allowed_auth_backends):
continue
backend_signature = inspect.signature(backend.authenticate)
try:
backend_signature.bind(request, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(request, **credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# 如果是 None, 证明没有检查过, 需要再次检查
if allowed_auth_backends is None:
# 有些 authentication 参数中不带 username, 之后还要再检查
allowed_auth_backends = user.get_allowed_auth_backends()
if not check_backend_can_auth(user.username, backend_path, allowed_auth_backends):
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials), request=request)
auth.authenticate = authenticate
class PasswordEncryptionViewMixin:
request = None
def get_decrypted_password(self, password=None, username=None):
request = self.request
if hasattr(request, 'data'):
data = request.data
else:
data = request.POST
username = username or data.get('username')
password = password or data.get('password')
password = self.decrypt_passwd(password)
if not password:
self.raise_password_decrypt_failed(username=username)
return password
def raise_password_decrypt_failed(self, username):
ip = self.get_request_ip()
raise errors.CredentialError(
error=errors.reason_password_decrypt_failed,
username=username, ip=ip, request=self.request
)
def decrypt_passwd(self, raw_passwd):
# 获取解密密钥,对密码进行解密
rsa_private_key = self.request.session.get(RSA_PRIVATE_KEY)
if rsa_private_key is not None:
try:
return rsa_decrypt(raw_passwd, rsa_private_key)
except Exception as e:
logger.error(e, exc_info=True)
logger.error(
f'Decrypt password failed: password[{raw_passwd}] '
f'rsa_private_key[{rsa_private_key}]'
)
return None
return raw_passwd
def get_request_ip(self):
ip = ''
if hasattr(self.request, 'data'):
ip = self.request.data.get('remote_addr', '')
ip = ip or get_request_ip(self.request)
return ip
def get_context_data(self, **kwargs):
# 生成加解密密钥对public_key传递给前端private_key存入session中供解密使用
rsa_public_key = self.request.session.get(RSA_PUBLIC_KEY)
rsa_private_key = self.request.session.get(RSA_PRIVATE_KEY)
if not all((rsa_private_key, rsa_public_key)):
rsa_private_key, rsa_public_key = gen_key_pair()
rsa_public_key = rsa_public_key.replace('\n', '\\n')
self.request.session[RSA_PRIVATE_KEY] = rsa_private_key
self.request.session[RSA_PUBLIC_KEY] = rsa_public_key
kwargs.update({
'rsa_public_key': rsa_public_key,
})
return super().get_context_data(**kwargs)
class AuthMixin(PasswordEncryptionViewMixin):
request = None
partial_credential_error = None
key_prefix_captcha = "_LOGIN_INVALID_{}"
def get_user_from_session(self):
if self.request.session.is_empty():
raise errors.SessionEmptyError()
if all((self.request.user,
not self.request.user.is_anonymous,
BACKEND_SESSION_KEY in self.request.session)):
user = self.request.user
user.backend = self.request.session[BACKEND_SESSION_KEY]
return user
user_id = self.request.session.get('user_id')
if not user_id:
user = None
else:
user = get_object_or_none(User, pk=user_id)
if not user:
raise errors.SessionEmptyError()
user.backend = self.request.session.get("auth_backend")
return user
def _check_is_block(self, username, raise_exception=True):
ip = self.get_request_ip()
if LoginBlockUtil(username, ip).is_block():
logger.warn('Ip was blocked' + ': ' + username + ':' + ip)
exception = errors.BlockLoginError(username=username, ip=ip)
if raise_exception:
raise errors.BlockLoginError(username=username, ip=ip)
else:
return exception
def check_is_block(self, raise_exception=True):
if hasattr(self.request, 'data'):
username = self.request.data.get("username")
else:
username = self.request.POST.get("username")
self._check_is_block(username, raise_exception)
def raise_credential_error(self, error):
raise self.partial_credential_error(error=error)
def _set_partial_credential_error(self, username, ip, request):
self.partial_credential_error = partial(errors.CredentialError, username=username, ip=ip, request=request)
def get_auth_data(self, decrypt_passwd=False):
request = self.request
if hasattr(request, 'data'):
data = request.data
else:
data = request.POST
items = ['username', 'password', 'challenge', 'public_key', 'auto_login']
username, password, challenge, public_key, auto_login = bulk_get(data, *items, default='')
ip = self.get_request_ip()
self._set_partial_credential_error(username=username, ip=ip, request=request)
password = password + challenge.strip()
if decrypt_passwd:
password = self.get_decrypted_password()
return username, password, public_key, ip, auto_login
def _check_only_allow_exists_user_auth(self, username):
# 仅允许预先存在的用户认证
if settings.ONLY_ALLOW_EXIST_USER_AUTH:
exist = User.objects.filter(username=username).exists()
if not exist:
logger.error(f"Only allow exist user auth, login failed: {username}")
self.raise_credential_error(errors.reason_user_not_exist)
def _check_auth_user_is_valid(self, username, password, public_key):
user = authenticate(self.request, username=username, password=password, public_key=public_key)
if not user:
self.raise_credential_error(errors.reason_password_failed)
elif user.is_expired:
self.raise_credential_error(errors.reason_user_expired)
elif not user.is_active:
self.raise_credential_error(errors.reason_user_inactive)
return user
def _check_login_acl(self, user, ip):
# ACL 限制用户登录
from acls.models import LoginACL
is_allowed = LoginACL.allow_user_to_login(user, ip)
if not is_allowed:
raise errors.LoginIPNotAllowed(username=user.username, request=self.request)
def set_login_failed_mark(self):
ip = self.get_request_ip()
cache.set(self.key_prefix_captcha.format(ip), 1, 3600)
def set_passwd_verify_on_session(self, user: User):
self.request.session['user_id'] = str(user.id)
self.request.session['auth_password'] = 1
self.request.session['auth_password_expired_at'] = time.time() + settings.AUTH_EXPIRED_SECONDS
def check_is_need_captcha(self):
# 最近有登录失败时需要填写验证码
ip = get_request_ip(self.request)
need = cache.get(self.key_prefix_captcha.format(ip))
return need
def check_user_auth(self, decrypt_passwd=False):
self.check_is_block()
request = self.request
username, password, public_key, ip, auto_login = self.get_auth_data(decrypt_passwd=decrypt_passwd)
self._check_only_allow_exists_user_auth(username)
user = self._check_auth_user_is_valid(username, password, public_key)
# 校验login-acl规则
self._check_login_acl(user, ip)
self._check_password_require_reset_or_not(user)
self._check_passwd_is_too_simple(user, password)
self._check_passwd_need_update(user)
LoginBlockUtil(username, ip).clean_failed_count()
request.session['auth_password'] = 1
request.session['user_id'] = str(user.id)
request.session['auto_login'] = auto_login
request.session['auth_backend'] = getattr(user, 'backend', settings.AUTH_BACKEND_MODEL)
return user
def _check_is_local_user(self, user: User):
if user.source != User.Source.local:
raise self.raise_credential_error(error=errors.only_local_users_are_allowed)
def check_oauth2_auth(self, user: User, auth_backend):
ip = self.get_request_ip()
request = self.request
self._set_partial_credential_error(user.username, ip, request)
if user.is_expired:
self.raise_credential_error(errors.reason_user_expired)
elif not user.is_active:
self.raise_credential_error(errors.reason_user_inactive)
self._check_is_local_user(user)
self._check_is_block(user.username)
self._check_login_acl(user, ip)
LoginBlockUtil(user.username, ip).clean_failed_count()
MFABlockUtils(user.username, ip).clean_failed_count()
request.session['auth_password'] = 1
request.session['user_id'] = str(user.id)
request.session['auth_backend'] = auth_backend
return user
@classmethod
def generate_reset_password_url_with_flash_msg(cls, user, message):
reset_passwd_url = reverse('authentication:reset-password')
query_str = urlencode({
'token': user.generate_reset_token()
})
reset_passwd_url = f'{reset_passwd_url}?{query_str}'
message_data = {
'title': _('Please change your password'),
'message': message,
'interval': 3,
'redirect_url': reset_passwd_url,
}
return FlashMessageUtil.gen_message_url(message_data)
@classmethod
def _check_passwd_is_too_simple(cls, user: User, password):
if user.is_superuser and password == 'admin':
message = _('Your password is too simple, please change it for security')
url = cls.generate_reset_password_url_with_flash_msg(user, message=message)
raise errors.PasswdTooSimple(url)
@classmethod
def _check_passwd_need_update(cls, user: User):
if user.need_update_password:
message = _('You should to change your password before login')
url = cls.generate_reset_password_url_with_flash_msg(user, message)
raise errors.PasswdNeedUpdate(url)
@classmethod
def _check_password_require_reset_or_not(cls, user: User):
if user.password_has_expired:
message = _('Your password has expired, please reset before logging in')
url = cls.generate_reset_password_url_with_flash_msg(user, message)
raise errors.PasswordRequireResetError(url)
def check_user_auth_if_need(self, decrypt_passwd=False):
request = self.request
if request.session.get('auth_password') and \
request.session.get('user_id'):
user = self.get_user_from_session()
if user:
return user
return self.check_user_auth(decrypt_passwd=decrypt_passwd)
def check_user_mfa_if_need(self, user):
if self.request.session.get('auth_mfa'):
return
if not user.mfa_enabled:
return
unset, url = user.mfa_enabled_but_not_set()
if unset:
raise errors.MFAUnsetError(user, self.request, url)
raise errors.MFARequiredError(mfa_types=user.get_supported_mfa_types())
def mark_mfa_ok(self, mfa_type=MFAType.OTP):
self.request.session['auth_mfa'] = 1
self.request.session['auth_mfa_time'] = time.time()
self.request.session['auth_mfa_required'] = ''
self.request.session['auth_mfa_type'] = mfa_type
def check_mfa_is_block(self, username, ip, raise_exception=True):
if MFABlockUtils(username, ip).is_block():
logger.warn('Ip was blocked' + ': ' + username + ':' + ip)
exception = errors.BlockMFAError(username=username, request=self.request, ip=ip)
if raise_exception:
raise exception
else:
return exception
def check_user_mfa(self, code, mfa_type=MFAType.OTP):
user = self.get_user_from_session()
ip = self.get_request_ip()
self.check_mfa_is_block(user.username, ip)
ok = user.check_mfa(code, mfa_type=mfa_type)
if ok:
self.mark_mfa_ok()
return
raise errors.MFAFailedError(
username=user.username,
request=self.request,
ip=ip, mfa_type=mfa_type,
)
def get_ticket(self):
from tickets.models import Ticket
ticket_id = self.request.session.get("auth_ticket_id")
logger.debug('Login confirm ticket id: {}'.format(ticket_id))
if not ticket_id:
ticket = None
else:
ticket = Ticket.all().filter(id=ticket_id).first()
return ticket
def get_ticket_or_create(self, confirm_setting):
ticket = self.get_ticket()
if not ticket or ticket.status_closed:
ticket = confirm_setting.create_confirm_ticket(self.request)
self.request.session['auth_ticket_id'] = str(ticket.id)
return ticket
def check_user_login_confirm(self):
ticket = self.get_ticket()
if not ticket:
raise errors.LoginConfirmOtherError('', "Not found")
if ticket.status_open:
raise errors.LoginConfirmWaitError(ticket.id)
elif ticket.state_approve:
self.request.session["auth_confirm"] = "1"
return
elif ticket.state_reject:
raise errors.LoginConfirmOtherError(
ticket.id, ticket.get_action_display()
)
elif ticket.state_close:
raise errors.LoginConfirmOtherError(
ticket.id, ticket.get_action_display()
)
else:
raise errors.LoginConfirmOtherError(
ticket.id, ticket.get_status_display()
)
def check_user_login_confirm_if_need(self, user):
if not settings.LOGIN_CONFIRM_ENABLE:
return
confirm_setting = user.get_login_confirm_setting()
if self.request.session.get('auth_confirm') or not confirm_setting:
return
self.get_ticket_or_create(confirm_setting)
self.check_user_login_confirm()
def clear_auth_mark(self):
self.request.session['auth_password'] = ''
self.request.session['auth_user_id'] = ''
self.request.session['auth_confirm'] = ''
self.request.session['auth_ticket_id'] = ''
def send_auth_signal(self, success=True, user=None, username='', reason=''):
if success:
post_auth_success.send(
sender=self.__class__, user=user, request=self.request
)
else:
post_auth_failed.send(
sender=self.__class__, username=username,
request=self.request, reason=reason
)
def redirect_to_guard_view(self):
guard_url = reverse('authentication:login-guard')
args = self.request.META.get('QUERY_STRING', '')
if args:
guard_url = "%s?%s" % (guard_url, args)
return redirect(guard_url)