Eliminate Caldera Core Dependencies on Plugins: stockpile

app/obfuscators/base64_basic.py

@@ -0,0 +1,24 @@
+from base64 import b64encode
+
+from app.utility.base_obfuscator import BaseObfuscator
+
+
+class Obfuscation(BaseObfuscator):
+
+    @property
+    def supported_platforms(self):
+        return dict(
+            windows=['psh'],
+            darwin=['sh'],
+            linux=['sh']
+        )
+
+    """ EXECUTORS """
+
+    def psh(self, link):
+        recoded = b64encode(self.decode_bytes(link.command).encode('UTF-16LE'))
+        return 'powershell -Enc %s' % recoded.decode('utf-8')
+
+    @staticmethod
+    def sh(link):
+        return 'eval "$(echo %s | base64 --decode)"' % str(link.command.encode(), 'utf-8')

app/obfuscators/plain_text.py

@@ -0,0 +1,7 @@
+from app.utility.base_obfuscator import BaseObfuscator
+
+
+class Obfuscation(BaseObfuscator):
+
+    def run(self, link, **kwargs):
+        return self.decode_bytes(link.command)

server.py

@@ -20,6 +20,7 @@
 from app.api.v2.responses import apispec_request_validation_middleware
 from app.api.v2.security import pass_option_middleware
 from app.objects.c_agent import Agent
+from app.objects.c_obfuscator import Obfuscator
 from app.objects.secondclass.c_executor import Executor
 from app.objects.secondclass.c_link import Link
 from app.service.app_svc import AppService
@@ -98,6 +99,20 @@ def run_tasks(services, run_vue_server=False):
     loop.create_task(app_svc.watch_ability_files())
     loop.run_until_complete(start_server())
     loop.run_until_complete(event_svc.fire_event(exchange="system", queue="ready"))
+    loop.run_until_complete(
+        data_svc.store(
+            Obfuscator(name='plain-text',
+                       description='Does no obfuscation to any command, instead running it in plain text',
+                       module='app.obfuscators.plain_text')
+        )
+    )
+    loop.run_until_complete(
+        data_svc.store(
+            Obfuscator(name='base64',
+                       description='Obfuscates commands in base64',
+                       module='app.obfuscators.base64_basic')
+        )
+    )
     if run_vue_server:
         loop.run_until_complete(start_vue_dev_server())
     try:

tests/conftest.py

@@ -206,7 +206,7 @@ def obfuscator(event_loop, data_svc):
     event_loop.run_until_complete(data_svc.store(
         Obfuscator(name='plain-text',
                    description='Does no obfuscation to any command, instead running it in plain text',
-                   module='plugins.stockpile.app.obfuscators.plain_text')
+                   module='app.obfuscators.plain_text')
         )
     )
 

tests/obfuscators/test_base64.py

@@ -0,0 +1,76 @@
+import pytest
+
+from base64 import b64encode
+from app.utility.base_service import BaseService
+from app.objects.c_agent import Agent
+from app.objects.secondclass.c_link import Link
+from app.objects.c_ability import AbilitySchema
+from app.objects.secondclass.c_executor import ExecutorSchema
+from app.obfuscators.base64_basic import Obfuscation
+
+
+@pytest.fixture
+def b64basic_obfuscator():
+    def _generate_obfuscator(obfuscator_agent):
+        return Obfuscation(obfuscator_agent)
+
+    return _generate_obfuscator
+
+
+@pytest.fixture
+def test_windows_agent(event_loop):
+    agent = Agent(paw='123', sleep_min=2, sleep_max=8, watchdog=0, executors=['psh'], platform='windows')
+    event_loop.run_until_complete(BaseService.get_service('data_svc').store(agent))
+    return agent
+
+
+@pytest.fixture
+def test_windows_executor(test_windows_agent):
+    return ExecutorSchema().load(dict(timeout=60, platform=test_windows_agent.platform, name='psh', command='ls'))
+
+
+@pytest.fixture
+def test_windows_ability(test_windows_executor, event_loop):
+    ability = AbilitySchema().load(dict(ability_id='123',
+                                        tactic='discovery',
+                                        technique_id='auto-generated',
+                                        technique_name='auto-generated',
+                                        name='Manual Command',
+                                        description='test windows ability',
+                                        executors=[ExecutorSchema().dump(test_windows_executor)]))
+    event_loop.run_until_complete(BaseService.get_service('data_svc').store(ability))
+    return ability
+
+
+@pytest.fixture
+def finished_windows_link(test_windows_executor, test_windows_agent, test_windows_ability):
+    return {
+        'command': str(b64encode(test_windows_executor.command.encode()), 'utf-8'),
+        'paw': test_windows_agent.paw,
+        'ability': test_windows_ability,
+        'executor': test_windows_executor,
+        'host': test_windows_agent.host,
+        'deadman': False,
+        'used': [],
+        'id': '789',
+        'relationships': [],
+        'status': 0,
+        'output': 'test_dir'
+    }
+
+
+class TestBase64BasicObfuscator:
+    def test_b64basic_obfuscator(self, b64basic_obfuscator, test_agent, test_windows_agent, finished_link, finished_windows_link):
+        # sh
+        linux_obfuscator = b64basic_obfuscator(test_agent)
+        sh_link = Link.load(finished_link)
+        expected_encoded = 'eval "$(echo bHM= | base64 --decode)"'
+        decoded = linux_obfuscator.run(sh_link)
+        assert decoded == expected_encoded
+
+        # psh
+        windows_obfuscator = b64basic_obfuscator(test_windows_agent)
+        psh_link = Link.load(finished_windows_link)
+        expected_encoded = 'powershell -Enc bABzAA=='
+        decoded = windows_obfuscator.run(psh_link)
+        assert decoded == expected_encoded

tests/obfuscators/test_plaintext.py

@@ -0,0 +1,18 @@
+import pytest
+
+from base64 import b64decode
+from app.objects.secondclass.c_link import Link
+from app.obfuscators.plain_text import Obfuscation
+
+
+@pytest.fixture
+def plaintext_obfuscator(test_agent):
+    return Obfuscation(test_agent)
+
+
+class TestPlainTextObfuscator:
+    def test_plaintext_obfuscator(self, plaintext_obfuscator, finished_link):
+        link = Link.load(finished_link)
+        plaintext_cmd = b64decode(link.command).decode('utf-8')
+        decoded = plaintext_obfuscator.run(link)
+        assert decoded == plaintext_cmd

tests/services/test_planning_svc.py

@@ -91,7 +91,7 @@ async def setup_planning_test(executor, ability, agent, operation, data_svc, eve
     await data_svc.store(
         Obfuscator(name='plain-text',
                    description='Does no obfuscation to any command, instead running it in plain text',
-                   module='plugins.stockpile.app.obfuscators.plain_text')
+                   module='app.obfuscators.plain_text')
     )
 
     yield tability, tagent, toperation, cability

tests/services/test_rest_svc.py

@@ -60,7 +60,7 @@ async def setup_rest_svc_test(data_svc):
     await data_svc.store(
         Obfuscator(name='plain-text',
                    description='Does no obfuscation to any command, instead running it in plain text',
-                   module='plugins.stockpile.app.obfuscators.plain_text')
+                   module='app.obfuscators.plain_text')
     )