Read group and user information using getent to respect nsswitch.conf

pyinfra/facts/server.py

@@ -5,7 +5,7 @@
 import shutil
 from datetime import datetime
 from tempfile import mkdtemp
-from typing import Dict, List, Optional
+from typing import Dict, Iterable, List, Optional, Union
 
 from dateutil.parser import parse as parse_date
 from distro import distro
@@ -385,13 +385,71 @@ def process(self, output):
         return sysctls
 
 
+class GroupInfo(TypedDict):
+    name: str
+    password: str
+    gid: int
+    user_list: list[str]
+
+
+def _group_info_from_group_str(info: str) -> GroupInfo:
+    """
+    Parses an entry from /etc/group or a similar source, e.g.
+    'plugdev:x:46:sysadmin,user2' into a GroupInfo dict object
+    """
+
+    fields = info.split(":")
+
+    if len(fields) != 4:
+        raise ValueError(f"Error parsing group '{info}', expected exactly 4 fields separated by :")
+
+    return {
+        "name": fields[0],
+        "password": fields[1],
+        "gid": int(fields[2]),
+        "user_list": fields[3].split(","),
+    }
+
+
+class Group(FactBase[GroupInfo]):
+    """
+    Returns information on a specific group on the system.
+    """
+
+    def command(self, group):
+        # FIXME: the '|| true' ensures 'process' is called, even if
+        #        getent was unable to find information on the group
+        #        There must be a better way to do this !
+        #        e.g. allow facts 'process' method access to the process
+        #        return code ?
+        return f"getent group {group} || true"
+
+    default = None
+
+    def process(self, output: Iterable[str]) -> str:
+        group_string = next(iter(output), None)
+
+        if group_string is None:
+            # This will happen if the group was simply not found
+            return None
+
+        return _group_info_from_group_str(group_string)
+
+
 class Groups(FactBase[List[str]]):
     """
     Returns a list of groups on the system.
     """
 
     def command(self):
-        return "cat /etc/group"
+        # getent will return the same output as `cat /etc/groups` but will
+        # respect nsswitch.conf settings, e.g. for using LDAP as additional source
+        # Note, that LDAP e.g. using sssd might be configured to not enumerate
+        # all groups / users, in which case only the local groups will be returned
+        return "getent group"
+
+    def requires_command(self) -> str:
+        return "getent"
 
     default = list
 
@@ -410,7 +468,20 @@ def process(self, output) -> list[str]:
 Crontab = crontab.Crontab
 
 
-class Users(FactBase):
+class UserInfo(TypedDict):
+    name: str
+    comment: str
+    home: str
+    shell: str
+    group: str
+    groups: list[str]
+    uid: int
+    gid: int
+    lastlog: str
+    password: str
+
+
+class Users(FactBase[dict[str, UserInfo]]):
     """
     Returns a dictionary of users -> details.
 
@@ -434,11 +505,13 @@ class Users(FactBase):
         }
     """
 
-    def command(self):
-        return """
+    def requires_command(self) -> str:
+        return "getent"
 
-        for i in `cat /etc/passwd | cut -d: -f1`; do
-            ENTRY=`grep ^$i: /etc/passwd`;
+    def command(self) -> str:
+        return """
+        for i in `getent passwd | cut -d: -f1`; do
+            ENTRY=`getent passwd | grep ^$i:`;
             LASTLOG_RAW=`(lastlog -u $i 2> /dev/null || lastlogin $i 2> /dev/null)`;
             LASTLOG=`echo $LASTLOG_RAW | grep ^$i | tr -s ' '`;
             PASSWORD=`grep ^$i: /etc/shadow | cut -d: -f2`;
@@ -448,7 +521,7 @@ def command(self):
 
     default = dict
 
-    def process(self, output):
+    def process(self, output: Iterable[str]) -> dict[str, UserInfo]:
         users = {}
         rex = r"[A-Z][a-z]{2} [A-Z][a-z]{2} {1,2}\d+ .+$"
 

tests/facts/server.Group/group.json

@@ -0,0 +1,14 @@
+{
+    "arg": "plugdev",
+    "command": "getent group plugdev || true",
+    "requires_command": "getent",
+    "output": [
+        "plugdev:x:46:sysadmin,myuser,abc"
+    ],
+    "fact": {
+        "name": "plugdev",
+        "password": "x",
+        "gid": 46,
+        "user_list": ["sysadmin", "myuser", "abc"]
+    }
+}

tests/facts/server.Group/nogroup.json

@@ -0,0 +1,7 @@
+{
+    "arg": "doesnotexist",
+    "command": "getent group doesnotexist || true",
+    "requires_command": "getent",
+    "output": [],
+    "fact": null
+}

tests/facts/server.Groups/groups.json

@@ -1,5 +1,6 @@
 {
-    "command": "cat /etc/group",
+    "command": "getent group",
+    "requires_command": "getent",
     "output": [
         "lpadmin:x:114:vagrant",
         "sambashare:x:115:vagrant",

tests/facts/server.Users/mixed.json

@@ -8,7 +8,8 @@
         "freebsd:*:1001:1001:FreeBSD:/home/freebsd:/bin/sh|freebsd|freebsd|freebsd pts/0 10.1.10.10 Thu Aug 31 05:37:58 2023|$bsdpw$",
         "freebsdnologin:*:1001:1001:FreeBSD NoLogin:/home/freebsdnologin:/bin/sh|freebsdnologin|freebsdnologin||"
     ],
-    "command": "for i in `cat /etc/passwd | cut -d: -f1`; do\n            ENTRY=`grep ^$i: /etc/passwd`;\n            LASTLOG_RAW=`(lastlog -u $i 2> /dev/null || lastlogin $i 2> /dev/null)`;\n            LASTLOG=`echo $LASTLOG_RAW | grep ^$i | tr -s ' '`;\n            PASSWORD=`grep ^$i: /etc/shadow | cut -d: -f2`;\n            echo \"$ENTRY|`id -gn $i`|`id -Gn $i`|$LASTLOG|$PASSWORD\";\n        done",
+    "command": "for i in `getent passwd | cut -d: -f1`; do\n            ENTRY=`getent passwd | grep ^$i:`;\n            LASTLOG_RAW=`(lastlog -u $i 2> /dev/null || lastlogin $i 2> /dev/null)`;\n            LASTLOG=`echo $LASTLOG_RAW | grep ^$i | tr -s ' '`;\n            PASSWORD=`grep ^$i: /etc/shadow | cut -d: -f2`;\n            echo \"$ENTRY|`id -gn $i`|`id -Gn $i`|$LASTLOG|$PASSWORD\";\n        done",
+    "requires_command": "getent",
     "fact": {
         "root": {
             "home": "/root",
@@ -104,4 +105,4 @@
             "password": ""
         }
     }
-}
+}

tests/test_facts_utils.py

@@ -0,0 +1,21 @@
+import pytest
+
+from pyinfra.facts.server import _group_info_from_group_str
+
+
+def test__group_info_from_group_str() -> None:
+    test_str = "plugdev:x:46:sysadmin,user2"
+    group_info = _group_info_from_group_str(test_str)
+
+    assert group_info["name"] == "plugdev"
+    assert group_info["password"] == "x"
+    assert group_info["gid"] == 46
+    assert group_info["user_list"] == ["sysadmin", "user2"]
+
+
+def test__group_info_from_group_str_empty() -> None:
+    with pytest.raises(ValueError):
+        _group_info_from_group_str("")
+
+    with pytest.raises(ValueError):
+        _group_info_from_group_str("a:b:")