add sssd_info module (ansible-collections#11120)

* add sssd_info module

* fix f-stings and remove support python2

* fix imports custom lib

* fix whitespace and add missing_required_lib

* fix str and add version

* try add mock test

* fix module and mock tests check

* fix required in main module

* fix spaces

* fix linters

* add final newline

* fix version of module

* fix description and error handling

* swap literal to dict

* fix str

* remove comment in methods

* remove _get in methods

* fix name method in test

* add botmeta

* fix description of server_type

* fix name of maintainer

* remove choices

* fix author

* fix type hint

* fix result

* fix spaces

* fix choices and empty returns

* fix mypy test result

* fix result

* run andebox yaml-doc

* remake simple try/exc for result

* fix tests

* add any type for testing mypy

* ruff formated

* fix docs

* remove unittest.main

* rename acc on git for official name

---------

Co-authored-by: Александр Габидуллин <agabidullin@astralinux.ru>

Author: a-gabidullin

.github/BOTMETA.yml

@@ -1502,6 +1502,8 @@ files:
     maintainers: vbotka
   $tests/fqdn_valid.py:
     maintainers: vbotka
+  $modules/sssd_info.py:
+    maintainers: a-gabidullin
 #########################
   docs/docsite/rst/filter_guide.rst: {}
   docs/docsite/rst/filter_guide_abstract_informations.rst: {}

plugins/modules/sssd_info.py

@@ -0,0 +1,239 @@
+#!/usr/bin/python
+# Copyright (c) 2025 Aleksandr Gabidullin <qualittv@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: sssd_info
+version_added: 12.2.0
+short_description: Check SSSD domain status using D-Bus
+description:
+  - Check the online status of SSSD domains, list domains, and retrieve active servers using D-Bus.
+author: "Aleksandr Gabidullin (@a-gabidullin)"
+requirements:
+  - dbus
+attributes:
+  check_mode:
+    support: full
+  diff_mode:
+    support: none
+  platform:
+    platforms: posix
+    description: This action requires a system with D-Bus and SSSD running.
+    support: full
+options:
+  action:
+    description:
+      - The action to perform.
+    type: str
+    required: true
+    choices:
+      domain_status: Check if domain is online.
+      domain_list: List all configured domains.
+      active_servers: Get active servers for domain.
+      list_servers: List all servers for domain.
+  domain:
+    description:
+      - Domain name to check.
+      - Required unless O(action=domain_list).
+      - When O(action=domain_list), this parameter is ignored and the module returns a list of all configured domains.
+    type: str
+  server_type:
+    description:
+      - Required parameter when O(action=active_servers) and O(action=list_servers).
+      - Optional and ignored for all other actions.
+      - At this point, the module supports ONLY the types C(IPA) for FreeIPA servers and C(AD).
+    type: str
+    choices: ['IPA', 'AD']
+extends_documentation_fragment:
+  - community.general.attributes
+"""
+
+EXAMPLES = r"""
+- name: Check SSSD domain status
+  community.general.sssd_info:
+    action: domain_status
+    domain: example.com
+  register: sssd_status_result
+
+- name: Get domain list
+  community.general.sssd_info:
+    action: domain_list
+  register: domain_list_result
+
+- name: Get active IPA servers for a domain
+  community.general.sssd_info:
+    action: active_servers
+    domain: example.com
+    server_type: IPA
+  register: active_servers_result
+
+- name: List servers for a domain
+  community.general.sssd_info:
+    action: list_servers
+    domain: example.com
+    server_type: AD
+  register: list_servers_result
+"""
+
+RETURN = r"""
+online:
+  description: The online status of the SSSD domain.
+  type: str
+  returned: when O(action=domain_status)
+  sample: online
+domain_list:
+  description: List of SSSD domains.
+  type: list
+  elements: str
+  returned: when O(action=domain_list)
+  sample: ["ipa.domain", "winad.test"]
+servers:
+  description: Active servers for the specified domain and type.
+  type: dict
+  returned: when O(action=active_servers)
+  sample: {"Global Catalog": "server1.winad.test", "Domain Server": "server2.winad.test"}
+list_servers:
+  description: List of servers for the specified domain.
+  type: list
+  elements: str
+  returned: when O(action=list_servers)
+  sample: ["server1.winad.test", "server2.winad.test"]
+"""
+
+
+from ansible.module_utils.basic import AnsibleModule
+from typing import Any
+from ansible_collections.community.general.plugins.module_utils import deps
+
+with deps.declare("dbus"):
+    import dbus
+
+
+class SSSDHandler:
+    """SSSD D-Bus handler"""
+
+    BUS_NAME = "org.freedesktop.sssd.infopipe"
+    DOMAIN_INTERFACE = "org.freedesktop.sssd.infopipe.Domains.Domain"
+    INFOPIPE_INTERFACE = "org.freedesktop.sssd.infopipe"
+
+    def __init__(self) -> None:
+        """Initialize SSSD D-Bus connection."""
+        self.bus = dbus.SystemBus()
+        self.sssd_obj = self.bus.get_object(self.BUS_NAME, "/org/freedesktop/sssd/infopipe")
+        self.infopipe_iface = dbus.Interface(self.sssd_obj, dbus_interface=self.INFOPIPE_INTERFACE)
+
+    def domain_path(self, domain: str) -> str:
+        return f"/org/freedesktop/sssd/infopipe/Domains/{domain.replace('.', '_2e')}"
+
+    def domain_object(self, domain: str) -> dbus.proxies.ProxyObject:
+        domain_path = self.domain_path(domain)
+        try:
+            return self.bus.get_object(self.BUS_NAME, domain_path)
+        except dbus.exceptions.DBusException as e:
+            raise Exception(f"Domain not found: {domain}. Error: {e}") from e
+
+    def check_domain_status(self, domain: str) -> str:
+        domain_obj = self.domain_object(domain)
+        iface = dbus.Interface(domain_obj, dbus_interface=self.DOMAIN_INTERFACE)
+        return "online" if iface.IsOnline() else "offline"
+
+    def active_servers(self, domain: str, server_type: str) -> dict[str, str]:
+        """Get active servers for domain.
+
+        Args:
+            domain: Domain name to get servers for.
+            server_type: Type of servers ('IPA' or 'AD').
+
+        Returns:
+            Dictionary with server types as keys and server names as values.
+        """
+        domain_obj = self.domain_object(domain)
+        iface = dbus.Interface(domain_obj, dbus_interface=self.DOMAIN_INTERFACE)
+
+        if server_type == "IPA":
+            server_name = f"{server_type} Server"
+            return {server_name: iface.ActiveServer(server_type)}
+        else:
+            return {
+                "AD Global Catalog": iface.ActiveServer(f"sd_gc_{domain}"),
+                "AD Domain Controller": iface.ActiveServer(f"sd_{domain}"),
+            }
+
+    def list_servers(self, domain: str, server_type: str) -> list[str]:
+        """List all servers for domain.
+
+        Args:
+            domain: Domain name to list servers for.
+            server_type: Type of servers ('IPA' or 'AD').
+
+        Returns:
+            List of server names.
+        """
+        domain_obj = self.domain_object(domain)
+        iface = dbus.Interface(domain_obj, dbus_interface=self.DOMAIN_INTERFACE)
+        if server_type == "IPA":
+            return iface.ListServers(server_type)
+        else:
+            return iface.ListServers(f"sd_{domain}")
+
+    def domain_list(self) -> list[str]:
+        """Get list of all domains.
+
+        Returns:
+            List of domain names.
+        """
+        response = self.infopipe_iface.ListDomains()
+        return [domain.rsplit("/", maxsplit=1)[-1].replace("_2e", ".") for domain in response]
+
+
+def main() -> None:
+    """Main function for the Ansible module."""
+    module = AnsibleModule(
+        argument_spec=dict(
+            action=dict(
+                type="str",
+                required=True,
+                choices=["domain_status", "domain_list", "active_servers", "list_servers"],
+            ),
+            domain=dict(type="str"),
+            server_type=dict(type="str", choices=["IPA", "AD"]),
+        ),
+        supports_check_mode=True,
+        required_if=[
+            ("action", "domain_status", ["domain"]),
+            ("action", "list_servers", ["domain", "server_type"]),
+            ("action", "active_servers", ["domain", "server_type"]),
+        ],
+    )
+
+    deps.validate(module)
+
+    action = module.params["action"]
+    domain = module.params.get("domain")
+    server_type = module.params.get("server_type")
+
+    sssd = SSSDHandler()
+    result: dict[str, Any] = {}
+
+    try:
+        if action == "domain_status":
+            result["online"] = sssd.check_domain_status(domain)
+        elif action == "domain_list":
+            result["domain_list"] = sssd.domain_list()
+        elif action == "active_servers":
+            result["servers"] = sssd.active_servers(domain, server_type)
+        elif action == "list_servers":
+            result["list_servers"] = sssd.list_servers(domain, server_type)
+
+    except Exception as e:
+        module.fail_json(msg=f"Error: {e}")
+
+    module.exit_json(**result)
+
+
+if __name__ == "__main__":
+    main()