snap: add param "dangerous"

changelogs/fragments/6908-snap-dangerous.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - snap - add option ``dangerous`` to the module, that will map into the command line argument ``--dangerous``, allowing unsigned snap files to be installed (https://github.com/ansible-collections/community.general/pull/6908, https://github.com/ansible-collections/community.general/issues/5715).

plugins/module_utils/snap.py

@@ -39,6 +39,8 @@ def snap_runner(module, **kwargs):
             classic=cmd_runner_fmt.as_bool("--classic"),
             channel=cmd_runner_fmt.as_func(lambda v: [] if v == 'stable' else ['--channel', '{0}'.format(v)]),
             options=cmd_runner_fmt.as_list(),
+            info=cmd_runner_fmt.as_fixed("info"),
+            dangerous=cmd_runner_fmt.as_bool("--dangerous"),
         ),
         check_rc=False,
         **kwargs

plugins/modules/snap.py

@@ -17,7 +17,7 @@
 module: snap
 short_description: Manages snaps
 description:
-    - "Manages snaps packages."
+    - Manages snaps packages.
 extends_documentation_fragment:
     - community.general.attributes
 attributes:
@@ -28,7 +28,11 @@
 options:
     name:
         description:
-            - Name of the snaps.
+            - Name of the snaps to be installed.
+            - Any named snap accepted by the C(snap) command is valid.
+            - >
+              Notice that snap files might require O(dangerous=true) to ignore the error
+              "cannot find signatures with metadata for snap".
         required: true
         type: list
         elements: str
@@ -45,7 +49,7 @@
         description:
             - Confinement policy. The classic confinement allows a snap to have
               the same level of access to the system as "classic" packages,
-              like those managed by APT. This option corresponds to the --classic argument.
+              like those managed by APT. This option corresponds to the C(--classic) argument.
               This option can only be specified if there is a single snap in the task.
         type: bool
         required: false
@@ -69,6 +73,14 @@
         type: list
         elements: str
         version_added: 4.4.0
+    dangerous:
+        description:
+            - Install the given snap file even if there are no pre-acknowledged signatures for it,
+              meaning it was not verified and could be dangerous.
+        type: bool
+        required: false
+        default: false
+        version_added: 7.2.0
 
 author:
     - Victor Carceler (@vcarceler) <[email protected]>
@@ -179,6 +191,7 @@ class Snap(StateModuleHelper):
             'classic': dict(type='bool', default=False),
             'channel': dict(type='str'),
             'options': dict(type='list', elements='str'),
+            'dangerous': dict(type='bool', default=False),
         },
         supports_check_mode=True,
     )
@@ -193,7 +206,16 @@ def _first_non_zero(a):
 
     def __init_module__(self):
         self.runner = snap_runner(self.module)
-        self.vars.set("snap_status", self.snap_status(self.vars.name, self.vars.channel), output=False)
+        # if state=present there might be file names passed in 'name', in
+        # which case they must be converted to their actual snap names, which
+        # is done using the names_from_snaps() method calling 'snap info'.
+        if self.vars.state == "present":
+            self.vars.set("snapinfo_run_info", [], output=(self.verbosity >= 4))
+            self.vars.set("snap_names", self.names_from_snaps(self.vars.name))
+            status_var = "snap_names"
+        else:
+            status_var = "name"
+        self.vars.set("snap_status", self.snap_status(self.vars[status_var], self.vars.channel), output=False)
         self.vars.set("snap_status_map", dict(zip(self.vars.name, self.vars.snap_status)), output=False)
 
     def _run_multiple_commands(self, commands, actionable_names, bundle=True, refresh=False):
@@ -269,11 +291,44 @@ def retrieve_option_map(self, snap_name):
 
         try:
             option_map = self.convert_json_to_map(out)
+            return option_map
         except Exception as e:
             self.do_raise(
                 msg="Parsing option map returned by 'snap get {0}' triggers exception '{1}', output:\n'{2}'".format(snap_name, str(e), out))
 
-        return option_map
+    def names_from_snaps(self, snaps):
+        def process_one(rc, out, err):
+            res = [line for line in out.split("\n") if line.startswith("name:")]
+            name = res[0].split()[1]
+            return [name]
+
+        def process_many(rc, out, err):
+            outputs = out.split("---")
+            res = []
+            for sout in outputs:
+                res.extend(process_one(rc, sout, ""))
+            return res
+
+        def process(rc, out, err):
+            if len(snaps) == 1:
+                check_error = err
+                process_ = process_one
+            else:
+                check_error = out
+                process_ = process_many
+
+            if "warning: no snap found" in check_error:
+                self.do_raise("Snaps not found: {0}.".format([x.split()[-1]
+                                                              for x in out.split('\n')
+                                                              if x.startswith("warning: no snap found")]))
+            return process_(rc, out, err)
+
+        with self.runner("info name", output_process=process) as ctx:
+            try:
+                names = ctx.run(name=snaps)
+            finally:
+                self.vars.snapinfo_run_info.append(ctx.run_info)
+        return names
 
     def snap_status(self, snap_name, channel):
         def _status_check(name, channel, installed):
@@ -287,14 +342,14 @@ def _status_check(name, channel, installed):
 
         with self.runner("_list") as ctx:
             rc, out, err = ctx.run(check_rc=True)
-        out = out.split('\n')[1:]
-        out = [self.__list_re.match(x) for x in out]
-        out = [(m.group('name'), m.group('channel')) for m in out if m]
+        list_out = out.split('\n')[1:]
+        list_out = [self.__list_re.match(x) for x in list_out]
+        list_out = [(m.group('name'), m.group('channel')) for m in list_out if m]
         if self.verbosity >= 4:
-            self.vars.status_out = out
+            self.vars.status_out = list_out
             self.vars.status_run_info = ctx.run_info
 
-        return [_status_check(n, channel, out) for n in snap_name]
+        return [_status_check(n, channel, list_out) for n in snap_name]
 
     def is_snap_enabled(self, snap_name):
         with self.runner("_list name") as ctx:
@@ -315,7 +370,7 @@ def _present(self, actionable_snaps, refresh=False):
         if self.check_mode:
             return
 
-        params = ['state', 'classic', 'channel']  # get base cmd parts
+        params = ['state', 'classic', 'channel', 'dangerous']  # get base cmd parts
         has_one_pkg_params = bool(self.vars.classic) or self.vars.channel != 'stable'
         has_multiple_snaps = len(actionable_snaps) > 1
 

tests/integration/targets/snap/meta/main.yml

@@ -5,3 +5,4 @@
 
 dependencies:
   - setup_snap
+  - setup_remote_tmp_dir

tests/integration/targets/snap/tasks/main.yml

@@ -15,3 +15,5 @@
       ansible.builtin.include_tasks: test.yml
     - name: Include test_channel
       ansible.builtin.include_tasks: test_channel.yml
+    - name: Include test_dangerous
+      ansible.builtin.include_tasks: test_dangerous.yml

tests/integration/targets/snap/tasks/test_dangerous.yml

@@ -0,0 +1,51 @@
+---
+# Copyright (c) Ansible Project
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+- name: Make sure package is not installed (cider)
+  community.general.snap:
+    name: cider
+    state: absent
+
+- name: Download cider snap
+  ansible.builtin.get_url:
+    url: https://github.com/ciderapp/cider-releases/releases/download/v1.6.0/cider_1.6.0_amd64.snap
+    dest: "{{ remote_tmp_dir }}/cider_1.6.0_amd64.snap"
+    mode: "0644"
+
+# Test for https://github.com/ansible-collections/community.general/issues/5715
+- name: Install package from file (check)
+  community.general.snap:
+    name: "{{ remote_tmp_dir }}/cider_1.6.0_amd64.snap"
+    dangerous: true
+    state: present
+  check_mode: true
+  register: install_dangerous_check
+
+- name: Install package from file
+  community.general.snap:
+    name: "{{ remote_tmp_dir }}/cider_1.6.0_amd64.snap"
+    dangerous: true
+    state: present
+  register: install_dangerous
+
+- name: Install package from file
+  community.general.snap:
+    name: "{{ remote_tmp_dir }}/cider_1.6.0_amd64.snap"
+    dangerous: true
+    state: present
+  register: install_dangerous_idempot
+
+- name: Remove package
+  community.general.snap:
+    name: cider
+    state: absent
+  register: remove_dangerous
+
+- assert:
+    that:
+      - install_dangerous_check is changed
+      - install_dangerous is changed
+      - install_dangerous_idempot is not changed
+      - remove_dangerous is changed

tests/unit/plugins/modules/test_snap.py

@@ -395,11 +395,46 @@ def mockie(self, path, *args, **kwargs):
 )
 
 TEST_CASES = [
+    ModuleTestCase(
+        id="simple case",
+        input={"name": ["hello-world"]},
+        output=dict(changed=True, snaps_installed=["hello-world"]),
+        run_command_calls=[
+            RunCmdCall(
+                command=['/testbin/snap', 'info', 'hello-world'],
+                environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
+                rc=0,
+                out='name: hello-world\n',
+                err="",
+            ),
+            RunCmdCall(
+                command=['/testbin/snap', 'list'],
+                environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
+                rc=0,
+                out="",
+                err="",
+            ),
+            RunCmdCall(
+                command=['/testbin/snap', 'install', 'hello-world'],
+                environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
+                rc=0,
+                out="hello-world (12345/stable) v12345 from Canonical** installed\n",
+                err="",
+            ),
+        ]
+    ),
     ModuleTestCase(
         id="issue_6803",
         input={"name": ["microk8s", "kubectl"], "classic": True},
         output=dict(changed=True, snaps_installed=["microk8s", "kubectl"]),
         run_command_calls=[
+            RunCmdCall(
+                command=['/testbin/snap', 'info', 'microk8s', 'kubectl'],
+                environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
+                rc=0,
+                out='name: microk8s\n---\nname: kubectl\n',
+                err="",
+            ),
             RunCmdCall(
                 command=['/testbin/snap', 'list'],
                 environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},