Merge pull request #34 from swimlane/0_6_1_release

0.7.0 Release

CHANGELOG.md

@@ -1,5 +1,17 @@
 # CHANGELOG
 
+## 0.7.0 - 2022-01-04
+
+    * Updated argument handling in get_atomics Retrieving Atomic Tests with specified destination in /opt throws unexpected keyword argument error #28
+    * Updated error catching and logging within state machine class when copying source files to remote system Logging and troubleshooting question #32
+    * Updated ConfigParser from instance variables to local method bound variables Using a second AtomicOperator instance executes the tests of the first instance too #33
+    * Added the ability to select specific tests for one or more provided techniques
+    * Updated documentation
+    * Added new Copier class to handle file transfer for remote connections
+    * Removed gathering of supporting_files and passing around with object
+    * Added new config_file_only parameter to only run the defined configuration within a configuration file
+    * Updated documentation around installation on macOS systems with M1 processors
+
 ## 0.6.0 - 2021-12-17
 
     * Updated documentation

README.md

@@ -30,6 +30,7 @@ Additionally, `atomic-operator` can be used in many other situations like:
 * Assist with downloading the atomic-red-team repository
 * Can be automated further based on a configuration file
 * A command-line and importable Python package
+* Select specific tests when one or more techniques are specified
 * Plus more
 
 ## Getting Started
@@ -62,6 +63,7 @@ pyyaml==5.4.1
 fire==0.4.0
 requests==2.26.0
 attrs==21.2.0
+pick==1.2.0
 ```
 
 ### macOS, Linux and Windows:
@@ -70,6 +72,22 @@ attrs==21.2.0
 pip install atomic-operator
 ```
 
+### macOS using M1 processor
+
+```bash
+git clone https://github.com/swimlane/atomic-operator.git
+cd atomic-operator
+
+# Satisfy ModuleNotFoundError: No module named 'setuptools_rust'
+brew install rust
+pip3 install --upgrade pip
+pip3 install setuptools_rust
+
+# Back to our regularly scheduled programming . . .  
+
+python setup.py install
+```
+
 ### Installing from source
 
 ```bash
@@ -108,6 +126,26 @@ In order to run a test you must provide some additional properties (and options
 atomic-operator run --atomics-path "/tmp/some_directory/redcanaryco-atomic-red-team-3700624"
 ```
 
+You can select individual tests when you provide one or more specific techniques. For example running the following on the command line:
+
+```bash
+atomic-operator run --techniques T1564.001 --select_tests
+```
+
+Will prompt the user with a selection list of tests associated with that technique. A user can select one or more tests by using the space bar to highlight the desired test:
+
+```text
+ Select Test(s) for Technique T1564.001 (Hide Artifacts: Hidden Files and Directories)
+
+ * Create a hidden file in a hidden directory (61a782e5-9a19-40b5-8ba4-69a4b9f3d7be)
+   Mac Hidden file (cddb9098-3b47-4e01-9d3b-6f5f323288a9)
+   Create Windows System File with Attrib (f70974c8-c094-4574-b542-2c545af95a32)
+   Create Windows Hidden File with Attrib (dadb792e-4358-4d8d-9207-b771faa0daa5)
+   Hidden files (3b7015f2-3144-4205-b799-b05580621379)
+   Hide a Directory (b115ecaf-3b24-4ed2-aefe-2fcb9db913d3)
+   Show all hidden files (9a1ec7da-b892-449f-ad68-67066d04380c)
+```
+
 ### Running Tests Remotely
 
 In order to run a test remotely you must provide some additional properties (and options if desired). The main method to run tests is named `run`.
@@ -133,6 +171,7 @@ atomic-operator run -- --help
 |--------------|----|-------|-----------|
 |techniques|list|all|One or more defined techniques by attack_technique ID.|
 |test_guids|list|None|One or more Atomic test GUIDs.|
+|select_tests|bool|False|Select one or more atomic tests to run when a techniques are specified.|
 |atomics_path|str|os.getcwd()|The path of Atomic tests.|
 |check_prereqs|bool|False|Whether or not to check for prereq dependencies (prereq_comand).|
 |get_prereqs|bool|False|Whether or not you want to retrieve prerequisites.|
@@ -143,6 +182,7 @@ atomic-operator run -- --help
 |prompt_for_input_args|bool|False|Whether you want to prompt for input arguments for each test.|
 |return_atomics|bool|False|Whether or not you want to return atomics instead of running them.|
 |config_file|str|None|A path to a conifg_file which is used to automate atomic-operator in environments.|
+|config_file_only|bool|False|Whether or not you want to run tests based on the provided config_file only.|
 |hosts|list|None|A list of one or more remote hosts to run a test on.|
 |username|str|None|Username for authentication of remote connections.|
 |password|str|None|Password for authentication of remote connections.|
@@ -177,6 +217,10 @@ FLAGS
         Type: list
         Default: []
         One or more Atomic test GUIDs. Defaults to None.
+    --select_tests=SELECT_TESTS
+        Type: bool
+        Default: False
+        Select one or more tests from provided techniques. Defaults to False.
     --atomics_path=ATOMICS_PATH
         Default: '/U...
         The path of Atomic tests. Defaults to os.getcwd().
@@ -208,6 +252,9 @@ FLAGS
         Type: Optional[]
         Default: None
         A path to a conifg_file which is used to automate atomic-operator in environments. Default to None.
+    --config_file_only=CONFIG_FILE_ONLY
+        Default: False
+        Whether or not you want to run tests based on the provided config_file only. Defaults to False.
     --hosts=HOSTS
         Default: []
         A list of one or more remote hosts to run a test on. Defaults to [].
@@ -311,3 +358,7 @@ See also the list of [contributors](https://github.com/swimlane/atomic-operator/
 ## License
 
 This project is licensed under the MIT License - see the [LICENSE](LICENSE.md) file for details
+
+## Shoutout
+
+- Thanks to [keithmccammon](https://github.com/keithmccammon) for helping identify issues with macOS M1 based proccesssor and providing a fix

atomic_operator/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.6.0"
+__version__ = "0.7.0"
 
 
 from .atomic_operator import AtomicOperator

atomic_operator/atomic/atomic.py

@@ -16,26 +16,10 @@ class Atomic:
     path                                  = attr.ib()
     atomic_tests: typing.List[AtomicTest] = attr.ib()
     hosts: typing.List[Host]              = attr.ib(default=None)
-    supporting_files: typing.List         = attr.ib(default=[])
 
     def __attrs_post_init__(self):
         if self.atomic_tests:
             test_list = []
             for test in self.atomic_tests:
                 test_list.append(AtomicTest(**test))
             self.atomic_tests = test_list
-        self.supporting_files = self.__gather_supporting_files()
-
-    def __gather_supporting_files(self):
-        return_list = []
-        for dirpath, dirnames, files in os.walk(self.path):
-            if files:
-                for file in files:
-                    if file.endswith('.yaml') or file.endswith('.md'):
-                        continue
-                    if '/' in dirpath:
-                        full_path = f"{dirpath}/{file}"
-                    else:
-                        full_path = f"{dirpath}\{file}"
-                    return_list.append(full_path)
-        return return_list

atomic_operator/atomic_operator.py

@@ -96,15 +96,15 @@ def __run_technique(self, technique, **kwargs):
                     self.__logger.debug(f"Description: {test.description}")
                     if test.executor.name in ['sh', 'bash']:
                         self.__test_responses[test.auto_generated_guid].update(
-                            RemoteRunner(test, technique.path, supporting_files=technique.supporting_files).start(host=host, executor='ssh')
+                            RemoteRunner(test, technique.path).start(host=host, executor='ssh')
                         )
                     elif test.executor.name in ['command_prompt']:
                         self.__test_responses[test.auto_generated_guid].update(
-                            RemoteRunner(test, technique.path, supporting_files=technique.supporting_files).start(host=host, executor='cmd')
+                            RemoteRunner(test, technique.path).start(host=host, executor='cmd')
                         )
                     elif test.executor.name in ['powershell']:
                         self.__test_responses[test.auto_generated_guid].update(
-                            RemoteRunner(test, technique.path, supporting_files=technique.supporting_files).start(host=host, executor='powershell')
+                            RemoteRunner(test, technique.path).start(host=host, executor='powershell')
                         )
                     else:
                         self.__logger.warning(f"Unable to execute test since the executor is {test.executor.name}. Skipping.....")
@@ -144,20 +144,26 @@ def get_atomics(self, desintation=os.getcwd(), **kwargs):
         """
         if not os.path.exists(desintation):
             os.makedirs(desintation)
-        folder_name = self.download_atomic_red_team_repo(desintation, **kwargs)
+        desintation = kwargs.pop('destination') if kwargs.get('destination') else desintation
+        folder_name = self.download_atomic_red_team_repo(
+            save_path=desintation, 
+            **kwargs
+        )
         return os.path.join(desintation, folder_name)
 
-    def run(self, techniques: list=['all'], test_guids: list=[], atomics_path=os.getcwd(), 
-                  check_prereqs=False, get_prereqs=False, cleanup=False, copy_source_files=True,
-                  command_timeout=20, debug=False, prompt_for_input_args=False,
-                  return_atomics=False, config_file=None, hosts=[], username=None,
-                  password=None, ssh_key_path=None, private_key_string=None,
-                  verify_ssl=False, ssh_port=22, ssh_timeout=5, *args, **kwargs) -> None:
+    def run(self, techniques: list=['all'], test_guids: list=[], select_tests=False,
+                  atomics_path=os.getcwd(), check_prereqs=False, get_prereqs=False, 
+                  cleanup=False, copy_source_files=True,command_timeout=20, debug=False, 
+                  prompt_for_input_args=False, return_atomics=False, config_file=None, 
+                  config_file_only=False, hosts=[], username=None, password=None, 
+                  ssh_key_path=None, private_key_string=None, verify_ssl=False, 
+                  ssh_port=22, ssh_timeout=5, *args, **kwargs) -> None:
         """The main method in which we run Atomic Red Team tests.
 
         Args:
             techniques (list, optional): One or more defined techniques by attack_technique ID. Defaults to 'all'.
             test_guids (list, optional): One or more Atomic test GUIDs. Defaults to None.
+            select_tests (bool, optional): Select one or more tests from provided techniques. Defaults to False.
             atomics_path (str, optional): The path of Atomic tests. Defaults to os.getcwd().
             check_prereqs (bool, optional): Whether or not to check for prereq dependencies (prereq_comand). Defaults to False.
             get_prereqs (bool, optional): Whether or not you want to retrieve prerequisites. Defaults to False.
@@ -168,6 +174,7 @@ def run(self, techniques: list=['all'], test_guids: list=[], atomics_path=os.get
             prompt_for_input_args (bool, optional): Whether you want to prompt for input arguments for each test. Defaults to False.
             return_atomics (bool, optional): Whether or not you want to return atomics instead of running them. Defaults to False.
             config_file (str, optional): A path to a conifg_file which is used to automate atomic-operator in environments. Default to None.
+            config_file_only (bool, optional): Whether or not you want to run tests based on the provided config_file only. Defaults to False.
             hosts (list, optional): A list of one or more remote hosts to run a test on. Defaults to [].
             username (str, optional): Username for authentication of remote connections. Defaults to None.
             password (str, optional): Password for authentication of remote connections. Defaults to None.
@@ -204,16 +211,17 @@ def run(self, techniques: list=['all'], test_guids: list=[], atomics_path=os.get
         # line to build a run_list of objects
         self.__config_parser = ConfigParser(
                 config_file=config_file,
-                techniques=self.parse_input_lists(techniques),
-                test_guids=self.parse_input_lists(test_guids),
-                host_list=self.parse_input_lists(hosts),
+                techniques=None if config_file_only else self.parse_input_lists(techniques),
+                test_guids=None if config_file_only else self.parse_input_lists(test_guids),
+                host_list=None if config_file_only else self.parse_input_lists(hosts),
                 username=username,
                 password=password,
                 ssh_key_path=ssh_key_path,
                 private_key_string=private_key_string,
                 verify_ssl=verify_ssl,
                 ssh_port=ssh_port,
-                ssh_timeout=ssh_timeout
+                ssh_timeout=ssh_timeout,
+                select_tests=select_tests
             )
         self.__run_list = self.__config_parser.run_list
 

atomic_operator/base.py

@@ -4,6 +4,7 @@
 from io import BytesIO
 import platform
 import requests
+from pick import pick
 from .utils.logger import LoggingBase
 
 
@@ -98,7 +99,7 @@ def parse_input_lists(self, value):
             value_list = set(value)
         return list(value_list)
 
-    def __path_replacement(self, string, path):
+    def _path_replacement(self, string, path):
         try:
             string = string.replace('$PathToAtomicsFolder', path)
         except:
@@ -111,7 +112,7 @@ def __path_replacement(self, string, path):
 
     def _replace_command_string(self, command: str, path:str, input_arguments: list=[]):
         if command:
-            command = self.__path_replacement(command, path)
+            command = self._path_replacement(command, path)
             if input_arguments:
                 for input in input_arguments:
                     for string in self._replacement_strings:
@@ -120,7 +121,7 @@ def _replace_command_string(self, command: str, path:str, input_arguments: list=
                         except:
                             # catching errors since some inputs are actually integers but defined as strings
                             pass
-        return self.__path_replacement(command, path)
+        return self._path_replacement(command, path)
 
     def _check_if_aws(self, test):
         if 'iaas:aws' in test.supported_platforms and self.get_local_system_platform() in ['macos', 'linux']:
@@ -148,3 +149,20 @@ def _set_input_arguments(self, test, **kwargs):
             for input in test.input_arguments:
                 if input.value == None:
                     input.value = input.default
+
+    def select_atomic_tests(self, technique):
+        options = None
+        test_list = []
+        for test in technique.atomic_tests:
+            test_list.append(test)
+        if test_list:
+            options = pick(
+                test_list, 
+                title=f"Select Test(s) for Technique {technique.attack_technique} ({technique.display_name})", 
+                multiselect=True, 
+                options_map_func=self.format_pick_options
+            )
+        return [i[0] for i in options] if options else []
+
+    def format_pick_options(self, option):
+        return f"{option.name} ({option.auto_generated_guid})"