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
176
177
178
179
180
181
182
183
184
|
import contextlib
import os
import subprocess
import tempfile
import ldap
@contextlib.contextmanager
def ldap_client(config):
client = ldap.initialize(config["uri"], trace_level=0)
client.set_option(ldap.OPT_REFERRALS, 0)
client.simple_bind(config["binddn"], config["bindpw"])
try:
yield client
finally:
client.unbind()
def build_config():
config = {}
LDAP_CONF = __file__.rsplit(".", 1)[0] + ".conf"
with open(LDAP_CONF) as f:
for line in f:
if line.startswith('binddn'):
if "binddn" not in config:
config["binddn"] = line.split(' ', 1)[1].strip()
elif line.startswith('bindpw'):
if "bindpw" not in config:
config["bindpw"] = line.split(' ', 1)[1].strip()
elif line.startswith('base'):
if "basedn" not in config:
config["basedn"] = line.split(' ', 1)[1].strip()
elif line.startswith('uri'):
if "uri" not in config:
config["uri"] = line.split(' ', 1)[1].strip()
return config
def validate_key(pubkey):
with tempfile.NamedTemporaryFile(delete=True) as f:
f.write(pubkey.encode('utf-8'))
f.flush()
try:
args = ['ssh-keygen', '-l', '-f', f.name]
subprocess.check_output(args, stderr=subprocess.PIPE)
except:
return False
return True
def do_query(search_attr='uid', search_pat='*', attrlist=[]):
config = build_config()
with ldap_client(config) as client:
result = client.search_s(
config["basedn"],
ldap.SCOPE_SUBTREE,
'(%s=%s)' % (search_attr, search_pat),
attrlist)
return result
def do_complex_query(base = None, search_filter='(uid=*)', \
attrlist=[], scope=ldap.SCOPE_SUBTREE):
"""This allows you to perform more complex LDAP queries by letting
you specify your own LDAP filter, change the basedn for the query,
or change the scope of the query.
Without any args, this will return the same result as
a call to do_query().
Examples:
search for uid's start with a 'k' but end with an 'n':
do_complex_query(search_filter="(&(uid=k*)(uid=*n))")
get a list of all groups in Linaro that start with an l:
do_complex_query(
base="ou=groups,dc=linaro,dc=org",
search_filter="(cn=l*)"
)
lookup on a specific DN:
do_complex_query(
base="uid=some.person,ou=staff,ou=accounts,dc=linaro,dc=org",
search_filter="(objectClass=*)"
scope=linaro_ldap.ldap.SCOPE_BASE,
attrlist=['displayName']
)
"""
config = build_config()
if base is None:
base = config["basedn"]
with ldap_client(config) as client:
result = client.search_s(
base,
scope,
search_filter,
attrlist)
return result
def get_users_and_keys(only_validated=False):
"""Gets all the users and their associated SSH key.
:return A list of tuples (uid, ssh_key), only if the user has an SSH
key.
"""
result = do_query(attrlist=['uid', 'sshPublicKey'])
all_users = {}
for row in result:
try:
# Just pick the first UID, it does not really matter how the
# user is called, it will access the git repository via the
# 'git' user.
uid = row[1]['uid'][0].decode('utf-8')
ssh_keys = row[1]['sshPublicKey']
for index, ssh_key in enumerate(ssh_keys):
ssh_key=ssh_key.decode('utf-8')
if not only_validated or validate_key(ssh_key):
key_name = '{0}@key_{1}.pub'.format(uid, index)
all_users.setdefault(uid, []).append((key_name, ssh_key))
except KeyError:
# If there are no SSH keys, skip this user.
pass
return all_users
def get_groups_and_users(ignore_list=[]):
result = do_query(attrlist=['uid', 'memberOf'])
all_groups = {}
for row in result:
try:
uid = row[1]['uid'][0].decode('utf-8')
groups = row[1]['memberOf']
if (type(ignore_list) is list) and (uid in ignore_list):
continue
for group in groups:
group = group.decode('utf-8').split(",")[0].split("=")[1]
all_groups.setdefault(group, []).append((uid))
except KeyError:
# This user is not in any groups
pass
return all_groups
def get_employees_by_team(ignore_list=[]):
"""Returns a dict of all current employees organized by the
department and team based on the departmentNumber field
on the employee user account.
Entries are in the form of:
entry[department][team] - [list of users on the team]
"""
results = do_complex_query(
search_filter='(&(objectClass=inetOrgPerson)(employeeType=*))',
attrlist=['*'],
base='ou=staff,ou=accounts,dc=linaro,dc=org'
)
TEAMS = {}
for entry in results:
uid = entry[1]['uid'].pop().decode('utf-8')
if uid in ignore_list:
continue
department = entry[1]['departmentNumber'].pop().decode('utf-8')
if department is None or department == 'None|None':
continue
segment,group = department.split('|')
if segment not in TEAMS:
TEAMS[segment] = {}
if group not in TEAMS[segment]:
TEAMS[segment][group] = []
user_info = {}
for k in list(entry[1].keys()):
if len(entry[1][k]) > 0:
user_info[k] = entry[1][k].pop().decode('utf-8')
TEAMS[segment][group].append(user_info)
return TEAMS
|
