Tests

lisa/sut_orchestrator/azure/tools.py

@@ -2,9 +2,10 @@
 # Licensed under the MIT license.
 
 import re
+import os 
 import sys
 import json
-from pathlib import PurePath
+from pathlib import PurePath, Path
 from typing import Any, Dict, List, Optional, Tuple, Type
 
 from assertpy import assert_that
@@ -26,6 +27,7 @@
 from lisa.operating_system import Posix
 
 
+
 class Waagent(Tool):
     __version_pattern = re.compile(r"(?<=\-)([^\s]+)")
 
@@ -575,51 +577,25 @@ def command(self) -> str:
     @property
     def can_install(self) -> bool:
         self._log.info(f"Inside the func: [{sys._getframe().f_code.co_name}]")
-        return True
+        return False
 
     def _install(self) -> bool:
-        self._log.info(f"Inside the func: [{sys._getframe().f_code.co_name}]")
-        posix_os: Posix = cast(Posix, self.node.os)
-        cur_repo = posix_os.get_repositories();
-        self._log.info(f"{cur_repo}")
-
-        #posix_os.add_azure_core_repo(code_name="bionic")
-        arch_name = "arm64,armhf,amd64"
-        repo_url = "https://packages.microsoft.com/ubuntu/18.04/prod"
-        code_name = "insiders-fast"
-        posix_os.add_repository(
-            repo=(f"deb [arch={arch_name}] {repo_url} {code_name} main"),
-            keys_location=[
-                "https://packages.microsoft.com/keys/microsoft.asc",
-                "https://packages.microsoft.com/keys/msopentech.asc",
-            ],
-        )
-
-
-        cur_repo = posix_os.get_repositories();
-        self._log.info(f"{cur_repo}")
-
-        if posix_os.is_package_in_repo(self.command):
-            posix_os.install_packages(self.command)
-            self._log.info(f"{self.command} is installed successfully")
-        else:
-            raise UnsupportedDistroException(
-                    self.node.os, f"The distro doesn't have {self.command} in its repo")
         return self._check_exists()
 
     def get_result(
         self,
         arg: str,
         json_out: bool = False,
         sudo: bool = False,
-    ) -> None:
+    ) -> Any:
         self._log.info(f"Inside the func: [{sys._getframe().f_code.co_name}]")
         if json_out:
             arg += ' --output json'
         result = self.run(
             arg,
             sudo=sudo,
             shell=True,
+            force_run=True,
         )
 
         result.assert_exit_code()

microsoft/runbook/my.yml

@@ -42,14 +42,15 @@ platform:
     admin_password: $(admin_password)
     keep_environment: $(keep_environment)
     azure:
-      resource_group_name: $(resource_group_name)
+      resource_group_name: "lisa-test-zakhter"
       deploy: $(deploy)
       subscription_id: $(subscription_id)
       wait_delete: $(wait_delete)
     requirement:
       core_count:
         min: 2
       azure:
+        #marketplace: "Debian:debian-10-daily:10-gen2:0.20231218.1599" #$(marketplace_image)
         marketplace: $(marketplace_image)
         vhd: $(vhd)
         location: $(location)

microsoft/testsuites/vm_extensions/mde.py

@@ -1,8 +1,11 @@
+import os
 import time
-from typing import cast
+import requests
+
+from typing import Any
+from pathlib import Path, PurePath
 
 from assertpy import assert_that
-from retry import retry
 
 from lisa import (
     Logger,
@@ -12,20 +15,12 @@
     TestSuiteMetadata,
     simple_requirement,
 )
-from lisa.operating_system import BSD, Posix
-from lisa.sut_orchestrator.azure.common import (
-    add_tag_for_vm,
-    add_user_assign_identity,
-    get_managed_service_identity_client,
-    get_node_context,
-)
-from lisa.sut_orchestrator.azure.features import AzureExtension
-from lisa.sut_orchestrator.azure.platform_ import AzurePlatform
-from lisa.sut_orchestrator.azure.tools import Azsecd
+from lisa.operating_system import BSD
 from lisa.sut_orchestrator.azure.tools import mdatp
 from lisa.testsuite import TestResult
-from lisa.tools import Cat, Journalctl, Service
-from lisa.util import LisaException, UnsupportedDistroException
+from lisa.tools import  RemoteCopy, Whoami, Curl
+from lisa import CustomScriptBuilder, CustomScript
+from lisa.util import LisaException
 
 
 @TestSuiteMetadata(
@@ -36,37 +31,120 @@
     """,
 )
 class MDE(TestSuite):
+
+    def before_case(self, log: Logger, **kwargs: Any) -> None:
+        response = requests.get("https://raw.githubusercontent.com/microsoft/mdatp-xplat/master/linux/installation/mde_installer.sh")
+        if response.ok:
+            script = response.text
+            import tempfile
+            _, self.mde_installer = tempfile.mkstemp(prefix='mde_installer', suffix='.sh')
+            with open(self.mde_installer, 'w') as writer:
+                writer.write(script)
+            self._echo_script = CustomScriptBuilder(Path(os.path.dirname(self.mde_installer)),
+                                                [os.path.basename(self.mde_installer)])
+        else:
+            log.error('Unable to download mde_installer.sh script')
+
+    @TestCaseMetadata(
+        description="""
+            Verify MDE installation
+        """,
+        priority=1,
+        requirement=simple_requirement(min_core_count=2,
+                                       min_memory_mb=1024,
+                                       unsupported_os=[BSD])
+    )
+    def verify_install(self, node: Node, log: Logger, result: TestResult) -> None:
+        script: CustomScript = node.tools[self._echo_script]
+        log.info('Installing MDE')
+        result1 = script.run(parameters="--install", sudo=True)
+        log.info(result1)
+
+        try:
+            output = node.tools[mdatp]._check_exists()
+        except LisaException as e:
+            log.error(e)
+            output = False
+
+        assert_that(output).described_as('Unable to install MDE').is_equal_to(True)
+
     @TestCaseMetadata(
         description="""
             Verify if MDE is healthy
         """,
         priority=1,
-        requirement=simple_requirement(
-            #supported_features=[AzureExtension], unsupported_os=[BSD]
-        ),
+        requirement=simple_requirement(min_core_count=2,
+                                       min_memory_mb=1024,
+                                       unsupported_os=[BSD])
     )
-    def verify_health(self, node: Node, log: Logger, result: TestResult) -> None:
+    def verify_onboard(self, node: Node, log: Logger, result: TestResult) -> None:
+        username = node.tools[Whoami].get_username()
+
+        remote_copy = node.tools[RemoteCopy]
+        remote_copy.copy_to_remote(
+            PurePath("/home/zakhter/projects/lab/MicrosoftDefenderATPOnboardingLinuxServer.py"), PurePath(f"/home/{username}/MicrosoftDefenderATPOnboardingLinuxServer.py"))
+
+        script: CustomScript = node.tools[self._echo_script]
+
+        log.info('Onboarding MDE')
+        result1 = script.run(parameters=f"--onboard /home/{username}/MicrosoftDefenderATPOnboardingLinuxServer.py/MicrosoftDefenderATPOnboardingLinuxServer.py", sudo=True)
+        log.info(result1)
 
-        environment = result.environment
-        assert environment, "fail to get environment from testresult"
-        platform = environment.platform
-        log.info(platform)
-        #assert isinstance(platform, AzurePlatform)
-        #rm_client = platform._rm_client
-        #assert rm_client
-        #msi_client = get_managed_service_identity_client(platform)
-
-        node_context = get_node_context(node)
-        resource_group_name = node_context.resource_group_name
-        location = node_context.location
-        vm_name = node_context.vm_name
-
-        # Add resource tag for AzSecPack
-        tag = {"exemptPolicy": True}
-        add_tag_for_vm(platform, resource_group_name, vm_name, tag, log)
+        output = node.tools[mdatp].get_result('health --field licensed')
+
+        log.info(output)
+
+        assert_that(output).is_equal_to(['true'])
+
+    @TestCaseMetadata(
+        description="""
+            Verify if MDE is healthy
+        """,
+        priority=1,
+        requirement=simple_requirement(min_core_count=2,
+                                       min_memory_mb=1024,
+                                       unsupported_os=[BSD])
+    )
+    def verify_health(self, node: Node, log: Logger, result: TestResult) -> None:
         output = node.tools[mdatp].get_result('health', json_out=True)
 
         log.info(output)
 
         assert_that(output['healthy']).is_equal_to(True)
 
+    @TestCaseMetadata(
+        description="""
+            Verify if MDE is healthy
+        """,
+        priority=1,
+        requirement=simple_requirement(min_core_count=2,
+                                       min_memory_mb=1024,
+                                       unsupported_os=[BSD])
+    )
+    def eicar_test(self, node: Node, log: Logger, result: TestResult) -> None:
+        log.info('Running EICAR test')
+
+        output = node.tools[mdatp].get_result('health --field real_time_protection_enabled')
+        if output == ['false']:
+            output = node.tools[mdatp].get_result('config real-time-protection --value enabled', sudo=True)
+            assert_that(' '.join(output)).is_equal_to('Configuration property updated.')
+
+        current_threat_list= node.tools[mdatp].get_result('threat list')
+        log.info(current_threat_list)
+
+        node.tools[Curl].fetch(arg="-o /tmp/eicar.com.txt",
+                               execute_arg="",
+                               url="https://secure.eicar.org/eicar.com.txt")
+
+        time.sleep(5) #Wait for remediation
+
+        new_threat_list = node.tools[mdatp].get_result('threat list')
+        log.info(new_threat_list)
+
+        eicar_detect = ' '.join(new_threat_list).replace(' '.join(current_threat_list), '')
+
+        log.info(eicar_detect)
+        log.info(eicar_detect.find('Name: Virus:DOS/EICAR_Test_File'))
+        assert_that('Name: Virus:DOS/EICAR_Test_File' in eicar_detect).is_equal_to(True)
+
+