Edit file File name : ipachangeconf.py Content :""" ipachangeconf - configuration file manipulation classes and functions partially based on authconfig code Copyright (c) 1999-2007 Red Hat, Inc. Author: Simo Sorce <ssorce@redhat.com> 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. 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/>. """ import fcntl import os import shutil import re def openLocked(filename, perms, create=True): fd = -1 flags = os.O_RDWR if create: flags = flags | os.O_CREAT try: fd = os.open(filename, flags, perms) fcntl.lockf(fd, fcntl.LOCK_EX) except OSError as err: errno, strerr = err.args if fd != -1: try: os.close(fd) except OSError: pass raise IOError(errno, strerr) return os.fdopen(fd, "r+") class IPAChangeConf(object): def __init__(self, name): self.progname = name self.indent = ("", "", "") self.assign = (" = ", "=") self.dassign = self.assign[0] self.comment = ("#",) self.dcomment = self.comment[0] self.eol = ("\n",) self.deol = self.eol[0] self.sectnamdel = ("[", "]") self.subsectdel = ("{", "}") self.backup_suffix = ".ipabkp" def setProgName(self, name): self.progname = name def setIndent(self, indent): if type(indent) is tuple: self.indent = indent elif type(indent) is str: self.indent = (indent, ) else: raise ValueError('Indent must be a list of strings') def setOptionAssignment(self, assign): if type(assign) is tuple: self.assign = assign else: self.assign = (assign, ) self.dassign = self.assign[0] def setCommentPrefix(self, comment): if type(comment) is tuple: self.comment = comment else: self.comment = (comment, ) self.dcomment = self.comment[0] def setEndLine(self, eol): if type(eol) is tuple: self.eol = eol else: self.eol = (eol, ) self.deol = self.eol[0] def setSectionNameDelimiters(self, delims): self.sectnamdel = delims def setSubSectionDelimiters(self, delims): self.subsectdel = delims def matchComment(self, line): for v in self.comment: if line.lstrip().startswith(v): return line.lstrip()[len(v):] return False def matchEmpty(self, line): if line.strip() == "": return True return False def matchSection(self, line): cl = "".join(line.strip().split()) if len(self.sectnamdel) != 2: return False if not cl.startswith(self.sectnamdel[0]): return False if not cl.endswith(self.sectnamdel[1]): return False return cl[len(self.sectnamdel[0]):-len(self.sectnamdel[1])] def matchSubSection(self, line): if self.matchComment(line): return False parts = line.split(self.dassign, 1) if len(parts) < 2: return False if parts[1].strip() == self.subsectdel[0]: return parts[0].strip() return False def matchSubSectionEnd(self, line): if self.matchComment(line): return False if line.strip() == self.subsectdel[1]: return True return False def getSectionLine(self, section): if len(self.sectnamdel) != 2: return section return self.sectnamdel[0] + section + self.sectnamdel[1] + self.deol @staticmethod def _get_debug_level_val(value): if value > 16: value = hex(value) return value def dump(self, options, level=0): output = "" if level >= len(self.indent): level = len(self.indent) - 1 for o in options: if o['type'] == "section": output += self.sectnamdel[0] + o['name'] + self.sectnamdel[1] + self.deol output += self.dump(o['value'], level + 1) continue if o['type'] == "subsection": output += self.indent[level] + o['name'] + self.dassign + self.subsectdel[0] + self.deol output += self.dump(o['value'], level + 1) output += self.indent[level] + self.subsectdel[1] + self.deol continue if o['type'] == "option": output += self.indent[level] + o['name'] + self.dassign + o['value'] + self.deol continue if o['type'] == "comment": output += self.dcomment + o['value'] + self.deol continue if o['type'] == "empty": output += self.deol continue raise SyntaxError('Unknown type: [' + o['type'] + ']') return output def parseLine(self, line): if self.matchEmpty(line): return {'name': 'empty', 'type': 'empty'} value = self.matchComment(line) if value: return {'name': 'comment', 'type': 'comment', 'value': value.rstrip()} parts = line.split(self.dassign, 1) if len(parts) < 2: raise SyntaxError('Syntax Error: Unknown line format') return {'name': parts[0].strip(), 'type': 'option', 'value': parts[1].rstrip()} def findOpts(self, opts, type, name, exclude_sections=False): num = 0 for o in opts: if o['type'] == type and o['name'] == name: return (num, o) if exclude_sections and (o['type'] == "section" or o['type'] == "subsection"): return (num, None) num += 1 return (num, None) def commentOpts(self, inopts, level=0): opts = [] if level >= len(self.indent): level = len(self.indent) - 1 for o in inopts: if o['type'] == 'section': no = self.commentOpts(o['value'], level + 1) val = self.dcomment + self.sectnamdel[0] + o['name'] + self.sectnamdel[1] opts.append({'name': 'comment', 'type': 'comment', 'value': val}) for n in no: opts.append(n) continue if o['type'] == 'subsection': no = self.commentOpts(o['value'], level + 1) val = self.indent[level] + o['name'] + self.dassign + self.subsectdel[0] opts.append({'name': 'comment', 'type': 'comment', 'value': val}) for n in no: opts.append(n) val = self.indent[level] + self.subsectdel[1] opts.append({'name': 'comment', 'type': 'comment', 'value': val}) continue if o['type'] == 'option': val = self.indent[level] + o['name'] + self.dassign + o['value'] opts.append({'name': 'comment', 'type': 'comment', 'value': val}) continue if o['type'] == 'comment': opts.append(o) continue if o['type'] == 'empty': opts.append({'name': 'comment', 'type': 'comment', 'value': ''}) continue raise SyntaxError('Unknown type: [' + o['type'] + ']') return opts def mergeOld(self, oldopts, newopts): opts = [] for o in oldopts: if o['type'] == "section" or o['type'] == "subsection": (num, no) = self.findOpts(newopts, o['type'], o['name']) if not no: opts.append(o) continue if no['action'] == "set": mo = self.mergeOld(o['value'], no['value']) opts.append({'name': o['name'], 'type': o['type'], 'value': mo}) continue if no['action'] == "comment": co = self.commentOpts(o['value']) for c in co: opts.append(c) continue if no['action'] == "remove": continue raise SyntaxError('Unknown action: [' + no['action'] + ']') if o['type'] == "comment" or o['type'] == "empty": opts.append(o) continue if o['type'] == "option": (num, no) = self.findOpts(newopts, 'option', o['name'], True) if not no: opts.append(o) continue if no['action'] == 'comment' or no['action'] == 'remove': if no['value'] is not None and o['value'] != no['value']: opts.append(o) continue if no['action'] == 'comment': opts.append({'name': 'comment', 'type': 'comment', 'value': self.dcomment + o['name'] + self.dassign + o['value']}) continue if no['action'] == 'set': opts.append(no) continue raise SyntaxError('Unknown action: [' + o['action'] + ']') raise SyntaxError('Unknown type: [' + o['type'] + ']') return opts def mergeNew(self, opts, newopts): cline = 0 for no in newopts: if no['type'] == "section" or no['type'] == "subsection": (num, o) = self.findOpts(opts, no['type'], no['name']) if not o: if no['action'] == 'set': opts.append(no) continue if no['action'] == "set": self.mergeNew(o['value'], no['value']) continue cline = num + 1 continue if no['type'] == "option": (num, o) = self.findOpts(opts, no['type'], no['name'], True) if not o: if no['action'] == 'set': opts.append(no) continue cline = num + 1 continue if no['type'] == "comment" or no['type'] == "empty": opts.insert(cline, no) cline += 1 continue raise SyntaxError('Unknown type: [' + no['type'] + ']') def merge(self, oldopts, newopts): """ Use a two pass strategy First we create a new opts tree from oldopts removing/commenting the options as indicated by the contents of newopts Second we fill in the new opts tree with options as indicated in the newopts tree (this is because entire (sub)sections may exist in the newopts that do not exist in oldopts) """ opts = self.mergeOld(oldopts, newopts) self.mergeNew(opts, newopts) return opts # TODO: Make parse() recursive? def parse(self, f): opts = [] sectopts = [] section = None subsectopts = [] subsection = None curopts = opts fatheropts = opts # Read in the old file. for line in f: # It's a section start. value = self.matchSection(line) if value: if section is not None: opts.append({'name': section, 'type': 'section', 'value': sectopts}) sectopts = [] curopts = sectopts fatheropts = sectopts section = value continue # It's a subsection start. value = self.matchSubSection(line) if value: if subsection is not None: raise SyntaxError('nested subsections are not supported yet') subsectopts = [] curopts = subsectopts subsection = value continue value = self.matchSubSectionEnd(line) if value: if subsection is None: raise SyntaxError('Unmatched end subsection terminator found') fatheropts.append({'name': subsection, 'type': 'subsection', 'value': subsectopts}) subsection = None curopts = fatheropts continue # Copy anything else as is. curopts.append(self.parseLine(line)) # Add last section if any if sectopts: opts.append({'name': section, 'type': 'section', 'value': sectopts}) return opts def changeConf(self, file, newopts): """Write settings to configuration file file is a path options is a set of dictionaries in the form: [{'name': 'foo', 'value': 'bar', 'action': 'set/comment'}] section is a section name like 'global' """ output = "" f = None try: # Do not catch an unexisting file error, we want to fail in that case shutil.copy2(file, file + self.backup_suffix) f = openLocked(file, 0o644) oldopts = self.parse(f) options = self.merge(oldopts, newopts) output = self.dump(options) # Write it out and close it. f.seek(0) f.truncate(0) f.write(output) finally: try: if f: f.close() except IOError: pass return True def newConf(self, file, options): """Write settings to new file, backup old file is a path options is a set of dictionaries in the form: [{'name': 'foo', 'value': 'bar', 'action': 'set/comment'}] section is a section name like 'global' """ output = "" f = None try: try: shutil.copy2(file, file + self.backup_suffix) except IOError as err: if err.errno == 2: # The orign file did not exist pass f = openLocked(file, 0o644) # Truncate f.seek(0) f.truncate(0) output = self.dump(options) f.write(output) finally: try: if f: f.close() except IOError: pass return True class SSSDChangeConf(IPAChangeConf): """An SSSD-specific subclass of IPAChangeConf""" OPTCRE = re.compile( r'\s*(?P<option>[^:=\s][^:=]*)' # very permissive! r'\s*=\s*' # any number of space/tab, # followed by separator # followed by any # space/tab r'(?P<value>.*)$' # everything up to EOL ) def __init__(self): IPAChangeConf.__init__(self, "SSSD") self.comment = ("#", ";") self.backup_suffix = ".bak" self.opts = [] def parseLine(self, line): """ Overrides IPAChangeConf parseLine so that lines are split using any separator in self.assign, not just the default one """ if self.matchEmpty(line): return {'name': 'empty', 'type': 'empty'} value = self.matchComment(line) if value: return {'name': 'comment', 'type': 'comment', 'value': value.rstrip()} mo = self.OPTCRE.match(line) if not mo: raise SyntaxError('Syntax Error: Unknown line format') try: name, value = mo.group('option', 'value') except IndexError: raise SyntaxError('Syntax Error: Unknown line format') return {'name': name.strip(), 'type': 'option', 'value': value.strip()} def readfp(self, fd): self.opts.extend(self.parse(fd)) def read(self, filename): fd = open(filename, 'r') self.readfp(fd) fd.close() def get(self, section, name): index, item = self.get_option_index(section, name) if item: return item['value'] def set(self, section, name, value): modkw = {'type': 'section', 'name': section, 'value': [{ 'type': 'option', 'name': name, 'value': value, 'action': 'set', }], 'action': 'set', } self.opts = self.merge(self.opts, [modkw]) def add_section(self, name, optkw, index=0): optkw.append({'type': 'empty', 'value': 'empty'}) addkw = {'type': 'section', 'name': name, 'value': optkw} self.opts.insert(index, addkw) def delete_section(self, name): self.delete_option('section', name) def sections(self): return [o for o in self.opts if o['type'] == 'section'] def has_section(self, section): return len([o for o in self.opts if o['type'] == 'section' if o['name'] == section]) > 0 def options(self, section): for opt in self.opts: if opt['type'] == 'section' and opt['name'] == section: return opt['value'] def delete_option(self, type, name, exclude_sections=False): return self.delete_option_subtree(self.opts, type, name) def delete_option_subtree(self, subtree, type, name, exclude_sections=False): index, item = self.findOpts(subtree, type, name, exclude_sections) if item: del subtree[index] return index def has_option(self, section, name): index, item = self.get_option_index(section, name) if index != -1 and item is not None: return True return False def strip_comments_empty(self, optlist): retlist = [] for opt in optlist: if opt['type'] in ('comment', 'empty'): continue retlist.append(opt) return retlist def get_option_index(self, parent_name, name, type='option'): subtree = None if parent_name: pindex, pdata = self.findOpts(self.opts, 'section', parent_name) if not pdata: return (-1, None) subtree = pdata['value'] else: subtree = self.opts return self.findOpts(subtree, type, name) Save