mirror of https://github.com/jumpserver/jumpserver
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
6.3 KiB
224 lines
6.3 KiB
# ~*~ coding: utf-8 ~*~
|
|
#
|
|
from __future__ import unicode_literals
|
|
import logging
|
|
import os
|
|
import re
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.mixins import UserPassesTestMixin
|
|
from django.urls import reverse_lazy
|
|
from django.utils.translation import ugettext as _
|
|
|
|
from paramiko.rsakey import RSAKey
|
|
|
|
from common.tasks import send_mail_async
|
|
from common.utils import reverse, get_object_or_none
|
|
from .models import User
|
|
|
|
|
|
try:
|
|
import cStringIO as StringIO
|
|
except ImportError:
|
|
import StringIO
|
|
|
|
|
|
logger = logging.getLogger('jumpserver')
|
|
|
|
|
|
class AdminUserRequiredMixin(UserPassesTestMixin):
|
|
login_url = reverse_lazy('users:login')
|
|
|
|
def test_func(self):
|
|
return self.request.user.is_staff
|
|
|
|
|
|
def ssh_key_gen(length=2048, password=None, username='root', hostname=None):
|
|
"""Generate user ssh private and public key
|
|
|
|
Use paramiko RSAKey generate it.
|
|
|
|
"""
|
|
|
|
if hostname is None:
|
|
hostname = os.uname()[1]
|
|
|
|
f = StringIO.StringIO()
|
|
|
|
try:
|
|
logger.debug(_('Begin to generate ssh private key ...'))
|
|
private_key_obj = RSAKey.generate(length)
|
|
private_key_obj.write_private_key(f, password=password)
|
|
private_key = f.getvalue()
|
|
|
|
public_key = "%(key_type)s %(key_content)s %(username)s@%(hostname)s" % {
|
|
'key_type': private_key_obj.get_name(),
|
|
'key_content': private_key_obj.get_base64(),
|
|
'username': username,
|
|
'hostname': hostname,
|
|
}
|
|
|
|
logger.debug(_('Finish to generate ssh private key ...'))
|
|
return private_key, public_key
|
|
|
|
except IOError:
|
|
raise IOError(_('These is error when generate ssh key.'))
|
|
|
|
|
|
def user_add_success_next(user):
|
|
subject = _('Create account successfully')
|
|
recipient_list = [user.email]
|
|
message = _("""
|
|
Hello %(name)s:
|
|
</br>
|
|
Your account has been created successfully
|
|
</br>
|
|
<a href="%(rest_password_url)s?token=%(rest_password_token)s">click here to set your password</a>
|
|
</br>
|
|
This link is valid for 1 hour. After it expires, <a href="%(forget_password_url)s?email=%(email)s">request new one</a>
|
|
|
|
</br>
|
|
---
|
|
|
|
</br>
|
|
<a href="%(login_url)s">Login direct</a>
|
|
|
|
</br>
|
|
""") % {
|
|
'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('users:login', external=True),
|
|
}
|
|
|
|
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:
|
|
</br>
|
|
Please click the link below to reset your password, if not your request, concern your account security
|
|
</br>
|
|
<a href="%(rest_password_url)s?token=%(rest_password_token)s">Click here reset password</a>
|
|
</br>
|
|
This link is valid for 1 hour. After it expires, <a href="%(forget_password_url)s?email=%(email)s">request new one<</a>
|
|
|
|
</br>
|
|
---
|
|
|
|
</br>
|
|
<a href="%(login_url)s">Login direct</a>
|
|
|
|
</br>
|
|
""") % {
|
|
'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('users:login', external=True),
|
|
}
|
|
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:
|
|
</br>
|
|
Your ssh public key has been reset by site administrator.
|
|
Please login and reset your ssh public key.
|
|
</br>
|
|
<a href="%(login_url)s">Login direct</a>
|
|
|
|
</br>
|
|
""") % {
|
|
'name': user.name,
|
|
'login_url': reverse('users:login', external=True),
|
|
}
|
|
if settings.DEBUG:
|
|
logger.debug(message)
|
|
|
|
send_mail_async.delay(subject, message, recipient_list, html_message=message)
|
|
|
|
|
|
def validate_ssh_pk(text):
|
|
"""
|
|
Expects a SSH private key as string.
|
|
Returns a boolean and a error message.
|
|
If the text is parsed as private key successfully,
|
|
(True,'') is returned. Otherwise,
|
|
(False, <message describing the error>) is returned.
|
|
|
|
from https://github.com/githubnemo/SSH-private-key-validator/blob/master/validate.py
|
|
|
|
"""
|
|
|
|
if not text:
|
|
return False, 'No text given'
|
|
|
|
startPattern = re.compile("^-----BEGIN [A-Z]+ PRIVATE KEY-----")
|
|
optionPattern = re.compile("^.+: .+")
|
|
contentPattern = re.compile("^([a-zA-Z0-9+/]{64}|[a-zA-Z0-9+/]{1,64}[=]{0,2})$")
|
|
endPattern = re.compile("^-----END [A-Z]+ PRIVATE KEY-----")
|
|
|
|
def contentState(text):
|
|
for i in range(0, len(text)):
|
|
line = text[i]
|
|
|
|
if endPattern.match(line):
|
|
if i == len(text) - 1 or len(text[i + 1]) == 0:
|
|
return True, ''
|
|
else:
|
|
return False, 'At end but content coming'
|
|
|
|
elif not contentPattern.match(line):
|
|
return False, 'Wrong string in content section'
|
|
|
|
return False, 'No content or missing end line'
|
|
|
|
def optionState(text):
|
|
for i in range(0, len(text)):
|
|
line = text[i]
|
|
|
|
if line[-1:] == '\\':
|
|
return optionState(text[i + 2:])
|
|
|
|
if not optionPattern.match(line):
|
|
return contentState(text[i + 1:])
|
|
|
|
return False, 'Expected option, found nothing'
|
|
|
|
def startState(text):
|
|
if len(text) == 0 or not startPattern.match(text[0]):
|
|
return False, 'Header is wrong'
|
|
return optionState(text[1:])
|
|
|
|
return startState([n.strip() for n in text.splitlines()])
|
|
|
|
|
|
def check_user_is_valid(**kwargs):
|
|
password = kwargs.pop('password', None)
|
|
public_key = kwargs.pop('public_key', None)
|
|
user = get_object_or_none(User, **kwargs)
|
|
|
|
if password and not user.check_password(password):
|
|
user = None
|
|
|
|
if public_key and not user.public_key == public_key:
|
|
user = None
|
|
|
|
if user and user.is_valid:
|
|
return user
|
|
|
|
return None
|