snap_connect: new module to manage snap interface connections (#11682)

* snap_connect: new module to manage snap interface connections

Fixes #7722

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* simplify _get_connections()

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

.github/BOTMETA.yml

@@ -1338,6 +1338,9 @@ files:
   $modules/snap_alias.py:
     labels: snap
     maintainers: russoz
+  $modules/snap_connect.py:
+    labels: snap
+    maintainers: russoz
   $modules/snmp_facts.py:
     maintainers: ogenstad ujwalkomarla lalten
   $modules/solaris_zone.py:

plugins/module_utils/snap.py

@@ -45,6 +45,10 @@ def snap_runner(module: AnsibleModule, **kwargs) -> CmdRunner:
             info=cmd_runner_fmt.as_fixed("info"),
             dangerous=cmd_runner_fmt.as_bool("--dangerous"),
             version=cmd_runner_fmt.as_fixed("version"),
+            _connect=cmd_runner_fmt.as_func(lambda v: ["connect", v]),
+            _disconnect=cmd_runner_fmt.as_func(lambda v: ["disconnect", v]),
+            _connections=cmd_runner_fmt.as_fixed("connections"),
+            slot=cmd_runner_fmt.as_list(),
         ),
         check_rc=False,
         **kwargs,

plugins/modules/snap_connect.py

@@ -0,0 +1,163 @@
+#!/usr/bin/python
+
+# Copyright (c) 2026, Alexei Znamensky (russoz) <russoz@gmail.com>
+# 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
+
+from __future__ import annotations
+
+DOCUMENTATION = r"""
+module: snap_connect
+short_description: Manages snap interface connections
+version_added: "12.6.0"
+description:
+  - Manages connections between snap plugs and slots.
+  - Snaps run in a sandbox and need explicit interface connections to access system resources
+    or communicate with other snaps.
+extends_documentation_fragment:
+  - community.general.attributes
+attributes:
+  check_mode:
+    support: full
+  diff_mode:
+    support: full
+options:
+  state:
+    description:
+      - Desired state of the connection.
+    type: str
+    choices: [absent, present]
+    default: present
+  plug:
+    description:
+      - The plug endpoint in the format C(<snap>:<plug>).
+    type: str
+    required: true
+  slot:
+    description:
+      - The slot endpoint in the format C(<snap>:<slot>) or C(:<slot>) for system slots.
+      - If not specified, snapd will attempt to find a matching slot automatically.
+    type: str
+notes:
+  - Privileged operations require root privileges.
+author:
+  - Alexei Znamensky (@russoz) <russoz@gmail.com>
+seealso:
+  - module: community.general.snap
+  - module: community.general.snap_alias
+"""
+
+EXAMPLES = r"""
+- name: Connect firefox camera plug to system camera slot
+  community.general.snap_connect:
+    plug: firefox:camera
+    slot: ":camera"
+
+- name: Connect snap plug (slot resolved automatically by snapd)
+  community.general.snap_connect:
+    plug: my-app:network
+
+- name: Disconnect a specific connection
+  community.general.snap_connect:
+    plug: firefox:camera
+    slot: ":camera"
+    state: absent
+"""
+
+RETURN = r"""
+snap_connections:
+  description: The list of snap connections after execution.
+  type: list
+  elements: dict
+  returned: always
+  contains:
+    interface:
+      description: The interface name.
+      type: str
+    plug:
+      description: The plug endpoint.
+      type: str
+    slot:
+      description: The slot endpoint.
+      type: str
+version:
+  description: Versions of snap components as reported by C(snap version).
+  type: dict
+  returned: always
+"""
+
+import re
+
+from ansible_collections.community.general.plugins.module_utils.module_helper import StateModuleHelper
+from ansible_collections.community.general.plugins.module_utils.snap import get_version, snap_runner
+
+
+class SnapConnect(StateModuleHelper):
+    _RE_CONNECTIONS = re.compile(r"^(?P<interface>\S+)\s+(?P<plug>\S+)\s+(?P<slot>\S+)\s+.*$")
+
+    module = dict(
+        argument_spec={
+            "state": dict(type="str", choices=["absent", "present"], default="present"),
+            "plug": dict(type="str", required=True),
+            "slot": dict(type="str"),
+        },
+        supports_check_mode=True,
+    )
+
+    def __init_module__(self):
+        self.runner = snap_runner(self.module)
+        self.vars.version = get_version(self.runner)
+        self.vars.set("snap_connections", self._get_connections(), change=True, diff=True)
+
+    def __quit_module__(self):
+        self.vars.snap_connections = self._get_connections()
+
+    def _get_connections(self):
+        def process(rc, out, err):
+            if rc != 0:
+                return []
+            connections = []
+            for line in out.splitlines()[1:]:
+                match = self._RE_CONNECTIONS.match(line.strip())
+                if match:
+                    connections.append(match.groupdict())
+            return connections
+
+        with self.runner("_connections", output_process=process) as ctx:
+            return ctx.run()
+
+    def _is_connected(self):
+        plug = self.vars.plug
+        slot = self.vars.slot
+        return any(
+            conn["slot"] != "-" and conn["plug"] == plug and (slot is None or conn["slot"] == slot)
+            for conn in self.vars.snap_connections
+        )
+
+    def state_present(self):
+        if not self._is_connected():
+            self.changed = True
+            if self.check_mode:
+                return
+            with self.runner("_connect slot") as ctx:
+                rc, dummy, err = ctx.run(_connect=self.vars.plug, slot=self.vars.slot)
+            if rc != 0:
+                self.do_raise(msg=f"snap connect failed: {err}")
+
+    def state_absent(self):
+        if self._is_connected():
+            self.changed = True
+            if self.check_mode:
+                return
+            with self.runner("_disconnect slot") as ctx:
+                rc, dummy, err = ctx.run(_disconnect=self.vars.plug, slot=self.vars.slot)
+            if rc != 0:
+                self.do_raise(msg=f"snap disconnect failed: {err}")
+
+
+def main():
+    SnapConnect.execute()
+
+
+if __name__ == "__main__":
+    main()

tests/integration/targets/snap_connect/aliases

@@ -0,0 +1,13 @@
+# 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
+
+azp/posix/1
+azp/posix/vm
+skip/alpine
+skip/fedora
+skip/freebsd
+skip/osx
+skip/macos
+skip/docker
+skip/rhel8.8  # TODO: fix

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

@@ -0,0 +1,7 @@
+---
+# 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
+
+dependencies:
+  - setup_snap

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

@@ -0,0 +1,13 @@
+---
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+# 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: Test
+  include_tasks: test.yml
+  when: has_snap

tests/integration/targets/snap_connect/tasks/test.yml

@@ -0,0 +1,99 @@
+---
+# 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
+
+# test-snapd-upower-observe-consumer declares a single plug: upower-observe
+# which auto-connects to the :upower-observe system slot.
+# We use this well-known test snap to exercise connect/disconnect.
+
+- name: Ensure snap 'test-snapd-upower-observe-consumer' is not installed
+  community.general.snap:
+    name: test-snapd-upower-observe-consumer
+    state: absent
+
+- name: Ensure snap 'test-snapd-upower-observe-consumer' is installed
+  community.general.snap:
+    name: test-snapd-upower-observe-consumer
+
+################################################################################
+
+- name: Disconnect upower-observe plug (ensure clean state)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+    state: absent
+
+- name: Connect upower-observe plug (check mode)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+  check_mode: true
+  register: connect_0
+
+- name: Connect upower-observe plug
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+  register: connect_1
+
+- name: Connect upower-observe plug (check mode idempotent)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+  check_mode: true
+  register: connect_2
+
+- name: Connect upower-observe plug (idempotent)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+  register: connect_3
+
+- name: Assert connect results
+  assert:
+    that:
+      - connect_0 is changed
+      - connect_1 is changed
+      - connect_2 is not changed
+      - connect_3 is not changed
+
+################################################################################
+
+- name: Disconnect upower-observe plug (check mode)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+    state: absent
+  check_mode: true
+  register: disconnect_0
+
+- name: Disconnect upower-observe plug
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+    state: absent
+  register: disconnect_1
+
+- name: Disconnect upower-observe plug (check mode idempotent)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+    state: absent
+  check_mode: true
+  register: disconnect_2
+
+- name: Disconnect upower-observe plug (idempotent)
+  community.general.snap_connect:
+    plug: test-snapd-upower-observe-consumer:upower-observe
+    slot: ":upower-observe"
+    state: absent
+  register: disconnect_3
+
+- name: Assert disconnect results
+  assert:
+    that:
+      - disconnect_0 is changed
+      - disconnect_1 is changed
+      - disconnect_2 is not changed
+      - disconnect_3 is not changed