fix(sys-svc-compose): probe provider networks by their exact name

The network probe used community.docker.docker_network_info, whose
get_network() scans GET /networks and returns the first entry whose Name
equals the requested name OR whose Id starts with it, in a list order that
changes between calls. In main run 38038541631 (Zammad#1, swarm) the
central redis overlay got the ID n8nwzb1q39m3u4k7awwwmt6yg, the probe for
the not yet created provider network "n8n" returned it as existing, the
create step was skipped, docker stack deploy attached hermes to the redis
overlay as "n8n", and the oauth2-proxy then reached the password-protected
central redis by its bare host name: "NOAUTH Authentication required".
The same scan can also return the ID-prefix twin of a network that does
exist, so comparing the returned name alone would turn that network into
"missing" and fail the create with "already exists".

Both probes now run `container network ls --filter name=<entity> --format
'{{json .}}'`, whose name filter never matches an ID. The new lookup
network_driver picks the entry named exactly after the entity and returns
its driver, '' when there is none; the drift checks, their messages and the
create steps all read that one entry. No docker module is left in the file,
so its two docker-module-uses-container-cli suppressions go too.

Verified: on the host daemon GET /networks/<id prefix> returns a network of
another name and docker network ls --filter name=<id prefix> returns
nothing. In ansible-playbook against real docker network ls output the
lookup returned the exact network's driver in 20 of 20 probes beside an
ID-prefix twin, an empty driver (create path) for a missing network and for
a name only contained in a sibling, and the drift path for a foreign
driver. Unit tests cover the same cases; with a substring match instead of
the exact name the two sibling cases fail.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

plugins/lookup/network_driver.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.lookup import LookupBase
+
+from utils.roles.entity.name import get_entity_name
+
+
+class LookupModule(LookupBase):
+    """
+    Usage:
+      {{ lookup('network_driver', probe, network_role_id) }}
+
+    Args:
+      probe: the registered result of
+        `container network ls --filter name=<entity> --format '{{json .}}'`.
+      network_role_id: the role whose entity name names the network.
+
+    Returns:
+      The driver of the network named exactly after that entity, '' when none exists.
+    """
+
+    def run(self, terms, variables: dict[str, Any] | None = None, **kwargs):
+        if not terms or len(terms) != 2:
+            raise AnsibleError(
+                "lookup('network_driver', probe, network_role_id) expects exactly 2 terms"
+            )
+
+        probe, network_role_id = terms
+        name = get_entity_name(str(network_role_id))
+        lines = probe.get("stdout_lines") if isinstance(probe, dict) else None
+        for line in lines or []:
+            try:
+                network = json.loads(line)
+            except json.JSONDecodeError as exc:
+                raise AnsibleError(
+                    f"lookup('network_driver'): unparsable probe line {line!r}"
+                ) from exc
+            if network.get("Name") == name:
+                return [str(network.get("Driver", ""))]
+        return [""]

roles/sys-svc-compose/tasks/utils/network/create.yml

@@ -1,8 +1,16 @@
 ---
 - name: "🔍 Probe existing overlay network for '{{ network_role_id }}' (swarm scope)"
-  community.docker.docker_network_info: # nocheck: docker-module-uses-container-cli  read-only info module; returns structured network facts the wrapper's raw JSON would need parsed
-    name: "{{ network_role_id | get_entity_name }}"
+  ansible.builtin.command:
+    argv:
+      - container
+      - network
+      - ls
+      - --filter
+      - "name={{ network_role_id | get_entity_name }}"
+      - --format
+      - "{{ '{{json .}}' }}"
   register: sys_svc_compose_overlay_probe
+  changed_when: false
   when: (SWARM_CLUSTER_AVAILABLE | bool) and not (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
   run_once: true
   delegate_to: "{{ STACK_HOST }}"
@@ -11,12 +19,11 @@
   ansible.builtin.fail:
     msg: >-
       Network '{{ network_role_id | get_entity_name }}' exists with driver
-      '{{ sys_svc_compose_overlay_probe.network.Driver | default('unknown') }}' but 'overlay' is required.
+      '{{ lookup('network_driver', sys_svc_compose_overlay_probe, network_role_id) }}' but 'overlay' is required.
       Remove the stale network or revert the networks.local.force_bridge change.
   when:
     - (SWARM_CLUSTER_AVAILABLE | bool) and not (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
-    - sys_svc_compose_overlay_probe.exists | default(false) | bool
-    - (sys_svc_compose_overlay_probe.network.Driver | default('')) != 'overlay'
+    - lookup('network_driver', sys_svc_compose_overlay_probe, network_role_id) not in ['', 'overlay']
   run_once: true
   delegate_to: "{{ STACK_HOST }}"
 
@@ -44,27 +51,34 @@
       - "{{ network_role_id | get_entity_name }}"
   when:
     - (SWARM_CLUSTER_AVAILABLE | bool) and not (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
-    - not (sys_svc_compose_overlay_probe.exists | default(false) | bool)
+    - lookup('network_driver', sys_svc_compose_overlay_probe, network_role_id) == ''
   run_once: true
   delegate_to: "{{ STACK_HOST }}"
   changed_when: true
 
 - name: "🔍 Probe existing bridge network for '{{ network_role_id }}' (compose)"
-  community.docker.docker_network_info: # nocheck: docker-module-uses-container-cli  read-only info module; returns structured network facts the wrapper's raw JSON would need parsed
-    name: "{{ network_role_id | get_entity_name }}"
+  ansible.builtin.command:
+    argv:
+      - container
+      - network
+      - ls
+      - --filter
+      - "name={{ network_role_id | get_entity_name }}"
+      - --format
+      - "{{ '{{json .}}' }}"
   register: sys_svc_compose_bridge_probe
+  changed_when: false
   when: (not (SWARM_CLUSTER_AVAILABLE | bool)) or (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
 
 - name: "🛑 Fail on network driver drift for '{{ network_role_id }}' (compose)"
   ansible.builtin.fail:
     msg: >-
       Network '{{ network_role_id | get_entity_name }}' exists with driver
-      '{{ sys_svc_compose_bridge_probe.network.Driver | default('unknown') }}' but 'bridge' is required.
+      '{{ lookup('network_driver', sys_svc_compose_bridge_probe, network_role_id) }}' but 'bridge' is required.
       Remove the stale network or revert the networks.local.force_bridge change.
   when:
     - (not (SWARM_CLUSTER_AVAILABLE | bool)) or (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
-    - sys_svc_compose_bridge_probe.exists | default(false) | bool
-    - (sys_svc_compose_bridge_probe.network.Driver | default('')) != 'bridge'
+    - lookup('network_driver', sys_svc_compose_bridge_probe, network_role_id) not in ['', 'bridge']
 
 - name: "✅ Ensure docker bridge network for '{{ network_role_id }}' exists (compose)"
   ansible.builtin.command:
@@ -81,5 +95,5 @@
       - "{{ network_role_id | get_entity_name }}"
   when:
     - (not (SWARM_CLUSTER_AVAILABLE | bool)) or (lookup('config', network_role_id, 'networks.local.force_bridge', false) | bool)
-    - not (sys_svc_compose_bridge_probe.exists | default(false) | bool)
+    - lookup('network_driver', sys_svc_compose_bridge_probe, network_role_id) == ''
   changed_when: true

tests/unit/python/plugins/lookup/test_network_driver.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import unittest
+
+from ansible.errors import AnsibleError
+
+from . import PROJECT_ROOT
+
+
+def _load_lookup():
+    spec = importlib.util.spec_from_file_location(
+        "lookup_network_driver",
+        str(PROJECT_ROOT / "plugins/lookup/network_driver.py"),
+    )
+    mod = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    spec.loader.exec_module(mod)
+    return mod.LookupModule
+
+
+def _probe(*networks):
+    return {"rc": 0, "stdout_lines": [json.dumps(network) for network in networks]}
+
+
+class TestNetworkDriverLookup(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.LookupModule = _load_lookup()
+
+    def _driver(self, probe):
+        return self.LookupModule().run([probe, "web-app-n8n"])
+
+    def test_the_network_named_after_the_entity_gives_its_driver(self):
+        probe = _probe({"Name": "n8n", "Driver": "overlay"})
+        self.assertEqual(self._driver(probe), ["overlay"])
+
+    def test_a_sibling_listed_first_does_not_shadow_the_exact_name(self):
+        probe = _probe(
+            {"Name": "n8n-worker", "Driver": "bridge"},
+            {"Name": "n8n", "Driver": "overlay"},
+        )
+        self.assertEqual(self._driver(probe), ["overlay"])
+
+    def test_only_a_sibling_containing_the_name_means_no_network(self):
+        probe = _probe({"Name": "n8n-worker", "Driver": "overlay"})
+        self.assertEqual(self._driver(probe), [""])
+
+    def test_an_empty_listing_means_no_network(self):
+        self.assertEqual(self._driver(_probe()), [""])
+
+    def test_a_skipped_probe_means_no_network(self):
+        self.assertEqual(self._driver({"skipped": True, "changed": False}), [""])
+
+    def test_an_unparsable_line_is_rejected(self):
+        with self.assertRaises(AnsibleError):
+            self._driver({"rc": 0, "stdout_lines": ["not json"]})
+
+    def test_a_wrong_term_count_is_rejected(self):
+        with self.assertRaises(AnsibleError):
+            self.LookupModule().run([_probe()])
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()