Merge utils: Allow writing to profile includes

Allow writing to local profiles

This notably allows aa-notify to write to local profiles instead of the main profile with the new `--local` option. This keeps the base profile clean, avoiding breakages when the system updates profiles.

Signed-off-by: Maxime Bélair <maxime.belair@canonical.com>

MR: https://gitlab.com/apparmor/apparmor/-/merge_requests/1764
Approved-by: John Johansen <john@jjmx.net>
Merged-by: John Johansen <john@jjmx.net>

utils/aa-notify

@@ -612,7 +612,8 @@ def add_to_profile(rule, profile_name):
         return
 
     update_profile_path = update_profile.__file__
-    command = ['pkexec', '--keep-cwd', update_profile_path,  'add_rule', rule, profile_name]
+
+    command = ['pkexec', '--keep-cwd', update_profile_path, 'add_rule', args.local, rule, profile_name]
     try:
         subprocess.run(command, check=True)
     except subprocess.CalledProcessError as e:
@@ -647,7 +648,7 @@ def allow_rules(clean_rules, allow_all=False):
     with open(tmp.name, mode='w') as f:
 
         for profile_name, profile_rules in clean_rules.items():
-            written += f.write(profile_rules.get_writable_rules(template_path))
+            written += f.write(profile_rules.get_writable_rules(template_path, args.local))
 
     if written > 0:
         create_from_file(tmp.name)
@@ -849,6 +850,7 @@ def main():
     parser.add_argument('-w', '--wait', type=int, metavar=('NUM'), help=_('wait NUM seconds before displaying notifications (with -p)'))
     parser.add_argument('-m', '--merge-notifications', action='store_true', help=_('Merge notification for improved readability (with -p)'))
     parser.add_argument('-F', '--foreground', action='store_true', help=_('Do not fork to the background'))
+    parser.add_argument('-L', '--local', nargs='?', const='yes', default='auto', choices=['yes', 'no', 'auto'], help=_('Add to local profile'))
     parser.add_argument('--prompt-filter', type=str, metavar=('PF'), help=_('kind of operations which display a popup prompt'))
     parser.add_argument('--debug', action='store_true', help=_('debug mode'))
     parser.add_argument('--configdir', type=str, help=argparse.SUPPRESS)
@@ -938,6 +940,7 @@ def main():
     - prompt_filter
     - maximum_number_notification_profiles
     - keys_to_aggregate
+    - use_local_profiles
     - filter.profile,
     - filter.operation,
     - filter.name,
@@ -960,6 +963,7 @@ def main():
         'message_footer',
         'maximum_number_notification_profiles',
         'keys_to_aggregate',
+        'use_local_profiles',
         'filter.profile',
         'filter.operation',
         'filter.name',
@@ -1068,6 +1072,15 @@ def main():
         keys_to_aggregate = config['']['keys_to_aggregate'].strip().split(',')
     else:
         keys_to_aggregate = {'operation', 'class', 'name', 'denied', 'target'}
+    if 'use_local_profiles' in config['']:
+        if config['']['use_local_profiles'] in {'auto', 'yes', 'no'}:
+            args.local = config['']['use_local_profiles']
+        elif config['']['use_local_profiles'] is None:
+            args.local = 'yes'
+        else:
+            sys.exit(_('ERROR: using an invalid value for use_local_profiles in config {}\nSupported values: {}').format(
+                config['']['use_local_profiles'], ', '.join({'yes', 'auto', 'no'})
+            ))
 
     if args.file:
         logfile = args.file

utils/aa-notify.pod

@@ -71,6 +71,14 @@ This has no effect when running under sudo.
 
 wait NUM seconds before displaying notifications (for use with -p)
 
+=item -L, --local [{yes,no,auto}]
+
+add rules to a local profiles instead of the real profiles.
+This simplify profiles' deployment by keeping local modifications self-contained.
+ - B<yes>: always use a local profile
+ - B<no>: never use a local profile
+ - B<auto>: use a local profile if the main profile already relies on a local profile
+
 =item -v, --verbose
 
 show messages with summaries.
@@ -98,6 +106,9 @@ System-wide configuration for B<aa-notify> is done via
   # Binaries for which we ignore userns-related capability denials
   ignore_denied_capability="sudo,su"
 
+  # Write change to local profiles if enabled to preserve regular profiles and simplify upgrades
+  use_local_profiles
+
   # OPTIONAL - kind of operations which display a popup prompt.
   prompt_filter="userns"
 

utils/apparmor/aa.py

@@ -1703,6 +1703,72 @@ def read_profile(file, is_active_profile, read_error_fatal=False):
             extra_profiles.add_profile(filename, profile, attachment, profile_data[profile])
 
 
+def get_local_include(profile_name):
+    # If a local profile already exists, we use it.
+    for rule in active_profiles[profile_name]['inc_ie'].rules:
+        if rule.path.startswith("local/"):
+            return rule.path
+    return None
+
+
+def create_local_profile_if_needed(profile_name):
+    base_profile = profile_name.split("/", 1)[0]
+    local_include = get_local_include(profile_name)
+
+    # Not found: we add a mention of the local profile in the main profile
+    if not local_include:
+        local_include = "local/" + profile_name.replace('/', '.')
+        active_profiles[profile_name]['inc_ie'].add(IncludeRule(local_include, True, True))
+        write_profile_ui_feedback(base_profile)
+
+    inc_file = profile_dir + '/' + local_include
+
+    # Create the include if needed
+    if not include.get(inc_file, {}).get(inc_file, False):
+        include[inc_file] = dict()
+        include[inc_file][inc_file] = ProfileStorage(inc_file, inc_file, "create_local_profile_if_needed")
+
+    return inc_file
+
+
+def serialize_include(prof_storage, include_metadata=True):
+    lines = []
+    if include_metadata:
+        lines.append('# Last Modified: %s' % time.asctime())
+
+    if prof_storage.get('initial_comment'):
+        lines.append(prof_storage['initial_comment'].rstrip())
+
+    lines.extend(prof_storage.get_rules_clean(0))
+
+    return '\n'.join(lines) + '\n'
+
+
+def write_include_ui_feedback(include_data, incfile, out_dir=None, include_metadata=True):
+    aaui.UI_Info(_('Writing updated include file %s') % incfile)
+    write_include(include_data, incfile, out_dir, include_metadata)
+
+
+def write_include(include_data, incfile, out_dir=None, include_metadata=True):
+    target_file = incfile if incfile.startswith('/') else os.path.join(profile_dir, incfile)
+    if out_dir:
+        target_file = os.path.join(out_dir, os.path.basename(target_file))
+
+    include_string = serialize_include(include_data, include_metadata=include_metadata)
+
+    with NamedTemporaryFile('w', suffix='~', delete=False) as tmp:
+        if os.path.exists(target_file):
+            shutil.copymode(target_file, tmp.name)
+        else:
+            pass  # 0o600 (NamedTemporaryFile default)
+        tmp.write(include_string)
+
+    try:
+        shutil.move(tmp.name, target_file)
+    except PermissionError:
+        aaui.UI_Important(_('WARNING: Can\'t write to %s. Please run this script with elevated privileges') % target_file)
+
+
 def attach_profile_data(profiles, profile_data):
     profile_data = merged_to_split(profile_data)
     # Make deep copy of data to avoid changes to

utils/apparmor/gui.py

@@ -100,12 +100,12 @@ class ProfileRules:
         for raw_rule in raw_rules:
             self.rules.append(SelectableRule(raw_rule, self.selectable))
 
-    def get_writable_rules(self, template_path, allow_all=False):
+    def get_writable_rules(self, template_path, local='yes', allow_all=False):
         out = ''
         for rule in self.rules:
             if allow_all or rule.selected.get():
                 if not self.is_userns_profile:
-                    out += 'add_rule\t{}\t{}\n'.format(rule.rule, self.profile_name)
+                    out += 'add_rule\t{}\t{}\t{}\n'.format(local, rule.rule, self.profile_name)
                 else:
                     out += 'create_userns\t{}\t{}\t{}\t{}\t{}\n'.format(template_path, self.profile_name, self.bin_path, self.profile_path, 'allow')
         return out

utils/apparmor/update_profile.py

@@ -8,8 +8,19 @@ from apparmor import aa
 from apparmor.logparser import ReadLog
 
 from apparmor.translations import init_translation
+
 _ = init_translation()
 
+is_aa_inited = False
+
+
+def init_if_needed():
+    global is_aa_inited
+    if not is_aa_inited:
+        aa.init_aa()
+        aa.read_profiles()
+        is_aa_inited = True
+
 
 def create_userns(template_path, name, bin_path, profile_path, decision):
     with open(template_path, 'r') as f:
@@ -27,27 +38,48 @@ def create_userns(template_path, name, bin_path, profile_path, decision):
         exit(_('Cannot reload updated profile'))
 
 
-def add_to_profile(rule, profile_name):
-    aa.init_aa()
-    aa.update_profiles()
-
-    rule_type, rule_class = ReadLog('', '', '').get_rule_type(rule)
-
-    rule_obj = rule_class.create_instance(rule)
-
-    if not aa.active_profiles.profile_exists(profile_name):
-        exit(_('Cannot find {} in profiles').format(profile_name))
-    aa.active_profiles[profile_name][rule_type].add(rule_obj, cleanup=True)
+def add_to_profile(rule_obj, profile_name):
+    aa.active_profiles[profile_name][rule_obj.rule_name].add(rule_obj, cleanup=True)
 
     # Save changes
     aa.write_profile_ui_feedback(profile_name)
+
+
+def add_to_local_profile(rule_obj, profile_name):
+    inc_file = aa.create_local_profile_if_needed(profile_name, cleanup=True)
+
+    aa.include[inc_file][inc_file].data[rule_obj.rule_name].add(rule_obj)
+    aa.write_include_ui_feedback(aa.include[inc_file][inc_file], inc_file)
+
+
+def add_rule(mode, rule, profile_name):
+    init_if_needed()
+
+    if not aa.active_profiles.profile_exists(profile_name):
+        exit(_('Cannot find {} in profiles').format(profile_name))
+
+    rule_type, rule_class = ReadLog('', '', '').get_rule_type(rule)
+    rule_obj = rule_class.create_instance(rule)
+
+    if mode == 'yes':
+        add_to_local_profile(rule_obj, profile_name)
+    elif mode == 'no':
+        add_to_profile(rule_obj, profile_name)
+    elif mode == 'auto':
+        if aa.get_local_include(profile_name):
+            add_to_local_profile(rule_obj, profile_name)
+        else:
+            add_to_profile(rule_obj, profile_name)
+    else:
+        usage(False)
+
     aa.reload_base(profile_name)
 
 
 def usage(is_help):
     print('This tool is a low level tool - do not use it directly')
     print('{} create_userns <template_path> <name> <bin_path> <profile_path> <decision>'.format(sys.argv[0]))
-    print('{} add_rule <rule> <profile_name>'.format(sys.argv[0]))
+    print('{} add_rule <mode=yes|no|auto> <rule> <profile_name>'.format(sys.argv[0]))
     print('{} from_file <file>'.format(sys.argv[0]))
     if is_help:
         exit(0)
@@ -76,9 +108,9 @@ def do_command(command, args):
             usage(False)
         create_userns(args[1], args[2], args[3], args[4], args[5])
     elif command == 'add_rule':
-        if not len(args) == 3:
+        if not len(args) == 4:
             usage(False)
-        add_to_profile(args[1], args[2])
+        add_rule(args[1], args[2], args[3])
     elif command == 'help':
         usage(True)
     else:

utils/notify.conf

@@ -20,6 +20,9 @@ interface_theme="ubuntu"
 # Binaries for which we ignore userns-related capability denials
 ignore_denied_capability="sudo,su"
 
+# OPTIONAL - Write changes to local profiles to preserve regular profiles and simplify upgrades (yes, no, auto)
+# use_local_profiles="yes"
+
 # OPTIONAL - kind of operations which display a popup prompt.
 # prompt_filter="userns"
 

utils/test/test-aa-notify.py

@@ -168,10 +168,11 @@ class AANotifyTest(AANotifyBase):
         expected_return_code = 0
         expected_output_1 = \
 '''usage: aa-notify [-h] [-p] [--display DISPLAY] [-f FILE] [-l] [-s NUM] [-v]
-                 [-u USER] [-w NUM] [-m] [-F] [--prompt-filter PF] [--debug]
-                 [--filter.profile PROFILE] [--filter.operation OPERATION]
-                 [--filter.name NAME] [--filter.denied DENIED]
-                 [--filter.family FAMILY] [--filter.socket SOCKET]
+                 [-u USER] [-w NUM] [-m] [-F] [-L [{yes,no,auto}]]
+                 [--prompt-filter PF] [--debug] [--filter.profile PROFILE]
+                 [--filter.operation OPERATION] [--filter.name NAME]
+                 [--filter.denied DENIED] [--filter.family FAMILY]
+                 [--filter.socket SOCKET]
 
 Display AppArmor notifications or messages for DENIED entries.
 '''  # noqa: E128
@@ -193,6 +194,8 @@ Display AppArmor notifications or messages for DENIED entries.
   -m, --merge-notifications
                         Merge notification for improved readability (with -p)
   -F, --foreground      Do not fork to the background
+  -L, --local [{yes,no,auto}]
+                        Add to local profile
   --prompt-filter PF    kind of operations which display a popup prompt
   --debug               debug mode
 
@@ -231,6 +234,11 @@ Filtering options:
             ), (
                 ', --wait NUM    ',
                 ' NUM, --wait NUM',
+            ), (
+                ' -L, --local [{yes,no,auto}]\n'
+                + '                        Add to local profile',
+                ' -L [{yes,no,auto}], --local [{yes,no,auto}]\n'
+                + '                        Add to local profile'
             )]
             for patch in patches:
                 expected_output_2 = expected_output_2.replace(patch[0], patch[1])