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
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/env python
import requests
import sys
admin_username = 'admin'
admin_password = 'admin'
domain_url = 'http://localhost:8080'
class UserCreation:
headers = {}
def __init__(self, username, password, domain):
self.username = username
self.password = password
self.domain = domain
def auth(self):
url = "{}/api/users/v1/auth/".format(self.domain)
data = {"username": self.username, "password": self.password}
resp = requests.post(url, data=data)
if resp.status_code == 200:
data = resp.json()
self.headers.update({
'Authorization': '{} {}'.format('Bearer', data['token'])
})
else:
print("用户名 或 密码 或 地址 不对")
sys.exit(2)
def get_user_detail(self, name, url):
resp = requests.get(url, headers=self.headers)
if resp.status_code == 200:
data = resp.json()
if len(data) < 1:
return None
for d in data:
if d['name'] == name:
return d
return None
return None
def get_system_user_detail(self, name):
url = '{}/api/assets/v1/system-user/?name={}'.format(self.domain, name)
return self.get_user_detail(name, url)
def create_system_user(self, info):
system_user = self.get_system_user_detail(info.get('name'))
if system_user:
return system_user
url = '{}/api/assets/v1/system-user/'.format(self.domain)
resp = requests.post(url, data=info, headers=self.headers, json=False)
if resp.status_code == 201:
return resp.json()
else:
print("创建系统用户失败: {} {}".format(info['name'], resp.content))
return None
def set_system_user_auth(self, system_user, info):
url = '{}/api/assets/v1/system-user/{}/auth-info/'.format(
self.domain, system_user['id']
)
data = {'password': info.get('password')}
resp = requests.patch(url, data=data, headers=self.headers)
if resp.status_code > 300:
print("设置系统用户密码失败: {} {}".format(
system_user.get('name'), resp.content.decode()
))
else:
return True
def get_admin_user_detail(self, name):
url = '{}/api/assets/v1/admin-user/?name={}'.format(self.domain, name)
return self.get_user_detail(name, url)
def create_admin_user(self, info):
admin_user = self.get_admin_user_detail(info.get('name'))
if admin_user:
return admin_user
url = '{}/api/assets/v1/admin-user/'.format(self.domain)
resp = requests.post(url, data=info, headers=self.headers, json=False)
if resp.status_code == 201:
return resp.json()
else:
print("创建管理用户失败: {} {}".format(info['name'], resp.content.decode()))
return None
def set_admin_user_auth(self, admin_user, info):
url = '{}/api/assets/v1/admin-user/{}/auth/'.format(
self.domain, admin_user['id']
)
data = {'password': info.get('password')}
resp = requests.patch(url, data=data, headers=self.headers)
if resp.status_code > 300:
print("设置管理用户密码失败: {} {}".format(
admin_user.get('name'), resp.content.decode()
))
else:
return True
def create_system_users(self):
print("#"*10, " 开始创建系统用户 ", "#"*10)
users = []
f = open('system_users.txt')
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
name, username, password, protocol, auto_push = line.split()[:5]
info = {
"name": name,
"username": username,
"password": password,
"protocol": protocol,
"auto_push": bool(int(auto_push)),
"login_mode": "auto"
}
users.append(info)
for i, info in enumerate(users, start=1):
system_user = self.create_system_user(info)
if system_user and self.set_system_user_auth(system_user, info):
print("[{}] 创建系统用户成功: {}".format(i, system_user['name']))
def create_admin_users(self):
print("\n", "#"*10, " 开始创建管理用户 ", "#"*10)
users = []
f = open('admin_users.txt')
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
name, username, password = line.split()[:3]
info = {
"name": name,
"username": username,
"password": password,
}
users.append(info)
for i, info in enumerate(users, start=1):
admin_user = self.create_admin_user(info)
if admin_user and self.set_admin_user_auth(admin_user, info):
print("[{}] 创建管理用户成功: {}".format(i, admin_user['name']))
def main():
api = UserCreation(username=admin_username,
password=admin_password,
domain=domain_url)
api.auth()
api.create_system_users()
api.create_admin_users()
if __name__ == '__main__':
main()