Merge pull request #48 from swimlane/0_8_4_release

0.8.4 Release

CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## 0.8.4 - 2022-03-25
+
+    * Updated formatting of executor for AWS and local runners
+    * Updated documentation
+    * Added formatting constants to base class to improve updating of windows variables on command line runners
+
 ## 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

README.md

@@ -84,7 +84,7 @@ pip3 install --upgrade pip
 pip3 install setuptools_rust
 
 # Back to our regularly scheduled programming . . .  
-
+pip install -r requirements.txt
 python setup.py install
 ```
 
@@ -93,6 +93,7 @@ python setup.py install
 ```bash
 git clone https://github.com/swimlane/atomic-operator.git
 cd atomic-operator
+pip install -r requirements.txt
 python setup.py install
 ```
 

atomic_operator/__init__.py

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

atomic_operator/__meta__.py

@@ -0,0 +1,10 @@
+__title__ = 'atomic-operator'
+__description__ = 'A python package to execute Atomic tests'
+__url__ = 'https://github.com/swimlane/atomic-operator'
+__version__ = '0.8.4'
+__author__ = 'Swimlane'
+__author_email__ = 'info@swimlane.com'
+__maintainer__ = 'MSAdministrator'
+__maintainer_email__ = 'rickardja@live.com'
+__license__ = 'MIT'
+__copyright__ = 'Copyright 2022 Swimlane'

atomic_operator/atomic/loader.py

@@ -45,9 +45,14 @@ def load_technique(self, path_to_dir) -> dict:
             with open(self.get_abs_path(path_to_dir), 'r', encoding="utf-8") as f:
                 return yaml.safe_load(f.read())
         except:
+            self.__logger.warning(f"Unable to load technique in '{path_to_dir}'")
+
+        try:
             # windows does not like get_abs_path so casting to string
             with open(str(path_to_dir), 'r', encoding="utf-8") as f:
                 return yaml.safe_load(f.read())
+        except OSError as oe:
+            self.__logger.warning(f"Unable to load technique in '{path_to_dir}': {oe}")
 
     def load_techniques(self) -> dict:
         """The main entrypoint when loading techniques from disk.
@@ -74,6 +79,7 @@ def load_techniques(self) -> dict:
             technique = self.__get_file_name(atomic_entry)
             if not self.__techniques.get(technique):
                 loaded_technique = self.load_technique(str(atomic_entry))
-                loaded_technique.update({'path': os.path.dirname(str(atomic_entry))})
-                self.__techniques[technique] = Atomic(**loaded_technique)
+                if loaded_technique:
+                    loaded_technique.update({'path': os.path.dirname(str(atomic_entry))})
+                    self.__techniques[technique] = Atomic(**loaded_technique)
         return self.__techniques

atomic_operator/base.py

@@ -31,6 +31,11 @@ class Base(metaclass=LoggingBase):
             'macos': '/bin/bash'
         }
     }
+    VARIABLE_REPLACEMENTS = {
+        'command_prompt': {
+            '%temp%': "$env:TEMP"
+        }
+    }
     _replacement_strings = [
         '#{{{0}}}',
         '${{{0}}}'
@@ -110,7 +115,7 @@ def _path_replacement(self, string, path):
             pass
         return string
 
-    def _replace_command_string(self, command: str, path:str, input_arguments: list=[]):
+    def _replace_command_string(self, command: str, path:str, input_arguments: list=[], executor=None):
         if command:
             command = self._path_replacement(command, path)
             if input_arguments:
@@ -121,6 +126,12 @@ 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
+                    if executor and self.VARIABLE_REPLACEMENTS.get(executor):
+                        for key,val in self.VARIABLE_REPLACEMENTS[executor].items():
+                            try:
+                                command = command.replace(key, val)
+                            except:
+                                pass
         return self._path_replacement(command, path)
 
     def _check_if_aws(self, test):
@@ -146,6 +157,12 @@ def _set_input_arguments(self, test, **kwargs):
             if Base.CONFIG.prompt_for_input_args:
                 for input in test.input_arguments:
                     input.value = self.prompt_user_for_input(test.name, input)
+            for key,val in self.VARIABLE_REPLACEMENTS.items():
+                if test.executor.name == key:
+                    for k,v in val.items():
+                        for input in test.input_arguments:
+                            if k in input.default:
+                                input.value = input.default.replace(k,v)
             for input in test.input_arguments:
                 if input.value == None:
                     input.value = input.default

atomic_operator/execution/awsrunner.py

@@ -41,12 +41,13 @@ def execute_process(self, command, executor=None, host=None, cwd=None, elevation
             if executor in ['powershell']:
                 command = f"Start-Process PowerShell -Verb RunAs; {command}"
             elif executor in ['cmd', 'command_prompt']:
-                command = f'cmd.exe /c "{command}"'
+                command = f'{self.command_map.get(executor).get(self.__local_system_platform)} /c "{command}"'
             elif executor in ['sh', 'bash', 'ssh']:
                 command = f"sudo {command}"
             else:
                 self.__logger.warning(f"Elevation is required but the executor '{executor}' is unknown!")
-        command = self._replace_command_string(command, self.CONFIG.atomics_path, input_arguments=self.test.input_arguments)
+        command = self._replace_command_string(command, self.CONFIG.atomics_path, input_arguments=self.test.input_arguments, executor=executor)
+        executor = self.command_map.get(executor).get(self.__local_system_platform)
         p = subprocess.Popen(
             executor, 
             shell=False, 
@@ -89,5 +90,5 @@ def _get_executor_command(self):
     def start(self):
         response = self.__check_for_aws_cli()
         if not response.get('error'):
-            return self.execute(executor=self._get_executor_command())
+            return self.execute(executor=self.test.executor.name)
         return response

atomic_operator/execution/localrunner.py

@@ -39,7 +39,8 @@ def execute_process(self, command, executor=None, host=None, cwd=None, elevation
                 command = f"sudo {command}"
             else:
                 self.__logger.warning(f"Elevation is required but the executor '{executor}' is unknown!")
-        command = self._replace_command_string(command, self.CONFIG.atomics_path, input_arguments=self.test.input_arguments)
+        command = self._replace_command_string(command, self.CONFIG.atomics_path, input_arguments=self.test.input_arguments, executor=executor)
+        executor = self.command_map.get(executor).get(self.__local_system_platform)
         p = subprocess.Popen(
             executor, 
             shell=False, 
@@ -81,4 +82,4 @@ def _get_executor_command(self):
         return __executor
 
     def start(self):
-        return self.execute(executor=self._get_executor_command())
+        return self.execute(executor=self.test.executor.name)

atomic_operator/execution/statemachine.py

@@ -146,7 +146,7 @@ def __invoke_cmd(self, command, input_arguments=None, elevation_required=False):
             self.__create_win_client(self.hostinfo)
         # TODO: NEED TO ADD LOGIC TO TRANSFER FILES TO WINDOWS SYSTEMS USING CMD
         Copier(windows_client=self.__win_client, elevation_required=elevation_required).copy(input_arguments)
-        command = self._replace_command_string(command, path='c:/temp', input_arguments=input_arguments)
+        command = self._replace_command_string(command, path='c:/temp', input_arguments=input_arguments, executor='command_prompt')
         if elevation_required:
             command = f'runas /user:{self.hostinfo.username}:{self.hostinfo.password} cmd.exe; {command}'
         # TODO: NEED TO ADD LOGIC TO TRANSFER FILES TO WINDOWS SYSTEMS USING CMD
@@ -173,7 +173,7 @@ def __invoke_powershell(self, command, input_arguments=None, elevation_required=
 
         # TODO: NEED TO ADD LOGIC TO TRANSFER FILES TO WINDOWS SYSTEMS USING POWERSHELL
         Copier(windows_client=self.__win_client, elevation_required=elevation_required).copy(input_arguments=input_arguments)
-        command = self._replace_command_string(command, path='c:/temp', input_arguments=input_arguments)
+        command = self._replace_command_string(command, path='c:/temp', input_arguments=input_arguments, executor='powershell')
         # TODO: NEED TO ADD LOGIC TO TRANSFER FILES TO WINDOWS SYSTEMS USING POWERSHELL
         if elevation_required:
             command = f'Start-Process PowerShell -Verb RunAs; {command}'

docs/CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## 0.8.4 - 2022-03-25
+
+    * Updated formatting of executor for AWS and local runners
+    * Updated documentation
+    * Added formatting constants to base class to improve updating of windows variables on command line runners
+
 ## 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
@@ -22,11 +28,14 @@
     * Updating handling of passing --help to the run command
     * Updated docs to reflect change
 
-## 0.5.0 - 2021-11-19
+## 0.5.0 - 2021-11-18
 
-    * Updated and expanded tests
-    * Added CI to build docs and deploy to pypi
-    * Updated localrunner to support windows based on changes made previously
+    * Updated handling of versioning
+    * Updated CI to handle versioning of docs and deployment on release
+    * Added better handling of extracting zip file
+    * Added safer loading of yaml files
+    * Update docs
+    * Improved logging across the board and implemented a debug switch
 
 ## 0.4.0 - 2021-11-15
 

docs/index.md

@@ -79,7 +79,7 @@ pip3 install --upgrade pip
 pip3 install setuptools_rust
 
 # Back to our regularly scheduled programming . . .  
-
+pip install -r requirements.txt
 python setup.py install
 ```
 
@@ -88,6 +88,7 @@ python setup.py install
 ```bash
 git clone https://github.com/swimlane/atomic-operator.git
 cd atomic-operator
+pip install -r requirements.txt
 python setup.py install
 ```
 

requirements.txt

@@ -3,5 +3,5 @@ fire==0.4.0
 requests==2.26.0
 attrs==21.2.0
 pypsrp==0.5.0
-paramiko==2.7.2
+paramiko>=2.10.1
 pick==1.2.0

setup.py

@@ -15,13 +15,11 @@ def read(rel_path: str) -> str:
         return fp.read()
 
 
-def get_version(rel_path: str) -> str:
-    for line in read(rel_path).splitlines():
-        if line.startswith("__version__"):
-            # __version__ = "0.9"
-            delim = '"' if '"' in line else "'"
-            return line.split(delim)[1]
-    raise RuntimeError("Unable to find version string.")
+here = os.path.abspath(os.path.dirname(__file__))
+
+about = {}
+with open(os.path.join(here, 'atomic_operator', '__meta__.py'), 'r', 'utf-8') as f:
+    exec(f.read(), about)
 
 
 PROJECT_URLS = {
@@ -52,21 +50,36 @@ def get_version(rel_path: str) -> str:
 ]
 
 setup(
-    name='atomic-operator',
-    version=get_version("atomic_operator/__init__.py"),
+    name=about['__title__'],
+    version=about['__version__'],
     packages=find_packages(exclude=['tests*']),
-    license='MIT',
-    description='A python package to execute Atomic tests',
+    license=about['__license__'],
+    description=about['__description__'],
     long_description=open('README.md').read(),
     long_description_content_type="text/markdown",
     project_urls=PROJECT_URLS,
-    install_requires=parse_requirements('./requirements.txt'),
+    install_requires=[
+        'requests>=2.20.0; python_version >= "3.6"',
+        'charset_normalizer~=2.0.0; python_version >= "3"',
+        'chardet>=3.0.2,<5; python_version < "3"',
+        'idna>=2.5,<3; python_version < "3"',
+        'idna>=2.5,<4; python_version >= "3"',
+        'urllib3>=1.21.1,<1.27',
+        'certifi>=2017.4.17',
+        'windows-curses>=2.2.0,<3.0.0; platform_system=="Windows" and python_version >= "3.6"',
+        'attrs>=21.3.0; python_version >= "3.6"',
+        'pyyaml>=6.0; python_version >= "3.6"',
+        'fire>=0.4.0; python_version >= "3.6"',
+        'pypsrp>=0.5.0; python_version >= "3.6"',
+        'paramiko>=2.7.2; python_version >= "3.6"',
+        'pick>=1.2.0; python_version >= "3.6"'
+    ],
     keywords=['atomic-red-team', 'att&ck', 'test', 'mitre', 'executor'],
-    url='https://github.com/swimlane/atomic-operator',
-    author='MSAdministrator',
-    author_email='rickardja@live.com',
-    maintainer='MSAdministrator',
-    maintainer_email='rickardja@live.com',
+    url=about['__url__'],
+    author=about['__author__'],
+    author_email=about['__author_email__'],
+    maintainer=about['__maintainer__'],
+    maintainer_email=about['__maintainer_email__'],
     python_requires='>=3.6, <4',
     classifiers=CLASSIFIERS,
     package_data={