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
|
#!/usr/bin/env python
import sys
import httplib
import urllib
import base64
import json
import optparse
import getpass
import logging
from pprint import pprint
#from BeautifulSoup import BeautifulSoup
BASE_URL = "https://login.linaro.org:8443/crowd/rest/usermanagement/1"
AUTH = (None, None)
log = logging.getLogger()
class CrowdRequestError(Exception):
pass
def strip_html(html):
return html
html = BeautifulSoup(html, convertEntities=BeautifulSoup.HTML_ENTITIES)
if not html.body:
return html
m = ' '.join(html.body.findAll(text=True))
return m
def rest(uri, params):
c = httplib.HTTPSConnection("login.linaro.org", 8443)
auth = base64.encodestring('%s:%s' % AUTH)
headers = {"Authorization": "Basic " + auth}
url = "/crowd/rest/usermanagement/1" + uri + "?" + urllib.urlencode(params)
log.debug("REST request: %s", url)
c.request("GET", url, headers=headers)
resp = c.getresponse()
if resp.status != 200:
data = resp.read()
log.error("Non-successful response code %d from Crowd: %s", resp.status, strip_html(data))
raise CrowdRequestError(resp, data)
data = json.load(resp)
return data
if __name__ == "__main__":
optparser = optparse.OptionParser(usage="%prog users|groups|user|alises|group|usergroups|ismember|groupusers|groupgroups|members")
optparser.add_option("-u", "--user", default="rest-test", help="Crowd username")
optparser.add_option("-p", "--passwd", help="Crowd password")
optparser.add_option("-P", "--ask-passwd", action="store_true", help="Ask Crowd password")
optparser.add_option("-n", "--nested", action="store_true", help="Process nested groups")
optparser.add_option("-r", "--raw", action="store_true", help="Show raw JSON response")
optparser.add_option("-d", "--debug", action="store_true", help="Debug logging")
options, args = optparser.parse_args(sys.argv[1:])
if len(args) < 1:
optparser.error("Wrong number of arguments")
logging.basicConfig(level=(logging.DEBUG if options.debug else logging.INFO))
if options.ask_passwd:
options.passwd = getpass.getpass("Crowd password: ")
AUTH = (options.user, options.passwd)
if args[0] == "user":
pprint(rest("/user.json", {"username": args[1], "expand": "attributes"}))
elif args[0] == "aliases":
pprint(rest("/aliases.json", {"username": args[1]}))
elif args[0] == "group":
pprint(rest("/group.json", {"groupname": args[1], "expand": "attributes"}))
elif args[0] in ["users", "groups"]:
entity = args[0]
params = {"entity-type": entity[:-1]}
if len(args) > 1:
params["restriction"] = args[1]
data = rest("/search.json", params)
if options.raw:
pprint(data)
else:
names = sorted([x["name"] for x in data[entity]])
for n in names:
print n
elif args[0] == "usergroups":
url = "/user/group/nested.json" if options.nested else "/user/group/direct.json"
data = rest(url, {"username": args[1]})
if options.raw:
pprint(data)
else:
names = [x["name"] for x in data["groups"]]
names.sort()
for n in names:
print n
elif args[0] == "groupusers":
url = "/group/user/nested.json" if options.nested else "/group/user/direct.json"
data = rest(url, {"groupname": args[1]})
if options.raw:
pprint(data)
else:
names = [x["name"] for x in data["users"]]
names.sort()
for n in names:
print n
elif args[0] == "groupgroups":
url = "/group/child-group/nested.json" if options.nested else "/group/child-group/direct.json"
data = rest(url, {"groupname": args[1]})
if options.raw:
pprint(data)
else:
names = [x["name"] for x in data["groups"]]
names.sort()
for n in names:
print n
elif args[0] == "members":
def print_names(header, list):
if list:
print header
for n in list:
print "\t" + n
data = rest("/group/user/direct.json", {"groupname": args[1]})
names = sorted([x["name"] for x in data["users"]])
print_names("Users:", names)
data = rest("/group/child-group/direct.json", {"groupname": args[1]})
names = sorted([x["name"] for x in data["groups"]])
print_names("Groups:", names)
elif args[0] == "ismember":
url = "/user/group/nested.json" if options.nested else "/user/group/direct.json"
try:
data = rest(url, {"username": args[1], "groupname": args[2]})
assert data["name"] == args[2]
print "yes"
except CrowdRequestError, e:
print e.args[1]
print "no"
|
