import requests import struct import time import base64 import hashlib import argparse import getpass import os import json import threading import tkinter as tk from tkinter import ttk from tkinter import scrolledtext # --- AMF3 Encoder Helpers --- def encode_u29(n): if n < 0x80: return bytes([n]) elif n < 0x4000: return bytes([((n >> 7) & 0x7F) | 0x80, n & 0x7F]) elif n < 0x200000: return bytes([((n >> 14) & 0x7F) | 0x80, ((n >> 7) & 0x7F) | 0x80, n & 0x7F]) else: # Simplified, usually max 4 bytes return bytes([((n >> 22) & 0x7F) | 0x80, ((n >> 15) & 0x7F) | 0x80, ((n >> 8) & 0x7F) | 0x80, n & 0xFF]) def encode_string(s): if s == "": return b'\x06\x01' # Empty string b = s.encode('utf-8') l = len(b) # String marker 0x06 + U29 length ((len << 1) | 1) return b'\x06' + encode_u29((l << 1) | 1) + b def encode_integer(n): # AMF3 Integer (0x04) is actually a U29, but let's stick to spec. # If it fits in U29 (29 bits), use 0x04. Else use Double 0x05. if -268435456 <= n <= 268435455: return b'\x04' + encode_u29(n & 0x1FFFFFFF) # Mask for negative handling else: return encode_double(float(n)) def encode_double(n): return b'\x05' + struct.pack(">d", n) def decode_u29(data, offset): res = 0 for _ in range(4): b = data[offset] offset += 1 res = (res << 7) | (b & 0x7F) if not (b & 0x80): break return res, offset def read_amf3_string(data, offset): head = data[offset] offset += 1 if (head & 1) == 0: return None, offset l = head >> 1 s = data[offset:offset+l].decode('utf-8', errors='ignore') offset += l return s, offset def parse_amf3_object(data, offset): obj = {} if offset >= len(data) or data[offset] != 0x0A: return obj, offset offset += 1 traits = data[offset] offset += 1 cname, offset = read_amf3_string(data, offset) while offset < len(data): key, offset = read_amf3_string(data, offset) if key is None or key == "": break t = data[offset] offset += 1 val = None if t == 0x04: v, offset = decode_u29(data, offset) val = v elif t == 0x05: val = struct.unpack(">d", data[offset:offset+8])[0] offset += 8 elif t == 0x06: val, offset = read_amf3_string(data, offset) elif t == 0x01: val = None elif t == 0x00: val = None obj[key] = val return obj, offset def encode_undefined(): return b'\x00' def encode_null(): return b'\x01' def create_amf_packet(target_method, args): # AMF0 Header packet = b'\x00\x03' # AMF3 packet += b'\x00\x00' # Header Count 0 packet += b'\x00\x01' # Body Count 1 # Target URI target = target_method.encode('utf-8') packet += struct.pack(">H", len(target)) + target # Response URI response = b"/1" packet += struct.pack(">H", len(response)) + response # Content Length (Placeholder) packet += b'\xff\xff\xff\xff' # AMF0 Strict Array Wrapper (as seen in HAR) # 0A (Strict Array) + 00 00 00 01 (Count 1) body = b'\x0a\x00\x00\x00\x01' # AMF3 Body (The single element of the AMF0 array) body += b'\x11' # AMF3 encoding # Array of Arguments (AMF3 Array) # Marker 0x09 body += b'\x09' # Length of array (U29) count = len(args) body += encode_u29((count << 1) | 1) # Empty associative part body += b'\x01' # Encode each argument for arg in args: if arg is None: # We'll use Undefined (0x00) if specified, or Null (0x01) # Based on HAR analysis, it looked like 0x00 body += encode_undefined() elif isinstance(arg, str): body += encode_string(arg) elif isinstance(arg, int): # Check if it should be encoded as double (like timestamp) if arg > 268435455 or arg < -268435456: body += encode_double(float(arg)) else: body += encode_integer(arg) elif isinstance(arg, float): body += encode_double(arg) elif isinstance(arg, bytes): # Raw bytes injection (for debugging or specific AMF types) body += arg else: print(f"Warning: Unsupported type {type(arg)}") # Add body length and content packet = packet[:-4] + struct.pack(">I", len(body)) + body return packet def parse_response(data): print("\n[Response Analysis]") # Decode base64 if needed (sometimes HAR has base64, but requests.content is raw bytes) # The server returns raw bytes, usually. # Check for 'status' if b'status' in data: print("Found 'status' in response") if b'sessionkey' in data: # Extract session key idx = data.find(b'sessionkey') # Format: 0A (len) sessionkey 06 (String) ... # String 'sessionkey' len is 10. # It's likely an AMF3 object key. # Simple extraction heuristic # Look for the string value after sessionkey # 'sessionkey' (0x06 ...) ... value (0x06 ...) pass def start_ui(): return _start_ui_impl() def main(): ui_mode = False try: p = argparse.ArgumentParser() p.add_argument("--ui", action="store_true") p.add_argument("--user") p.add_argument("--pass", dest="password") p.add_argument("--config") a, unknown = p.parse_known_args() ui_mode = a.ui except: ui_mode = False if ui_mode: return start_ui() # 1. Get Token first print("--- Step 1: Check Version (Get Token) ---") # We can import check_version logic or reimplement simply # For speed, I'll just use the logic from check_version.py but using our new encoder ver_args = ["Public 0.53.1"] packet = create_amf_packet("SystemLogin.checkVersion", ver_args) url = "https://play.ninjasage.id/amf" headers = { "Content-Type": "application/x-amf", "User-Agent": "Mozilla/5.0 (Windows; U; en) AppleWebKit/533.19.4 (KHTML, like Gecko) AdobeAIR/51.1", "Referer": "app:/NinjaSage.swf", "x-flash-version": "51,1,3,10", "x-air-appid": "pta5yAteNUXGZU30lxDv+BRLP9xn3HZsknJnVEyQZU4=" } session = requests.Session() session.verify = False try: resp = session.post(url, headers=headers, data=packet) # Parse Token and Timestamp token = None timestamp = None # Token (__): 05 5f 5f 06 token_marker = resp.content.find(b'\x05\x5f\x5f\x06') if token_marker != -1: len_byte = resp.content[token_marker+4] token_len = len_byte >> 1 token = resp.content[token_marker+5 : token_marker+5+token_len].decode('utf-8') print(f"Got Token: {token}") # Timestamp (_): 03 5f 05 ts_marker = resp.content.find(b'\x03\x5f\x05') if ts_marker != -1: double_bytes = resp.content[ts_marker+3 : ts_marker+3+8] timestamp = struct.unpack(">d", double_bytes)[0] print(f"Got Timestamp: {timestamp}") if not token or not timestamp: print("Failed to get token/timestamp. Exiting.") return # 2. Login User print("\n--- Step 2: Login User ---") parser = argparse.ArgumentParser() parser.add_argument("--user") parser.add_argument("--pass", dest="password") parser.add_argument("--config") parser.add_argument("--ui", action="store_true") args = parser.parse_args() if args.ui: return start_ui() # Precedence: CLI > ENV > CONFIG > Prompt username = args.user password = args.password if not username: username = os.environ.get("NS_USERNAME") if not password: password = os.environ.get("NS_PASSWORD") if (not username or not password) and args.config: try: with open(args.config, "r", encoding="utf-8") as cf: cfg = json.load(cf) username = username or cfg.get("username") password = password or cfg.get("password") except Exception as e: print(f"Gagal membaca config '{args.config}': {e}") if not username: username = input("Username: ") if not password: password = getpass.getpass("Password: ") # Calculate Password Hash (Arg 1) # Assumed to be Base64(MD5(password)) pass_hash = base64.b64encode(hashlib.md5(password.encode('utf-8')).digest()).decode('utf-8') print(f"Generated Password Hash for '{password}': {pass_hash}") # Arg 7 Construction # Based on HAR analysis of successful logins: # Structure: Timestamp + CONSTANT_HASH # The constant hash part observed in HAR is: "40367c3cc999a9f9e951a1d33211545b84b2d5a6" # It appears to be a SHA1 hash (40 chars) that is static across different users. ts_str = str(int(timestamp)) static_hash = "40367c3cc999a9f9e951a1d33211545b84b2d5a6" # Combined Arg 7 final_arg7 = f"{ts_str}{static_hash}" print(f"Generated Arg 7: {final_arg7}") id_str = "2078177199839932112023317301122772909" login_args = [ username, pass_hash, float(timestamp), 1011067, # Arg 4 (from HAR) 1011067, # Arg 5 (from HAR) token, final_arg7, id_str, 11 ] print("Sending Login Request with args:") print(login_args) packet = create_amf_packet("SystemLogin.loginUser", login_args) resp = session.post(url, headers=headers, data=packet) print(f"Status Code: {resp.status_code}") # Print raw response for debugging print(f"Response Hex: {resp.content.hex()}") try: print(f"Response Text Preview: {resp.content[:100]}") except: pass # Parse Response Structure offset = 0 login_obj = None try: version = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 header_count = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 # Skip Headers (Assume 0 for now) body_count = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 print(f"AMF Version: {version}, Header Count: {header_count}, Body Count: {body_count}") for i in range(body_count): # Target URI t_len = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 target_uri = resp.content[offset:offset+t_len].decode('utf-8', errors='ignore') offset += t_len # Response URI r_len = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 response_uri = resp.content[offset:offset+r_len].decode('utf-8', errors='ignore') offset += r_len # Content Length content_len = struct.unpack(">I", resp.content[offset:offset+4])[0] offset += 4 print(f"Body {i+1}: Target='{target_uri}', Response='{response_uri}', Len={content_len}") # Body Content Marker marker = resp.content[offset] offset += 1 if marker == 0x11: obj, _ = parse_amf3_object(resp.content, offset) for k, v in obj.items(): print(f" {k}: {v}") status = obj.get("status") if status == 1: print("*** LOGIN SUCCESS ***") login_obj = obj else: print("*** LOGIN FAILED ***") elif marker == 0x03: # AMF0 Object print("Content is AMF0 Object") # Parse AMF0 Object properties while True: # Key length k_len = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 if k_len == 0: # Check for Object End (0x09) if resp.content[offset] == 0x09: offset += 1 break # End of Object key = resp.content[offset:offset+k_len].decode('utf-8') offset += k_len val_type = resp.content[offset] offset += 1 print(f" Key: {key}, Type: {hex(val_type)}") if val_type == 0x02: # String s_len = struct.unpack(">H", resp.content[offset:offset+2])[0] offset += 2 val = resp.content[offset:offset+s_len].decode('utf-8') offset += s_len print(f" Value: '{val}'") elif val_type == 0x00: # Number val = struct.unpack(">d", resp.content[offset:offset+8])[0] offset += 8 print(f" Value: {val}") elif val_type == 0x01: # Boolean val = resp.content[offset] offset += 1 print(f" Value: {val != 0}") else: print(f" Value: (Unknown Type {hex(val_type)})") break # Safety break except Exception as e: print(f"Parsing Error: {e}") if login_obj and login_obj.get("status") == 1: acc_id = login_obj.get("account_id") or login_obj.get("user_id") tok2 = login_obj.get("token") or token if acc_id and tok2: print("\n--- Step 3: getAllCharacters ---") packet2 = create_amf_packet("SystemLogin.getAllCharacters", [int(acc_id), str(tok2)]) resp2 = session.post(url, headers=headers, data=packet2) print(f"getAllCharacters Status: {resp2.status_code}") off = 0 try: ver = struct.unpack(">H", resp2.content[off:off+2])[0]; off += 2 hdr = struct.unpack(">H", resp2.content[off:off+2])[0]; off += 2 bc = struct.unpack(">H", resp2.content[off:off+2])[0]; off += 2 tlen = struct.unpack(">H", resp2.content[off:off+2])[0]; off += 2 off += tlen rlen = struct.unpack(">H", resp2.content[off:off+2])[0]; off += 2 off += rlen clen = struct.unpack(">I", resp2.content[off:off+4])[0]; off += 4 if resp2.content[off] == 0x11: off += 1 obj2, _ = parse_amf3_object(resp2.content, off) for k, v in obj2.items(): print(f" {k}: {v}") except Exception as e: print(f"Parse getAllCharacters error: {e}") else: print("Missing account_id/token; skipping character calls") except Exception as e: print(f"Error: {e}") def _start_ui_impl(): def do_login(): u = entry_user.get() p = entry_pass.get() btn.config(state="disabled") status_var.set("Memulai...") log("Memulai proses login...") def worker(): try: r = perform_login(u, p) status_var.set(r.get("message") or "") if r.get("status") == 1: status_var.set("LOGIN SUCCESS") log("LOGIN SUCCESS") if r.get("obj"): log(f"Obj: {r['obj']}") else: status_var.set(f"LOGIN FAILED: {r.get('result')}") log(f"LOGIN FAILED: {r.get('result')}") if r.get("raw_hex"): log("Response Hex:") log(r["raw_hex"]) except Exception as e: status_var.set(f"Error: {e}") log(f"Error: {e}") finally: btn.config(state="normal") threading.Thread(target=worker, daemon=True).start() root = tk.Tk() root.title("NinjaSage AMF Login") frame = ttk.Frame(root, padding=16) frame.pack(fill="both", expand=True) ttk.Label(frame, text="Username").grid(row=0, column=0, sticky="w") entry_user = ttk.Entry(frame, width=30) entry_user.grid(row=0, column=1, sticky="ew") ttk.Label(frame, text="Password").grid(row=1, column=0, sticky="w") entry_pass = ttk.Entry(frame, width=30, show="*") entry_pass.grid(row=1, column=1, sticky="ew") btn = ttk.Button(frame, text="Login", command=do_login) btn.grid(row=2, column=0, columnspan=2, pady=8) status_var = tk.StringVar() ttk.Label(frame, textvariable=status_var).grid(row=3, column=0, columnspan=2, sticky="w") log_box = scrolledtext.ScrolledText(frame, width=60, height=12) log_box.grid(row=4, column=0, columnspan=2, sticky="nsew", pady=8) def log(msg): log_box.insert("end", str(msg) + "\n") log_box.see("end") frame.columnconfigure(1, weight=1) frame.rowconfigure(4, weight=1) root.mainloop() def perform_login(username, password): url = "https://play.ninjasage.id/amf" headers = { "Content-Type": "application/x-amf", "User-Agent": "Mozilla/5.0 (Windows; U; en) AppleWebKit/533.19.4 (KHTML, like Gecko) AdobeAIR/51.1", "Referer": "app:/NinjaSage.swf", "x-flash-version": "51,1,3,10", "x-air-appid": "pta5yAteNUXGZU30lxDv+BRLP9xn3HZsknJnVEyQZU4=" } session = requests.Session() session.verify = False ver_args = ["Public 0.53.1"] packet = create_amf_packet("SystemLogin.checkVersion", ver_args) r1 = session.post(url, headers=headers, data=packet) token = None timestamp = None tm = r1.content.find(b'\x05\x5f\x5f\x06') if tm != -1: lb = r1.content[tm+4] tl = lb >> 1 token = r1.content[tm+5:tm+5+tl].decode('utf-8') ts = r1.content.find(b'\x03\x5f\x05') if ts != -1: db = r1.content[ts+3:ts+3+8] timestamp = struct.unpack(">d", db)[0] if not token or not timestamp: return {"status": 0, "message": "Gagal ambil token/timestamp"} pass_hash = base64.b64encode(hashlib.md5(password.encode('utf-8')).digest()).decode('utf-8') ts_str = str(int(timestamp)) static_hash = "40367c3cc999a9f9e951a1d33211545b84b2d5a6" final_arg7 = f"{ts_str}{static_hash}" id_str = "2078177199839932112023317301122772909" login_args = [ username, pass_hash, float(timestamp), 1011067, 1011067, token, final_arg7, id_str, 11 ] packet2 = create_amf_packet("SystemLogin.loginUser", login_args) r2 = session.post(url, headers=headers, data=packet2) off = 0 try: struct.unpack(">H", r2.content[off:off+2])[0]; off += 2 struct.unpack(">H", r2.content[off:off+2])[0]; off += 2 bc = struct.unpack(">H", r2.content[off:off+2])[0]; off += 2 for _ in range(bc): tl = struct.unpack(">H", r2.content[off:off+2])[0]; off += 2 off += tl rl = struct.unpack(">H", r2.content[off:off+2])[0]; off += 2 off += rl cl = struct.unpack(">I", r2.content[off:off+4])[0]; off += 4 m = r2.content[off]; off += 1 if m == 0x11: obj, _ = parse_amf3_object(r2.content, off) s = obj.get("status") return {"status": s or 0, "result": obj.get("result"), "error": obj.get("error"), "obj": obj, "raw_hex": r2.content.hex()} except: return {"status": 0, "message": "Gagal parsing response", "raw_hex": r2.content.hex()} return {"status": 0, "message": "Tidak ada body response", "raw_hex": r2.content.hex()} if __name__ == "__main__": main()