freeipa/ipalib/aci.py

270 lines
10 KiB
Python
Raw Normal View History

2008-10-10 23:49:05 -05:00
# Authors:
# Rob Crittenden <rcritten@redhat.com>
#
# Copyright (C) 2008 Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
2008-10-10 23:49:05 -05:00
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
2008-10-10 23:49:05 -05:00
import shlex
import re
import six
2008-10-10 23:49:05 -05:00
# The Python re module doesn't do nested parenthesis
# Break the ACI into 3 pieces: target, name, permissions/bind_rules
ACIPat = re.compile(r'\(version\s+3.0\s*;\s*ac[li]\s+\"([^\"]*)\"\s*;'
r'\s*(.*);\s*\)', re.UNICODE)
2008-10-10 23:49:05 -05:00
# Break the permissions/bind_rules out
PermPat = re.compile(r'(\w+)\s*\(([^()]*)\)\s*(.*)', re.UNICODE)
2008-10-10 23:49:05 -05:00
# Break the bind rule out
BindPat = re.compile(r'\(?([a-zA-Z0-9;\.]+)\s*(\!?=)\s*\"(.*)\"\)?',
re.UNICODE)
ACTIONS = ["allow", "deny"]
PERMISSIONS = ["read", "write", "add", "delete", "search", "compare",
"selfwrite", "proxy", "all"]
2008-10-10 23:49:05 -05:00
class ACI:
2008-10-10 23:49:05 -05:00
"""
Holds the basic data for an ACI entry, as stored in the cn=accounts
entry in LDAP. Has methods to parse an ACI string and export to an
ACI String.
"""
__hash__ = None
2008-10-10 23:49:05 -05:00
def __init__(self,acistr=None):
self.name = None
2011-04-21 03:13:06 -05:00
self.source_group = None
self.dest_group = None
2008-10-10 23:49:05 -05:00
self.orig_acistr = acistr
self.target = {}
self.action = "allow"
self.permissions = ["write"]
self.bindrule = {}
2008-10-10 23:49:05 -05:00
if acistr is not None:
self._parse_acistr(acistr)
def __getitem__(self,key):
"""Fake getting attributes by key for sorting"""
if key == 0:
return self.name
if key == 1:
return self.source_group
if key == 2:
return self.dest_group
raise TypeError("Unknown key value %s" % key)
def __repr__(self):
"""An alias for export_to_string()"""
return self.export_to_string()
def export_to_string(self):
"""Output a Directory Server-compatible ACI string"""
self.validate()
aci = ""
for t, v in sorted(self.target.items()):
op = v['operator']
if type(v['expression']) in (tuple, list):
2008-10-10 23:49:05 -05:00
target = ""
for l in v['expression']:
2008-10-10 23:49:05 -05:00
target = target + l + " || "
target = target[:-4]
aci = aci + "(%s %s \"%s\")" % (t, op, target)
2008-10-10 23:49:05 -05:00
else:
aci = aci + "(%s %s \"%s\")" % (t, op, v['expression'])
aci = aci + "(version 3.0;acl \"%s\";%s (%s) %s %s \"%s\"" % (self.name, self.action, ",".join(self.permissions), self.bindrule['keyword'], self.bindrule['operator'], self.bindrule['expression']) + ";)"
2008-10-10 23:49:05 -05:00
return aci
def _remove_quotes(self, s):
# Remove leading and trailing quotes
if s.startswith('"'):
s = s[1:]
if s.endswith('"'):
s = s[:-1]
return s
def _parse_target(self, aci):
if six.PY2:
aci = aci.encode('utf-8')
lexer = shlex.shlex(aci)
2008-10-10 23:49:05 -05:00
lexer.wordchars = lexer.wordchars + "."
var = False
op = "="
2008-10-10 23:49:05 -05:00
for token in lexer:
# We should have the form (a = b)(a = b)...
if token == "(":
var = next(lexer).strip()
operator = next(lexer)
if operator not in ("=", "!="):
# Peek at the next char before giving up
operator = operator + next(lexer)
if operator not in ("=", "!="):
raise SyntaxError("No operator in target, got '%s'" % operator)
op = operator
val = next(lexer).strip()
2008-10-10 23:49:05 -05:00
val = self._remove_quotes(val)
end = next(lexer)
2008-10-10 23:49:05 -05:00
if end != ")":
raise SyntaxError('No end parenthesis in target, got %s' % end)
if var == 'targetattr':
# Make a string of the form attr || attr || ... into a list
t = re.split(r'[^a-zA-Z0-9;\*]+', val)
self.target[var] = {}
self.target[var]['operator'] = op
self.target[var]['expression'] = t
2008-10-10 23:49:05 -05:00
else:
self.target[var] = {}
self.target[var]['operator'] = op
self.target[var]['expression'] = val
2008-10-10 23:49:05 -05:00
def _parse_acistr(self, acistr):
vstart = acistr.find('version 3.0')
if vstart < 0:
raise SyntaxError("malformed ACI, unable to find version %s" % acistr)
acimatch = ACIPat.match(acistr[vstart-1:])
if not acimatch or len(acimatch.groups()) < 2:
raise SyntaxError("malformed ACI, match for version and bind rule failed %s" % acistr)
self._parse_target(acistr[:vstart-1])
self.name = acimatch.group(1)
bindperms = PermPat.match(acimatch.group(2))
2008-10-10 23:49:05 -05:00
if not bindperms or len(bindperms.groups()) < 3:
raise SyntaxError("malformed ACI, permissions match failed %s" % acistr)
2008-10-10 23:49:05 -05:00
self.action = bindperms.group(1)
self.permissions = bindperms.group(2).replace(' ','').split(',')
self.set_bindrule(bindperms.group(3))
2008-10-10 23:49:05 -05:00
def validate(self):
"""Do some basic verification that this will produce a
valid LDAP ACI.
returns True if valid
"""
if type(self.permissions) not in (tuple, list):
raise SyntaxError("permissions must be a list")
2008-10-10 23:49:05 -05:00
for p in self.permissions:
if p.lower() not in PERMISSIONS:
raise SyntaxError("invalid permission: '%s'" % p)
2008-10-10 23:49:05 -05:00
if not self.name:
raise SyntaxError("name must be set")
if not isinstance(self.name, str):
raise SyntaxError("name must be a string")
2008-10-10 23:49:05 -05:00
if not isinstance(self.target, dict) or len(self.target) == 0:
raise SyntaxError("target must be a non-empty dictionary")
if not isinstance(self.bindrule, dict):
raise SyntaxError("bindrule must be a dictionary")
if not self.bindrule.get('operator') or not self.bindrule.get('keyword') or not self.bindrule.get('expression'):
raise SyntaxError("bindrule is missing a component")
return True
def set_target_filter(self, filter, operator="="):
self.target['targetfilter'] = {}
if not filter.startswith("("):
filter = "(" + filter + ")"
self.target['targetfilter']['expression'] = filter
self.target['targetfilter']['operator'] = operator
def set_target_attr(self, attr, operator="="):
if not attr:
if 'targetattr' in self.target:
del self.target['targetattr']
return
if type(attr) not in (tuple, list):
attr = [attr]
self.target['targetattr'] = {}
self.target['targetattr']['expression'] = attr
self.target['targetattr']['operator'] = operator
def set_target(self, target, operator="="):
assert target.startswith("ldap:///")
self.target['target'] = {}
self.target['target']['expression'] = target
self.target['target']['operator'] = operator
def set_bindrule(self, bindrule):
if bindrule.startswith('(') != bindrule.endswith(')'):
raise SyntaxError("non-matching parentheses in bindrule")
match = BindPat.match(bindrule)
if not match or len(match.groups()) < 3:
raise SyntaxError("malformed bind rule")
self.set_bindrule_keyword(match.group(1))
self.set_bindrule_operator(match.group(2))
self.set_bindrule_expression(match.group(3).replace('"',''))
def set_bindrule_keyword(self, keyword):
self.bindrule['keyword'] = keyword
def set_bindrule_operator(self, operator):
self.bindrule['operator'] = operator
def set_bindrule_expression(self, expression):
self.bindrule['expression'] = expression
def isequal(self, b):
"""
Compare the current ACI to another one to see if they are
the same.
returns True if equal, False if not.
"""
assert isinstance(b, ACI)
try:
2009-06-01 12:04:01 -05:00
if self.name.lower() != b.name.lower():
return False
if set(self.permissions) != set(b.permissions):
return False
if self.bindrule.get('keyword') != b.bindrule.get('keyword'):
return False
if self.bindrule.get('operator') != b.bindrule.get('operator'):
return False
if self.bindrule.get('expression') != b.bindrule.get('expression'):
return False
if self.target.get('targetfilter',{}).get('expression') != b.target.get('targetfilter',{}).get('expression'):
return False
if self.target.get('targetfilter',{}).get('operator') != b.target.get('targetfilter',{}).get('operator'):
return False
if set(self.target.get('targetattr', {}).get('expression', ())) != set(b.target.get('targetattr',{}).get('expression', ())):
return False
if self.target.get('targetattr',{}).get('operator') != b.target.get('targetattr',{}).get('operator'):
return False
if self.target.get('target',{}).get('expression') != b.target.get('target',{}).get('expression'):
return False
if self.target.get('target',{}).get('operator') != b.target.get('target',{}).get('operator'):
return False
except Exception:
# If anything throws up then they are not equal
return False
# We got this far so lets declare them the same
2008-10-10 23:49:05 -05:00
return True
__eq__ = isequal
def __ne__(self, b):
return not self == b