diff --git a/lisa/microsoft/testsuites/fedora/selinux_validation.py b/lisa/microsoft/testsuites/fedora/selinux_validation.py new file mode 100644 index 0000000000..3e4d4a5ca9 --- /dev/null +++ b/lisa/microsoft/testsuites/fedora/selinux_validation.py @@ -0,0 +1,405 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +from datetime import datetime, timedelta + +from assertpy.assertpy import assert_that + +from lisa import ( + Node, + TestCaseMetadata, + TestSuite, + TestSuiteMetadata, + simple_requirement, +) +from lisa.operating_system import Fedora +from lisa.util import check_till_timeout + + +@TestSuiteMetadata( + area="fedora", + category="functional", + description=""" + Fedora SELinux Validation Tests. + + Validates SELinux configuration, policy enforcement, and security + tool functionality on Fedora cloud images. + """, +) +class FedoraSELinuxValidation(TestSuite): + _SELINUX_FS_MOUNT = "/sys/fs/selinux" + + def _assert_rpm_installed(self, node: Node, packages: list[str]) -> None: + """Assert that packages are already installed.""" + for pkg in packages: + result = node.execute(f"rpm -q {pkg}") + assert_that(result.exit_code).described_as( + f"{pkg} package must be installed" + ).is_equal_to(0) + + def _backup_audit_rules(self, node: Node) -> str: + """Backup current audit rules and clear them.""" + backup = node.execute( + "auditctl -l | sed 's/^No rules$/# No rules/'", shell=True, sudo=True + ) + node.execute("auditctl -D", sudo=True) + node.execute("auditctl -w /etc/shadow -p w", sudo=True) + return backup.stdout + + def _restore_audit_rules(self, node: Node, backup: str) -> None: + """Restore audit rules from backup.""" + node.execute("auditctl -D", sudo=True, no_error_log=True) + if backup.strip(): + node.execute( + f"echo '{backup}' > /tmp/audit_backup && auditctl -R /tmp/audit_backup", + shell=True, + sudo=True, + no_error_log=True, + ) + node.execute("rm -f /tmp/audit_backup", sudo=True, no_error_log=True) + + def _wait_for_service_active( + self, node: Node, service: str, timeout: int = 30 + ) -> None: + """Wait for a systemd service to become active.""" + + def check_active() -> bool: + result = node.execute(f"systemctl is-active {service}", no_error_log=True) + return result.stdout.strip() == "active" + + check_till_timeout( + check_active, + timeout_message=f"{service} service did not become active", + timeout=timeout, + ) + + def _assert_cmd_fails_with_error( + self, node: Node, cmd: str, error_keywords: list[str], desc: str + ) -> None: + """Assert command fails or outputs error keywords.""" + result = node.execute(cmd, shell=True, sudo=True, no_error_log=True) + combined = (result.stdout + result.stderr).lower() + has_error = ( + any(kw in combined for kw in error_keywords) or result.exit_code != 0 + ) + assert_that(has_error).described_as(f"{desc} (got: {combined[:200]})").is_true() + + def _setenforce_and_verify(self, node: Node, value: str) -> None: + """Run setenforce and verify both exit code and enforce file.""" + r = node.execute(f"setenforce {value}", sudo=True) + assert_that(r.exit_code).described_as( + f"setenforce {value} must succeed" + ).is_equal_to(0) + grep = node.execute(f"grep {value} {self._SELINUX_FS_MOUNT}/enforce") + assert_that(grep.exit_code).described_as( + f"enforce file must contain {value} after setenforce {value}" + ).is_equal_to(0) + + @TestCaseMetadata( + description=""" + Verify setenforce command works correctly. + + Tests switching between SELinux enforcing and permissive modes, + validates mode changes are reflected in the enforce file and + audit logs, and verifies error handling for edge cases. + """, + priority=1, + requirement=simple_requirement(supported_os=[Fedora]), + use_new_environment=True, + ) + def verify_setenforce(self, node: Node) -> None: + self._assert_rpm_installed(node, ["libselinux", "libselinux-utils"]) + + # setenforce binary must exist and be executable + help_result = node.execute("setenforce --help", no_error_log=True) + assert_that(help_result.exit_code).described_as( + "setenforce binary must exist and be executable (--help exits 0 or 1)" + ).is_in(0, 1) + + # auditctl must be present + which_auditctl = node.execute("which auditctl") + assert_that(which_auditctl.exit_code).described_as( + "auditctl must be present for proper functioning" + ).is_equal_to(0) + + # auditd must be active on boot; not started manually + self._wait_for_service_active(node, "auditd", timeout=60) + + backup_rules = "" + try: + backup_rules = self._backup_audit_rules(node) + node.log.info(f"Original audit rules:\n{backup_rules}") + + # timestamp - 1s mirrors bash `sleep 1`; time.sleep() not permitted in LISA + raw_time = node.execute("date '+%m/%d/%Y %T'").stdout.strip() + start_time = ( + datetime.strptime(raw_time, "%m/%d/%Y %H:%M:%S") - timedelta(seconds=1) + ).strftime("%m/%d/%Y %H:%M:%S") + + # setenforce 1 → 0 → 1 with enforce file verification at each step + self._setenforce_and_verify(node, "1") + self._setenforce_and_verify(node, "0") + self._setenforce_and_verify(node, "1") # restore + + # poll until MAC_STATUS record appears, scoped to start_time + def check_mac_status_enforcing1() -> bool: + result = node.execute( + f"ausearch --input-logs -m MAC_STATUS -i -ts {start_time}" + ' | grep "type=MAC_STATUS"' + ' | grep "enforcing=1" | grep "old_enforcing=0"', + sudo=True, + no_error_log=True, + shell=True, + ) + return result.exit_code == 0 + + check_till_timeout( + check_mac_status_enforcing1, + timeout_message=( + "MAC_STATUS record enforcing=1 old_enforcing=0" + " not found in audit log" + ), + timeout=30, + ) + + # Verify enforcing=0 record exists + avc0_result = node.execute( + f"ausearch --input-logs -m MAC_STATUS -i -ts {start_time}" + ' | grep "type=MAC_STATUS"' + ' | grep "enforcing=0" | grep "old_enforcing=1"', + sudo=True, + no_error_log=True, + shell=True, + ) + assert_that(avc0_result.exit_code).described_as( + "MAC_STATUS record enforcing=0 old_enforcing=1 must exist in audit log" + ).is_equal_to(0) + + # Associated SYSCALL record must show comm=setenforce + syscall_result = node.execute( + f"ausearch --input-logs -m MAC_STATUS -i -ts {start_time}" + ' | grep "type=SYSCALL" | grep "comm=setenforce"', + sudo=True, + no_error_log=True, + shell=True, + ) + assert_that(syscall_result.exit_code).described_as( + "MAC_STATUS event with comm=setenforce must exist in audit log" + ).is_equal_to(0) + + node.log.info("MAC_STATUS/SYSCALL audit records verified") + + # Edge case: selinuxfs unmounted + umount_result = node.execute( + f"umount --lazy {self._SELINUX_FS_MOUNT}", sudo=True, no_error_log=True + ) + if umount_result.exit_code == 0: + for val in ["1", "0", "Enforcing", "Permissive"]: + self._assert_cmd_fails_with_error( + node, + f"setenforce {val} 2>&1", + ["selinux is disabled"], + f"setenforce {val} with selinuxfs unmounted" + " must indicate disabled", + ) + node.execute( + f"mount -t selinuxfs none {self._SELINUX_FS_MOUNT}", + sudo=True, + ) + node.log.info("Edge case: selinuxfs unmounted passed") + else: + node.log.info( + "Edge case: selinuxfs unmounted skipped:" + " lazy umount not supported" + ) + + # Edge case: enforce file locked + node.execute("touch /var/tmp/selinux_enforce_bind_test", sudo=True) + node.execute("chattr +i /var/tmp/selinux_enforce_bind_test", sudo=True) + node.execute( + f"mount --bind /var/tmp/selinux_enforce_bind_test" + f" {self._SELINUX_FS_MOUNT}/enforce", + sudo=True, + ) + try: + for val in ["1", "0", "Enforcing", "Permissive"]: + self._assert_cmd_fails_with_error( + node, + f"setenforce {val} 2>&1", + ["failed"], + f"setenforce {val} with bound enforce file must report failure", + ) + finally: + node.execute( + f"umount {self._SELINUX_FS_MOUNT}/enforce 2>/dev/null || true", + shell=True, + sudo=True, + no_error_log=True, + ) + node.execute( + "chattr -i /var/tmp/selinux_enforce_bind_test 2>/dev/null || true", + shell=True, + sudo=True, + no_error_log=True, + ) + node.execute( + "rm -f /var/tmp/selinux_enforce_bind_test", + sudo=True, + no_error_log=True, + ) + node.log.info("Edge case: enforce file locked passed") + + node.log.info("setenforce validation passed") + + finally: + self._restore_audit_rules(node, backup_rules) + node.log.info("audit rules restored after setenforce test") + + @TestCaseMetadata( + description=""" + Verify SELinux info commands: sestatus, getenforce, and avcstat. + + Tests that getenforce accurately reports mode changes, sestatus + displays all required policy fields and contexts, and avcstat + shows live AVC cache statistics. + """, + priority=1, + requirement=simple_requirement(supported_os=[Fedora]), + use_new_environment=True, + ) + def verify_selinux_info(self, node: Node) -> None: + self._assert_rpm_installed(node, ["policycoreutils"]) + + # === getenforce: mode read === + node.log.info("Checking getenforce...") + getenforce = node.execute("getenforce") + assert_that(getenforce.exit_code).described_as( + "getenforce command must succeed" + ).is_equal_to(0) + + mode = getenforce.stdout.strip() + valid_modes = ["Enforcing", "Permissive", "Disabled"] + assert_that(mode).described_as( + f"getenforce must return valid mode, got: {mode}" + ).is_in(*valid_modes) + + original_mode = mode + + if mode != "Disabled": + enforce_file = node.execute(f"cat {self._SELINUX_FS_MOUNT}/enforce") + expected_value = "1" if mode == "Enforcing" else "0" + assert_that(enforce_file.stdout.strip()).described_as( + f"enforce file must match getenforce mode {mode}" + ).is_equal_to(expected_value) + + assert_that(mode).described_as( + "Fedora cloud image should have SELinux in Enforcing mode" + ).is_equal_to("Enforcing") + + # === getenforce: mode toggle === + try: + # setenforce 0 → getenforce must report Permissive + node.execute("setenforce 0", sudo=True) + perm_check = node.execute("getenforce") + assert_that(perm_check.stdout.strip()).described_as( + "getenforce must report 'Permissive' after setenforce 0" + ).is_equal_to("Permissive") + + # setenforce 1 → getenforce must report Enforcing + node.execute("setenforce 1", sudo=True) + enf_check = node.execute("getenforce") + assert_that(enf_check.stdout.strip()).described_as( + "getenforce must report 'Enforcing' after setenforce 1" + ).is_equal_to("Enforcing") + + finally: + restore_val = "1" if original_mode == "Enforcing" else "0" + node.execute(f"setenforce {restore_val}", sudo=True, no_error_log=True) + + # === sestatus: standard fields === + node.log.info("Checking sestatus...") + sestatus = node.execute("sestatus") + assert_that(sestatus.exit_code).described_as( + "sestatus command must succeed" + ).is_equal_to(0) + + output = sestatus.stdout + + sestatus_fields = [ + (r"SELinux status:\s+enabled", "SELinux status: enabled"), + (r"SELinuxfs mount:\s+/sys/fs/selinux", "SELinuxfs mount"), + ( + r"SELinux mount point:\s+/sys/fs/selinux" + r"|SELinux root directory:\s+\S+", + "SELinux root directory", + ), + (r"Current mode:\s+enforcing", "Current mode: enforcing"), + (r"Mode from config file:\s+\S+", "Mode from config file"), + (r"Loaded policy name:\s+\S+", "Loaded policy name"), + (r"Policy MLS status:\s+\S+", "Policy MLS status"), + (r"Policy deny_unknown status:\s+\S+", "Policy deny_unknown status"), + (r"Max kernel policy version:\s+\d+", "Max kernel policy version"), + ] + for pattern, desc in sestatus_fields: + assert_that(output).described_as(f"sestatus must show {desc}").matches( + pattern + ) + + # === sestatus: policy booleans === + sestatus_b = node.execute("sestatus -b") + assert_that(sestatus_b.exit_code).described_as( + "sestatus -b must succeed" + ).is_equal_to(0) + assert_that(sestatus_b.stdout).described_as( + "sestatus -b must show 'Policy booleans:' header" + ).contains("Policy booleans:") + # mount_anyfile should be on; deny_ptrace should be off on default Fedora + assert_that(sestatus_b.stdout).described_as( + "sestatus -b must contain mount_anyfile on" + ).matches(r"mount_anyfile\s+on") + assert_that(sestatus_b.stdout).described_as( + "sestatus -b must contain deny_ptrace off" + ).matches(r"deny_ptrace\s+off") + + # === sestatus: security contexts === + sestatus_v = node.execute("sestatus -v") + assert_that(sestatus_v.exit_code).described_as( + "sestatus -v must succeed" + ).is_equal_to(0) + _con = r"[a-z_]+_u:[a-z_]+_r:[a-z_]+_t:s\d+" + for pattern, desc in [ + (r"Process contexts:", "Process contexts section"), + (r"Current context:\s+" + _con, "Current context with valid SELinux label"), + (r"Init context:\s+system_u:system_r:init_t:s0", "Init context"), + (r"File contexts:", "File contexts section"), + (r"/etc/passwd\s+" + _con, "/etc/passwd context with valid SELinux label"), + ]: + assert_that(sestatus_v.stdout).described_as( + f"sestatus -v must show {desc}" + ).matches(pattern) + + # === avcstat: cache statistics === + node.log.info("Checking avcstat...") + avcstat1 = node.execute("avcstat") + assert_that(avcstat1.exit_code).described_as( + "avcstat command must succeed" + ).is_equal_to(0) + + avc_output = avcstat1.stdout + expected_columns = ["lookups", "hits", "misses", "allocs", "reclaims", "frees"] + for col in expected_columns: + assert_that(avc_output.lower()).described_as( + f"avcstat must show '{col}' column" + ).contains(col) + assert_that(avc_output).described_as( + "avcstat must show a row of numeric values" + ).matches(r"\s+\d+\s+\d+\s+\d+\s+\d+\s+\d+\s+\d+") + + # Trigger a filesystem access to ensure AVC lookups increment between runs + node.execute("ls /etc/passwd", no_error_log=True) + avcstat2 = node.execute("avcstat") + assert_that(avcstat2.stdout).described_as( + "Two consecutive avcstat outputs must differ (AVC cache is active)" + ).is_not_equal_to(avcstat1.stdout) + + node.log.info("selinux-info validation passed")