#! /usr/bin/env python # ---------------------------------------------------------------------- # Copyright (C) 2013 Kshitij Gupta # # This program is free software; you can redistribute it and/or # modify it under the terms of version 2 of the GNU General Public # License as published by the Free Software Foundation. # # 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. # # ---------------------------------------------------------------------- import argparse import re import apparmor.aa import apparmor.aamode import apparmor.severity import apparmor.cleanprofile as cleanprofile import apparmor.ui as aaui # setup module translations from apparmor.translations import init_translation _ = init_translation() parser = argparse.ArgumentParser(description=_('Perform a 3way merge on the given profiles')) parser.add_argument('mine', type=str, help=_('your profile')) parser.add_argument('base', type=str, help=_('base profile')) parser.add_argument('other', type=str, help=_('other profile')) parser.add_argument('-d', '--dir', type=str, help=_('path to profiles')) parser.add_argument('-a', '--auto', action='store_true', help=_('Automatically merge profiles, exits incase of *x conflicts')) args = parser.parse_args() profiles = [args.mine, args.base, args.other] def main(): mergeprofiles = Merge(profiles) #Get rid of common/superfluous stuff mergeprofiles.clear_common() if not args.auto: mergeprofiles.ask_the_questions('other') mergeprofiles.clear_common() mergeprofiles.ask_the_questions('base') q = apparmor.aa.hasher() q['title'] = 'Changed Local Profiles' q['headers'] = [] q['explanation'] = _('The following local profiles were changed. Would you like to save them?') q['functions'] = ['CMD_SAVE_CHANGES', 'CMD_VIEW_CHANGES', 'CMD_ABORT'] q['default'] = 'CMD_VIEW_CHANGES' q['options'] = [] q['selected'] = 0 ans = '' arg = None programs = list(mergeprofiles.user.aa.keys()) program = programs[0] while ans != 'CMD_SAVE_CHANGES': ans, arg = aaui.UI_PromptUser(q) if ans == 'CMD_SAVE_CHANGES': apparmor.aa.write_profile_ui_feedback(program) apparmor.aa.reload_base(program) elif ans == 'CMD_VIEW_CHANGES': for program in programs: apparmor.aa.original_aa[program] = apparmor.aa.deepcopy(apparmor.aa.aa[program]) #oldprofile = apparmor.serialize_profile(apparmor.original_aa[program], program, '') newprofile = apparmor.aa.serialize_profile(mergeprofiles.user.aa[program], program, '') apparmor.aa.display_changes_with_comments(mergeprofiles.user.filename, newprofile) class Merge(object): def __init__(self, profiles): user, base, other = profiles #Read and parse base profile and save profile data, include data from it and reset them apparmor.aa.read_profile(base, True) self.base = cleanprofile.Prof(base) self.reset() #Read and parse other profile and save profile data, include data from it and reset them apparmor.aa.read_profile(other, True) self.other = cleanprofile.Prof(other) self.reset() #Read and parse user profile apparmor.aa.read_profile(profiles[0], True) self.user = cleanprofile.Prof(user) def reset(self): apparmor.aa.aa = apparmor.aa.hasher() apparmor.aa.filelist = apparmor.aa.hasher() apparmor.aa.include = dict() apparmor.aa.existing_profiles = apparmor.aa.hasher() apparmor.aa.original_aa = apparmor.aa.hasher() def clear_common(self): deleted = 0 #Remove off the parts in other profile which are common/superfluous from user profile user_other = cleanprofile.CleanProf(False, self.user, self.other) deleted += user_other.compare_profiles() #Remove off the parts in base profile which are common/superfluous from user profile user_base = cleanprofile.CleanProf(False, self.user, self.base) deleted += user_base.compare_profiles() #Remove off the parts in other profile which are common/superfluous from base profile # base_other = cleanprofile.CleanProf(False, self.base, self.other) # XXX base_other not used? deleted += user_base.compare_profiles() def conflict_mode(self, profile, hat, allow, path, mode, new_mode, old_mode): m = new_mode o = old_mode new_mode = apparmor.aa.flatten_mode(new_mode) old_mode = apparmor.aa.flatten_mode(old_mode) conflict_modes = set('uUpPcCiIxX') conflict_x= (old_mode | new_mode) & conflict_modes if conflict_x: #We may have conflicting x modes if conflict_x & set('x'): conflict_x.remove('x') if conflict_x & set('X'): conflict_x.remove('X') if len(conflict_x) > 1: q = apparmor.aa.hasher() q['headers'] = [_('Path'), path] q['headers'] += [_('Select the appropriate mode'), ''] options = [] options.append('%s: %s' %(mode, apparmor.aa.mode_to_str_user(new_mode)))# - (old_mode & conflict_x)))) options.append('%s: %s' %(mode, apparmor.aa.mode_to_str_user(old_mode)))#(old_mode | new_mode) - (new_mode & conflict_x)))) q['options'] = options q['functions'] = ['CMD_ALLOW', 'CMD_ABORT'] done = False while not done: ans, selected = aaui.UI_PromptUser(q) if ans == 'CMD_ALLOW': if selected == 0: self.user.aa[profile][hat][allow]['path'][path][mode] = m#apparmor.aa.owner_flatten_mode(new_mode)#(old_mode | new_mode) - (old_mode & conflict_x) return m elif selected == 1: return o pass#self.user.aa[profile][hat][allow][path][mode] = (old_mode | new_mode) - (new_mode & conflict_x) else: raise apparmor.aa.AppArmorException(_('Unknown selection')) done = True def ask_the_questions(self, other): if other == 'other': other = self.other else: other = self.base #print(other.aa) #Add the file-wide includes from the other profile to the user profile done = False options = list(map(lambda inc: '#include <%s>' %inc, sorted(other.filelist[other.filename]['include'].keys()))) q = apparmor.aa.hasher() q['options'] = options default_option = 1 q['selected'] = default_option - 1 q['headers'] = [_('File includes'), _('Select the ones you wish to add')] q['functions'] = ['CMD_ALLOW', 'CMD_IGNORE_ENTRY', 'CMD_ABORT', 'CMD_FINISHED'] q['default'] = 'CMD_ALLOW' while not done and options: ans, selected = aaui.UI_PromptUser(q) if ans == 'CMD_IGNORE_ENTRY': done = True elif ans == 'CMD_ALLOW': selection = options[selected] inc = apparmor.aa.re_match_include(selection) self.user.filelist[self.user.filename]['include'][inc] = True options.pop(selected) aaui.UI_Info(_('Adding %s to the file.') % selection) sev_db = apparmor.aa.sev_db if not sev_db: sev_db = apparmor.severity.Severity(apparmor.aa.CONFDIR + '/severity.db', _('unknown')) for profile in sorted(other.aa.keys()): for hat in sorted(other.aa[profile].keys()): #Add the includes from the other profile to the user profile done = False options = list(map(lambda inc: '#include <%s>' %inc, sorted(other.aa[profile][hat]['include'].keys()))) q = apparmor.aa.hasher() q['options'] = options default_option = 1 q['selected'] = default_option - 1 q['headers'] = [_('File includes'), _('Select the ones you wish to add')] q['functions'] = ['CMD_ALLOW', 'CMD_IGNORE_ENTRY', 'CMD_ABORT', 'CMD_FINISHED'] q['default'] = 'CMD_ALLOW' while not done and options: ans, selected = aaui.UI_PromptUser(q) if ans == 'CMD_IGNORE_ENTRY': done = True elif ans == 'CMD_ALLOW': selection = options[selected] inc = apparmor.aa.re_match_include(selection) deleted = apparmor.aa.delete_duplicates(self.user.aa[profile][hat], inc) self.user.aa[profile][hat]['include'][inc] = True options.pop(selected) aaui.UI_Info(_('Adding %s to the file.') % selection) if deleted: aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) #Add the capabilities for allow in ['allow', 'deny']: if other.aa[profile][hat].get(allow, False): continue for capability in sorted(other.aa[profile][hat][allow]['capability'].keys()): severity = sev_db.rank('CAP_%s' % capability) default_option = 1 options = [] newincludes = apparmor.aa.match_cap_includes(self.user.aa[profile][hat], capability) q = apparmor.aa.hasher() if newincludes: options += list(map(lambda inc: '#include <%s>' %inc, sorted(set(newincludes)))) if options: options.append('capability %s' % capability) q['options'] = [options] q['selected'] = default_option - 1 q['headers'] = [_('Profile'), apparmor.aa.combine_name(profile, hat)] q['headers'] += [_('Capability'), capability] q['headers'] += [_('Severity'), severity] audit_toggle = 0 q['functions'] = ['CMD_ALLOW', 'CMD_DENY', 'CMD_IGNORE_ENTRY', 'CMD_ABORT', 'CMD_FINISHED'] q['default'] = 'CMD_ALLOW' done = False while not done: ans, selected = aaui.UI_PromptUser(q) # Ignore the log entry if ans == 'CMD_IGNORE_ENTRY': done = True break if ans == 'CMD_ALLOW': selection = '' if options: selection = options[selected] match = apparmor.aa.re_match_include(selection) if match: deleted = False inc = match deleted = apparmor.aa.delete_duplicates(self.user.aa[profile][hat], inc) self.user.aa[profile][hat]['include'][inc] = True aaui.UI_Info(_('Adding %s to profile.') % selection) if deleted: aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) self.user.aa[profile][hat]['allow']['capability'][capability]['set'] = True self.user.aa[profile][hat]['allow']['capability'][capability]['audit'] = other.aa[profile][hat]['allow']['capability'][capability]['audit'] apparmor.aa.changed[profile] = True aaui.UI_Info(_('Adding capability %s to profile.'), capability) done = True elif ans == 'CMD_DENY': self.user.aa[profile][hat]['deny']['capability'][capability]['set'] = True apparmor.aa.changed[profile] = True aaui.UI_Info(_('Denying capability %s to profile.') % capability) done = True else: done = False # Process all the path entries. for allow in ['allow', 'deny']: for path in sorted(other.aa[profile][hat][allow]['path'].keys()): #print(path, other.aa[profile][hat][allow]['path'][path]) mode = other.aa[profile][hat][allow]['path'][path]['mode'] if self.user.aa[profile][hat][allow]['path'].get(path, False): mode = self.conflict_mode(profile, hat, allow, path, 'mode', other.aa[profile][hat][allow]['path'][path]['mode'], self.user.aa[profile][hat][allow]['path'][path]['mode']) self.conflict_mode(profile, hat, allow, path, 'audit', other.aa[profile][hat][allow]['path'][path]['audit'], self.user.aa[profile][hat][allow]['path'][path]['audit']) apparmor.aa.changed[profile] = True continue # Lookup modes from profile allow_mode = set() allow_audit = set() deny_mode = set() deny_audit = set() fmode, famode, fm = apparmor.aa.rematchfrag(self.user.aa[profile][hat], 'allow', path) if fmode: allow_mode |= fmode if famode: allow_audit |= famode cm, cam, m = apparmor.aa.rematchfrag(self.user.aa[profile][hat], 'deny', path) if cm: deny_mode |= cm if cam: deny_audit |= cam imode, iamode, im = apparmor.aa.match_prof_incs_to_path(self.user.aa[profile][hat], 'allow', path) if imode: allow_mode |= imode if iamode: allow_audit |= iamode cm, cam, m = apparmor.aa.match_prof_incs_to_path(self.user.aa[profile][hat], 'deny', path) if cm: deny_mode |= cm if cam: deny_audit |= cam if deny_mode & apparmor.aamode.AA_MAY_EXEC: deny_mode |= apparmor.aamode.ALL_AA_EXEC_TYPE # Mask off the denied modes mode = mode - deny_mode # If we get an exec request from some kindof event that generates 'PERMITTING X' # check if its already in allow_mode # if not add ix permission if mode & apparmor.aamode.AA_MAY_EXEC: # Remove all type access permission mode = mode - apparmor.aamode.ALL_AA_EXEC_TYPE if not allow_mode & apparmor.aamode.AA_MAY_EXEC: mode |= apparmor.aa.str_to_mode('ix') # m is not implied by ix ### If we get an mmap request, check if we already have it in allow_mode ##if mode & AA_EXEC_MMAP: ## # ix implies m, so we don't need to add m if ix is present ## if contains(allow_mode, 'ix'): ## mode = mode - AA_EXEC_MMAP if not mode: continue matches = [] if fmode: matches += fm if imode: matches += im if not apparmor.aa.mode_contains(allow_mode, mode): default_option = 1 options = [] newincludes = [] include_valid = False for incname in apparmor.aa.include.keys(): include_valid = False # If already present skip if self.user.aa[profile][hat][incname]: continue if incname.startswith(apparmor.aa.profile_dir): incname = incname.replace(apparmor.aa.profile_dir+'/', '', 1) include_valid = apparmor.aa.valid_include('', incname) if not include_valid: continue cm, am, m = apparmor.aa.match_include_to_path(incname, 'allow', path) if cm and apparmor.aa.mode_contains(cm, mode): dm = apparmor.aa.match_include_to_path(incname, 'deny', path)[0] # If the mode is denied if not mode & dm: if not list(filter(lambda s: '/**' == s, m)): newincludes.append(incname) # Add new includes to the options if newincludes: options += list(map(lambda s: '#include <%s>' % s, sorted(set(newincludes)))) # We should have literal the path in options list too options.append(path) # Add any the globs matching path from logprof globs = apparmor.aa.glob_common(path) if globs: matches += globs # Add any user entered matching globs for user_glob in apparmor.aa.user_globs: if apparmor.aa.matchliteral(user_glob, path): matches.append(user_glob) matches = list(set(matches)) if path in matches: matches.remove(path) options += apparmor.aa.order_globs(matches, path) default_option = len(options) sev_db.unload_variables() sev_db.load_variables(apparmor.aa.get_profile_filename(profile)) severity = sev_db.rank(path, apparmor.aa.mode_to_str(mode)) sev_db.unload_variables() audit_toggle = 0 owner_toggle = 0 if apparmor.aa.cfg['settings']['default_owner_prompt']: owner_toggle = apparmor.aa.cfg['settings']['default_owner_prompt'] done = False while not done: q = apparmor.aa.hasher() q['headers'] = [_('Profile'), apparmor.aa.combine_name(profile, hat), _('Path'), path] if allow_mode: mode |= allow_mode tail = '' s = '' prompt_mode = None if owner_toggle == 0: prompt_mode = apparmor.aa.flatten_mode(mode) tail = ' ' + _('(owner permissions off)') elif owner_toggle == 1: prompt_mode = mode elif owner_toggle == 2: prompt_mode = allow_mode | apparmor.aa.owner_flatten_mode(mode - allow_mode) tail = ' ' + _('(force new perms to owner)') else: prompt_mode = apparmor.aa.owner_flatten_mode(mode) tail = ' ' + _('(force all rule perms to owner)') if audit_toggle == 1: s = apparmor.aa.mode_to_str_user(allow_mode) if allow_mode: s += ', ' s += 'audit ' + apparmor.aa.mode_to_str_user(prompt_mode - allow_mode) + tail elif audit_toggle == 2: s = 'audit ' + apparmor.aa.mode_to_str_user(prompt_mode) + tail else: s = apparmor.aa.mode_to_str_user(prompt_mode) + tail q['headers'] += [_('Old Mode'), apparmor.aa.mode_to_str_user(allow_mode), _('New Mode'), s] else: s = '' tail = '' prompt_mode = None if audit_toggle: s = 'audit' if owner_toggle == 0: prompt_mode = apparmor.aa.flatten_mode(mode) tail = ' ' + _('(owner permissions off)') elif owner_toggle == 1: prompt_mode = mode else: prompt_mode = apparmor.aa.owner_flatten_mode(mode) tail = ' ' + _('(force perms to owner)') s = apparmor.aa.mode_to_str_user(prompt_mode) q['headers'] += [_('Mode'), s] q['headers'] += [_('Severity'), severity] q['options'] = options q['selected'] = default_option - 1 q['functions'] = ['CMD_ALLOW', 'CMD_DENY', 'CMD_IGNORE_ENTRY', 'CMD_GLOB', 'CMD_GLOBEXT', 'CMD_NEW', 'CMD_ABORT', 'CMD_FINISHED', 'CMD_OTHER'] q['default'] = 'CMD_ALLOW' ans, selected = aaui.UI_PromptUser(q) if ans == 'CMD_IGNORE_ENTRY': done = True break if ans == 'CMD_OTHER': audit_toggle, owner_toggle = aaui.UI_ask_mode_toggles(audit_toggle, owner_toggle, allow_mode) elif ans == 'CMD_USER_TOGGLE': owner_toggle += 1 if not allow_mode and owner_toggle == 2: owner_toggle += 1 if owner_toggle > 3: owner_toggle = 0 elif ans == 'CMD_ALLOW': path = options[selected] done = True match = apparmor.aa.re_match_include(path) if match: inc = match deleted = 0 deleted = apparmor.aa.delete_duplicates(self.user.aa[profile][hat], inc) self.user.aa[profile][hat]['include'][inc] = True apparmor.aa.changed[profile] = True aaui.UI_Info(_('Adding %s to profile.') % path) if deleted: aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) else: if self.user.aa[profile][hat]['allow']['path'][path].get('mode', False): mode |= self.user.aa[profile][hat]['allow']['path'][path]['mode'] deleted = [] for entry in self.user.aa[profile][hat]['allow']['path'].keys(): if path == entry: continue if apparmor.aa.matchregexp(path, entry): if apparmor.aa.mode_contains(mode, self.user.aa[profile][hat]['allow']['path'][entry]['mode']): deleted.append(entry) for entry in deleted: self.user.aa[profile][hat]['allow']['path'].pop(entry) deleted = len(deleted) if owner_toggle == 0: mode = apparmor.aa.flatten_mode(mode) #elif owner_toggle == 1: # mode = mode elif owner_toggle == 2: mode = allow_mode | apparmor.aa.owner_flatten_mode(mode - allow_mode) elif owner_toggle == 3: mode = apparmor.aa.owner_flatten_mode(mode) if not self.user.aa[profile][hat]['allow'].get(path, False): self.user.aa[profile][hat]['allow']['path'][path]['mode'] = self.user.aa[profile][hat]['allow']['path'][path].get('mode', set()) | mode tmpmode = set() if audit_toggle == 1: tmpmode = mode- allow_mode elif audit_toggle == 2: tmpmode = mode self.user.aa[profile][hat]['allow']['path'][path]['audit'] = self.user.aa[profile][hat]['allow']['path'][path].get('audit', set()) | tmpmode apparmor.aa.changed[profile] = True aaui.UI_Info(_('Adding %s %s to profile') % (path, apparmor.aa.mode_to_str_user(mode))) if deleted: aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) elif ans == 'CMD_DENY': path = options[selected].strip() # Add new entry? self.user.aa[profile][hat]['deny']['path'][path]['mode'] = self.user.aa[profile][hat]['deny']['path'][path].get('mode', set()) | (mode - allow_mode) self.user.aa[profile][hat]['deny']['path'][path]['audit'] = self.user.aa[profile][hat]['deny']['path'][path].get('audit', set()) apparmor.aa.changed[profile] = True done = True elif ans == 'CMD_NEW': arg = options[selected] if not apparmor.aa.re_match_include(arg): ans = aaui.UI_GetString(_('Enter new path: '), arg) # if ans: # if not matchliteral(ans, path): # ynprompt = _('The specified path does not match this log entry:\n\n Log Entry: %s\n Entered Path: %s\nDo you really want to use this path?') % (path,ans) # key = aaui.UI_YesNo(ynprompt, 'n') # if key == 'n': # continue apparmor.aa.user_globs.append(ans) options.append(ans) default_option = len(options) elif ans == 'CMD_GLOB': newpath = options[selected].strip() if not apparmor.aa.re_match_include(newpath): newpath = apparmor.aa.glob_path(newpath) if newpath not in options: options.append(newpath) default_option = len(options) else: default_option = options.index(newpath) + 1 elif ans == 'CMD_GLOBEXT': newpath = options[selected].strip() if not apparmor.aa.re_match_include(newpath): newpath = apparmor.aa.glob_path_withext(newpath) if newpath not in options: options.append(newpath) default_option = len(options) else: default_option = options.index(newpath) + 1 elif re.search('\d', ans): default_option = ans # for allow in ['allow', 'deny']: for family in sorted(other.aa[profile][hat][allow]['netdomain']['rule'].keys()): # severity handling for net toggles goes here for sock_type in sorted(other.aa[profile][hat][allow]['netdomain']['rule'][family].keys()): if apparmor.aa.profile_known_network(self.user.aa[profile][hat], family, sock_type): continue default_option = 1 options = [] newincludes = apparmor.aa.match_net_includes(self.user.aa[profile][hat], family, sock_type) q = apparmor.aa.hasher() if newincludes: options += list(map(lambda s: '#include <%s>'%s, sorted(set(newincludes)))) if True:#options: options.append('network %s %s' % (family, sock_type)) q['options'] = options q['selected'] = default_option - 1 q['headers'] = [_('Profile'), apparmor.aa.combine_name(profile, hat)] q['headers'] += [_('Network Family'), family] q['headers'] += [_('Socket Type'), sock_type] audit_toggle = 0 q['functions'] = ['CMD_ALLOW', 'CMD_DENY', 'CMD_IGNORE_ENTRY', 'CMD_AUDIT_NEW', 'CMD_ABORT', 'CMD_FINISHED'] q['default'] = 'CMD_ALLOW' done = False while not done: ans, selected = aaui.UI_PromptUser(q) if ans == 'CMD_IGNORE_ENTRY': done = True break if ans.startswith('CMD_AUDIT'): audit_toggle = not audit_toggle audit = '' if audit_toggle: audit = 'audit' q['functions'] = ['CMD_ALLOW', 'CMD_DENY', 'CMD_AUDIT_OFF', 'CMD_ABORT', 'CMD_FINISHED'] else: q['functions'] = ['CMD_ALLOW', 'CMD_DENY', 'CMD_AUDIT_NEW', 'CMD_ABORT', 'CMD_FINISHED'] q['headers'] = [_('Profile'), apparmor.aa.combine_name(profile, hat)] q['headers'] += [_('Network Family'), audit + family] q['headers'] += [_('Socket Type'), sock_type] elif ans == 'CMD_ALLOW': #print(options, selected) selection = options[selected] done = True if apparmor.aa.re_match_include(selection): #re.search('#include\s+<.+>$', selection): inc = apparmor.aa.re_match_include(selection) #re.search('#include\s+<(.+)>$', selection).groups()[0] deleted = 0 deleted = apparmor.aa.delete_duplicates(self.user.aa[profile][hat], inc) self.user.aa[profile][hat]['include'][inc] = True apparmor.aa.changed[profile] = True aaui.UI_Info(_('Adding %s to profile') % selection) if deleted: aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) else: self.user.aa[profile][hat]['allow']['netdomain']['audit'][family][sock_type] = audit_toggle self.user.aa[profile][hat]['allow']['netdomain']['rule'][family][sock_type] = True apparmor.aa.changed[profile] = True aaui.UI_Info(_('Adding network access %s %s to profile.') % (family, sock_type)) elif ans == 'CMD_DENY': done = True self.user.aa[profile][hat]['deny']['netdomain']['rule'][family][sock_type] = True apparmor.aa.changed[profile] = True aaui.UI_Info(_('Denying network access %s %s to profile') % (family, sock_type)) else: done = False if __name__ == '__main__': main()