Onboarding MDE for Linux to LISA

lisa/sut_orchestrator/azure/tools.py

@@ -2,7 +2,10 @@
 # Licensed under the MIT license.
 
 import re
-from pathlib import PurePath
+import os 
+import sys
+import json
+from pathlib import PurePath, Path
 from typing import Any, Dict, List, Optional, Tuple, Type
 
 from assertpy import assert_that
@@ -20,6 +23,9 @@
     get_matched_str,
 )
 from lisa.util.process import ExecutableResult
+from typing import Optional, cast
+from lisa.operating_system import Posix
+
 
 
 class Waagent(Tool):
@@ -561,3 +567,39 @@ def get_pool_records(self, pool_id: int, force_run: bool = False) -> Dict[str, s
             records[content_split[i]] = content_split[i + 1]
 
         return records
+
+class mdatp(Tool):
+    @property
+    def command(self) -> str:
+        self._log.info(f"Inside the func: [{sys._getframe().f_code.co_name}]")
+        return "mdatp"
+
+    @property
+    def can_install(self) -> bool:
+        self._log.info(f"Inside the func: [{sys._getframe().f_code.co_name}]")
+        return False
+
+    def _install(self) -> bool:
+        return self._check_exists()
+
+    def get_result(
+        self,
+        arg: str,
+        json_out: bool = False,
+        sudo: bool = False,
+    ) -> 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()
+        if json_out:
+            return json.loads(result.stdout)
+        return result.stdout.split()
+

microsoft/runbook/my.yml

@@ -0,0 +1,58 @@
+name: azure default
+include:
+  - path: ./debug.yml
+variable:
+  - name: origin
+    value: tiers/tier.yml
+  - name: case
+    value: verify_cpu_count
+  - name: location
+    value: "westus3"
+  - name: keep_environment
+    value: "no"
+  - name: resource_group_name
+    value: ""
+  - name: marketplace_image
+    value: ""
+  - name: vhd
+    value: ""
+  - name: vm_size
+    value: ""
+  - name: deploy
+    value: true
+  - name: wait_delete
+    value: false
+  - name: concurrency
+    value: 5
+  - name: admin_private_key_file
+    value: ""
+    is_secret: true
+  - name: admin_password
+    value: ""
+    is_secret: true
+  - name: case
+    value: verify_cpu_count
+concurrency: $(concurrency)
+notifier:
+  - type: html
+  - type: env_stats
+platform:
+  - type: azure
+    admin_private_key_file: $(admin_private_key_file)
+    admin_password: $(admin_password)
+    keep_environment: $(keep_environment)
+    azure:
+      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)
+        vm_size: $(vm_size)
+

microsoft/testsuites/vm_extensions/mde.py

@@ -0,0 +1,150 @@
+import os
+import time
+import requests
+
+from typing import Any
+from pathlib import Path, PurePath
+
+from assertpy import assert_that
+
+from lisa import (
+    Logger,
+    Node,
+    TestCaseMetadata,
+    TestSuite,
+    TestSuiteMetadata,
+    simple_requirement,
+)
+from lisa.operating_system import BSD
+from lisa.sut_orchestrator.azure.tools import mdatp
+from lisa.testsuite import TestResult
+from lisa.tools import  RemoteCopy, Whoami, Curl
+from lisa import CustomScriptBuilder, CustomScript
+from lisa.util import LisaException
+
+
+@TestSuiteMetadata(
+    area="vm_extension",
+    category="functional",
+    description="""
+    MDE Test Suite
+    """,
+)
+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(min_core_count=2,
+                                       min_memory_mb=1024,
+                                       unsupported_os=[BSD])
+    )
+    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)
+
+        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)
+
+