1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# ~*~ coding: utf-8 ~*~
#
from __future__ import unicode_literals
import os
import logging
from paramiko.rsakey import RSAKey
from django.contrib.auth.mixins import UserPassesTestMixin
from django.urls import reverse_lazy
from django.utils.translation import ugettext as _
from common.tasks import send_mail_async
from common.utils import reverse
from users.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),
}
send_mail_async.delay(subject, message, recipient_list, html_message=message)