Make checks smarter about environments in which they should not run

tests/test/check/test-dmesg.sh

@@ -38,36 +38,37 @@ rlJournalStart
 
  rlRun "cat $results"
 
- rlAssertExists "$dump_before"
  if [ "$method" = "container" ]; then
- assert_check_result "dmesg as a before-test should fail with containers" "error" "before-test" "harmless"
+ assert_check_result "dmesg as a before-test should skip with containers" "skip" "before-test" "harmless"
+
+ rlAssertNotExists "$dump_before"
 
- if rlIsFedora ">=38"; then
- rlAssertGrep "dmesg: read kernel buffer failed: Operation not permitted" "$dump_before"
- else
- rlAssertGrep "dmesg: read kernel buffer failed: Permission denied" "$dump_before"
- fi
  else
  assert_check_result "dmesg as a before-test should pass" "pass" "before-test" "harmless"
+
+ rlAssertExists "$dump_before"
+ rlLogInfo "$(cat $dump_before)"
  fi
- rlLogInfo "$(cat $dump_before)"
 
- rlAssertExists "$dump_after"
  if [ "$method" = "container" ]; then
- assert_check_result "dmesg as an after-test should fail with containers" "error" "after-test" "harmless"
+ assert_check_result "dmesg as an after-test should skip with containers" "skip" "after-test" "harmless"
+
+ rlAssertNotExists "$dump_after"
 
- if rlIsFedora ">=38"; then
- rlAssertGrep "dmesg: read kernel buffer failed: Operation not permitted" "$dump_after"
- else
- rlAssertGrep "dmesg: read kernel buffer failed: Permission denied" "$dump_after"
- fi
  elif [ "$method" = "virtual" ]; then
  assert_check_result "dmesg as an after-test should fail" "fail" "after-test" "segfault"
+
+ rlAssertExists "$dump_after"
+ rlLogInfo "$(cat $dump_after)"
+
  rlAssertGrep "Some segfault happened" "$segfault/checks/dmesg-after-test.txt"
+
  else
  assert_check_result "dmesg as an after-test should pass" "pass" "after-test" "harmless"
+
+ rlAssertExists "$dump_after"
+ rlLogInfo "$(cat $dump_after)"
  fi
- rlLogInfo "$(cat $dump_after)"
  rlPhaseEnd
  done
 

tmt/checks/avc.py

@@ -293,7 +293,8 @@ def before_test(
  invocation: 'TestInvocation',
  environment: Optional[tmt.utils.EnvironmentType] = None,
  logger: tmt.log.Logger) -> list[CheckResult]:
- create_ausearch_timestamp(invocation, logger)
+ if invocation.guest.facts.has_selinux:
+ create_ausearch_timestamp(invocation, logger)
 
  return []
 
@@ -305,6 +306,11 @@ def after_test(
  invocation: 'TestInvocation',
  environment: Optional[tmt.utils.EnvironmentType] = None,
  logger: tmt.log.Logger) -> list[CheckResult]:
+ if not invocation.guest.facts.has_selinux:
+ return [CheckResult(
+ name='avc',
+ result=ResultOutcome.SKIP)]
+
  assert invocation.phase.step.workdir is not None # narrow type
 
  outcome, path = create_final_report(invocation, logger)

tmt/checks/dmesg.py

@@ -8,6 +8,7 @@
 import tmt.utils
 from tmt.checks import Check, CheckEvent, CheckPlugin, provides_check
 from tmt.result import CheckResult, ResultOutcome
+from tmt.steps.provision import GuestCapability
 from tmt.utils import Path, render_run_exception_streams
 
 if TYPE_CHECKING:
@@ -62,7 +63,16 @@ def _test_output_logger(
  level=level,
  topic=topic)
 
- return guest.execute(tmt.utils.ShellScript('dmesg -c'), log=_test_output_logger)
+ if guest.facts.has_capability(GuestCapability.SYSLOG_ACTION_READ_CLEAR):
+ script = tmt.utils.ShellScript('dmesg -c')
+
+ else:
+ script = tmt.utils.ShellScript('dmesg')
+
+ if not guest.facts.is_superuser:
+ script = tmt.utils.ShellScript(f'sudo {script.to_shell_command()}')
+
+ return guest.execute(script, log=_test_output_logger)
 
  @classmethod
  def _save_dmesg(
@@ -106,6 +116,9 @@ def before_test(
  invocation: 'TestInvocation',
  environment: Optional[tmt.utils.EnvironmentType] = None,
  logger: tmt.log.Logger) -> list[CheckResult]:
+ if not invocation.guest.facts.has_capability(GuestCapability.SYSLOG_ACTION_READ_ALL):
+ return [CheckResult(name='dmesg', result=ResultOutcome.SKIP)]
+
  outcome, path = cls._save_dmesg(invocation, CheckEvent.BEFORE_TEST, logger)
 
  return [CheckResult(name='dmesg', result=outcome, log=[path])]
@@ -118,6 +131,9 @@ def after_test(
  invocation: 'TestInvocation',
  environment: Optional[tmt.utils.EnvironmentType] = None,
  logger: tmt.log.Logger) -> list[CheckResult]:
+ if not invocation.guest.facts.has_capability(GuestCapability.SYSLOG_ACTION_READ_ALL):
+ return [CheckResult(name='dmesg', result=ResultOutcome.SKIP)]
+
  outcome, path = cls._save_dmesg(invocation, CheckEvent.AFTER_TEST, logger)
 
  return [CheckResult(name='dmesg', result=outcome, log=[path])]

tmt/steps/provision/__init__.py

@@ -109,6 +109,16 @@ class GuestPackageManager(enum.Enum):
 T = TypeVar('T')
 
 
+class GuestCapability(enum.Enum):
+ """ Various Linux capabilities """
+
+ # See man 2 syslog:
+ #: Read all messages remaining in the ring buffer.
+ SYSLOG_ACTION_READ_ALL = 'syslog-action-read-all'
+ #: Read and clear all messages remaining in the ring buffer.
+ SYSLOG_ACTION_READ_CLEAR = 'syslog-action-read-clear'
+
+
 @dataclasses.dataclass
 class GuestFacts(SerializableContainer):
  """
@@ -135,9 +145,29 @@ class GuestFacts(SerializableContainer):
  has_selinux: Optional[bool] = None
  is_superuser: Optional[bool] = None
 
+ #: Various Linux capabilities and whether they are permitted to
+ #: commands executed on this guest.
+ capabilities: dict[GuestCapability, bool] = field(
+ default_factory=cast(Callable[[], dict[GuestCapability, bool]], dict),
+ serialize=lambda capabilities: {
+ capability.value: enabled
+ for capability, enabled in capabilities.items()
+ } if capabilities else {},
+ unserialize=lambda raw_value: {
+ GuestCapability(raw_capability): enabled
+ for raw_capability, enabled in raw_value.items()
+ }
+ )
+
  os_release_content: dict[str, str] = field(default_factory=dict)
  lsb_release_content: dict[str, str] = field(default_factory=dict)
 
+ def has_capability(self, cap: GuestCapability) -> bool:
+ if not self.capabilities:
+ return False
+
+ return self.capabilities.get(cap, False)
+
  # TODO nothing but a fancy helper, to check for some special errors that
  # may appear this soon in provisioning. But, would it make sense to put
  # this detection into the `GuestSsh.execute()` method?
@@ -339,6 +369,14 @@ def _query_is_superuser(self, guest: 'Guest') -> Optional[bool]:
 
  return output.stdout.strip() == 'root'
 
+ def _query_capabilities(self, guest: 'Guest') -> dict[GuestCapability, bool]:
+ # TODO: there must be a canonical way of getting permitted capabilities.
+ # For now, we're interested in whether we can access kernel message buffer.
+ return {
+ GuestCapability.SYSLOG_ACTION_READ_ALL: True,
+ GuestCapability.SYSLOG_ACTION_READ_CLEAR: True
+ }
+
  def sync(self, guest: 'Guest') -> None:
  """ Update stored facts to reflect the given guest """
 
@@ -351,6 +389,7 @@ def sync(self, guest: 'Guest') -> None:
  self.package_manager = self._query_package_manager(guest)
  self.has_selinux = self._query_has_selinux(guest)
  self.is_superuser = self._query_is_superuser(guest)
+ self.capabilities = self._query_capabilities(guest)
 
  self.in_sync = True
 

tmt/steps/provision/podman.py

@@ -9,6 +9,7 @@
 import tmt.steps
 import tmt.steps.provision
 import tmt.utils
+from tmt.steps.provision import GuestCapability
 from tmt.utils import Command, OnProcessStartCallback, Path, ShellScript, field, retry
 
 # Timeout in seconds of waiting for a connection
@@ -456,6 +457,11 @@ def go(self) -> None:
  parent=self.step)
  self._guest.start()
 
+ # TODO: this might be allowed with `--privileged`...
+ self._guest.facts.capabilities[GuestCapability.SYSLOG_ACTION_READ_ALL] = False
+ # ... while this seems to be forbidden completely.
+ self._guest.facts.capabilities[GuestCapability.SYSLOG_ACTION_READ_CLEAR] = False
+
  def guest(self) -> Optional[GuestContainer]:
  """ Return the provisioned guest """
  return self._guest