mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2026-08-04 10:23:03 -05:00
Adds delegation listing and creation to the GUI.
This commit is contained in:
+8
-6
@@ -16,6 +16,7 @@
|
||||
#
|
||||
|
||||
import re
|
||||
import urllib
|
||||
|
||||
class ACI:
|
||||
"""
|
||||
@@ -25,10 +26,10 @@ class ACI:
|
||||
"""
|
||||
|
||||
def __init__(self,acistr=None):
|
||||
self.name = ''
|
||||
self.source_group = ''
|
||||
self.dest_group = ''
|
||||
self.attrs = []
|
||||
self.name = ''
|
||||
if acistr is not None:
|
||||
self.parse_acistr(acistr)
|
||||
|
||||
@@ -40,15 +41,15 @@ class ACI:
|
||||
# dn's aren't typed in, but searched for, and the search results
|
||||
# will return escaped dns
|
||||
|
||||
acistr = ('(targetattr = "%s")' +
|
||||
acistr = ('(targetattr="%s")' +
|
||||
'(targetfilter="(memberOf=%s)")' +
|
||||
'(version 3.0;' +
|
||||
'acl "%s";' +
|
||||
'allow (write) ' +
|
||||
'groupdn="%s";)') % (attrs_str,
|
||||
'groupdn="ldap:///%s";)') % (attrs_str,
|
||||
self.dest_group,
|
||||
self.name,
|
||||
self.source_group)
|
||||
urllib.quote(self.source_group, "/=, "))
|
||||
return acistr
|
||||
|
||||
def _match(self, prefix, inputstr):
|
||||
@@ -89,7 +90,7 @@ class ACI:
|
||||
def parse_acistr(self, acistr):
|
||||
"""Parses the acistr. If the string isn't recognized, a SyntaxError
|
||||
is raised."""
|
||||
acistr = self._match('(targetattr = ', acistr)
|
||||
acistr = self._match('(targetattr=', acistr)
|
||||
(attrstr, acistr) = self._match_str(acistr)
|
||||
self.attrs = attrstr.split(' || ')
|
||||
|
||||
@@ -107,7 +108,8 @@ class ACI:
|
||||
|
||||
acistr = self._match(';allow (write) groupdn=', acistr)
|
||||
(src_dn_str, acistr) = self._match_str(acistr)
|
||||
self.source_group = src_dn_str
|
||||
src_dn_str = self._match('ldap:///', src_dn_str)
|
||||
self.source_group = urllib.unquote(src_dn_str)
|
||||
|
||||
acistr = self._match(';)', acistr)
|
||||
if len(acistr) > 0:
|
||||
|
||||
@@ -54,6 +54,14 @@ class IPAClient:
|
||||
if self.local:
|
||||
self.transport.set_krbccache(krbccache)
|
||||
|
||||
# Higher-level API
|
||||
|
||||
def get_aci_entry(self, sattrs=None):
|
||||
"""Returns the entry containing access control ACIs."""
|
||||
|
||||
result = self.transport.get_aci_entry(sattrs)
|
||||
return entity.Entity(result)
|
||||
|
||||
# General searches
|
||||
|
||||
def get_entry_by_dn(self,dn,sattrs=None):
|
||||
|
||||
@@ -67,6 +67,23 @@ class RPCClient:
|
||||
|
||||
return obj
|
||||
|
||||
# Higher-level API
|
||||
|
||||
def get_aci_entry(self, sattrs=None):
|
||||
"""Returns the entry containing access control ACIs."""
|
||||
server = self.setup_server()
|
||||
if sattrs is None:
|
||||
sattrs = "__NONE__"
|
||||
try:
|
||||
result = server.get_aci_entry(sattrs)
|
||||
except xmlrpclib.Fault, fault:
|
||||
raise ipaerror.gen_exception(fault.faultCode, fault.faultString)
|
||||
except socket.error, (value, msg):
|
||||
raise xmlrpclib.Fault(value, msg)
|
||||
|
||||
return ipautil.unwrap_binary_data(result)
|
||||
|
||||
|
||||
# General searches
|
||||
|
||||
def get_entry_by_dn(self,dn,sattrs=None):
|
||||
|
||||
@@ -22,15 +22,16 @@ sys.path.insert(0, ".")
|
||||
|
||||
import unittest
|
||||
import aci
|
||||
import urllib
|
||||
|
||||
|
||||
class TestACI(unittest.TestCase):
|
||||
acitemplate = ('(targetattr = "%s")' +
|
||||
acitemplate = ('(targetattr="%s")' +
|
||||
'(targetfilter="(memberOf=%s)")' +
|
||||
'(version 3.0;' +
|
||||
'acl "%s";' +
|
||||
'allow (write) ' +
|
||||
'groupdn="%s";)')
|
||||
'groupdn="ldap:///%s";)')
|
||||
|
||||
def setUp(self):
|
||||
self.aci = aci.ACI()
|
||||
@@ -52,6 +53,20 @@ class TestACI(unittest.TestCase):
|
||||
|
||||
self.assertEqual(aci, exportaci)
|
||||
|
||||
def testURLEncodedExport(self):
|
||||
self.aci.source_group = 'cn=foo " bar, dc=freeipa, dc=org'
|
||||
self.aci.dest_group = 'cn=bar, dc=freeipa, dc=org'
|
||||
self.aci.name = 'this is a "name'
|
||||
self.aci.attrs = ['field1', 'field2', 'field3']
|
||||
|
||||
exportaci = self.aci.export_to_string()
|
||||
aci = TestACI.acitemplate % ('field1 || field2 || field3',
|
||||
self.aci.dest_group,
|
||||
'this is a "name',
|
||||
urllib.quote(self.aci.source_group, "/=, "))
|
||||
|
||||
self.assertEqual(aci, exportaci)
|
||||
|
||||
def testSimpleParse(self):
|
||||
attr_str = 'field3 || field4 || field5'
|
||||
dest_dn = 'cn=dest\\"group, dc=freeipa, dc=org'
|
||||
@@ -66,6 +81,21 @@ class TestACI(unittest.TestCase):
|
||||
self.assertEqual(name, self.aci.name)
|
||||
self.assertEqual(src_dn, self.aci.source_group)
|
||||
|
||||
def testUrlEncodedParse(self):
|
||||
attr_str = 'field3 || field4 || field5'
|
||||
dest_dn = 'cn=dest\\"group, dc=freeipa, dc=org'
|
||||
name = 'my name'
|
||||
src_dn = 'cn=src " group, dc=freeipa, dc=org'
|
||||
|
||||
acistr = TestACI.acitemplate % (attr_str, dest_dn, name,
|
||||
urllib.quote(src_dn, "/=, "))
|
||||
self.aci.parse_acistr(acistr)
|
||||
|
||||
self.assertEqual(['field3', 'field4', 'field5'], self.aci.attrs)
|
||||
self.assertEqual(dest_dn, self.aci.dest_group)
|
||||
self.assertEqual(name, self.aci.name)
|
||||
self.assertEqual(src_dn, self.aci.source_group)
|
||||
|
||||
def testInvalidParse(self):
|
||||
try:
|
||||
self.aci.parse_acistr('foo bar')
|
||||
|
||||
Reference in New Issue
Block a user