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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import re
from rest_framework import serializers
from django.utils.translation import ugettext_lazy as _
from common.serializers import AdaptedBulkListSerializer
from common.utils import ssh_pubkey_gen
from orgs.mixins.serializers import BulkOrgResourceModelSerializer
from ..models import SystemUser
from ..const import (
GENERAL_FORBIDDEN_SPECIAL_CHARACTERS_PATTERN,
GENERAL_FORBIDDEN_SPECIAL_CHARACTERS_ERROR_MSG
)
from .base import AuthSerializer, AuthSerializerMixin
class SystemUserSerializer(AuthSerializerMixin, BulkOrgResourceModelSerializer):
"""
系统用户
"""
auto_generate_key = serializers.BooleanField(initial=True, required=False, write_only=True)
class Meta:
model = SystemUser
list_serializer_class = AdaptedBulkListSerializer
fields = [
'id', 'name', 'username', 'password', 'public_key', 'private_key',
'login_mode', 'login_mode_display', 'priority', 'protocol',
'auto_push', 'cmd_filters', 'sudo', 'shell', 'comment',
'assets_amount', 'nodes_amount', 'auto_generate_key'
]
extra_kwargs = {
'password': {"write_only": True},
'public_key': {"write_only": True},
'private_key': {"write_only": True},
'nodes_amount': {'label': _('Node')},
'assets_amount': {'label': _('Asset')},
'login_mode_display': {'label': _('Login mode display')},
'created_by': {'read_only': True},
}
@staticmethod
def validate_name(name):
pattern = GENERAL_FORBIDDEN_SPECIAL_CHARACTERS_PATTERN
res = re.search(pattern, name)
if res is not None:
msg = GENERAL_FORBIDDEN_SPECIAL_CHARACTERS_ERROR_MSG
raise serializers.ValidationError(msg)
return name
def validate_auto_push(self, value):
login_mode = self.initial_data.get("login_mode")
protocol = self.initial_data.get("protocol")
if login_mode == SystemUser.LOGIN_MANUAL or \
protocol in [SystemUser.PROTOCOL_TELNET,
SystemUser.PROTOCOL_VNC]:
value = False
return value
def validate_auto_generate_key(self, value):
login_mode = self.initial_data.get("login_mode")
protocol = self.initial_data.get("protocol")
if self.context["request"].method.lower() != "post":
value = False
elif self.instance:
value = False
elif login_mode == SystemUser.LOGIN_MANUAL:
value = False
elif protocol in [SystemUser.PROTOCOL_TELNET, SystemUser.PROTOCOL_VNC]:
value = False
return value
def validate_username(self, username):
if username:
return username
login_mode = self.initial_data.get("login_mode")
protocol = self.initial_data.get("protocol")
if login_mode == SystemUser.LOGIN_AUTO and \
protocol != SystemUser.PROTOCOL_VNC:
msg = _('* Automatic login mode must fill in the username.')
raise serializers.ValidationError(msg)
return username
def validate_password(self, password):
super().validate_password(password)
auto_gen_key = self.initial_data.get("auto_generate_key", False)
private_key = self.initial_data.get("private_key")
login_mode = self.initial_data.get("login_mode")
if not self.instance and not auto_gen_key and not password and \
not private_key and login_mode == SystemUser.LOGIN_AUTO:
raise serializers.ValidationError(_("Password or private key required"))
return password
def validate(self, attrs):
username = attrs.get("username", "manual")
protocol = attrs.get("protocol")
auto_gen_key = attrs.get("auto_generate_key", False)
if auto_gen_key:
password = SystemUser.gen_password()
attrs["password"] = password
if protocol == SystemUser.PROTOCOL_SSH:
private_key, public_key = SystemUser.gen_key(username)
attrs["private_key"] = private_key
attrs["public_key"] = public_key
# 如果设置了private key，没有设置public key则生成
elif attrs.get("private_key", None):
private_key = attrs["private_key"]
password = attrs.get("password")
public_key = ssh_pubkey_gen(private_key, password=password,
username=username)
attrs["public_key"] = public_key
attrs.pop("auto_generate_key", None)
return attrs
@classmethod
def setup_eager_loading(cls, queryset):
""" Perform necessary eager loading of data. """
queryset = queryset.prefetch_related('cmd_filters', 'nodes')
return queryset
class SystemUserAuthSerializer(AuthSerializer):
"""
系统用户认证信息
"""
class Meta:
model = SystemUser
fields = [
"id", "name", "username", "protocol",
"login_mode", "password", "private_key",
]
class SystemUserSimpleSerializer(serializers.ModelSerializer):
"""
系统用户最基本信息的数据结构
"""
class Meta:
model = SystemUser
fields = ('id', 'name', 'username')