forked from CLCMacTeam/besapi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
besapi.py
157 lines (116 loc) · 4.14 KB
/
besapi.py
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
#!/usr/bin/env python
#
# Copyright 2014 The Pennsylvania State University.
#
"""
besapi.py
Created by Matt Hansen ([email protected]) on 2014-03-20.
Library for communicating with the BES (BigFix) REST API.
"""
import requests
from lxml import etree, objectify
from pkg_resources import resource_filename
class BESConnection():
def __init__(self, username, password, rootserver, verify=False):
self.session = requests.Session()
self.session.auth = (username, password)
self.rootserver = rootserver
self.verify = verify
if not self.login():
self.get('login').request.raise_for_status()
def url(self, path):
if path.startswith(self.rootserver):
url = path
else:
url = "%s/api/%s" % (self.rootserver, path)
return url
def get(self, path='help', **kwargs):
return RESTResult(self.session.get(self.url(path),
verify=self.verify,
**kwargs))
def post(self, path, data, **kwargs):
return RESTResult(self.session.post(self.url(path),
data=data,
verify=self.verify,
**kwargs))
def put(self, path, data, **kwargs):
return RESTResult(self.session.put(self.url(path),
data=data,
verify=self.verify,
**kwargs))
def delete(self, path, **kwargs):
return RESTResult(self.session.delete(self.url(path),
verify=self.verify,
**kwargs))
def login(self):
return bool(self.get('login').request.status_code == 200)
def logout(self):
self.session.auth = None
self.session.cookies.clear()
self.session.close()
__call__ = login
class RESTResult():
def __init__(self, request):
self.request = request
self.text = request.text
self._besxml = None
self._besobj = None
if ('content-type' in request.headers and
request.headers['content-type'] == 'application/xml'):
self.valid = True
elif (type(request.text) is unicode and
self.validate_xsd(request.text.encode('utf-8'))):
self.valid = True
else:
if self.validate_xsd(request.text):
self.valid = True
else:
self.valid = False
def __str__(self):
if self.valid:
return self.besxml
else:
return self.text
def __call__(self):
return self.besobj
@property
def besxml(self):
if self.valid and self._besxml is None:
self._besxml = self.xmlparse_text(self.text)
return self._besxml
@property
def besobj(self):
if self.valid and self._besobj is None:
self._besobj = self.objectify_text(self.text)
return self._besobj
def validate_xsd(self, doc):
try:
xmldoc = etree.fromstring(doc)
except:
return False
for xsd in ['BES.xsd', 'BESAPI.xsd', 'BESActionSettings.xsd']:
xmlschema_doc = etree.parse(
resource_filename(__name__, "schemas/%s" % xsd)
)
xmlschema = etree.XMLSchema(xmlschema_doc)
if xmlschema.validate(xmldoc):
return True
return False
def xmlparse_text(self, text):
if type(text) is unicode:
root_xml = etree.fromstring(text.encode('utf-8'))
else:
root_xml = text
return etree.tostring(root_xml, encoding='utf-8',
xml_declaration=True)
def objectify_text(self, text):
if type(text) is unicode:
root_xml = text.encode('utf-8')
else:
root_xml = text
return objectify.fromstring(root_xml)
def main():
import bescli
bescli.main()
if __name__ == '__main__':
main()