164 lines
5.9 KiB
Python
164 lines
5.9 KiB
Python
"""
|
|
OpenWrt LuCI Firmware Flasher (Barrier Breaker 14.07)
|
|
|
|
Automates the 3-step firmware flash process via LuCI web interface:
|
|
|
|
Step 1: POST /cgi-bin/luci → username=root&password= → get sysauth cookie + stok
|
|
Step 2: POST /cgi-bin/luci/;stok=XXX/admin/system/flashops
|
|
→ multipart: image=firmware.bin → get verification page
|
|
Step 3: POST /cgi-bin/luci/;stok=XXX/admin/system/flashops
|
|
→ step=2&keep= → device flashes and reboots
|
|
"""
|
|
|
|
import requests
|
|
import re
|
|
import time
|
|
import os
|
|
|
|
|
|
def flash_device(ip, firmware_path, username="root", password="",
|
|
keep_settings=False, status_cb=None):
|
|
"""
|
|
Flash firmware to an OpenWrt device via LuCI.
|
|
|
|
Returns:
|
|
"DONE" on success, "FAIL: reason" on error
|
|
"""
|
|
base_url = f"http://{ip}"
|
|
session = requests.Session()
|
|
|
|
try:
|
|
# ═══════════════════════════════════════════
|
|
# STEP 1: Login
|
|
# ═══════════════════════════════════════════
|
|
if status_cb:
|
|
status_cb("Logging in...")
|
|
|
|
# GET login page to detect form field names
|
|
login_url = f"{base_url}/cgi-bin/luci"
|
|
try:
|
|
get_resp = session.get(login_url, timeout=10)
|
|
page_html = get_resp.text
|
|
except Exception:
|
|
page_html = ""
|
|
|
|
# Barrier Breaker uses "username"/"password"
|
|
# Newer LuCI uses "luci_username"/"luci_password"
|
|
if 'name="luci_username"' in page_html:
|
|
login_data = {"luci_username": username, "luci_password": password}
|
|
else:
|
|
login_data = {"username": username, "password": password}
|
|
|
|
resp = session.post(login_url, data=login_data,
|
|
timeout=10, allow_redirects=True)
|
|
|
|
if resp.status_code == 403:
|
|
return "FAIL: Login denied (403)"
|
|
|
|
# Extract stok from response body or URL
|
|
stok = None
|
|
for source in [resp.url, resp.text]:
|
|
match = re.search(r";stok=([a-f0-9]+)", source)
|
|
if match:
|
|
stok = match.group(1)
|
|
break
|
|
|
|
# Also check redirect history
|
|
if not stok:
|
|
for hist in resp.history:
|
|
match = re.search(r";stok=([a-f0-9]+)",
|
|
hist.headers.get("Location", ""))
|
|
if match:
|
|
stok = match.group(1)
|
|
break
|
|
|
|
# Verify login succeeded
|
|
has_cookie = "sysauth" in str(session.cookies)
|
|
if not stok and not has_cookie:
|
|
return "FAIL: Login failed — no session"
|
|
|
|
# Build flash URL with stok
|
|
if stok:
|
|
flash_url = f"{base_url}/cgi-bin/luci/;stok={stok}/admin/system/flashops"
|
|
else:
|
|
flash_url = f"{base_url}/cgi-bin/luci/admin/system/flashops"
|
|
|
|
# ═══════════════════════════════════════════
|
|
# STEP 2: Upload firmware image
|
|
# ═══════════════════════════════════════════
|
|
if status_cb:
|
|
status_cb("Uploading firmware...")
|
|
|
|
filename = os.path.basename(firmware_path)
|
|
with open(firmware_path, "rb") as f:
|
|
# Don't send "keep" field = uncheck "Keep settings"
|
|
# Send "keep": "on" only if keep_settings is True
|
|
data = {}
|
|
if keep_settings:
|
|
data["keep"] = "on"
|
|
|
|
resp = session.post(
|
|
flash_url,
|
|
data=data,
|
|
files={"image": (filename, f, "application/octet-stream")},
|
|
timeout=300,
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
return f"FAIL: Upload HTTP {resp.status_code}"
|
|
|
|
resp_lower = resp.text.lower()
|
|
|
|
# Check for errors
|
|
if "invalid image" in resp_lower or "bad image" in resp_lower:
|
|
return "FAIL: Invalid firmware image"
|
|
if "unsupported" in resp_lower or "not compatible" in resp_lower:
|
|
return "FAIL: Firmware not compatible"
|
|
|
|
# Verify we got the "Flash Firmware - Verify" page
|
|
if "verify" not in resp_lower or "proceed" not in resp_lower:
|
|
# Still on flash form = upload was ignored
|
|
if 'name="image"' in resp.text:
|
|
return "FAIL: Upload ignored by server"
|
|
return "FAIL: Unexpected response after upload"
|
|
|
|
# ═══════════════════════════════════════════
|
|
# STEP 3: Click "Proceed" to start flash
|
|
# ═══════════════════════════════════════════
|
|
if status_cb:
|
|
status_cb("Confirming (Proceed)...")
|
|
|
|
# The Proceed form from the verification page:
|
|
# <input type="hidden" name="step" value="2" />
|
|
# <input type="hidden" name="keep" value="" />
|
|
confirm_data = {
|
|
"step": "2",
|
|
"keep": "",
|
|
}
|
|
|
|
if keep_settings:
|
|
confirm_data["keep"] = "on"
|
|
|
|
try:
|
|
session.post(flash_url, data=confirm_data, timeout=300)
|
|
except requests.exceptions.ConnectionError:
|
|
# Device rebooting — this is expected and means SUCCESS
|
|
pass
|
|
except requests.exceptions.ReadTimeout:
|
|
# Also expected during reboot
|
|
pass
|
|
|
|
if status_cb:
|
|
status_cb("Rebooting...")
|
|
time.sleep(3)
|
|
|
|
return "DONE"
|
|
|
|
except requests.ConnectionError:
|
|
return "FAIL: Cannot connect"
|
|
except requests.Timeout:
|
|
return "DONE (rebooting)"
|
|
except Exception as e:
|
|
return f"FAIL: {e}"
|
|
finally:
|
|
session.close() |