X-Git-Url: http://xvm.mit.edu/gitweb/invirt/packages/invirt-web.git/blobdiff_plain/b26a7d35e4ac578595f88eef03303cfbfd42a319..e37dd15bbf8683bab8454ea0eec8e9968ce982b2:/code/getafsgroups.py diff --git a/code/getafsgroups.py b/code/getafsgroups.py new file mode 100644 index 0000000..899de81 --- /dev/null +++ b/code/getafsgroups.py @@ -0,0 +1,96 @@ +#!/usr/bin/python +import pprint +import subprocess + +# import ldap +# l = ldap.open("W92-130-LDAP-2.mit.edu") +# # ldap.mit.edu is 1/2 broken right now so we're going to the working backend +# l.simple_bind_s("", "") + +# def getLdapGroups(user): +# """ +# getLdapGroups(user): returns a generator for the list of LDAP groups containing user +# """ +# for user_data in l.search_s("ou=affiliates,dc=mit,dc=edu", ldap.SCOPE_ONELEVEL, "uid=" + user, []): +# for group_data in l.search_s("ou=groups,dc=mit,dc=edu", ldap.SCOPE_ONELEVEL, "uniqueMember="+user_data[0], ['cn']): +# yield group_data[1]['cn'][0] + +# def checkLdapGroups(user, group): +# """ +# checkLdapGroups(user, group): returns True if and only if user is in LDAP group group +# """ +# for result_data in l.search_s("ou=affiliates,dc=mit,dc=edu", ldap.SCOPE_ONELEVEL, "uid=" + user, []): +# if l.search_s("ou=groups,dc=mit,dc=edu", ldap.SCOPE_ONELEVEL, "(&(cn=" + group + ")(uniqueMember="+result_data[0] + "))", []) != []: +# return True +# return False + +class MyException(Exception): + pass + +def getAfsGroupMembers(group, cell): + p = subprocess.Popen(["pts", "membership", group, '-c', cell], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if p.wait(): + return [] + return [line.strip() for line in p.stdout.readlines()[1:]] + +def checkAfsGroup(user, group, cell): + """ + checkAfsGroup(user, group) returns True if and only if user is in AFS group group in cell cell + """ + return user in getAfsGroupMembers(group, cell) + +def getCell(locker): + p = subprocess.Popen(["fs", "whichcell", "/mit/" + locker], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if p.wait(): + raise MyException(p.stderr.read()) + return p.stdout.read().split()[-1][1:-1] + +def getLockerAcl(locker): + p = subprocess.Popen(["fs", "listacl", "/mit/" + locker], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if p.wait(): + raise MyException(p.stderr.read()) + lines = p.stdout.readlines() + values = [] + for line in lines[1:]: + fields = line.split() + if fields[0] == 'Negative': + break + if 'rlidwka' in fields[1]: + values.append(fields[0]) + return values + +def notLockerOwner(user, locker): + """ + notLockerOwner(user, locker) returns false if and only if user administers locker. + + If the user does not own the locker, returns the string reason for + the failure. + """ + try: + cell = getCell(locker) + values = getLockerAcl(locker) + except MyException, e: + return str(e) + + for entry in values: + if entry == user or (entry[0:6] == "system" and + checkAfsGroup(user, entry, cell)): + return False + return "You don't have admin bits on /mit/" + locker + + +if __name__ == "__main__": +# print list(getldapgroups("tabbott")) + print checkAfsGroup("tabbott", "system:debathena", 'athena.mit.edu') + print checkAfsGroup("tabbott", "system:debathena", 'sipb.mit.edu') + print checkAfsGroup("tabbott", "system:debathena-root", 'athena.mit.edu') + print checkAfsGroup("tabbott", "system:hmmt-request", 'athena.mit.edu') + print notLockerOwner("tabbott", "tabbott") + print notLockerOwner("tabbott", "debathena") + print notLockerOwner("tabbott", "sipb") + print notLockerOwner("tabbott", "lsc") + print notLockerOwner("tabbott", "scripts") + print notLockerOwner("ecprice", "hmmt")