forked from delian/pythonUnifiAPI
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathCloudAPI.py
More file actions
158 lines (144 loc) · 5.29 KB
/
Copy pathCloudAPI.py
File metadata and controls
158 lines (144 loc) · 5.29 KB
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
import http.client
import http.cookiejar
import urllib.request
import urllib.parse
import urllib.error
import json
import ssl
import time
import inspect
class CloudAPI:
username = ''
password = ''
version = '4.8.20'
baseurl = 'https://sso.ubnt.com/api/sso/v1'
loggedin = False
debug = False
cookies = ''
requesttype = 'POST'
site = 'default'
def log(self, *args):
if self.debug:
print(*args)
def __init__(self, username=None, password=None, debug=None,
requesttype=None, baseurl=None):
if username:
self.username = username
if password:
self.password = password
if debug:
self.debug = debug
if requesttype:
self.requesttype = requesttype
if baseurl:
self.baseurl = baseurl
ssl._create_default_https_context = ssl._create_unverified_context # This is the way to allow unverified SSL
self.cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPHandler(debuglevel=1 if self.debug else 0),
urllib.request.HTTPSHandler(debuglevel=1 if self.debug else 0),
urllib.request.HTTPCookieProcessor(self.cj))
opener.addheaders = [('User-agent', 'Mozilla/5.0')]
urllib.request.install_opener(opener)
def request(self, url, data=None, headers=None, method='POST', baseurl = None):
# req = None
headers = headers or {
'Content-type': 'application/json',
'Referer': 'https://account.ubnt.com/login?redirect=https%3A%2F%2Funifi.ubnt.com',
'Origin': 'https://account.ubnt.com',
'dnt': 1
}
if not baseurl:
baseurl = self.baseurl
self.log('Request to %s with data %s' % (baseurl + url, data))
if data:
req = urllib.request.Request(url=baseurl + url, data=json.dumps(data).encode("utf8"), headers=headers, method=method)
else:
req = urllib.request.Request(url=baseurl + url, headers=headers, method='GET')
return urllib.request.urlopen(req)
def reqjson(self, url, data=None, headers=None, method='POST', autologin=True, baseurl=None):
if autologin:
self.login()
resp = self.request(url, data, headers, method, baseurl)
content = json.loads(resp.read().decode('utf8'))
return content
def response(self, content, type="UnifiError", description="UnifiError"):
if 'meta' in content and 'rc' in content['meta'] and content['meta']['rc'] == 'ok':
self.log('%s, %s completed successfully' % (type, description))
return content
self.log('%s, %s completed not successfully' % (type, description))
raise Exception(type, description, content)
def login(self, username=None, password=None):
"""
Login the system
:param username:
:param password:
:return:
"""
if self.loggedin:
return
if username:
self.username = username
if password:
self.password = password
try:
content = self.reqjson("/login",
{'user': self.username, 'password': self.password},
autologin=False
)
self.log("Successful logged in to %s with %s" % (self.baseurl, self.username))
self.log("Response: %s" % content)
self.log("Cookies: %s" % self.cj)
self.loggedin = True
return content
except urllib.error.HTTPError as e:
self.log("Cannot log in to %s with %s" % (self.baseurl, self.username))
self.log("Problem is %s" % e)
self.loggedin = False
return {}
def logout(self):
"""
Logout
:return:
"""
if not self.loggedin:
return
resp = self.request("/logout"," ")
self.log('Resp is %s' % resp.read())
self.loggedin = False
return {}
def self(self):
"""
Get information about yourself
:return:
"""
content = self.reqjson("/user/self")
return content
def devices(self):
"""
List the registered devices
:return:
"""
content = self.reqjson("/devices", baseurl="https://device-airos.svc.ubnt.com/api/airos/v1/unifi")
return content
def delete_device(self, device_id):
"""
Delete device by device id
:param device_id:
:return:
"""
content = self.reqjson("/devices/"+str(device_id),
baseurl="https://device-airos.svc.ubnt.com/api/airos/v1/unifi",
method = "DELETE"
)
return content
def launch_dashboard(self, d):
"""
Return info about the dashboard
:param d: dashboard id
:return:
"""
content = self.reqjson("/turn/creds?username="+str(d),
baseurl="https://device-airos.svc.ubnt.com/api/airos/v1/unifi",
method = "GET"
)
return content