Add support for multi-token management in SmartCardUtils

Author: krishnavema

sssd_test_framework/misc/globals.py

@@ -3,3 +3,12 @@
 test_venv = "/opt/test_venv"
 test_venv_bin = f"{test_venv}/bin"
 scauto_path = f"{test_venv_bin}/scauto"
+
+USER_RESOLVABLE_ATTEMPTS: int = 15
+"""Maximum number of polling attempts when waiting for a user to become resolvable by SSSD."""
+
+USER_RESOLVABLE_INTERVAL_S: int = 2
+"""Seconds to sleep between each polling attempt when waiting for a user to be resolvable."""
+
+USER_RESOLVABLE_CACHE_EXPIRY_ATTEMPT: int = 3
+"""Attempt number at which ``sss_cache -E`` is called to flush the SSSD cache."""

sssd_test_framework/utils/authentication.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import time
 from datetime import datetime
 from enum import Enum
 from typing import Any
@@ -11,7 +12,12 @@
 from pytest_mh.utils.fs import LinuxFileSystem
 
 from ..misc.errors import ExpectScriptError
-from ..misc.globals import test_venv_bin
+from ..misc.globals import (
+    USER_RESOLVABLE_ATTEMPTS,
+    USER_RESOLVABLE_CACHE_EXPIRY_ATTEMPT,
+    USER_RESOLVABLE_INTERVAL_S,
+    test_venv_bin,
+)
 from .idp import IdpAuthenticationUtils
 
 __all__ = [
@@ -844,6 +850,54 @@ def vfido_passkey(
         rc, _, _, _ = self.vfido_passkey_with_output(username=username, pin=pin, command=command)
         return rc == 0
 
+    def smartcard_with_su_output(self, username: str, pin: str, *, num_certs: int = 1) -> ProcessResult:
+        """
+        Wait for the user to become resolvable then authenticate via ``su`` with the smart card PIN.
+
+        :param username: Username.
+        :type username: str
+        :param pin: Smart card PIN.
+        :type pin: str
+        :param num_certs: Number of certificates that map to the user, defaults to 1.
+        :type num_certs: int, optional
+        :return: Result of the ``su`` command.
+        :rtype: ProcessResult
+        """
+        for attempt in range(USER_RESOLVABLE_ATTEMPTS):
+            time.sleep(USER_RESOLVABLE_INTERVAL_S)
+            check = self.host.conn.run(f"getent passwd {username}", raise_on_error=False)
+            if check.rc == 0:
+                break
+            if attempt == USER_RESOLVABLE_CACHE_EXPIRY_ATTEMPT:
+                self.host.conn.run("sss_cache -E", raise_on_error=False)
+        else:
+            raise AssertionError(
+                f"User '{username}' was not resolvable by SSSD after {USER_RESOLVABLE_ATTEMPTS} attempts"
+            )
+
+        su_input = f"1\n{pin}" if num_certs > 1 else pin
+        return self.host.conn.run(
+            f"su - {username} -c 'su - {username} -c whoami'",
+            input=su_input,
+            raise_on_error=False,
+        )
+
+    def smartcard_with_su(self, username: str, pin: str, *, num_certs: int = 1) -> bool:
+        """
+        Wait for the user to become resolvable then authenticate via ``su`` with the smart card PIN.
+
+        :param username: Username.
+        :type username: str
+        :param pin: Smart card PIN.
+        :type pin: str
+        :param num_certs: Number of certificates that map to the user, defaults to 1.
+        :type num_certs: int, optional
+        :return: True if authentication was successful, False otherwise.
+        :rtype: bool
+        """
+        result = self.smartcard_with_su_output(username, pin, num_certs=num_certs)
+        return result.rc == 0 and "PIN" in result.stderr and username in result.stdout
+
 
 class SSHAuthenticationUtils(MultihostUtility[MultihostHost]):
     """

sssd_test_framework/utils/smartcard.py

@@ -47,24 +47,34 @@ def __init__(self, host: MultihostHost, fs: LinuxFileSystem, svc: SystemdService
         self.svc: SystemdServices = svc
         """Systemd utility to manage and interact with svc."""
 
-    def initialize_card(self, label: str = "sc_test", so_pin: str = "12345678", user_pin: str = "123456") -> None:
+    def initialize_card(
+        self,
+        label: str = "sc_test",
+        so_pin: str = "12345678",
+        user_pin: str = "123456",
+        reset: bool = True,
+    ) -> None:
         """
-        Initializes a SoftHSM token with the given label and PINs.
+        Initialize a SoftHSM token with the given label and PINs.
 
-        Cleans cache directories and prepares the token directory.
+        When *reset* is ``True`` (default), existing token storage and OpenSC
+        caches are removed first.  Pass ``False`` to add a token alongside
+        existing ones (multi-token / multi-card setup).
 
         :param label: Token label, defaults to "sc_test"
         :type label: str, optional
         :param so_pin: Security Officer PIN, defaults to "12345678"
         :type so_pin: str, optional
         :param user_pin: User PIN, defaults to "123456"
         :type user_pin: str, optional
+        :param reset: Remove existing tokens before initializing, defaults to True
+        :type reset: bool, optional
         """
-        for path in self.OPENSC_CACHE_PATHS:
-            self.fs.rm(path)
-
-        self.fs.rm(self.TOKEN_STORAGE_PATH)
-        self.fs.mkdir_p(self.TOKEN_STORAGE_PATH)
+        if reset:
+            for path in self.OPENSC_CACHE_PATHS:
+                self.fs.rm(path)
+            self.fs.rm(self.TOKEN_STORAGE_PATH)
+            self.fs.mkdir_p(self.TOKEN_STORAGE_PATH)
 
         args: CLIBuilderArgs = {
             "label": (self.cli.option.VALUE, label),
@@ -82,6 +92,8 @@ def add_cert(
         cert_id: str = "01",
         pin: str = "123456",
         private: bool | None = False,
+        token_label: str | None = None,
+        label: str | None = None,
     ) -> None:
         """
         Adds a certificate or private key to the smart card.
@@ -94,6 +106,15 @@ def add_cert(
         :type pin: str, optional
         :param private: Whether the object is a private key. Defaults to False.
         :type private: bool, optional
+        :param token_label: Label of the target token. When ``None`` (the
+            default) ``pkcs11-tool`` writes to the first available token.
+            Set this when multiple tokens exist to target a specific one.
+        :type token_label: str | None, optional
+        :param label: Label for the PKCS#11 object being written.  Required
+            when ``p11_child`` accesses the token directly (i.e. without
+            ``virt_cacard``), because the response parser expects a
+            non-empty label.
+        :type label: str | None, optional
         """
         obj_type = "privkey" if private else "cert"
         args: CLIBuilderArgs = {
@@ -104,9 +125,20 @@ def add_cert(
             "type": (self.cli.option.VALUE, obj_type),
             "id": (self.cli.option.VALUE, cert_id),
         }
+        if token_label is not None:
+            args["token-label"] = (self.cli.option.VALUE, token_label)
+        if label is not None:
+            args["label"] = (self.cli.option.VALUE, label)
         self.host.conn.run(self.cli.command("pkcs11-tool", args), env={"SOFTHSM2_CONF": self.SOFTHSM2_CONF_PATH})
 
-    def add_key(self, key_path: str, key_id: str = "01", pin: str = "123456") -> None:
+    def add_key(
+        self,
+        key_path: str,
+        key_id: str = "01",
+        pin: str = "123456",
+        token_label: str | None = None,
+        label: str | None = None,
+    ) -> None:
         """
         Adds a private key to the smart card.
 
@@ -116,8 +148,12 @@ def add_key(self, key_path: str, key_id: str = "01", pin: str = "123456") -> Non
         :type key_id: str, optional
         :param pin: User PIN, defaults to "123456"
         :type pin: str, optional
+        :param token_label: Label of the target token (see :meth:`add_cert`).
+        :type token_label: str | None, optional
+        :param label: Label for the PKCS#11 object (see :meth:`add_cert`).
+        :type label: str | None, optional
         """
-        self.add_cert(cert_path=key_path, cert_id=key_id, pin=pin, private=True)
+        self.add_cert(cert_path=key_path, cert_id=key_id, pin=pin, private=True, token_label=token_label, label=label)
 
     def generate_cert(
         self,

sssd_test_framework/utils/sssd.py

@@ -23,6 +23,7 @@
     from ..roles.base import BaseRole
     from ..roles.kdc import KDC
     from .authselect import AuthselectUtils
+    from .smartcard import SmartCardUtils
 
 
 __all__ = [
@@ -1178,3 +1179,31 @@ def subid(self) -> None:
         Configure SSSD for subid.
         """
         self.sssd.authselect.select("sssd", ["with-subid"])
+
+    def smartcard_with_softhsm(self, smartcard: SmartCardUtils) -> None:
+        """
+        Configure SSSD for smart card authentication with SoftHSM multi-token support.
+
+        :param smartcard: SmartCardUtils instance.
+        :type smartcard: SmartCardUtils
+        """
+        conf = smartcard.SOFTHSM2_CONF_PATH
+        token_storage = smartcard.TOKEN_STORAGE_PATH
+        module = "/usr/lib64/pkcs11/libsofthsm2.so"
+
+        smartcard.host.conn.run(f"grep -q 'slots.removable' {conf} || echo 'slots.removable = true' >> {conf}")
+        smartcard.host.conn.run(f"cp {conf} /etc/softhsm2.conf")
+        smartcard.host.conn.run(f'echo "module: {module}" > /etc/pkcs11/modules/softhsm2.module')
+        smartcard.fs.mkdir_p("/etc/systemd/system/sssd.service.d")
+        smartcard.host.conn.run(
+            f'printf "[Service]\\nEnvironment=SOFTHSM2_CONF={conf}\\n" '
+            f"> /etc/systemd/system/sssd.service.d/softhsm.conf"
+        )
+        smartcard.host.conn.run("systemctl daemon-reload")
+        smartcard.host.conn.run("chmod -R o+rX /opt/test_ca/")
+        smartcard.host.conn.run(f"chown -R sssd:sssd {token_storage}/ && chmod -R 770 {token_storage}/")
+
+        self.sssd.authselect.select("sssd", ["with-smartcard", "with-mkhomedir"])
+        self.sssd.pam["pam_cert_auth"] = "True"
+        self.sssd.domain["local_auth_policy"] = "enable:smartcard"
+        self.sssd.start()