# ~*~ 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 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:
Your account has been created successfully
click here to set your 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('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:
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('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, ) 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()])