Config files management

empire.py

@@ -3,9 +3,12 @@
 import sys
 
 from empire import arguments
+from empire import config_manager
+
 
 if __name__ == "__main__":
     args = arguments.args
+    config_manager.config_init()
 
     if args.subparser_name == "server":
         from empire.server import server
@@ -16,7 +19,7 @@
 
         from empire.scripts.sync_starkiller import sync_starkiller
 
-        with open("empire/server/config.yaml") as f:
+        with open(config_manager.CONFIG_SERVER_PATH) as f:
             config = yaml.safe_load(f)
 
         sync_starkiller(config)

empire/client/src/EmpireCliConfig.py

@@ -3,6 +3,8 @@
 
 import yaml
 
+from empire import config_manager
+
 log = logging.getLogger(__name__)
 
 
@@ -15,7 +17,8 @@ def __init__(self):
             self.set_yaml(location)
         if len(self.yaml.items()) == 0:
             log.info("Loading default config")
-            self.set_yaml("./empire/client/config.yaml")
+            self.set_yaml(config_manager.CONFIG_CLIENT_PATH)
+        config_manager.check_config_permission(self.yaml, "client")
 
     def set_yaml(self, location: str):
         try:
@@ -26,5 +29,4 @@ def set_yaml(self, location: str):
         except FileNotFoundError as exc:
             log.error(exc)
 
-
 empire_config = EmpireCliConfig()

empire/config_manager.py

@@ -0,0 +1,85 @@
+import logging
+import os
+import shutil
+import yaml
+
+from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+user_home = Path.home()
+SOURCE_CONFIG_CLIENT = Path("empire/client/config.yaml")
+SOURCE_CONFIG_SERVER = Path("empire/server/config.yaml")
+CONFIG_DIR = user_home / ".empire"
+CONFIG_CLIENT_PATH = CONFIG_DIR / "client" / "config.yaml"
+CONFIG_SERVER_PATH = CONFIG_DIR / "server" / "config.yaml"
+
+def config_init():
+    CONFIG_CLIENT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    CONFIG_SERVER_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+    if not CONFIG_CLIENT_PATH.exists():
+        shutil.copy(SOURCE_CONFIG_CLIENT, CONFIG_CLIENT_PATH)
+        log.info(f"Copied {SOURCE_CONFIG_CLIENT} to {CONFIG_CLIENT_PATH}")
+    else:
+        log.info(f"{CONFIG_CLIENT_PATH} already exists.")
+
+    if not CONFIG_SERVER_PATH.exists():
+        shutil.copy(SOURCE_CONFIG_SERVER, CONFIG_SERVER_PATH)
+        log.info(f"Copied {SOURCE_CONFIG_SERVER} to {CONFIG_SERVER_PATH}")
+    else:
+        log.info(f"{CONFIG_SERVER_PATH} already exists.")
+
+
+def check_config_permission(config_dict: dict, config_type: str):
+    """
+    Check if the specified directories in config.yaml are writable. If not, switches to a fallback directory.
+    Handles both server and client configurations.
+    
+    Args:
+        config_dict (dict): The configuration dictionary loaded from YAML.
+        config_type (str): The type of configuration ("server" or "client").
+    """
+    # Define paths to check based on config type
+    if config_type == "server":
+        paths_to_check = {
+            ("api", "cert_path"): config_dict["api"]["cert_path"],
+            ("database", "sqlite", "location"): config_dict["database"]["sqlite"]["location"],
+            ("starkiller", "directory"): config_dict["starkiller"]["directory"],
+            ("logging", "directory"): config_dict["logging"]["directory"],
+            ("debug", "last_task", "file"): config_dict["debug"]["last_task"]["file"],
+            ("directories", "downloads"): config_dict["directories"].get("downloads"),
+        }
+        config_path = CONFIG_SERVER_PATH  # Use the server config path
+
+    elif config_type == "client":
+        paths_to_check = {
+            ("logging", "directory"): config_dict["logging"]["directory"],
+            ("directories", "downloads"): config_dict["directories"].get("downloads"),
+            ("directories", "generated-stagers"): config_dict["directories"].get("generated-stagers"),
+        }
+        config_path = CONFIG_CLIENT_PATH  # Use the client config path
+
+    else:
+        raise ValueError("Invalid config_type. Expected 'server' or 'client'.")
+
+    # Check permissions and update paths as needed
+    for keys, dir_path in paths_to_check.items():
+        if not os.access(dir_path, os.W_OK):
+            log.info("No write permission for %s. Switching to fallback directory.", dir_path)
+            user_home = Path.home()
+            fallback_dir = os.path.join(user_home, ".empire", dir_path.removeprefix("empire/"))
+
+            # Update the directory in config_dict
+            target = config_dict  # target is a reference to config_dict
+            for key in keys[:-1]:
+                target = target[key]
+            target[keys[-1]] = fallback_dir
+
+            log.info("Updated %s to fallback directory: %s", "->".join(keys), fallback_dir)
+
+    # Write the updated configuration back to the correct YAML file
+    with open(config_path, 'w') as config_file:
+        yaml.safe_dump(config_dict, config_file)
+
+    log.info("Updated $config_type config.yaml to use fallback directory: %s", config_type, fallback_dir)

empire/server/core/config.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 import yaml
+from empire import config_manager
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 log = logging.getLogger(__name__)
@@ -133,6 +134,7 @@ def set_yaml(location: str):
     config_dict = set_yaml(location)
 if len(config_dict.items()) == 0:
     log.info("Loading default config")
-    config_dict = set_yaml("./empire/server/config.yaml")
+    config_dict = set_yaml(config_manager.CONFIG_SERVER_PATH)
 
+config_manager.check_config_permission(config_dict, "server")
 empire_config = EmpireConfig(config_dict)

empire/server/server.py

@@ -70,7 +70,9 @@ def setup_logging(args):
 INVOKE_OBFS_SRC_DIR_BASE = os.path.join(
     os.path.dirname(__file__), "data/Invoke-Obfuscation"
 )
-INVOKE_OBFS_DST_DIR_BASE = "/usr/local/share/powershell/Modules/Invoke-Obfuscation"
+
+user_home = Path.home()
+INVOKE_OBFS_DST_DIR_BASE = user_home / ".local" / "share" / "powershell" / "Modules" / "Invoke-Obfuscation"
 
 
 def reset():