Config path improvements (BC-SECURITY#724)

* Use path for config and config properties. Always expanduser and resolve to the absolute path

* fix obfuscation

* fix download test

* allow starkiller to be outside the dir too

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 -   Updated listeners to consistently use port 80 and 443 for HTTP traffic by default (@Cx01N)
 -   Make the installation of donut conditional on architecture since it doesn't work on ARM (@Vinnybod)
     -   When donut is invoked but not installed, give a useful warning (@Vinnybod)
+-   Allow a config to be loaded from an outside directory and the downloads/logs/etc to be stored in an outside directory (@Vinnybod)
 -   Drop support for Python 3.8 and 3.9
 - Update install script (@Vinnybod)
     -   Use pyenv to install Python
@@ -23,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     -   Bump all OS to use Python 3.12
     -   Refactor the script to be a bit more readable
     -   Condense the test_install_script job
+    -   Added option to start MySQL service on boot (@Cx01N)
 -   Update Docker build (@Vinnybod)
     -   Use the official Poetry installer
     -   Fix Starkiller trying to auto-update inside the container
@@ -36,7 +38,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     -   Use docopt-ng for Python 3.12 support
     -   Add packaging as a runtime dependency
 -   Correct more deprecation warnings for SQLAlchemy and invalid escape sequences (@Vinnybod)
--   Added option to start MySQL service on boot to install script (@Cx01N)
 -   Remove unneeded condition statement from all listeners (@Vinnybod)
 -   Updated the ruff minimum Python version to 3.10 and applied fixes to get codebase compliant (@Vinnybod)
 

empire/server/common/agents.py

@@ -42,7 +42,6 @@
 import threading
 import time
 import warnings
-from pathlib import Path
 
 from sqlalchemy import and_, or_
 from sqlalchemy.orm import Session
@@ -255,7 +254,7 @@ def save_file(
         parts = path.split("\\")
 
         # construct the appropriate save path
-        download_dir = Path(empire_config.directories.downloads)
+        download_dir = empire_config.directories.downloads
         save_path = download_dir / sessionID / "/".join(parts[0:-1])
         filename = os.path.basename(parts[-1])
         save_file = save_path / filename
@@ -345,7 +344,7 @@ def save_module_file(self, sessionID, path, data, language: str):
         parts = path.split("/")
 
         # construct the appropriate save path
-        download_dir = Path(empire_config.directories.downloads)
+        download_dir = empire_config.directories.downloads
         save_path = download_dir / sessionID / "/".join(parts[0:-1])
         filename = parts[-1]
         save_file = save_path / filename
@@ -403,7 +402,7 @@ def save_agent_log(self, session_id, data):
         if isinstance(data, bytes):
             data = data.decode("UTF-8")
 
-        save_path = Path(empire_config.directories.downloads) / session_id
+        save_path = empire_config.directories.downloads / session_id
 
         # make the recursive directory structure if it doesn't already exist
         if not save_path.exists():
@@ -1642,7 +1641,7 @@ def process_agent_packet(
         elif response_name == "TASK_CMD_JOB":
             # check if this is the powershell keylogging task, if so, write output to file instead of screen
             if key_log_task_id and key_log_task_id == task_id:
-                download_dir = Path(empire_config.directories.downloads)
+                download_dir = empire_config.directories.downloads
                 safe_path = download_dir.absolute()
                 save_path = download_dir / session_id / "keystrokes.txt"
 

empire/server/core/config.py

@@ -1,28 +1,37 @@
 import logging
 import sys
+from pathlib import Path
 
 import yaml
-from pydantic import BaseModel, Extra, Field
+from pydantic import BaseModel, Extra, Field, validator
 
 log = logging.getLogger(__name__)
 
 
-class StarkillerConfig(BaseModel):
+class EmpireBaseModel(BaseModel):
+    @validator("*")
+    def set_path(cls, v):
+        if isinstance(v, Path):
+            return v.expanduser().resolve()
+        return v
+
+
+class StarkillerConfig(EmpireBaseModel):
     repo: str = "bc-security/starkiller"
-    directory: str = "empire/server/api/v2/starkiller"
+    directory: Path = "empire/server/api/v2/starkiller"
     ref: str = "main"
     auto_update: bool = True
 
 
-class DatabaseDefaultObfuscationConfig(BaseModel):
+class DatabaseDefaultObfuscationConfig(EmpireBaseModel):
     language: str = "powershell"
     enabled: bool = False
     command: str = r"Token\All\1"
     module: str = "invoke-obfuscation"
     preobfuscatable: bool = True
 
 
-class DatabaseDefaultsConfig(BaseModel):
+class DatabaseDefaultsConfig(EmpireBaseModel):
     staging_key: str = "RANDOM"
     username: str = "empireadmin"
     password: str = "password123"
@@ -32,18 +41,18 @@ class DatabaseDefaultsConfig(BaseModel):
     ip_blacklist: str = Field("", alias="ip-blacklist")
 
 
-class SQLiteDatabaseConfig(BaseModel):
-    location: str = "empire/server/data/empire.db"
+class SQLiteDatabaseConfig(EmpireBaseModel):
+    location: Path = "empire/server/data/empire.db"
 
 
-class MySQLDatabaseConfig(BaseModel):
+class MySQLDatabaseConfig(EmpireBaseModel):
     url: str = "localhost:3306"
     username: str = ""
     password: str = ""
     database_name: str = "empire"
 
 
-class DatabaseConfig(BaseModel):
+class DatabaseConfig(EmpireBaseModel):
     use: str = "sqlite"
     sqlite: SQLiteDatabaseConfig
     mysql: MySQLDatabaseConfig
@@ -53,28 +62,28 @@ def __getitem__(self, key):
         return getattr(self, key)
 
 
-class DirectoriesConfig(BaseModel):
-    downloads: str
-    module_source: str
-    obfuscated_module_source: str
+class DirectoriesConfig(EmpireBaseModel):
+    downloads: Path
+    module_source: Path
+    obfuscated_module_source: Path
 
 
-class LoggingConfig(BaseModel):
+class LoggingConfig(EmpireBaseModel):
     level: str = "INFO"
-    directory: str = "empire/server/downloads/logs/"
+    directory: Path = "empire/server/downloads/logs/"
     simple_console: bool = True
 
 
-class LastTaskConfig(BaseModel):
+class LastTaskConfig(EmpireBaseModel):
     enabled: bool = False
-    file: str = "empire/server/data/last_task.txt"
+    file: Path = "empire/server/data/last_task.txt"
 
 
-class DebugConfig(BaseModel):
+class DebugConfig(EmpireBaseModel):
     last_task: LastTaskConfig
 
 
-class EmpireConfig(BaseModel):
+class EmpireConfig(EmpireBaseModel):
     supress_self_cert_warning: bool = Field(
         alias="supress-self-cert-warning", default=True
     )
@@ -95,13 +104,14 @@ class Config:
 
 
 def set_yaml(location: str):
+    location = Path(location).expanduser().resolve()
     try:
-        with open(location) as stream:
+        with location.open() as stream:
             return yaml.safe_load(stream)
     except yaml.YAMLError as exc:
-        print(exc)
+        log.warning(exc)
     except FileNotFoundError as exc:
-        print(exc)
+        log.warning(exc)
 
 
 config_dict = {}

empire/server/core/download_service.py

@@ -129,10 +129,7 @@ def create_download_from_text(
         """
         subdirectory = subdirectory or f"user/{user.username}"
         location = (
-            Path(empire_config.directories.downloads)
-            / "uploads"
-            / subdirectory
-            / filename
+            empire_config.directories.downloads / "uploads" / subdirectory / filename
         )
         location.parent.mkdir(parents=True, exist_ok=True)
 
@@ -157,10 +154,7 @@ def create_download(self, db: Session, user: models.User, file: UploadFile | Pat
             filename = file.filename
 
         location = (
-            Path(empire_config.directories.downloads)
-            / "uploads"
-            / user.username
-            / filename
+            empire_config.directories.downloads / "uploads" / user.username / filename
         )
         location.parent.mkdir(parents=True, exist_ok=True)
 

empire/server/core/obfuscation_service.py

@@ -100,7 +100,6 @@ def preobfuscate_modules(
         files = self._get_module_source_files(db_obf_config.language)
 
         for file in files:
-            file = os.getcwd() + "/" + file
             if reobfuscate or not self._is_obfuscated(file):
                 message = f"Obfuscating {os.path.basename(file)}..."
                 log.info(message)
@@ -133,8 +132,8 @@ def obfuscate_module(
         obfuscated_code = self.obfuscate(module_code, obfuscation_command)
 
         obfuscated_source = module_source.replace(
-            empire_config.directories.module_source,
-            empire_config.directories.obfuscated_module_source,
+            str(empire_config.directories.module_source),
+            str(empire_config.directories.obfuscated_module_source),
         )
 
         try:
@@ -230,12 +229,16 @@ def _get_obfuscated_module_source_files(self, language: str):
 
         return paths
 
-    def _is_obfuscated(self, module_source):
+    def _is_obfuscated(self, module_source: str | Path):
+        if isinstance(module_source, Path):
+            module_source = str(module_source)
+
         obfuscated_source = module_source.replace(
-            empire_config.directories.module_source,
-            empire_config.directories.obfuscated_module_source,
+            str(empire_config.directories.module_source),
+            str(empire_config.directories.obfuscated_module_source),
         )
-        return os.path.isfile(obfuscated_source)
+
+        return Path(obfuscated_source).exists()
 
     def _convert_obfuscation_command(self, obfuscate_command):
         return (

empire/server/core/stager_service.py

@@ -1,7 +1,6 @@
 import copy
 import os
 import uuid
-from pathlib import Path
 from typing import Any
 
 from sqlalchemy.orm import Session
@@ -169,7 +168,7 @@ def generate_stager(self, template_instance):
             file_name = f"{uuid.uuid4()}.txt"
 
         file_name = (
-            Path(empire_config.directories.downloads) / "generated-stagers" / file_name
+            empire_config.directories.downloads / "generated-stagers" / file_name
         )
         file_name.parent.mkdir(parents=True, exist_ok=True)
         mode = "w" if isinstance(resp, str) else "wb"

empire/server/server.py

@@ -32,8 +32,7 @@ def setup_logging(args):
     else:
         log_level = logging.getLevelName(empire_config.logging.level.upper())
 
-    logging_dir = empire_config.logging.directory
-    log_dir = Path(logging_dir)
+    log_dir = empire_config.logging.directory
     log_dir.mkdir(parents=True, exist_ok=True)
     root_log_file = log_dir / "empire_server.log"
     root_logger = logging.getLogger()

empire/test/test_download_service.py

@@ -36,11 +36,9 @@ def test_create_download_from_path(main, session_local, models):
         download = download_service.create_download(db, user, test_upload)
 
         assert download.id > 0
-        assert download.filename.startswith(
-            "test-upload"
-        ) and download.filename.endswith(".yaml")
-        assert download.location.startswith(
-            f"empire/test/downloads/uploads/{user.username}/test-upload"
-        ) and download.location.endswith(".yaml")
+        assert download.filename.startswith("test-upload")
+        assert download.filename.endswith(".yaml")
+        assert f"empire/test/downloads/uploads/{user.username}/" in download.location
+        assert download.location.endswith(".yaml")
 
         db.delete(download)

empire/test/test_logs.py

@@ -85,7 +85,7 @@ def test_log_level_by_config(monkeypatch):
     assert stream_handler.level == logging.WARNING
 
 
-def test_log_level_by_arg(monkeypatch):
+def test_log_level_by_arg():
     logging.getLogger().handlers.clear()
     os.chdir(Path(os.path.dirname(os.path.abspath(__file__))).parent.parent)
     sys.argv = [
@@ -97,16 +97,13 @@ def test_log_level_by_arg(monkeypatch):
         "ERROR",
     ]
 
-    monkeypatch.setattr("empire.server.server.empire", MagicMock())
-
     from empire import arguments
     from empire.server.server import setup_logging
 
     config_mock = MagicMock()
     test_config = _load_test_config()
     test_config["logging"]["level"] = "WaRNiNG"  # Should be overwritten by arg
     config_mock.yaml = test_config
-    monkeypatch.setattr("empire.server.server.empire_config", config_mock)
 
     args = arguments.parent_parser.parse_args()  # Force reparse of args between runs
     setup_logging(args)
@@ -118,21 +115,18 @@ def test_log_level_by_arg(monkeypatch):
     assert stream_handler.level == logging.ERROR
 
 
-def test_log_level_by_debug_arg(monkeypatch):
+def test_log_level_by_debug_arg():
     logging.getLogger().handlers.clear()
     os.chdir(Path(os.path.dirname(os.path.abspath(__file__))).parent.parent)
     sys.argv = ["", "server", "--config", SERVER_CONFIG_LOC, "--debug"]
 
-    monkeypatch.setattr("empire.server.server.empire", MagicMock())
-
     from empire import arguments
     from empire.server.server import setup_logging
 
     config_mock = MagicMock()
     test_config = _load_test_config()
     test_config["logging"]["level"] = "WaRNiNG"  # Should be overwritten by arg
     config_mock.yaml = test_config
-    monkeypatch.setattr("empire.server.server.empire_config", config_mock)
 
     args = arguments.parent_parser.parse_args()  # Force reparse of args between runs
     setup_logging(args)

empire/test/test_obfuscation_api.py

@@ -1,5 +1,6 @@
 import os
 from contextlib import contextmanager
+from pathlib import Path
 
 import pytest
 
@@ -12,7 +13,9 @@ def patch_config(empire_config):
     """
     orig_src_dir = empire_config.directories.module_source
     try:
-        empire_config.directories.module_source = "empire/test/data/module_source/"
+        empire_config.directories.module_source = Path(
+            "empire/test/data/module_source/"
+        ).resolve()
         yield empire_config
     finally:
         empire_config.directories.module_source = orig_src_dir
@@ -231,8 +234,8 @@ def test_preobfuscate_post(client, admin_auth_header, empire_config):
         count = 0
         for root, _dirs, files in os.walk(module_dir):
             for file in files:
-                root_rep = root.replace(module_dir, obf_module_dir)
-                assert os.path.exists(root_rep + "/" + file)
+                root_rep = root.replace(str(module_dir), str(obf_module_dir))
+                assert (Path(root_rep) / file).exists()
                 count += 1
 
         assert count > 0
@@ -266,5 +269,5 @@ def test_preobfuscate_delete(client, admin_auth_header, empire_config):
 
         for root, _dirs, files in os.walk(module_dir):
             for file in files:
-                root_rep = root.replace(module_dir, obf_module_dir)
+                root_rep = root.replace(str(module_dir), str(obf_module_dir))
                 assert not os.path.exists(root_rep + "/" + file)