AppArmor tests and fix profile name in flag (#4109)

supervisor/utils/apparmor.py

@@ -1,5 +1,6 @@
 """Some functions around AppArmor profiles."""
 import logging
+from pathlib import Path
 import re
 
 from ..exceptions import AppArmorFileError, AppArmorInvalidError
@@ -9,7 +10,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
 RE_PROFILE = re.compile(r"^profile ([^ ]+).*$")
 
 
-def get_profile_name(profile_file):
+def get_profile_name(profile_file: Path) -> str:
     """Read the profile name from file."""
     profiles = set()
 
@@ -39,14 +40,14 @@ def get_profile_name(profile_file):
     return profiles.pop()
 
 
-def validate_profile(profile_name, profile_file):
+def validate_profile(profile_name: str, profile_file: Path) -> bool:
     """Check if profile from file is valid with profile name."""
     if profile_name == get_profile_name(profile_file):
         return True
     return False
 
 
-def adjust_profile(profile_name, profile_file, profile_new):
+def adjust_profile(profile_name: str, profile_file: Path, profile_new: Path) -> None:
     """Fix the profile name."""
     org_profile = get_profile_name(profile_file)
     profile_data = []
@@ -59,7 +60,7 @@ def adjust_profile(profile_name, profile_file, profile_new):
                 if not match:
                     profile_data.append(line)
                 else:
-                    profile_data.append(line.replace(org_profile, profile_name))
+                    profile_data.append(line.replace(org_profile, profile_name, 1))
     except OSError as err:
         raise AppArmorFileError(
             f"Can't adjust origin profile: {err}", _LOGGER.error

tests/common.py

@@ -66,31 +66,36 @@ def fire_property_change_signal(
     )
 
 
+def get_fixture_path(filename: str) -> Path:
+    """Get path for fixture."""
+    return Path(Path(__file__).parent.joinpath("fixtures"), filename)
+
+
 def load_json_fixture(filename: str) -> Any:
     """Load a json fixture."""
-    path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
+    path = get_fixture_path(filename)
     return json.loads(path.read_text(encoding="utf-8"))
 
 
 def load_yaml_fixture(filename: str) -> Any:
     """Load a YAML fixture."""
-    path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
+    path = get_fixture_path(filename)
     return read_yaml_file(path)
 
 
 def load_fixture(filename: str) -> str:
     """Load a fixture."""
-    path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
+    path = get_fixture_path(filename)
     return path.read_text(encoding="utf-8")
 
 
 def load_binary_fixture(filename: str) -> bytes:
     """Load a fixture without decoding."""
-    path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
+    path = get_fixture_path(filename)
     return path.read_bytes()
 
 
 def exists_fixture(filename: str) -> bool:
     """Check if a fixture exists."""
-    path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
+    path = get_fixture_path(filename)
     return path.exists()

tests/fixtures/apparmor_multiple_profiles.txt

@@ -0,0 +1,10 @@
+
+#include <tunables/global>
+
+profile example flags=(attach_disconnected,mediate_deleted) {
+  #include <abstractions/base>
+}
+
+profile example_2 flags=(attach_disconnected,mediate_deleted) {
+  #include <abstractions/base>
+}

tests/fixtures/apparmor_no_profile.txt

@@ -0,0 +1 @@
+#include <tunables/global>

tests/fixtures/apparmor_valid.txt

@@ -0,0 +1,16 @@
+#include <tunables/global>
+
+profile example flags=(attach_disconnected,mediate_deleted) {
+  #include <abstractions/base>
+
+  # Capabilities
+  file,
+  signal (send) set=(kill,term,int,hup,cont),
+
+  # Start new profile for service
+  /usr/bin/my_program cx -> my_program,
+
+  profile my_program flags=(attach_disconnected,mediate_deleted) {
+    #include <abstractions/base>
+  }
+}

tests/fixtures/apparmor_valid_mediate.txt

@@ -0,0 +1,16 @@
+#include <tunables/global>
+
+profile mediate flags=(attach_disconnected,mediate_deleted) {
+  #include <abstractions/base>
+
+  # Capabilities
+  file,
+  signal (send) set=(kill,term,int,hup,cont),
+
+  # Start new profile for service
+  /usr/bin/my_program cx -> my_program,
+
+  profile my_program flags=(attach_disconnected,mediate_deleted) {
+    #include <abstractions/base>
+  }
+}

tests/utils/test_apparmor.py

@@ -0,0 +1,71 @@
+"""Tests for apparmor utility."""
+
+from pathlib import Path
+
+import pytest
+
+from supervisor.exceptions import AppArmorInvalidError
+from supervisor.utils.apparmor import adjust_profile, validate_profile
+
+from tests.common import get_fixture_path
+
+TEST_PROFILE = """
+#include <tunables/global>
+
+profile test flags=(attach_disconnected,mediate_deleted) {
+  #include <abstractions/base>
+
+  # Capabilities
+  file,
+  signal (send) set=(kill,term,int,hup,cont),
+
+  # Start new profile for service
+  /usr/bin/my_program cx -> my_program,
+
+  profile my_program flags=(attach_disconnected,mediate_deleted) {
+    #include <abstractions/base>
+  }
+}
+""".strip()
+
+
+async def test_valid_apparmor_file():
+    """Test a valid apparmor file."""
+    assert validate_profile("example", get_fixture_path("apparmor_valid.txt"))
+
+
+async def test_apparmor_missing_profile(caplog: pytest.LogCaptureFixture):
+    """Test apparmor file missing profile."""
+    with pytest.raises(AppArmorInvalidError):
+        validate_profile("example", get_fixture_path("apparmor_no_profile.txt"))
+
+    assert (
+        "Missing AppArmor profile inside file: apparmor_no_profile.txt" in caplog.text
+    )
+
+
+async def test_apparmor_multiple_profiles(caplog: pytest.LogCaptureFixture):
+    """Test apparmor file with too many profiles."""
+    with pytest.raises(AppArmorInvalidError):
+        validate_profile("example", get_fixture_path("apparmor_multiple_profiles.txt"))
+
+    assert (
+        "Too many AppArmor profiles inside file: apparmor_multiple_profiles.txt"
+        in caplog.text
+    )
+
+
+async def test_apparmor_profile_adjust(tmp_path: Path):
+    """Test apparmor profile adjust."""
+    profile_out = tmp_path / "apparmor_out.txt"
+    adjust_profile("test", get_fixture_path("apparmor_valid.txt"), profile_out)
+
+    assert profile_out.read_text(encoding="utf-8") == TEST_PROFILE
+
+
+async def test_apparmor_profile_adjust_mediate(tmp_path: Path):
+    """Test apparmor profile adjust when name matches a flag."""
+    profile_out = tmp_path / "apparmor_out.txt"
+    adjust_profile("test", get_fixture_path("apparmor_valid_mediate.txt"), profile_out)
+
+    assert profile_out.read_text(encoding="utf-8") == TEST_PROFILE