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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Date: Wed Aug 23 14:01:06 2023
# Author: liuliancao <liuliancao@gmail.com>
"""Description: used for grant permission for jumpserver."""
import argparse
# pip install requests drf-httpsig
import requests, datetime, json
from httpsig.requests_auth import HTTPSignatureAuth
import sys
class JumpserverAPI:
auth = None
jms_url = ''
headers = None
def __init__(self, jms_url, app_id, app_key):
try:
self.jms_url = jms_url
signature_headers = ['(request-target)', 'accept', 'date']
self.auth = HTTPSignatureAuth(key_id=app_id, secret=app_key, algorithm='hmac-sha256', headers=signature_headers)
gmt_form = '%a, %d %b %Y %H:%M:%S GMT'
self.headers = {
'Accept': 'application/json',
'X-JMS-ORG': '00000000-0000-0000-0000-000000000002',
'Date': datetime.datetime.utcnow().strftime(gmt_form)
}
except Exception as e:
print("jumpserver auth failed "+str(e))
finally:
pass
def get_user(self, username):
url = self.jms_url + '/api/v1/users/users/?format=json&username='+username
response = requests.get(url, auth=self.auth, headers=self.headers)
users = json.loads(response.text)
if users and users[0]['username'] == username:
return users[0]
else:
return None
def get_host(self, private_ip):
url = self.jms_url + '/api/v1/assets/assets/?format=json&is_active=true&ip='+private_ip
response = requests.get(url, auth=self.auth, headers=self.headers)
host = json.loads(response.text)
if host:
return host[0]
else:
return None
def get_system_user(self, name):
url = self.jms_url + '/api/v1/assets/system-users/?format=json&name='+name
response = requests.get(url, auth=self.auth, headers=self.headers)
system_users = json.loads(response.text)
if system_users:
return system_users[0]
else:
return None
def get_permission_rule(self, name):
url = self.jms_url + '/api/v1/perms/asset-permissions/?format=json&name='+name
response = requests.get(url, auth=self.auth, headers=self.headers)
rules = json.loads(response.text)
if rules:
return rules[0]
else:
return None
def update_rule(self, rule):
url = self.jms_url + '/api/v1/perms/asset-permissions/{id}/'.format(id=rule['id'])
data = {
"name": rule['name'],
"date_expired": rule['date_expired'],
"users": rule['users'],
"assets": rule['assets'],
"system_users": rule['system_users'],
"is_active": rule['is_active'],
"is_valid": rule['is_valid'],
"actions": rule['actions']
}
response = requests.put(url, auth=self.auth, headers=self.headers, data=data)
res = json.loads(response.text)
def create_rule(self, name, user_ids, system_user_ids, asset_ids, expires_at='2323/08/23 13:34:16 +0800', is_active="true", is_valid="true"):
# @name: rule name
# @user_ids: the grant users ids ["xxx"]
# @system_user_ids: the user to assets, root/administrator etc, like ["xxx"]
# @asset_ids: the hosts
# @expires_at: the rule's expire time
url = self.jms_url + '/api/v1/perms/asset-permissions/?format=json'
rule = {
"name": name,
"date_expired": expires_at,
"users": user_ids,
"assets": asset_ids,
"system_users": system_user_ids,
"is_active": is_active,
"is_valid": is_valid,
"actions": [
"all"
],
}
response = requests.post(url, auth=self.auth, headers=self.headers, data=rule)
res = json.loads(response.text)
def grant_host(self, private_ip, username, system_user, expires_day):
# @expires_day: should be like '2023-08-23'
# expires_at: should be like '2023/08/23 13:34:16 +0800'
# user, system_user, host should be ready
user = self.get_user(username)
if not user:
print("username not found, please first login into "+self.jms_url)
return
host = self.get_host(private_ip)
if not host:
print("asset not found or not active, please check "+private_ip)
return
system_user = self.get_system_user(name=system_user)
if not system_user:
print("system user not found or not active, please check "+system_user)
return
# rule will be used if it's expired
rule_name = ("awx-create-{private_ip}-{username}-{system_user}").format(private_ip=private_ip,username=username,system_user=system_user['name'])
rule = self.get_permission_rule(rule_name)
expires_at = expires_day.replace('-','/') + ' 23:59:59 +0800'
if rule:
rule['date_expired'] = expires_at
rule['is_active'] = "true"
rule['is_expired'] = "false"
rule['is_valid'] = "true"
self.update_rule(rule)
print("update the rule to valid"+rule)
else:
self.create_rule(
name=rule_name,
asset_ids=[host['id']],
user_ids=[user['id']],
system_user_ids=[system_user['id']],
expires_at=expires_at
)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='jumpserver grant permission')
parser.add_argument('--jmsUrl', nargs='?', default='', required=True, help="jumpserver root url https:/xxxx.com")
parser.add_argument('--appID', nargs='?', default='', required=True, help="jumpserver app id")
parser.add_argument('--appKey', nargs='?', default='', required=True, help="jump server app key")
parser.add_argument('--username', nargs='?', default='', required=True, help="user name like nick")
parser.add_argument('--asset', nargs='?', default='', required=True, help="asset like 192.168.0.111")
parser.add_argument('--expiresDay', nargs='?', default='', required=True, help="expires day like 2023-08-24")
parser.add_argument('--systemUser', nargs='?', default='', required=True, help="system user name like jenkins-administrator")
args = parser.parse_args()
jms_url = args.jmsUrl
app_id = args.appID
app_key = args.appKey
jp = JumpserverAPI(jms_url, app_id, app_key)
# print(jp.get_user('lqx'))
# print(jp.get_user('lqx2'))
# print(jp.get_host('192.168.0.177'))
# print(jp.get_host('192.168.8.8'))
# print(jp.get_system_user(name='ssjjmb-ops-win10'))
# print(jp.get_system_user(name='ssjjmb-ops-win11'))
# print(jp.get_permission_rule(name='191-single-battle-dev'))
# print(jp.get_permission_rule(name='191-single-battle-dev1'))
# print(jp.grant_host(private_ip="192.168.0.177",username="lqx",system_user="lqx_windows_share",expires_day="2023-08-27"))
print(jp.grant_host(private_ip=args.asset,username=args.username,system_user=args.systemUser,expires_day=args.expiresDay))
|