New upstream version 4.18.0

tests/system/framework/hosts/base.py

@@ -47,6 +47,7 @@ class BaseLinuxHost(MultihostHost[ShadowMultihostDomain]):
         self._distro_name: str = "unknown"
         self._distro_major: int = 0
         self._distro_minor: int = 0
+        self._revision: int = 0
 
     def _distro_information(self):
         """
@@ -54,14 +55,19 @@ class BaseLinuxHost(MultihostHost[ShadowMultihostDomain]):
         """
         self.logger.info(f"Detecting distro information on {self.hostname}")
         os_release = self.fs.read("/etc/os-release")
-        self._os_release = dict(csv.reader([x for x in os_release.splitlines() if x], delimiter="="))
+        valid_lines = [line for line in os_release.splitlines() if line and not line.startswith("#")]
+        self._os_release = dict(csv.reader(valid_lines, delimiter="="))
         if "NAME" in self._os_release:
             self._distro_name = self._os_release["NAME"]
         if "VERSION_ID" not in self._os_release:
             return
-        if "." in self._os_release["VERSION_ID"]:
-            self._distro_major = int(self._os_release["VERSION_ID"].split(".", maxsplit=1)[0])
-            self._distro_minor = int(self._os_release["VERSION_ID"].split(".", maxsplit=1)[1])
+        if self._os_release["VERSION_ID"].count(".") == 2:
+            self._distro_major = int(self._os_release["VERSION_ID"].split(".")[0])
+            self._distro_minor = int(self._os_release["VERSION_ID"].split(".")[1])
+            self._revision = int(self._os_release["VERSION_ID"].split(".")[2])
+        elif self._os_release["VERSION_ID"].count(".") == 1:
+            self._distro_major = int(self._os_release["VERSION_ID"].split(".")[0])
+            self._distro_minor = int(self._os_release["VERSION_ID"].split(".")[1])
         else:
             self._distro_major = int(self._os_release["VERSION_ID"])
 

tests/system/framework/hosts/shadow.py

@@ -42,6 +42,9 @@ class ShadowHost(BaseHost, BaseLinuxHost):
         ]
         """Files to verify for mismatch."""
 
+        self._features: dict[str, bool] | None = None
+        """Features supported by the host."""
+
     def pytest_setup(self) -> None:
         super().pytest_setup()
 
@@ -61,6 +64,34 @@ class ShadowHost(BaseHost, BaseLinuxHost):
         """
         raise NotImplementedError("Stopping shadow service is not implemented.")
 
+    @property
+    def features(self) -> dict[str, bool]:
+        """
+        Features supported by the host.
+        """
+        if self._features is not None:
+            return self._features
+
+        self.logger.info(f"Detecting shadow features on {self.hostname}")
+        result = self.conn.run(
+            """
+            set -ex
+
+            getent gshadow > /dev/null 2>&1 && echo "gshadow" || :
+            """,
+            log_level=ProcessLogLevel.Error,
+        )
+
+        # Set default values
+        self._features = {
+            "gshadow": False,
+        }
+
+        self._features.update({k: True for k in result.stdout_lines})
+        self.logger.info("Detected features:", extra={"data": {"Features": self._features}})
+
+        return self._features
+
     def backup(self) -> Any:
         """
         Backup all shadow data.

tests/system/framework/misc/__init__.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import datetime
 from typing import Any
 
 
@@ -40,3 +41,14 @@ def to_list_of_strings(value: Any | list[Any] | None) -> list[str]:
     :rtype: list[str]
     """
     return [str(x) for x in to_list(value)]
+
+
+def days_since_epoch():
+    """
+    Gets the current date and returns the number of days since 1970-01-01 UTC, this time is also referred to as the
+    Unix epoch.
+    """
+    epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+    now_utc = datetime.datetime.now(datetime.timezone.utc)
+    delta = now_utc - epoch
+    return delta.days

tests/system/framework/roles/shadow.py

@@ -127,3 +127,16 @@ class Shadow(BaseLinuxRole[ShadowHost]):
         self.host.discard_file("/etc/gshadow")
 
         return cmd
+
+    def chage(self, *args) -> ProcessResult:
+        """
+        Change user password expiry information.
+        """
+        args_dict = self._parse_args(args)
+        self.logger.info(f'Changing user password expiry information on user "{args_dict["name"]}" on {self.host.hostname}')
+        cmd = self.host.conn.run("chage " + args[0], log_level=ProcessLogLevel.Error)
+
+        self.host.discard_file("/etc/passwd")
+        self.host.discard_file("/etc/shadow")
+
+        return cmd

tests/system/framework/utils/tools.py

@@ -14,7 +14,9 @@ __all__ = [
     "UnixGroup",
     "IdEntry",
     "PasswdEntry",
+    "ShadowEntry",
     "GroupEntry",
+    "GShadowEntry"
     "InitgroupsEntry",
     "LinuxToolsUtils",
     "KillCommand",
@@ -222,6 +224,98 @@ class PasswdEntry(object):
         return cls.FromDict(result[0])
 
 
+class ShadowEntry(object):
+    """
+    Result of ``getent shadow``
+    """
+
+    def __init__(
+        self,
+        name: str,
+        password: str,
+        last_changed: int,
+        min_days: int,
+        max_days: int,
+        warn_days: int,
+        inactivity_days: int,
+        expiration_date: int,
+    ) -> None:
+        self.name: str | None = name
+        """
+        User name.
+        """
+
+        self.password: str | None = password
+        """
+        User password.
+        """
+
+        self.last_changed: int = last_changed
+        """
+        Last password change.
+        """
+
+        self.min_days: int = min_days
+        """
+        Minimum number of days before a password change is allowed.
+        """
+
+        self.max_days: int = max_days
+        """
+        Maximum number of days a password is valid.
+        """
+
+        self.warn_days: int = warn_days
+        """
+        Number of days to warn the user before the password expires.
+        """
+
+        self.inactivity_days: int | None = inactivity_days
+        """
+        Number of days after a password expires before the account is disabled.
+        """
+
+        self.expiration_date: int | None = expiration_date
+        """
+        The account expiration date, expressed as the number of days since 1970-01-01 00:00:00 UTC.
+        """
+
+    def __str__(self) -> str:
+        return (
+            f"({self.name}:{self.password}:{self.last_changed}:"
+            f"{self.min_days}:{self.max_days}:{self.warn_days}:"
+            f"{self.inactivity_days}:{self.expiration_date}:)"
+        )
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    @classmethod
+    def FromDict(cls, d: dict[str, Any]) -> ShadowEntry:
+        return cls(
+            name=d.get("username", None),
+            password=d.get("password", None),
+            last_changed=d.get("last_changed", None),
+            min_days=d.get("minimum", None),
+            max_days=d.get("maximum", None),
+            warn_days=d.get("warn", None),
+            inactivity_days=d.get("inactive", None),
+            expiration_date=d.get("expire", None),
+        )
+
+    @classmethod
+    def FromOutput(cls, stdout: str) -> ShadowEntry:
+        result = jc.parse("shadow", stdout)
+
+        if not isinstance(result, list):
+            raise TypeError(f"Unexpected type: {type(result)}, expecting list")
+
+        if len(result) != 1:
+            raise ValueError("More then one entry was returned")
+
+        return cls.FromDict(result[0])
+
+
 class GroupEntry(object):
     """
     Result of ``getent group``
@@ -276,6 +370,69 @@ class GroupEntry(object):
         return cls.FromDict(result[0])
 
 
+class GShadowEntry(object):
+    """
+    Result of ``getent gshadow``
+    """
+
+    def __init__(
+        self,
+        name: str,
+        password: str,
+        administrators: str,
+        members: str,
+    ) -> None:
+        self.name: str | None = name
+        """
+        Group name.
+        """
+
+        self.password: str | None = password
+        """
+        Group password.
+        """
+
+        self.administrators: int = administrators
+        """
+        Group administrators.
+        """
+
+        self.members: int = members
+        """
+        Group members.
+        """
+
+    def __str__(self) -> str:
+        return (
+            f"({self.name}:{self.password}:{self.administrators}:"
+            f"{self.members})"
+        )
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    @classmethod
+    def FromDict(cls, d: dict[str, Any]) -> GShadowEntry:
+        return cls(
+            name=d.get("group_name", None),
+            password=d.get("password", None),
+            administrators=d.get("administrators", None),
+            members=d.get("members", []),
+        )
+
+    @classmethod
+    def FromOutput(cls, stdout: str) -> GShadowEntry:
+        result = jc.parse("gshadow", stdout)
+
+        if not isinstance(result, list):
+            raise TypeError(f"Unexpected type: {type(result)}, expecting list")
+
+        if len(result) != 1:
+            raise ValueError("More then one entry was returned")
+
+        return cls.FromDict(result[0])
+
+
 class InitgroupsEntry(object):
     """
     Result of ``getent initgroups``
@@ -435,6 +592,19 @@ class GetentUtils(MultihostUtility[MultihostHost]):
         """
         return self.__exec(PasswdEntry, "passwd", name, service)
 
+    def shadow(self, name: str | int, *, service: str | None = None) -> ShadowEntry | None:
+        """
+        Call ``getent shadow $name``
+
+        :param name: User name or id.
+        :type name: str | int
+        :param service: Service used, defaults to None
+        :type service: str | None
+        :return: shadow data, None if not found
+        :rtype: ShadowEntry | None
+        """
+        return self.__exec(ShadowEntry, "shadow", name, service)
+
     def group(self, name: str | int, *, service: str | None = None) -> GroupEntry | None:
         """
         Call ``getent group $name``
@@ -448,6 +618,19 @@ class GetentUtils(MultihostUtility[MultihostHost]):
         """
         return self.__exec(GroupEntry, "group", name, service)
 
+    def gshadow(self, name: str | int, *, service: str | None = None) -> GShadowEntry | None:
+        """
+        Call ``getent gshadow $name``
+
+        :param name: Group name or id.
+        :type name: str | int
+        :param service: Service used, defaults to None
+        :type service: str | None
+        :return: group data, None if not found
+        :rtype: GShadowEntry | None
+        """
+        return self.__exec(GShadowEntry, "gshadow", name, service)
+
     def initgroups(self, name: str, *, service: str | None = None) -> InitgroupsEntry:
         """
         Call ``getent initgroups $name``