# ~*~ coding: utf-8 ~*~
#
import os
import re
import pyotp
import base64
import logging
from django.http import Http404
from django.conf import settings
from django.utils.translation import ugettext as _
from django.core.cache import cache
from datetime import datetime
from common.tasks import send_mail_async
from common.utils import reverse, get_object_or_none
from .models import User
logger = logging.getLogger('jumpserver')
def construct_user_created_email_body(user):
default_body = _("""
Your account has been created successfully
""") % {
'username': user.username,
'rest_password_url': reverse('users:reset-password', external=True),
'rest_password_token': user.generate_reset_token(),
'forget_password_url': reverse('users:forgot-password', external=True),
'email': user.email,
'login_url': reverse('authentication:login', external=True),
}
if settings.EMAIL_CUSTOM_USER_CREATED_BODY:
custom_body = '' + settings.EMAIL_CUSTOM_USER_CREATED_BODY + '
'
else:
custom_body = ''
body = custom_body + default_body
return body
def send_user_created_mail(user):
recipient_list = [user.email]
subject = _('Create account successfully')
if settings.EMAIL_CUSTOM_USER_CREATED_SUBJECT:
subject = settings.EMAIL_CUSTOM_USER_CREATED_SUBJECT
honorific = '' + _('Hello %(name)s') % {'name': user.name} + ':
'
if settings.EMAIL_CUSTOM_USER_CREATED_HONORIFIC:
honorific = '' + settings.EMAIL_CUSTOM_USER_CREATED_HONORIFIC + ':
'
body = construct_user_created_email_body(user)
signature = 'jumpserver
'
if settings.EMAIL_CUSTOM_USER_CREATED_SIGNATURE:
signature = '' + settings.EMAIL_CUSTOM_USER_CREATED_SIGNATURE + '
'
message = honorific + body + signature
if settings.DEBUG:
try:
print(message)
except OSError:
pass
send_mail_async.delay(subject, message, recipient_list, html_message=message)
def send_reset_password_mail(user):
subject = _('Reset password')
recipient_list = [user.email]
message = _("""
Hello %(name)s:
Please click the link below to reset your password, if not your request, concern your account security
Click here reset password
This link is valid for 1 hour. After it expires, request new one
---
Login direct
""") % {
'name': user.name,
'rest_password_url': reverse('users:reset-password', external=True),
'rest_password_token': user.generate_reset_token(),
'forget_password_url': reverse('users:forgot-password', external=True),
'email': user.email,
'login_url': reverse('authentication:login', external=True),
}
if settings.DEBUG:
logger.debug(message)
send_mail_async.delay(subject, message, recipient_list, html_message=message)
def send_password_expiration_reminder_mail(user):
subject = _('Security notice')
recipient_list = [user.email]
message = _("""
Hello %(name)s:
Your password will expire in %(date_password_expired)s,
For your account security, please click on the link below to update your password in time
Click here update password
If your password has expired, please click
Password expired
to apply for a password reset email.
---
Login direct
""") % {
'name': user.name,
'date_password_expired': datetime.fromtimestamp(datetime.timestamp(
user.date_password_expired)).strftime('%Y-%m-%d %H:%M'),
'update_password_url': reverse('users:user-password-update', external=True),
'forget_password_url': reverse('users:forgot-password', external=True),
'email': user.email,
'login_url': reverse('authentication:login', external=True),
}
if settings.DEBUG:
logger.debug(message)
send_mail_async.delay(subject, message, recipient_list, html_message=message)
def send_user_expiration_reminder_mail(user):
subject = _('Expiration notice')
recipient_list = [user.email]
message = _("""
Hello %(name)s:
Your account will expire in %(date_expired)s,
In order not to affect your normal work, please contact the administrator for confirmation.
""") % {
'name': user.name,
'date_expired': datetime.fromtimestamp(datetime.timestamp(
user.date_expired)).strftime('%Y-%m-%d %H:%M'),
}
if settings.DEBUG:
logger.debug(message)
send_mail_async.delay(subject, message, recipient_list, html_message=message)
def send_reset_ssh_key_mail(user):
subject = _('SSH Key Reset')
recipient_list = [user.email]
message = _("""
Hello %(name)s:
Your ssh public key has been reset by site administrator.
Please login and reset your ssh public key.
Login direct
""") % {
'name': user.name,
'login_url': reverse('authentication:login', external=True),
}
if settings.DEBUG:
logger.debug(message)
send_mail_async.delay(subject, message, recipient_list, html_message=message)
def get_user_or_pre_auth_user(request):
user = request.user
if user.is_authenticated:
return user
pre_auth_user_id = request.session.get('user_id')
user = None
if pre_auth_user_id:
user = get_object_or_none(User, pk=pre_auth_user_id)
return user
def redirect_user_first_login_or_index(request, redirect_field_name):
if request.user.is_first_login:
return reverse('users:user-first-login')
url_in_post = request.POST.get(redirect_field_name)
if url_in_post:
return url_in_post
url_in_get = request.GET.get(redirect_field_name, reverse('index'))
return url_in_get
def generate_otp_uri(username, otp_secret_key=None, issuer="JumpServer"):
if otp_secret_key is None:
otp_secret_key = base64.b32encode(os.urandom(10)).decode('utf-8')
totp = pyotp.TOTP(otp_secret_key)
otp_issuer_name = settings.OTP_ISSUER_NAME or issuer
uri = totp.provisioning_uri(name=username, issuer_name=otp_issuer_name)
return uri, otp_secret_key
def check_otp_code(otp_secret_key, otp_code):
if not otp_secret_key or not otp_code:
return False
totp = pyotp.TOTP(otp_secret_key)
otp_valid_window = settings.OTP_VALID_WINDOW or 0
return totp.verify(otp=otp_code, valid_window=otp_valid_window)
def get_password_check_rules():
check_rules = []
for rule in settings.SECURITY_PASSWORD_RULES:
key = "id_{}".format(rule.lower())
value = getattr(settings, rule)
if not value:
continue
check_rules.append({'key': key, 'value': int(value)})
return check_rules
def check_password_rules(password):
pattern = r"^"
if settings.SECURITY_PASSWORD_UPPER_CASE:
pattern += '(?=.*[A-Z])'
if settings.SECURITY_PASSWORD_LOWER_CASE:
pattern += '(?=.*[a-z])'
if settings.SECURITY_PASSWORD_NUMBER:
pattern += '(?=.*\d)'
if settings.SECURITY_PASSWORD_SPECIAL_CHAR:
pattern += '(?=.*[`~!@#\$%\^&\*\(\)-=_\+\[\]\{\}\|;:\'\",\.<>\/\?])'
pattern += '[a-zA-Z\d`~!@#\$%\^&\*\(\)-=_\+\[\]\{\}\|;:\'\",\.<>\/\?]'
pattern += '.{' + str(settings.SECURITY_PASSWORD_MIN_LENGTH-1) + ',}$'
match_obj = re.match(pattern, password)
return bool(match_obj)
key_prefix_limit = "_LOGIN_LIMIT_{}_{}"
key_prefix_block = "_LOGIN_BLOCK_{}"
# def increase_login_failed_count(key_limit, key_block):
def increase_login_failed_count(username, ip):
key_limit = key_prefix_limit.format(username, ip)
count = cache.get(key_limit)
count = count + 1 if count else 1
limit_time = settings.SECURITY_LOGIN_LIMIT_TIME
cache.set(key_limit, count, int(limit_time)*60)
def get_login_failed_count(username, ip):
key_limit = key_prefix_limit.format(username, ip)
count = cache.get(key_limit, 0)
return count
def clean_failed_count(username, ip):
key_limit = key_prefix_limit.format(username, ip)
key_block = key_prefix_block.format(username)
cache.delete(key_limit)
cache.delete(key_block)
def is_block_login(username, ip):
count = get_login_failed_count(username, ip)
key_block = key_prefix_block.format(username)
limit_count = settings.SECURITY_LOGIN_LIMIT_COUNT
limit_time = settings.SECURITY_LOGIN_LIMIT_TIME
if count >= limit_count:
cache.set(key_block, 1, int(limit_time)*60)
if count and count >= limit_count:
return True
def is_need_unblock(key_block):
if not cache.get(key_block):
return False
return True
def construct_user_email(username, email):
if '@' not in email:
if '@' in username:
email = username
else:
email = '{}@{}'.format(username, settings.EMAIL_SUFFIX)
return email
def get_current_org_members(exclude=()):
from orgs.utils import current_org
return current_org.get_org_members(exclude=exclude)
def get_source_choices():
from .models import User
choices_all = dict(User.SOURCE_CHOICES)
choices = [
(User.SOURCE_LOCAL, choices_all[User.SOURCE_LOCAL]),
]
if settings.AUTH_LDAP:
choices.append((User.SOURCE_LDAP, choices_all[User.SOURCE_LDAP]))
if settings.AUTH_OPENID:
choices.append((User.SOURCE_OPENID, choices_all[User.SOURCE_OPENID]))
if settings.AUTH_RADIUS:
choices.append((User.SOURCE_RADIUS, choices_all[User.SOURCE_RADIUS]))
return choices