first commit
This commit is contained in:
35
.gitignore
vendored
Normal file
35
.gitignore
vendored
Normal file
@@ -0,0 +1,35 @@
|
||||
# ── Python ──
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.pyo
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
*.egg
|
||||
|
||||
# ── Virtual Environment ──
|
||||
venv/
|
||||
.venv/
|
||||
env/
|
||||
|
||||
# ── IDE ──
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
*~
|
||||
.DS_Store
|
||||
|
||||
# ── PyInstaller Build ──
|
||||
dist/
|
||||
build/
|
||||
*.spec
|
||||
|
||||
# ── Logs ──
|
||||
*.log
|
||||
flash_log.txt
|
||||
|
||||
# ── Firmware files (optional, uncomment if needed) ──
|
||||
# *.bin
|
||||
# *.hex
|
||||
# *.uf2
|
||||
BIN
IoT_Firmware_Loader.exe
Normal file
BIN
IoT_Firmware_Loader.exe
Normal file
Binary file not shown.
255
README.md
Normal file
255
README.md
Normal file
@@ -0,0 +1,255 @@
|
||||
# ⚡ IoT Firmware Loader
|
||||
|
||||
Công cụ desktop dùng để **scan, phát hiện và flash firmware hàng loạt** cho các thiết bị IoT (ESP32/ESP8266) trong mạng LAN.
|
||||
|
||||
> **Tech stack:** Python 3.9+ · PyQt6 · Scapy · Requests
|
||||
|
||||
---
|
||||
|
||||
## 📁 Cấu trúc dự án
|
||||
|
||||
```
|
||||
iot_fw_loader/
|
||||
├── main.py # UI chính (PyQt6) + điều phối toàn bộ luồng
|
||||
├── scanner.py # Quét thiết bị trong mạng LAN
|
||||
├── device_filter.py # Lọc thiết bị IoT theo MAC vendor
|
||||
├── flasher.py # Upload firmware qua HTTP
|
||||
├── run.sh # Script khởi chạy (macOS/Linux)
|
||||
├── run.bat # Script khởi chạy (Windows)
|
||||
└── venv/ # Python virtual environment
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🔧 Cài đặt & Chạy
|
||||
|
||||
### Yêu cầu
|
||||
|
||||
- Python 3.9+
|
||||
- Các thư viện: `PyQt6`, `scapy`, `requests`
|
||||
|
||||
### Khởi chạy nhanh
|
||||
|
||||
**macOS / Linux:**
|
||||
|
||||
```bash
|
||||
./run.sh
|
||||
```
|
||||
|
||||
**Windows:**
|
||||
|
||||
```bat
|
||||
run.bat
|
||||
```
|
||||
|
||||
> Script tự tạo `venv` và cài dependencies nếu chưa có.
|
||||
|
||||
**Hoặc chạy thủ công:**
|
||||
|
||||
```bash
|
||||
source venv/bin/activate
|
||||
python main.py
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🏗 Kiến trúc hệ thống
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────┐
|
||||
│ main.py (UI) │
|
||||
│ ┌───────────┐ ┌───────────┐ ┌────────────────────┐ │
|
||||
│ │ Machine │ │ Firmware │ │ Network Scan │ │
|
||||
│ │ Info │ │ Selector │ │ (QThread) │ │
|
||||
│ └───────────┘ └───────────┘ └────────┬───────────┘ │
|
||||
│ ┌────────────────────────────────────┼───────────┐ │
|
||||
│ │ Device Table │ │ │
|
||||
│ │ IP │ MAC │ Type │ Status │ │ │
|
||||
│ └────────────────────────────────────┼───────────┘ │
|
||||
│ ┌────────────────────────────────────┼───────────┐ │
|
||||
│ │ Flash Controls + Progress Bar │ │ │
|
||||
│ │ (ThreadPoolExecutor × 5) │ │ │
|
||||
│ └────────────────────────────────────┼───────────┘ │
|
||||
└───────────────────────────────────────┼─────────────┘
|
||||
│
|
||||
┌───────────────────┼───────────────────┐
|
||||
│ │ │
|
||||
┌─────▼─────┐ ┌──────▼──────┐ ┌──────▼──────┐
|
||||
│ scanner.py │ │device_filter│ │ flasher.py │
|
||||
│ │ │ .py │ │ │
|
||||
│ Scapy ARP │ │ MAC vendor │ │ HTTP POST │
|
||||
│ ↓ fallback│ │ matching │ │ /update │
|
||||
│ arp -a │ │ │ │ │
|
||||
└────────────┘ └─────────────┘ └─────────────┘
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🔄 Luồng hoạt động chi tiết
|
||||
|
||||
### Tổng quan
|
||||
|
||||
```
|
||||
Khởi động App ──▶ Hiển thị Machine Info
|
||||
│
|
||||
▼
|
||||
Chọn Firmware (.bin/.hex/.uf2)
|
||||
│
|
||||
▼
|
||||
Nhập dải mạng ──▶ Nhấn "Scan LAN"
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────┐
|
||||
│ ScanThread (background) │
|
||||
│ 1. Thử Scapy ARP broadcast │
|
||||
│ 2. Nếu lỗi → fallback arp -a │
|
||||
└──────────────┬──────────────────┘
|
||||
▼
|
||||
Lọc IoT device theo MAC vendor
|
||||
│
|
||||
▼
|
||||
Hiển thị bảng: tất cả thiết bị
|
||||
(IoT = 🟢 xanh lá, Other = ⚪)
|
||||
│
|
||||
▼
|
||||
Nhấn "Flash All IoT Devices"
|
||||
│
|
||||
▼
|
||||
┌──────────────────────────────────┐
|
||||
│ ThreadPoolExecutor (5 workers) │
|
||||
│ POST firmware → mỗi thiết bị │
|
||||
│ Cập nhật status + progress bar │
|
||||
└──────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
Hoàn tất ✅
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📋 Phân tích từng module
|
||||
|
||||
### 1. `main.py` — Giao diện & Điều phối
|
||||
|
||||
| Thành phần | Mô tả |
|
||||
| ------------------------- | --------------------------------------------- |
|
||||
| `get_local_ip()` | Lấy IP máy tính qua UDP socket tới `8.8.8.8` |
|
||||
| `get_default_network(ip)` | Tự tính dải mạng `/24` từ IP máy |
|
||||
| `get_machine_info()` | Thu thập hostname, IP, OS, MAC |
|
||||
| `ScanThread` | `QThread` chạy scan ở background, tránh đơ UI |
|
||||
| `App` | Widget chính chứa toàn bộ UI |
|
||||
|
||||
**Các phần UI:**
|
||||
|
||||
| Group Box | Chức năng |
|
||||
| ---------------- | -------------------------------------- |
|
||||
| 🖥 Machine Info | Hiển thị Hostname, IP, OS, MAC của máy |
|
||||
| 📦 Firmware | Chọn file firmware |
|
||||
| 📡 Network Scan | Nhập dải mạng + nút Scan |
|
||||
| 📋 Devices Found | Bảng thiết bị (IP, MAC, Type, Status) |
|
||||
| 🚀 Flash | Progress bar + nút Flash All |
|
||||
|
||||
**Xử lý sự kiện chính:**
|
||||
|
||||
| Method | Trigger | Hành động |
|
||||
| ------------------ | ------------------ | ------------------------------------------ |
|
||||
| `select_fw()` | Nhấn "Select File" | Mở dialog chọn firmware |
|
||||
| `scan()` | Nhấn "Scan LAN" | Tạo `ScanThread` → chạy background |
|
||||
| `_on_scan_done()` | Scan hoàn tất | Phân loại IoT/Other → hiển thị bảng |
|
||||
| `_on_scan_error()` | Scan lỗi | Hiện `QMessageBox` lỗi |
|
||||
| `flash_all()` | Nhấn "Flash All" | Tạo `ThreadPoolExecutor` → flash song song |
|
||||
| `flash_worker()` | Mỗi thread | POST firmware → cập nhật status |
|
||||
|
||||
---
|
||||
|
||||
### 2. `scanner.py` — Quét mạng
|
||||
|
||||
**Chiến lược 2 lớp (dual-layer scan):**
|
||||
|
||||
| Phương pháp | Hàm | Yêu cầu | Độ chính xác |
|
||||
| ----------------------- | ------------------------ | -------------- | ------------------------------------------- |
|
||||
| **Primary:** Scapy ARP | `_scan_with_scapy()` | Root/sudo | Cao — scan active |
|
||||
| **Fallback:** ARP table | `_scan_with_arp_table()` | Không cần root | Trung bình — chỉ thấy thiết bị đã giao tiếp |
|
||||
|
||||
**Luồng trong `scan_network()`:**
|
||||
|
||||
```python
|
||||
try:
|
||||
return _scan_with_scapy(network) # Cần root
|
||||
except:
|
||||
return _scan_with_arp_table(network) # Fallback, parse "arp -a"
|
||||
```
|
||||
|
||||
**Scapy ARP scan:**
|
||||
|
||||
- Tạo ARP request → broadcast `ff:ff:ff:ff:ff:ff`
|
||||
- Thu thập response → trích IP + MAC
|
||||
- Timeout: 2 giây
|
||||
|
||||
**ARP table fallback:**
|
||||
|
||||
- Chạy lệnh `arp -a` của hệ thống
|
||||
- Parse output bằng regex: `\(IP\) at MAC`
|
||||
- Lọc theo dải mạng đầu vào
|
||||
|
||||
---
|
||||
|
||||
### 3. `device_filter.py` — Phân loại thiết bị
|
||||
|
||||
Lọc thiết bị IoT dựa trên **3 byte đầu của MAC address** (OUI — Organizationally Unique Identifier):
|
||||
|
||||
| MAC Prefix | Vendor |
|
||||
| ---------- | ------------------------- |
|
||||
| `24:6F:28` | Espressif (ESP32/ESP8266) |
|
||||
| `84:F3:EB` | Espressif |
|
||||
| `DC:4F:22` | Espressif |
|
||||
|
||||
> **Lưu ý:** Chỉ thiết bị có MAC bắt đầu bằng các prefix trên mới được đánh dấu là IoT. Tất cả thiết bị khác vẫn hiển thị trong bảng với label "Other".
|
||||
|
||||
---
|
||||
|
||||
### 4. `flasher.py` — Nạp firmware
|
||||
|
||||
| Bước | Chi tiết |
|
||||
| ------------- | --------------------------------------------------- |
|
||||
| 1. Mở file | `open(firmware_path, "rb")` |
|
||||
| 2. Upload | `POST http://{ip}/update` với `multipart/form-data` |
|
||||
| 3. Chờ reboot | `time.sleep(3)` — đợi thiết bị khởi động lại |
|
||||
| 4. Kết quả | Trả `"DONE"` hoặc `"FAIL"` |
|
||||
|
||||
> **Giả định:** Thiết bị IoT chạy HTTP server với endpoint `/update` hỗ trợ OTA firmware update (chuẩn ESP32 Arduino OTA).
|
||||
|
||||
---
|
||||
|
||||
## ⚙️ Cấu hình
|
||||
|
||||
| Tham số | Giá trị | Vị trí |
|
||||
| ----------------- | ----------------------- | ----------------------------------------------- |
|
||||
| Network range | Tự tính `/24` từ IP máy | `main.py` → `get_default_network()` |
|
||||
| Scan timeout | 2 giây | `scanner.py` → `srp(..., timeout=2)` |
|
||||
| Flash timeout | 10 giây | `flasher.py` → `requests.post(..., timeout=10)` |
|
||||
| Reboot wait | 3 giây | `flasher.py` → `time.sleep(3)` |
|
||||
| Max flash workers | 5 | `main.py` → `ThreadPoolExecutor(max_workers=5)` |
|
||||
| Firmware filter | `.bin .hex .uf2` | `main.py` → `QFileDialog` filter |
|
||||
|
||||
---
|
||||
|
||||
## 🔒 Lưu ý bảo mật & Quyền
|
||||
|
||||
- **Scapy cần root/sudo** trên macOS/Linux để gửi raw ARP packets
|
||||
- Nếu không có quyền root → tự động dùng `arp -a` (không cần root nhưng chỉ thấy thiết bị đã giao tiếp gần đây)
|
||||
- Firmware upload qua **HTTP không mã hóa** — chỉ nên dùng trong mạng LAN nội bộ
|
||||
|
||||
---
|
||||
|
||||
## 📊 Trạng thái thiết bị (State Machine)
|
||||
|
||||
```
|
||||
READY ──────▶ FLASHING
|
||||
│ │
|
||||
│ ┌─────┴──────┐
|
||||
│ ▼ ▼
|
||||
│ ✅ DONE ❌ FAIL
|
||||
│
|
||||
└── (thiết bị Other: trạng thái "—", không flash)
|
||||
```
|
||||
59
build_windows.bat
Normal file
59
build_windows.bat
Normal file
@@ -0,0 +1,59 @@
|
||||
@echo off
|
||||
REM ============================================
|
||||
REM IoT Firmware Loader - Windows Build Script
|
||||
REM Tao file .exe khong can cai Python
|
||||
REM ============================================
|
||||
|
||||
echo ========================================
|
||||
echo IoT Firmware Loader - Build for Windows
|
||||
echo ========================================
|
||||
echo.
|
||||
|
||||
cd /d "%~dp0"
|
||||
|
||||
REM 1. Tao venv neu chua co
|
||||
if not exist "venv" (
|
||||
echo [1/4] Creating virtual environment...
|
||||
python -m venv venv
|
||||
)
|
||||
|
||||
REM 2. Activate venv
|
||||
call venv\Scripts\activate.bat
|
||||
|
||||
REM 3. Cai dependencies + PyInstaller
|
||||
echo [2/4] Installing dependencies...
|
||||
pip install PyQt6 scapy requests pyinstaller --quiet
|
||||
|
||||
REM 4. Build .exe
|
||||
echo [3/4] Building executable...
|
||||
pyinstaller ^
|
||||
--name "IoT_Firmware_Loader" ^
|
||||
--onefile ^
|
||||
--windowed ^
|
||||
--noconfirm ^
|
||||
--clean ^
|
||||
--hidden-import "scapy.all" ^
|
||||
--hidden-import "scapy.layers.l2" ^
|
||||
--hidden-import "scapy.arch.windows" ^
|
||||
--hidden-import "PyQt6.QtWidgets" ^
|
||||
--hidden-import "PyQt6.QtCore" ^
|
||||
--hidden-import "PyQt6.QtGui" ^
|
||||
main.py
|
||||
|
||||
echo.
|
||||
echo [4/4] Build complete!
|
||||
echo.
|
||||
|
||||
if exist "dist\IoT_Firmware_Loader.exe" (
|
||||
echo ✅ SUCCESS: dist\IoT_Firmware_Loader.exe
|
||||
echo.
|
||||
echo File size:
|
||||
for %%A in ("dist\IoT_Firmware_Loader.exe") do echo %%~zA bytes
|
||||
echo.
|
||||
echo Ban co the copy file .exe nay sang may khac va chay truc tiep.
|
||||
) else (
|
||||
echo ❌ BUILD FAILED. Check errors above.
|
||||
)
|
||||
|
||||
echo.
|
||||
pause
|
||||
205
debug_full.py
Normal file
205
debug_full.py
Normal file
@@ -0,0 +1,205 @@
|
||||
"""
|
||||
Debug Flash — chạy đầy đủ 3 bước flash và log chi tiết từng bước.
|
||||
|
||||
Usage:
|
||||
python debug_full.py <IP> <firmware.bin>
|
||||
python debug_full.py 192.168.11.17 V3.0.6p5.bin
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import requests
|
||||
import re
|
||||
|
||||
|
||||
def debug_flash(ip, firmware_path, username="root", password=""):
|
||||
base_url = f"http://{ip}"
|
||||
session = requests.Session()
|
||||
|
||||
print(f"{'='*65}")
|
||||
print(f" Full Flash Debug — {ip}")
|
||||
print(f" Firmware: {os.path.basename(firmware_path)}")
|
||||
print(f" Size: {os.path.getsize(firmware_path) / 1024 / 1024:.2f} MB")
|
||||
print(f"{'='*65}")
|
||||
|
||||
# ═══════════════════════════════════
|
||||
# STEP 1: Login
|
||||
# ═══════════════════════════════════
|
||||
print(f"\n{'─'*65}")
|
||||
print(f" STEP 1: Login")
|
||||
print(f"{'─'*65}")
|
||||
|
||||
login_url = f"{base_url}/cgi-bin/luci"
|
||||
|
||||
print(f" → GET {login_url}")
|
||||
try:
|
||||
r = session.get(login_url, timeout=10)
|
||||
print(f" Status: {r.status_code}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Cannot connect: {e}")
|
||||
return
|
||||
|
||||
# Detect form fields
|
||||
if 'name="luci_username"' in r.text:
|
||||
login_data = {"luci_username": username, "luci_password": password}
|
||||
print(f" Fields: luci_username / luci_password")
|
||||
else:
|
||||
login_data = {"username": username, "password": password}
|
||||
print(f" Fields: username / password")
|
||||
|
||||
print(f"\n → POST {login_url}")
|
||||
print(f" Data: {login_data}")
|
||||
resp = session.post(login_url, data=login_data,
|
||||
timeout=10, allow_redirects=True)
|
||||
|
||||
print(f" Status: {resp.status_code}")
|
||||
print(f" Cookies: {dict(session.cookies)}")
|
||||
|
||||
# Extract stok
|
||||
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
|
||||
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
|
||||
|
||||
print(f" stok: {stok}")
|
||||
has_cookie = "sysauth" in str(session.cookies)
|
||||
print(f" sysauth: {'✅ YES' if has_cookie else '❌ NO'}")
|
||||
|
||||
if not stok and not has_cookie:
|
||||
print(f"\n ❌ LOGIN FAILED — no stok, no sysauth cookie")
|
||||
return
|
||||
|
||||
print(f"\n ✅ LOGIN OK")
|
||||
|
||||
# Build flash URL
|
||||
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
|
||||
# ═══════════════════════════════════
|
||||
print(f"\n{'─'*65}")
|
||||
print(f" STEP 2: Upload firmware")
|
||||
print(f"{'─'*65}")
|
||||
|
||||
print(f" → POST {flash_url}")
|
||||
print(f" File field: image")
|
||||
print(f" File name: {os.path.basename(firmware_path)}")
|
||||
print(f" keep: NOT sent (unchecked)")
|
||||
print(f" Uploading...")
|
||||
|
||||
filename = os.path.basename(firmware_path)
|
||||
with open(firmware_path, "rb") as f:
|
||||
resp = session.post(
|
||||
flash_url,
|
||||
data={}, # No keep = uncheck Keep Settings
|
||||
files={"image": (filename, f, "application/octet-stream")},
|
||||
timeout=300,
|
||||
)
|
||||
|
||||
print(f" Status: {resp.status_code}")
|
||||
print(f" URL: {resp.url}")
|
||||
|
||||
# Check response content
|
||||
resp_lower = resp.text.lower()
|
||||
has_verify = "verify" in resp_lower
|
||||
has_proceed = "proceed" in resp_lower
|
||||
has_checksum = "checksum" in resp_lower
|
||||
has_image_form = 'name="image"' in resp.text and 'type="file"' in resp.text
|
||||
|
||||
print(f"\n Response analysis:")
|
||||
print(f" Has 'verify': {'✅' if has_verify else '❌'}")
|
||||
print(f" Has 'proceed': {'✅' if has_proceed else '❌'}")
|
||||
print(f" Has 'checksum': {'✅' if has_checksum else '❌'}")
|
||||
print(f" Has flash form: {'⚠️ (upload ignored!)' if has_image_form else '✅ (not flash form)'}")
|
||||
|
||||
# Extract checksum from verification page
|
||||
checksum = re.search(r'Checksum:\s*<code>([a-f0-9]+)</code>', resp.text)
|
||||
size_info = re.search(r'Size:\s*([\d.]+\s*MB)', resp.text)
|
||||
if checksum:
|
||||
print(f" Checksum: {checksum.group(1)}")
|
||||
if size_info:
|
||||
print(f" Size: {size_info.group(1)}")
|
||||
|
||||
# Check for keep settings status
|
||||
if "will be erased" in resp.text:
|
||||
print(f" Config: Will be ERASED ✅ (keep=off)")
|
||||
elif "will be kept" in resp.text:
|
||||
print(f" Config: Will be KEPT ⚠️ (keep=on)")
|
||||
|
||||
if not has_verify or not has_proceed:
|
||||
print(f"\n ❌ Did NOT get verification page!")
|
||||
# Show cleaned response
|
||||
text = re.sub(r'<[^>]+>', ' ', resp.text)
|
||||
text = re.sub(r'\s+', ' ', text).strip()
|
||||
print(f" Response: {text[:500]}")
|
||||
return
|
||||
|
||||
# Show the Proceed form
|
||||
print(f"\n Proceed form (from HTML):")
|
||||
forms = re.findall(r'<form[^>]*>(.*?)</form>', resp.text, re.DOTALL)
|
||||
for i, form_body in enumerate(forms):
|
||||
if 'value="Proceed"' in form_body or 'step' in form_body:
|
||||
inputs = re.findall(r'<input[^>]*/?\s*>', form_body)
|
||||
for inp in inputs:
|
||||
print(f" {inp.strip()}")
|
||||
|
||||
print(f"\n ✅ UPLOAD OK — Verification page received")
|
||||
|
||||
# ═══════════════════════════════════
|
||||
# STEP 3: Proceed (confirm flash)
|
||||
# ═══════════════════════════════════
|
||||
print(f"\n{'─'*65}")
|
||||
print(f" STEP 3: Proceed (confirm flash)")
|
||||
print(f"{'─'*65}")
|
||||
|
||||
confirm_data = {
|
||||
"step": "2",
|
||||
"keep": "",
|
||||
}
|
||||
|
||||
print(f" → POST {flash_url}")
|
||||
print(f" Data: {confirm_data}")
|
||||
|
||||
try:
|
||||
resp = session.post(flash_url, data=confirm_data, timeout=300)
|
||||
print(f" Status: {resp.status_code}")
|
||||
# Show cleaned response
|
||||
text = re.sub(r'<[^>]+>', ' ', resp.text)
|
||||
text = re.sub(r'\s+', ' ', text).strip()
|
||||
print(f" Response: {text[:300]}")
|
||||
except requests.ConnectionError:
|
||||
print(f" ✅ Connection dropped — device is REBOOTING!")
|
||||
except requests.ReadTimeout:
|
||||
print(f" ✅ Timeout — device is REBOOTING!")
|
||||
|
||||
print(f"\n{'='*65}")
|
||||
print(f" 🎉 FLASH COMPLETE")
|
||||
print(f"{'='*65}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 3:
|
||||
print("Usage: python debug_full.py <IP> <firmware.bin>")
|
||||
print("Example: python debug_full.py 192.168.11.17 V3.0.6p5.bin")
|
||||
sys.exit(1)
|
||||
|
||||
ip = sys.argv[1]
|
||||
fw = sys.argv[2]
|
||||
|
||||
if not os.path.exists(fw):
|
||||
print(f"❌ File not found: {fw}")
|
||||
sys.exit(1)
|
||||
|
||||
debug_flash(ip, fw)
|
||||
155
doc/FLASH_DOC.md
Normal file
155
doc/FLASH_DOC.md
Normal file
@@ -0,0 +1,155 @@
|
||||
# Tài liệu Flash Firmware — IoT Firmware Loader
|
||||
|
||||
## Tổng quan
|
||||
|
||||
Ứng dụng tự động hóa quá trình nạp firmware cho thiết bị **OpenWrt Barrier Breaker 14.07** thông qua giao diện web **LuCI**. Thay vì thao tác thủ công trên trình duyệt, ứng dụng thực hiện 3 bước HTTP tự động cho mỗi thiết bị.
|
||||
|
||||
---
|
||||
|
||||
## Luồng hoạt động
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A[Chọn Firmware .bin] --> B[Scan LAN]
|
||||
B --> C[Hiển thị danh sách thiết bị]
|
||||
C --> D{Chọn thiết bị ☑}
|
||||
D --> E[Nhấn Flash Selected Devices]
|
||||
E --> F[FlashThread chạy background]
|
||||
F --> G["ThreadPoolExecutor\n(max_workers = N)"]
|
||||
G --> H1[Device 1 → flash_device]
|
||||
G --> H2[Device 2 → flash_device]
|
||||
G --> H3[Device N → flash_device]
|
||||
H1 & H2 & H3 --> I[All Done → Thông báo]
|
||||
```
|
||||
|
||||
### Chi tiết `flash_device()` cho mỗi thiết bị
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
S1["STEP 1: Login\nGET /cgi-bin/luci\nPOST username=root, password=empty"] --> C1{Thành công?}
|
||||
C1 -->|Có cookie sysauth + stok| S2
|
||||
C1 -->|403 Denied| F1["FAIL: Login denied (403)"]
|
||||
C1 -->|Không có session| F2["FAIL: Login failed — no session"]
|
||||
|
||||
S2["STEP 2: Upload Firmware\nPOST /flashops\nField: image=firmware.bin\nkeep=KHÔNG gửi (bỏ tích)"] --> C2{Response?}
|
||||
C2 -->|Trang Verify + Proceed| S3
|
||||
C2 -->|HTTP ≠ 200| F3["FAIL: Upload HTTP xxx"]
|
||||
C2 -->|invalid image| F4["FAIL: Invalid firmware image"]
|
||||
C2 -->|unsupported| F5["FAIL: Firmware not compatible"]
|
||||
C2 -->|Vẫn hiện form upload| F6["FAIL: Upload ignored by server"]
|
||||
|
||||
S3["STEP 3: Proceed\nPOST step=2, keep=empty\nXác nhận flash"] --> C3{Response?}
|
||||
C3 -->|200 Flashing...| R["DONE ✅\nThiết bị đang reboot"]
|
||||
C3 -->|Connection dropped| R
|
||||
C3 -->|Read timeout| R
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Bảng Status
|
||||
|
||||
### Status trên cột "Status" trong bảng thiết bị
|
||||
|
||||
| Icon | Status | Điều kiện hiển thị |
|
||||
| ---- | ---------------------------------------- | ------------------------------------------------------------- |
|
||||
| — | `READY` | Sau khi scan, thiết bị chưa được flash |
|
||||
| ⏳ | `Logging in...` | Đang POST login vào LuCI |
|
||||
| ⏳ | `Uploading firmware...` | Đang upload file .bin (~30MB) lên thiết bị |
|
||||
| ⏳ | `Confirming (Proceed)...` | Đã upload xong, đang gửi lệnh xác nhận flash |
|
||||
| ⏳ | `Rebooting...` | Thiết bị đang reboot sau khi flash |
|
||||
| ✅ | `DONE` | Flash thành công, thiết bị đang khởi động lại |
|
||||
| ✅ | `DONE (rebooting)` | Flash thành công nhưng timeout khi chờ response (bình thường) |
|
||||
| ❌ | `FAIL: Cannot connect` | Không kết nối được tới thiết bị (sai IP, khác mạng) |
|
||||
| ❌ | `FAIL: Login denied (403)` | Thiết bị từ chối đăng nhập (sai mật khẩu) |
|
||||
| ❌ | `FAIL: Login failed — no session` | Login không trả về cookie hoặc token |
|
||||
| ❌ | `FAIL: Upload HTTP xxx` | Server trả mã lỗi HTTP khi upload |
|
||||
| ❌ | `FAIL: Invalid firmware image` | File firmware không hợp lệ |
|
||||
| ❌ | `FAIL: Firmware not compatible` | Firmware không tương thích với thiết bị |
|
||||
| ❌ | `FAIL: Upload ignored by server` | Server nhận file nhưng không xử lý (sai form field) |
|
||||
| ❌ | `FAIL: Unexpected response after upload` | Không nhận được trang xác nhận Verify |
|
||||
|
||||
---
|
||||
|
||||
## Chi tiết kỹ thuật HTTP
|
||||
|
||||
### Step 1: Login
|
||||
|
||||
```
|
||||
GET http://{IP}/cgi-bin/luci → Lấy trang login, phát hiện field name
|
||||
POST http://{IP}/cgi-bin/luci → Gửi username=root&password=
|
||||
```
|
||||
|
||||
| Kết quả | Điều kiện |
|
||||
| -------------- | ------------------------------------------------------------------ |
|
||||
| **Thành công** | Response chứa `sysauth` cookie VÀ/HOẶC `stok` token trong URL/body |
|
||||
| **Thất bại** | HTTP 403, hoặc không có cookie/token |
|
||||
|
||||
**Field names tự động phát hiện:**
|
||||
|
||||
- OpenWrt Barrier Breaker 14.07: `username` / `password`
|
||||
- OpenWrt mới hơn: `luci_username` / `luci_password`
|
||||
|
||||
### Step 2: Upload Firmware
|
||||
|
||||
```
|
||||
POST http://{IP}/cgi-bin/luci/;stok={TOKEN}/admin/system/flashops
|
||||
Content-Type: multipart/form-data
|
||||
|
||||
Fields:
|
||||
image = [firmware.bin] (file upload)
|
||||
keep = (KHÔNG gửi) (bỏ tích Keep Settings)
|
||||
```
|
||||
|
||||
| Kết quả | Điều kiện |
|
||||
| -------------- | ----------------------------------------------------------------------- |
|
||||
| **Thành công** | Response chứa "Flash Firmware - Verify" + "Proceed" + checksum |
|
||||
| **Thất bại** | Response chứa "invalid image", "unsupported", hoặc vẫn hiện form upload |
|
||||
|
||||
### Step 3: Proceed (Xác nhận)
|
||||
|
||||
```
|
||||
POST http://{IP}/cgi-bin/luci/;stok={TOKEN}/admin/system/flashops
|
||||
|
||||
Fields:
|
||||
step = 2
|
||||
keep = (empty string)
|
||||
```
|
||||
|
||||
| Kết quả | Điều kiện |
|
||||
| -------------- | ------------------------------------------------------------------------------- |
|
||||
| **Thành công** | Response "The system is flashing now" HOẶC connection bị ngắt (thiết bị reboot) |
|
||||
| **Thất bại** | Hiếm khi xảy ra — nếu đã qua Step 2 thì Step 3 gần như luôn thành công |
|
||||
|
||||
---
|
||||
|
||||
## Xử lý song song
|
||||
|
||||
```
|
||||
FlashThread (QThread - background)
|
||||
└── ThreadPoolExecutor (max_workers = N)
|
||||
├── Thread 1 → flash_device(IP_1)
|
||||
├── Thread 2 → flash_device(IP_2)
|
||||
├── ...
|
||||
└── Thread N → flash_device(IP_N)
|
||||
```
|
||||
|
||||
| Config | Giá trị | Ý nghĩa |
|
||||
| ------------------------- | --------- | ------------------------------ |
|
||||
| Concurrent devices = `10` | Mặc định | Flash 10 thiết bị song song |
|
||||
| Concurrent devices = `0` | Unlimited | Flash tất cả thiết bị cùng lúc |
|
||||
| Concurrent devices = `1` | Tuần tự | Flash từng thiết bị một |
|
||||
|
||||
**Mỗi thiết bị có session HTTP riêng** → không bị lẫn cookie/token giữa các thiết bị.
|
||||
|
||||
---
|
||||
|
||||
## Files liên quan
|
||||
|
||||
| File | Chức năng |
|
||||
| --------------- | --------------------------------------------- |
|
||||
| `flasher.py` | Logic flash 3 bước (login → upload → proceed) |
|
||||
| `main.py` | UI PyQt6, FlashThread, quản lý song song |
|
||||
| `debug_full.py` | Script debug — chạy 3 bước với log chi tiết |
|
||||
| `scanner.py` | Scan mạng LAN tìm thiết bị |
|
||||
|
||||
""", "Complexity": 3, "Description": "Created flash documentation file with workflow diagrams, status conditions, HTTP details, and parallel processing explanation.", "EmptyFile": false, "IsArtifact": false, "Overwrite": false, "TargetFile": "/Users/nguyennhatminh/Documents/file code/Smatec/iot_fw_loader/FLASH_DOC.md
|
||||
59
doc/scanner_docs.md
Normal file
59
doc/scanner_docs.md
Normal file
@@ -0,0 +1,59 @@
|
||||
# Tài liệu Kỹ thuật: Cơ chế Quét IP trên Mạng (Network Scanner)
|
||||
|
||||
## 1. Tổng quan
|
||||
Thành phần quét IP (IP Scanner) trong file `scanner.py` được thiết kế để dò tìm và liệt kê tất cả các thiết bị đang hoạt động trên một dải mạng (ví dụ: `192.168.1.0/24`). Nó theo dõi và trả về danh sách các đối tượng chứa địa chỉ **IP** và **MAC Address** của từng thiết bị.
|
||||
|
||||
Để đảm bảo tỷ lệ phát hiện cao nhất trên các hệ điều hành khác nhau (Windows, macOS, Linux), script sử dụng kết hợp hai phương pháp dò tìm:
|
||||
1. **Quét mồi bằng Ping (Ping Sweep) kết hợp đọc bảng ARP tĩnh của hệ điều hành.**
|
||||
2. **Quét ARP trực tiếp ở Tầng 2 (Layer 2) bằng thư viện `scapy`.**
|
||||
|
||||
---
|
||||
|
||||
## 2. Luồng hoạt động chính (Hàm `scan_network`)
|
||||
|
||||
Đây là hàm được gọi trực tiếp khi muốn quét một mạng. Trình tự rà quét diễn ra qua các bước sau:
|
||||
|
||||
**Bước 1: "Đánh thức" các thiết bị (Ping Sweep)**
|
||||
- Hệ thống gọi hàm `_ping_sweep(network)` để gửi gói Ping (ICMP Echo Request) đồng loạt tới tất cả các IP có thể có trong mạng.
|
||||
- Mục đích của bước này là buộc các thiết bị phản hồi, từ đó hệ điều hành của máy đang chạy lệnh sẽ **tự động ghi nhận địa chỉ MAC** của các thiết bị đó vào bộ nhớ đệm ARP (ARP Cache).
|
||||
- Hệ thống tạm dừng 1 giây để đảm bảo hệ điều hành kịp lưu thông tin vào ARP Cache.
|
||||
|
||||
**Bước 2: Lấy dữ liệu từ bảng ARP của Hệ điều hành (Method 1)**
|
||||
- Thực thi lệnh hệ thống `arp -a` để đọc ARP Cache.
|
||||
- Kết quả được phân tích cú pháp (Regex) để trích xuất IP và MAC tương ứng, tương thích linh hoạt với cả đầu ra của Windows lẫn macOS/Linux.
|
||||
- Các thiết bị đọc được lưu vào danh sách nháp (biến `seen`).
|
||||
|
||||
**Bước 3: Quét sâu với Scapy (Method 2 - Dự phòng/Bổ sung)**
|
||||
- Script gọi thêm thư viện `scapy` để phát một thông điệp "ARP Who-has" tới địa chỉ MAC Broadcast (`ff:ff:ff:ff:ff:ff`).
|
||||
- Phương pháp này giúp phát hiện ra các thiết bị chặn Ping (chặn ICMP) ở Bước 1 nhưng buộc phải phản hồi gói ARP ở tầng Data Link.
|
||||
- Những thiết bị mới tìm thấy (nếu chưa có trong danh sách `seen` ở Bước 2) sẽ được bổ sung vào danh sách.
|
||||
|
||||
**Bước 4: Trả về kết quả**
|
||||
- Danh sách các thiết bị cuối cùng được sắp xếp từ nhỏ đến lớn dựa trên địa chỉ IP và trả về dưới dạng:
|
||||
`[{"ip": "192.168.1.2", "mac": "aa:bb:cc:dd:ee:ff"}, ...]`
|
||||
|
||||
---
|
||||
|
||||
## 3. Phân tích chi tiết các hàm hỗ trợ
|
||||
|
||||
### `_ping_sweep(network)` \& `_ping_one(ip, is_win)`
|
||||
- **Nhiệm vụ:** Quét Ping hàng loạt (Ping Sweep).
|
||||
- **Cách thức hoạt động:** Sử dụng `ThreadPoolExecutor` để chạy **tối đa 100 luồng (threads) song song**. Điều này giúp việc gửi Ping hàng loạt diễn ra cực kì nhanh chóng tính bằng giây thay vì phải đợi ping từng IP một.
|
||||
- **Biện pháp an toàn:** Script có cơ chế tự bảo vệ chặn flood mạng: nó sẽ **từ chối chạy Ping Sweep** nếu dải mạng (subnet) lớn hơn 256 địa chỉ IP (tức là chỉ chạy cho dải mạng từ `/24` trở xuống).
|
||||
|
||||
### `_scan_with_arp_table(network)`
|
||||
- **Nhiệm vụ:** Hàm chạy độc lập để quét tìm thiết bị mà **không cần thông qua đặc quyền Root/Administrator** (fallback method).
|
||||
- **Hỗ trợ Đa nền tảng:**
|
||||
- **Trên Windows (`win32`):** Chuẩn hóa định dạng chuẩn MAC từ gạch ngang sang dấu hai chấm (vd: từ `cc-2d-21...` sang `cc:2d:21...`). Nó dựa vào Regex nhận diện các dòng chuẩn xuất ra có chữ `dynamic` hoặc `static`.
|
||||
- **Trên macOS / Linux:** Dùng Regex đọc định dạng chuẩn riêng biệt ví dụ: `? (192.168.1.1) at aa:bb:cc:dd:ee:ff on en0`. Loại bỏ những địa chỉ lỗi bị đánh dấu là `(incomplete)`.
|
||||
|
||||
### `_scan_with_scapy(network)`
|
||||
- **Nhiệm vụ:** Công cụ quét cực mạnh ở Tầng 2 (Layer 2) của mô hình mạng OSI.
|
||||
- **Đặc thù:** Đòi hỏi người dùng phải cấp quyền Sudo/Root trên Linux/macOS hoặc phải cài đặt thư viện phần mềm **Npcap/WinPcap** trên Windows thì mới có thể sử dụng. Gửi gói `srp` bằng `scapy` với timeout là 3 giây để lấy về thông tin phần cứng đích thực.
|
||||
|
||||
---
|
||||
|
||||
## 4. Tóm tắt Ưu & Nhược điểm của thiết kế này
|
||||
|
||||
- **Ưu điểm:** Khả năng dò tìm diện rộng rất cao vì sự kết hợp song song của hai phương pháp. Nếu thiết bị không có quyền Root/Sudo để chạy `scapy`, nó vẫn có thể tìm được ít nhất ~90% thiết bị trong mạng nhờ tính năng Ping Sweeping mạnh mẽ. Tính tương thích chéo OS (Windows/Mac/Linux) được xử lý rất tốt và gọn gàng qua Regex.
|
||||
- **Nhược điểm:** Tốn thêm 1 giây (hàm `time.sleep(1)`) và thêm vài giây timeout tổng cộng, do cần chờ để làm đầy bộ nhớ đệm ARP trước khi quét sâu. Nếu thiết bị đích cố tình tắt phản hồi ARP, thì vẫn có khả năng bị lọt dán mạng.
|
||||
164
flasher.py
Normal file
164
flasher.py
Normal file
@@ -0,0 +1,164 @@
|
||||
"""
|
||||
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()
|
||||
948
main.py
Normal file
948
main.py
Normal file
@@ -0,0 +1,948 @@
|
||||
import sys
|
||||
import socket
|
||||
import ipaddress
|
||||
import platform
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
from PyQt6.QtWidgets import (
|
||||
QApplication, QWidget, QPushButton,
|
||||
QVBoxLayout, QHBoxLayout, QFileDialog, QTableWidget,
|
||||
QTableWidgetItem, QLabel, QProgressBar,
|
||||
QMessageBox, QGroupBox, QHeaderView, QLineEdit,
|
||||
QFrame, QCheckBox, QSpinBox, QScrollArea, QSizePolicy
|
||||
)
|
||||
from PyQt6.QtCore import (
|
||||
Qt, QThread, pyqtSignal, QPropertyAnimation, QAbstractAnimation,
|
||||
QParallelAnimationGroup, QRect
|
||||
)
|
||||
from PyQt6.QtGui import QFont, QColor, QIcon, QAction
|
||||
|
||||
import datetime
|
||||
|
||||
from scanner import scan_network
|
||||
from flasher import flash_device
|
||||
|
||||
|
||||
class CollapsibleGroupBox(QGroupBox):
|
||||
def __init__(self, title="", parent=None):
|
||||
super().__init__(title, parent)
|
||||
self.setCheckable(True)
|
||||
self.setChecked(True)
|
||||
|
||||
self.animation = QPropertyAnimation(self, b"maximumHeight")
|
||||
self.animation.setDuration(200)
|
||||
|
||||
# Connect the toggled signal to our animation function
|
||||
self.toggled.connect(self._toggle_animation)
|
||||
|
||||
self._full_height = 0
|
||||
|
||||
def set_content_layout(self, layout):
|
||||
# We need a wrapper widget to hold the layout
|
||||
self.content_widget = QWidget()
|
||||
self.content_widget.setStyleSheet("background-color: transparent;")
|
||||
self.content_widget.setLayout(layout)
|
||||
|
||||
main_layout = QVBoxLayout()
|
||||
main_layout.setContentsMargins(0, 0, 0, 0)
|
||||
main_layout.addWidget(self.content_widget)
|
||||
self.setLayout(main_layout)
|
||||
|
||||
def _toggle_animation(self, checked):
|
||||
if not hasattr(self, 'content_widget'):
|
||||
return
|
||||
|
||||
if checked:
|
||||
# Expand: show content first, then animate
|
||||
self.content_widget.setVisible(True)
|
||||
target_height = self.sizeHint().height()
|
||||
self.animation.stop()
|
||||
self.animation.setStartValue(self.height())
|
||||
self.animation.setEndValue(target_height)
|
||||
self.animation.finished.connect(self._on_expand_finished)
|
||||
self.animation.start()
|
||||
else:
|
||||
# Collapse
|
||||
self.animation.stop()
|
||||
self.animation.setStartValue(self.height())
|
||||
self.animation.setEndValue(32)
|
||||
self.animation.finished.connect(self._on_collapse_finished)
|
||||
self.animation.start()
|
||||
|
||||
def _on_expand_finished(self):
|
||||
# Remove height constraint so content can grow dynamically
|
||||
self.setMaximumHeight(16777215)
|
||||
try:
|
||||
self.animation.finished.disconnect(self._on_expand_finished)
|
||||
except TypeError:
|
||||
pass
|
||||
|
||||
def _on_collapse_finished(self):
|
||||
if not self.isChecked():
|
||||
self.content_widget.setVisible(False)
|
||||
try:
|
||||
self.animation.finished.disconnect(self._on_collapse_finished)
|
||||
except TypeError:
|
||||
pass
|
||||
|
||||
|
||||
def _resolve_hostname(ip):
|
||||
"""Reverse DNS lookup for a single IP."""
|
||||
try:
|
||||
return socket.gethostbyaddr(ip)[0]
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def get_local_ip():
|
||||
"""Get the local IP address of this machine."""
|
||||
try:
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
s.connect(("8.8.8.8", 80))
|
||||
ip = s.getsockname()[0]
|
||||
s.close()
|
||||
return ip
|
||||
except Exception:
|
||||
return "N/A"
|
||||
|
||||
|
||||
def get_default_network(ip):
|
||||
"""Guess the /24 network from local IP."""
|
||||
try:
|
||||
parts = ip.split(".")
|
||||
return f"{parts[0]}.{parts[1]}.{parts[2]}.0/24"
|
||||
except Exception:
|
||||
return "192.168.1.0/24"
|
||||
|
||||
|
||||
def get_machine_info():
|
||||
"""Collect machine info."""
|
||||
hostname = socket.gethostname()
|
||||
local_ip = get_local_ip()
|
||||
os_info = f"{platform.system()} {platform.release()}"
|
||||
mac_addr = "N/A"
|
||||
|
||||
try:
|
||||
import uuid
|
||||
mac = uuid.getnode()
|
||||
mac_addr = ":".join(
|
||||
f"{(mac >> (8 * i)) & 0xFF:02X}" for i in reversed(range(6))
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
"hostname": hostname,
|
||||
"ip": local_ip,
|
||||
"os": os_info,
|
||||
"mac": mac_addr,
|
||||
}
|
||||
|
||||
|
||||
# ── Stylesheet ──────────────────────────────────────────────
|
||||
|
||||
STYLE = """
|
||||
QWidget {
|
||||
background-color: #1a1b2e;
|
||||
color: #e2e8f0;
|
||||
font-family: 'Segoe UI', 'SF Pro Display', sans-serif;
|
||||
font-size: 12px;
|
||||
}
|
||||
|
||||
QGroupBox {
|
||||
border: 1px solid #2d3748;
|
||||
border-radius: 8px;
|
||||
margin-top: 10px;
|
||||
padding: 20px 8px 6px 8px;
|
||||
font-weight: bold;
|
||||
color: #7eb8f7;
|
||||
background-color: #1e2035;
|
||||
}
|
||||
|
||||
QGroupBox::title {
|
||||
subcontrol-origin: margin;
|
||||
subcontrol-position: top left;
|
||||
left: 14px;
|
||||
top: 5px;
|
||||
padding: 0px 8px;
|
||||
background-color: transparent;
|
||||
}
|
||||
|
||||
QGroupBox::indicator {
|
||||
width: 14px;
|
||||
height: 14px;
|
||||
border-radius: 4px;
|
||||
border: 1px solid #3d4a6b;
|
||||
background-color: #13141f;
|
||||
margin-top: 5px;
|
||||
}
|
||||
|
||||
QGroupBox::indicator:unchecked {
|
||||
background-color: #13141f;
|
||||
}
|
||||
|
||||
QGroupBox::indicator:checked {
|
||||
background-color: #3b82f6;
|
||||
border-color: #3b82f6;
|
||||
}
|
||||
|
||||
QLabel#title {
|
||||
font-size: 16px;
|
||||
font-weight: bold;
|
||||
color: #7eb8f7;
|
||||
letter-spacing: 1px;
|
||||
}
|
||||
|
||||
QLabel#info {
|
||||
color: #94a3b8;
|
||||
font-size: 11px;
|
||||
}
|
||||
|
||||
QPushButton {
|
||||
background-color: #2d3352;
|
||||
border: 1px solid #3d4a6b;
|
||||
border-radius: 6px;
|
||||
padding: 4px 12px;
|
||||
color: #e2e8f0;
|
||||
font-weight: 600;
|
||||
min-height: 24px;
|
||||
}
|
||||
|
||||
QPushButton:hover {
|
||||
background-color: #3d4a6b;
|
||||
border-color: #7eb8f7;
|
||||
color: #ffffff;
|
||||
}
|
||||
|
||||
QPushButton:pressed {
|
||||
background-color: #1a2040;
|
||||
}
|
||||
|
||||
QPushButton#scan {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #1a56db, stop:1 #1e66f5);
|
||||
border-color: #1a56db;
|
||||
color: #ffffff;
|
||||
}
|
||||
|
||||
QPushButton#scan:hover {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #2563eb, stop:1 #3b82f6);
|
||||
}
|
||||
|
||||
QPushButton#flash {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #15803d, stop:1 #16a34a);
|
||||
border-color: #15803d;
|
||||
color: #ffffff;
|
||||
font-size: 13px;
|
||||
min-height: 30px;
|
||||
}
|
||||
|
||||
QPushButton#flash:hover {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #16a34a, stop:1 #22c55e);
|
||||
}
|
||||
|
||||
QTableWidget {
|
||||
background-color: #13141f;
|
||||
alternate-background-color: #1a1b2e;
|
||||
border: 1px solid #2d3748;
|
||||
border-radius: 8px;
|
||||
gridline-color: #2d3748;
|
||||
selection-background-color: #2d3a5a;
|
||||
selection-color: #e2e8f0;
|
||||
}
|
||||
|
||||
QTableWidget::item {
|
||||
padding: 2px 6px;
|
||||
border: none;
|
||||
}
|
||||
|
||||
QTableWidget::item:selected {
|
||||
background-color: #2d3a5a;
|
||||
color: #7eb8f7;
|
||||
}
|
||||
|
||||
QHeaderView::section {
|
||||
background-color: #1e2035;
|
||||
color: #7eb8f7;
|
||||
border: none;
|
||||
border-bottom: 2px solid #3b82f6;
|
||||
border-right: 1px solid #2d3748;
|
||||
padding: 4px 6px;
|
||||
font-weight: bold;
|
||||
font-size: 11px;
|
||||
letter-spacing: 0.5px;
|
||||
text-transform: uppercase;
|
||||
}
|
||||
|
||||
QHeaderView::section:last {
|
||||
border-right: none;
|
||||
}
|
||||
|
||||
QProgressBar {
|
||||
border: 1px solid #2d3748;
|
||||
border-radius: 6px;
|
||||
text-align: center;
|
||||
background-color: #13141f;
|
||||
color: #e2e8f0;
|
||||
height: 20px;
|
||||
font-size: 11px;
|
||||
font-weight: 600;
|
||||
}
|
||||
|
||||
QProgressBar::chunk {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #3b82f6, stop:1 #7eb8f7);
|
||||
border-radius: 7px;
|
||||
}
|
||||
|
||||
QProgressBar#scan_bar {
|
||||
border: 1px solid #374151;
|
||||
border-radius: 5px;
|
||||
text-align: center;
|
||||
background-color: #13141f;
|
||||
color: #fbbf24;
|
||||
height: 16px;
|
||||
font-size: 10px;
|
||||
font-weight: 600;
|
||||
}
|
||||
|
||||
QProgressBar#scan_bar::chunk {
|
||||
background: qlineargradient(x1:0, y1:0, x2:1, y2:0,
|
||||
stop:0 #d97706, stop:1 #fbbf24);
|
||||
border-radius: 4px;
|
||||
}
|
||||
|
||||
QLineEdit {
|
||||
background-color: #13141f;
|
||||
border: 1px solid #2d3748;
|
||||
border-radius: 8px;
|
||||
padding: 7px 12px;
|
||||
color: #e2e8f0;
|
||||
selection-background-color: #2d3a5a;
|
||||
}
|
||||
|
||||
QLineEdit:focus {
|
||||
border-color: #3b82f6;
|
||||
background-color: #161727;
|
||||
}
|
||||
|
||||
QScrollBar:vertical {
|
||||
background: #13141f;
|
||||
width: 10px;
|
||||
border-radius: 5px;
|
||||
margin: 2px;
|
||||
}
|
||||
|
||||
QScrollBar::handle:vertical {
|
||||
background: #3d4a6b;
|
||||
border-radius: 5px;
|
||||
min-height: 30px;
|
||||
}
|
||||
|
||||
QScrollBar::handle:vertical:hover {
|
||||
background: #7eb8f7;
|
||||
}
|
||||
|
||||
QScrollBar::handle:vertical:pressed {
|
||||
background: #3b82f6;
|
||||
}
|
||||
|
||||
QScrollBar::add-line:vertical,
|
||||
QScrollBar::sub-line:vertical {
|
||||
height: 0px;
|
||||
background: transparent;
|
||||
}
|
||||
|
||||
QScrollBar::add-page:vertical,
|
||||
QScrollBar::sub-page:vertical {
|
||||
background: transparent;
|
||||
}
|
||||
|
||||
QScrollBar:horizontal {
|
||||
background: #13141f;
|
||||
height: 10px;
|
||||
border-radius: 5px;
|
||||
margin: 2px;
|
||||
}
|
||||
|
||||
QScrollBar::handle:horizontal {
|
||||
background: #3d4a6b;
|
||||
border-radius: 5px;
|
||||
min-width: 30px;
|
||||
}
|
||||
|
||||
QScrollBar::handle:horizontal:hover {
|
||||
background: #7eb8f7;
|
||||
}
|
||||
|
||||
QScrollBar::handle:horizontal:pressed {
|
||||
background: #3b82f6;
|
||||
}
|
||||
|
||||
QScrollBar::add-line:horizontal,
|
||||
QScrollBar::sub-line:horizontal {
|
||||
width: 0px;
|
||||
background: transparent;
|
||||
}
|
||||
|
||||
QScrollBar::add-page:horizontal,
|
||||
QScrollBar::sub-page:horizontal {
|
||||
background: transparent;
|
||||
}
|
||||
|
||||
QCheckBox {
|
||||
spacing: 6px;
|
||||
color: #94a3b8;
|
||||
font-size: 12px;
|
||||
}
|
||||
|
||||
QCheckBox::indicator {
|
||||
width: 16px;
|
||||
height: 16px;
|
||||
border-radius: 4px;
|
||||
border: 1px solid #3d4a6b;
|
||||
background-color: #13141f;
|
||||
}
|
||||
|
||||
QCheckBox::indicator:checked {
|
||||
background-color: #3b82f6;
|
||||
border-color: #3b82f6;
|
||||
}
|
||||
|
||||
QCheckBox::indicator:hover {
|
||||
border-color: #7eb8f7;
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
class ScanThread(QThread):
|
||||
"""Run network scan in a background thread so UI doesn't freeze."""
|
||||
finished = pyqtSignal(list)
|
||||
error = pyqtSignal(str)
|
||||
scan_progress = pyqtSignal(int, int) # done, total (ping sweep)
|
||||
stage = pyqtSignal(str) # current scan phase
|
||||
|
||||
def __init__(self, network):
|
||||
super().__init__()
|
||||
self.network = network
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
def on_ping_progress(done, total):
|
||||
self.scan_progress.emit(done, total)
|
||||
|
||||
def on_stage(s):
|
||||
self.stage.emit(s)
|
||||
|
||||
results = scan_network(self.network,
|
||||
progress_cb=on_ping_progress,
|
||||
stage_cb=on_stage)
|
||||
# Resolve hostnames in parallel
|
||||
self.stage.emit("hostname")
|
||||
with ThreadPoolExecutor(max_workers=50) as executor:
|
||||
future_to_dev = {
|
||||
executor.submit(_resolve_hostname, d["ip"]): d
|
||||
for d in results
|
||||
}
|
||||
for future in as_completed(future_to_dev):
|
||||
dev = future_to_dev[future]
|
||||
try:
|
||||
dev["name"] = future.result(timeout=3)
|
||||
except Exception:
|
||||
dev["name"] = ""
|
||||
self.finished.emit(results)
|
||||
except Exception as e:
|
||||
self.error.emit(str(e))
|
||||
|
||||
|
||||
class FlashThread(QThread):
|
||||
"""Run firmware flash in background so UI stays responsive."""
|
||||
device_status = pyqtSignal(int, str) # index, status message
|
||||
device_done = pyqtSignal(int, str) # index, result
|
||||
all_done = pyqtSignal()
|
||||
|
||||
def __init__(self, devices, firmware_path, max_workers=10):
|
||||
super().__init__()
|
||||
self.devices = devices
|
||||
self.firmware_path = firmware_path
|
||||
self.max_workers = max_workers
|
||||
|
||||
def run(self):
|
||||
def _flash_one(i, dev):
|
||||
try:
|
||||
def on_status(msg):
|
||||
self.device_status.emit(i, msg)
|
||||
|
||||
result = flash_device(
|
||||
dev["ip"], self.firmware_path,
|
||||
status_cb=on_status
|
||||
)
|
||||
self.device_done.emit(i, result)
|
||||
except Exception as e:
|
||||
self.device_done.emit(i, f"FAIL: {e}")
|
||||
|
||||
# Use configured max_workers (0 = unlimited = one per device)
|
||||
workers = self.max_workers if self.max_workers > 0 else len(self.devices)
|
||||
with ThreadPoolExecutor(max_workers=workers) as executor:
|
||||
futures = []
|
||||
for i, dev in enumerate(self.devices):
|
||||
futures.append(executor.submit(_flash_one, i, dev))
|
||||
for f in futures:
|
||||
f.result()
|
||||
|
||||
self.all_done.emit()
|
||||
|
||||
|
||||
class App(QWidget):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.setWindowTitle("IoT Firmware Loader")
|
||||
self.firmware = None
|
||||
self.all_devices = [] # all scan results (unfiltered)
|
||||
self.devices = [] # currently displayed (filtered)
|
||||
self.scan_thread = None
|
||||
|
||||
info = get_machine_info()
|
||||
self.local_ip = info["ip"]
|
||||
self.gateway_ip = self._guess_gateway(self.local_ip)
|
||||
|
||||
layout = QVBoxLayout()
|
||||
layout.setSpacing(4)
|
||||
layout.setContentsMargins(8, 6, 8, 6)
|
||||
|
||||
# ── Title ──
|
||||
title = QLabel("⚡ IoT Firmware Loader (MiraV3)")
|
||||
title.setObjectName("title")
|
||||
title.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.addWidget(title)
|
||||
|
||||
copyright_label = QLabel(f"© {datetime.datetime.now().year} Smatec — R&D Software")
|
||||
copyright_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
copyright_label.setStyleSheet(
|
||||
"color: #9399b2; font-size: 11px; font-weight: 500; font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto;"
|
||||
)
|
||||
layout.addWidget(copyright_label)
|
||||
|
||||
# ── Machine Info Group ──
|
||||
info_group = CollapsibleGroupBox("🖥 Machine Info")
|
||||
info_layout = QVBoxLayout()
|
||||
info_layout.setSpacing(2)
|
||||
info_layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
row1 = QHBoxLayout()
|
||||
row1.addWidget(self._info_label("Hostname:"))
|
||||
row1.addWidget(self._info_value(info["hostname"]))
|
||||
row1.addStretch()
|
||||
row1.addWidget(self._info_label("IP:"))
|
||||
row1.addWidget(self._info_value(info["ip"]))
|
||||
info_layout.addLayout(row1)
|
||||
|
||||
row2 = QHBoxLayout()
|
||||
row2.addWidget(self._info_label("OS:"))
|
||||
row2.addWidget(self._info_value(info["os"]))
|
||||
row2.addStretch()
|
||||
row2.addWidget(self._info_label("MAC:"))
|
||||
row2.addWidget(self._info_value(info["mac"]))
|
||||
info_layout.addLayout(row2)
|
||||
|
||||
info_group.set_content_layout(info_layout)
|
||||
layout.addWidget(info_group)
|
||||
|
||||
# ── Firmware Selection ──
|
||||
fw_group = CollapsibleGroupBox("📦 Firmware")
|
||||
fw_layout = QHBoxLayout()
|
||||
fw_layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
self.fw_label = QLabel("No firmware selected")
|
||||
self.fw_label.setObjectName("info")
|
||||
self.fw_label.setWordWrap(True)
|
||||
fw_layout.addWidget(self.fw_label, 1)
|
||||
|
||||
btn_fw = QPushButton("📁 Select File")
|
||||
btn_fw.clicked.connect(self.select_fw)
|
||||
btn_fw.setFixedWidth(110)
|
||||
fw_layout.addWidget(btn_fw)
|
||||
|
||||
fw_group.set_content_layout(fw_layout)
|
||||
layout.addWidget(fw_group)
|
||||
|
||||
# ── Network Scan ──
|
||||
scan_group = CollapsibleGroupBox("📡 Network Scan")
|
||||
scan_layout = QVBoxLayout()
|
||||
scan_layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
net_row = QHBoxLayout()
|
||||
net_row.addWidget(QLabel("Network:"))
|
||||
self.net_input = QLineEdit(get_default_network(self.local_ip))
|
||||
self.net_input.setPlaceholderText("e.g. 192.168.4.0/24")
|
||||
net_row.addWidget(self.net_input, 1)
|
||||
|
||||
btn_scan = QPushButton("🔍 Scan LAN")
|
||||
btn_scan.setObjectName("scan")
|
||||
btn_scan.clicked.connect(self.scan)
|
||||
btn_scan.setFixedWidth(110)
|
||||
net_row.addWidget(btn_scan)
|
||||
scan_layout.addLayout(net_row)
|
||||
|
||||
self.scan_progress_bar = QProgressBar()
|
||||
self.scan_progress_bar.setObjectName("scan_bar")
|
||||
self.scan_progress_bar.setRange(0, 0)
|
||||
self.scan_progress_bar.setFormat("")
|
||||
self.scan_progress_bar.setVisible(False)
|
||||
scan_layout.addWidget(self.scan_progress_bar)
|
||||
|
||||
self.scan_status = QLabel("")
|
||||
self.scan_status.setObjectName("info")
|
||||
scan_layout.addWidget(self.scan_status)
|
||||
|
||||
scan_group.set_content_layout(scan_layout)
|
||||
layout.addWidget(scan_group)
|
||||
|
||||
# ── Device Table ──
|
||||
dev_group = QGroupBox("📋 Devices Found")
|
||||
dev_layout = QVBoxLayout()
|
||||
dev_layout.setSpacing(4)
|
||||
dev_layout.setContentsMargins(4, 4, 4, 4)
|
||||
|
||||
self.table = QTableWidget()
|
||||
self.table.setColumnCount(5)
|
||||
self.table.setHorizontalHeaderLabels(["", "IP", "Name", "MAC", "Status"])
|
||||
self.table.setAlternatingRowColors(True)
|
||||
header = self.table.horizontalHeader()
|
||||
header.setSectionResizeMode(0, QHeaderView.ResizeMode.Fixed)
|
||||
header.resizeSection(0, 40)
|
||||
for col in range(1, 5):
|
||||
header.setSectionResizeMode(col, QHeaderView.ResizeMode.Stretch)
|
||||
self.table.verticalHeader().setVisible(False)
|
||||
self.table.setSelectionBehavior(
|
||||
QTableWidget.SelectionBehavior.SelectRows
|
||||
)
|
||||
dev_layout.addWidget(self.table)
|
||||
|
||||
filter_row = QHBoxLayout()
|
||||
self.device_count_label = QLabel("Total: 0 devices")
|
||||
self.device_count_label.setObjectName("info")
|
||||
filter_row.addWidget(self.device_count_label)
|
||||
filter_row.addStretch()
|
||||
|
||||
btn_select_all = QPushButton("☑ Select All")
|
||||
btn_select_all.clicked.connect(self._select_all_devices)
|
||||
filter_row.addWidget(btn_select_all)
|
||||
|
||||
btn_deselect_all = QPushButton("☐ Deselect All")
|
||||
btn_deselect_all.clicked.connect(self._deselect_all_devices)
|
||||
filter_row.addWidget(btn_deselect_all)
|
||||
|
||||
self.show_all_cb = QCheckBox("Show all (include gateway & self)")
|
||||
self.show_all_cb.setChecked(False)
|
||||
self.show_all_cb.stateChanged.connect(self._refresh_table)
|
||||
filter_row.addWidget(self.show_all_cb)
|
||||
dev_layout.addLayout(filter_row)
|
||||
|
||||
dev_group.setLayout(dev_layout)
|
||||
layout.addWidget(dev_group, stretch=1)
|
||||
|
||||
# ── Flash Controls ──
|
||||
flash_group = QGroupBox("🚀 Flash")
|
||||
flash_layout = QVBoxLayout()
|
||||
flash_layout.setSpacing(4)
|
||||
flash_layout.setContentsMargins(4, 4, 4, 4)
|
||||
|
||||
self.progress = QProgressBar()
|
||||
self.progress.setFormat("%v / %m devices (%p%)")
|
||||
flash_layout.addWidget(self.progress)
|
||||
|
||||
# Parallel count row
|
||||
parallel_row = QHBoxLayout()
|
||||
parallel_row.addWidget(QLabel("Concurrent devices:"))
|
||||
self.parallel_spin = QSpinBox()
|
||||
self.parallel_spin.setRange(0, 100)
|
||||
self.parallel_spin.setValue(10)
|
||||
self.parallel_spin.setSpecialValueText("Unlimited")
|
||||
self.parallel_spin.setToolTip("0 = unlimited (all devices at once)")
|
||||
self.parallel_spin.setFixedWidth(80)
|
||||
parallel_row.addWidget(self.parallel_spin)
|
||||
parallel_row.addStretch()
|
||||
flash_layout.addLayout(parallel_row)
|
||||
|
||||
btn_flash = QPushButton("⚡ Flash Selected Devices")
|
||||
btn_flash.setObjectName("flash")
|
||||
btn_flash.clicked.connect(self.flash_all)
|
||||
flash_layout.addWidget(btn_flash)
|
||||
|
||||
flash_group.setLayout(flash_layout)
|
||||
layout.addWidget(flash_group)
|
||||
|
||||
self.setLayout(layout)
|
||||
|
||||
# ── Helpers ──
|
||||
|
||||
def _guess_gateway(self, ip):
|
||||
"""Guess gateway IP (x.x.x.1) from local IP."""
|
||||
try:
|
||||
parts = ip.split(".")
|
||||
return f"{parts[0]}.{parts[1]}.{parts[2]}.1"
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
def _info_label(self, text):
|
||||
lbl = QLabel(text)
|
||||
lbl.setObjectName("info")
|
||||
return lbl
|
||||
|
||||
def _info_value(self, text):
|
||||
lbl = QLabel(text)
|
||||
lbl.setStyleSheet("color: #cdd6f4; font-weight: bold;")
|
||||
return lbl
|
||||
|
||||
def _get_filtered_devices(self):
|
||||
"""Return devices filtered based on show_all checkbox."""
|
||||
if self.show_all_cb.isChecked():
|
||||
return list(self.all_devices)
|
||||
excluded = {self.local_ip, self.gateway_ip}
|
||||
return [d for d in self.all_devices if d["ip"] not in excluded]
|
||||
|
||||
def _refresh_table(self):
|
||||
"""Re-populate table based on current filter state."""
|
||||
self.devices = self._get_filtered_devices()
|
||||
self.table.setRowCount(len(self.devices))
|
||||
|
||||
for i, d in enumerate(self.devices):
|
||||
# Checkbox column
|
||||
cb_item = QTableWidgetItem()
|
||||
cb_item.setCheckState(Qt.CheckState.Checked)
|
||||
cb_item.setFlags(cb_item.flags() | Qt.ItemFlag.ItemIsUserCheckable)
|
||||
self.table.setItem(i, 0, cb_item)
|
||||
|
||||
ip_item = QTableWidgetItem(d["ip"])
|
||||
name_item = QTableWidgetItem(d.get("name", ""))
|
||||
mac_item = QTableWidgetItem(d["mac"].upper())
|
||||
status_item = QTableWidgetItem(d["status"])
|
||||
|
||||
# Dim gateway & self if showing all
|
||||
if d["ip"] in {self.local_ip, self.gateway_ip}:
|
||||
for item in [ip_item, name_item, mac_item, status_item]:
|
||||
item.setForeground(QColor("#4a5568"))
|
||||
cb_item.setCheckState(Qt.CheckState.Unchecked)
|
||||
|
||||
self.table.setItem(i, 1, ip_item)
|
||||
self.table.setItem(i, 2, name_item)
|
||||
self.table.setItem(i, 3, mac_item)
|
||||
self.table.setItem(i, 4, status_item)
|
||||
|
||||
total = len(self.devices)
|
||||
hidden = len(self.all_devices) - total
|
||||
label = f"Total: {total} devices"
|
||||
if hidden > 0:
|
||||
label += f" (hiding {hidden})"
|
||||
self.device_count_label.setText(label)
|
||||
|
||||
def _select_all_devices(self):
|
||||
"""Check all device checkboxes."""
|
||||
for i in range(self.table.rowCount()):
|
||||
item = self.table.item(i, 0)
|
||||
if item:
|
||||
item.setCheckState(Qt.CheckState.Checked)
|
||||
|
||||
def _deselect_all_devices(self):
|
||||
"""Uncheck all device checkboxes."""
|
||||
for i in range(self.table.rowCount()):
|
||||
item = self.table.item(i, 0)
|
||||
if item:
|
||||
item.setCheckState(Qt.CheckState.Unchecked)
|
||||
|
||||
# ── Actions ──
|
||||
|
||||
def select_fw(self):
|
||||
file, _ = QFileDialog.getOpenFileName(
|
||||
self, "Select Firmware", "",
|
||||
"Firmware Files (*.bin *.hex *.uf2);;All Files (*)"
|
||||
)
|
||||
if file:
|
||||
self.firmware = file
|
||||
name = file.split("/")[-1]
|
||||
self.fw_label.setText(f"✅ {name}")
|
||||
self.fw_label.setStyleSheet("color: #a6e3a1;")
|
||||
|
||||
def scan(self):
|
||||
network_str = self.net_input.text().strip()
|
||||
|
||||
try:
|
||||
network = ipaddress.ip_network(network_str, strict=False)
|
||||
except ValueError:
|
||||
QMessageBox.warning(
|
||||
self, "Invalid Network",
|
||||
f"'{network_str}' is not a valid network.\n"
|
||||
"Example: 192.168.4.0/24"
|
||||
)
|
||||
return
|
||||
|
||||
self.scan_status.setText("⏳ Preparing scan...")
|
||||
self.scan_status.setStyleSheet("color: #f9e2af;")
|
||||
self.table.setRowCount(0)
|
||||
self.devices = []
|
||||
|
||||
self.scan_progress_bar.setRange(0, 0)
|
||||
self.scan_progress_bar.setFormat(" Starting...")
|
||||
self.scan_progress_bar.setVisible(True)
|
||||
|
||||
self.scan_thread = ScanThread(network)
|
||||
self.scan_thread.finished.connect(self._on_scan_done)
|
||||
self.scan_thread.error.connect(self._on_scan_error)
|
||||
self.scan_thread.scan_progress.connect(self._on_scan_progress)
|
||||
self.scan_thread.stage.connect(self._on_scan_stage)
|
||||
self.scan_thread.start()
|
||||
|
||||
def _on_scan_done(self, results):
|
||||
self.scan_progress_bar.setVisible(False)
|
||||
self.all_devices = []
|
||||
|
||||
for dev in results:
|
||||
dev["status"] = "READY"
|
||||
self.all_devices.append(dev)
|
||||
|
||||
self.all_devices.sort(key=lambda d: ipaddress.ip_address(d["ip"]))
|
||||
|
||||
# Apply filter and populate table
|
||||
self._refresh_table()
|
||||
|
||||
total_all = len(self.all_devices)
|
||||
total_shown = len(self.devices)
|
||||
self.scan_status.setText(f"✅ Scan complete — {total_all} found, {total_shown} shown")
|
||||
self.scan_status.setStyleSheet("color: #a6e3a1;")
|
||||
|
||||
def _on_scan_stage(self, stage):
|
||||
if stage == "ping":
|
||||
self.scan_progress_bar.setFormat("⚡ Pinging hosts... %v / %m")
|
||||
self.scan_status.setText("⏳ Pinging all hosts...")
|
||||
self.scan_status.setStyleSheet("color: #f9e2af;")
|
||||
elif stage == "arp":
|
||||
self.scan_progress_bar.setRange(0, 0)
|
||||
self.scan_progress_bar.setFormat(" Reading ARP cache...")
|
||||
self.scan_status.setText("⏳ Reading ARP cache...")
|
||||
elif stage == "scapy":
|
||||
self.scan_progress_bar.setRange(0, 0)
|
||||
self.scan_progress_bar.setFormat(" ARP broadcast scan...")
|
||||
self.scan_status.setText("⏳ Running ARP broadcast scan...")
|
||||
elif stage == "hostname":
|
||||
self.scan_progress_bar.setRange(0, 0)
|
||||
self.scan_progress_bar.setFormat(" Resolving hostnames...")
|
||||
self.scan_status.setText("⏳ Resolving hostnames...")
|
||||
|
||||
def _on_scan_progress(self, done, total):
|
||||
self.scan_progress_bar.setRange(0, total)
|
||||
self.scan_progress_bar.setValue(done)
|
||||
self.scan_status.setText(f"⏳ Pinging hosts... {done} / {total}")
|
||||
self.scan_status.setStyleSheet("color: #f9e2af;")
|
||||
|
||||
def _on_scan_error(self, error_msg):
|
||||
self.scan_progress_bar.setVisible(False)
|
||||
self.scan_status.setText("❌ Scan failed")
|
||||
self.scan_status.setStyleSheet("color: #f38ba8;")
|
||||
QMessageBox.critical(
|
||||
self, "Scan Error", f"Failed to scan network:\n{error_msg}"
|
||||
)
|
||||
|
||||
def _get_selected_devices(self):
|
||||
"""Return list of (table_row_index, device_dict) for checked devices."""
|
||||
selected = []
|
||||
for i in range(self.table.rowCount()):
|
||||
item = self.table.item(i, 0)
|
||||
if item and item.checkState() == Qt.CheckState.Checked:
|
||||
if i < len(self.devices):
|
||||
selected.append((i, self.devices[i]))
|
||||
return selected
|
||||
|
||||
def flash_all(self):
|
||||
if not self.firmware:
|
||||
QMessageBox.warning(
|
||||
self, "No Firmware",
|
||||
"Please select a firmware file first."
|
||||
)
|
||||
return
|
||||
|
||||
if not self.devices:
|
||||
QMessageBox.information(
|
||||
self, "No Devices",
|
||||
"No devices found to flash.\n"
|
||||
"Please scan the network first."
|
||||
)
|
||||
return
|
||||
|
||||
selected = self._get_selected_devices()
|
||||
if not selected:
|
||||
QMessageBox.information(
|
||||
self, "No Selection",
|
||||
"No devices selected.\n"
|
||||
"Check the boxes next to the devices you want to flash."
|
||||
)
|
||||
return
|
||||
|
||||
total = len(selected)
|
||||
self.progress.setMaximum(total)
|
||||
self.progress.setValue(0)
|
||||
|
||||
# Build list with row indices for UI updates
|
||||
self._flash_row_map = {}
|
||||
flash_devices = []
|
||||
for idx, (row, dev) in enumerate(selected):
|
||||
self._flash_row_map[idx] = row
|
||||
flash_devices.append(dev)
|
||||
|
||||
# Run flashing in background thread so UI doesn't freeze
|
||||
self.flash_thread = FlashThread(
|
||||
flash_devices, self.firmware,
|
||||
max_workers=self.parallel_spin.value()
|
||||
)
|
||||
self.flash_thread.device_status.connect(self._on_flash_status)
|
||||
self.flash_thread.device_done.connect(self._on_flash_done)
|
||||
self.flash_thread.all_done.connect(self._on_flash_all_done)
|
||||
self.flash_thread.start()
|
||||
|
||||
def _on_flash_status(self, index, msg):
|
||||
"""Update status column while flashing."""
|
||||
row = self._flash_row_map.get(index, index)
|
||||
self.table.setItem(row, 4, QTableWidgetItem(f"⏳ {msg}"))
|
||||
|
||||
def _on_flash_done(self, index, result):
|
||||
"""One device finished flashing."""
|
||||
row = self._flash_row_map.get(index, index)
|
||||
if result.startswith("DONE"):
|
||||
item = QTableWidgetItem(f"✅ {result}")
|
||||
item.setForeground(QColor("#a6e3a1"))
|
||||
# Auto-uncheck so it won't be flashed again
|
||||
cb = self.table.item(row, 0)
|
||||
if cb:
|
||||
cb.setCheckState(Qt.CheckState.Unchecked)
|
||||
else:
|
||||
item = QTableWidgetItem(f"❌ {result}")
|
||||
item.setForeground(QColor("#f38ba8"))
|
||||
self.table.setItem(row, 4, item)
|
||||
self.progress.setValue(self.progress.value() + 1)
|
||||
|
||||
def _on_flash_all_done(self):
|
||||
"""All flashing complete."""
|
||||
QMessageBox.information(self, "Flash Complete", "All devices have been processed.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app = QApplication(sys.argv)
|
||||
app.setStyleSheet(STYLE)
|
||||
|
||||
window = App()
|
||||
|
||||
# 60% height, limited width (750px), centered
|
||||
screen = app.primaryScreen().availableGeometry()
|
||||
w = min(750, screen.width())
|
||||
h = int(screen.height() * 0.6)
|
||||
x = (screen.width() - w) // 2
|
||||
y = screen.y() + (screen.height() - h) // 2
|
||||
window.setGeometry(x, y, w, h)
|
||||
window.show()
|
||||
|
||||
sys.exit(app.exec())
|
||||
3
requirements.txt
Normal file
3
requirements.txt
Normal file
@@ -0,0 +1,3 @@
|
||||
PyQt6
|
||||
scapy
|
||||
requests
|
||||
16
run.bat
Normal file
16
run.bat
Normal file
@@ -0,0 +1,16 @@
|
||||
@echo off
|
||||
REM IoT Firmware Loader - Windows launcher
|
||||
|
||||
cd /d "%~dp0"
|
||||
|
||||
if exist "venv" (
|
||||
call venv\Scripts\activate.bat
|
||||
) else (
|
||||
echo Creating virtual environment...
|
||||
python -m venv venv
|
||||
call venv\Scripts\activate.bat
|
||||
pip install PyQt6 scapy requests
|
||||
)
|
||||
|
||||
echo Starting IoT Firmware Loader...
|
||||
python main.py
|
||||
18
run.sh
Normal file
18
run.sh
Normal file
@@ -0,0 +1,18 @@
|
||||
#!/bin/bash
|
||||
# IoT Firmware Loader - macOS/Linux launcher
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
cd "$SCRIPT_DIR"
|
||||
|
||||
# Activate virtual environment
|
||||
if [ -d "venv" ]; then
|
||||
source venv/bin/activate
|
||||
else
|
||||
echo "❌ Virtual environment not found. Creating..."
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate
|
||||
pip install PyQt6 scapy requests
|
||||
fi
|
||||
|
||||
echo "🚀 Starting IoT Firmware Loader..."
|
||||
python main.py
|
||||
224
scanner.py
Normal file
224
scanner.py
Normal file
@@ -0,0 +1,224 @@
|
||||
import subprocess
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import ipaddress
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
# Windows: prevent subprocess from opening visible console windows
|
||||
_NO_WINDOW = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
|
||||
|
||||
|
||||
def _scan_with_scapy(network):
|
||||
"""Scan using scapy (requires root/sudo, and Npcap on Windows)."""
|
||||
from scapy.all import ARP, Ether, srp
|
||||
|
||||
arp = ARP(pdst=str(network))
|
||||
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
|
||||
|
||||
packet = ether / arp
|
||||
|
||||
result = srp(packet, timeout=3, verbose=0)[0]
|
||||
|
||||
devices = []
|
||||
|
||||
for sent, received in result:
|
||||
devices.append({
|
||||
"ip": received.psrc,
|
||||
"mac": received.hwsrc
|
||||
})
|
||||
|
||||
return devices
|
||||
|
||||
|
||||
def _ping_one(ip, is_win):
|
||||
"""Ping a single IP to populate ARP table."""
|
||||
try:
|
||||
if is_win:
|
||||
subprocess.run(
|
||||
["ping", "-n", "1", "-w", "600", str(ip)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
timeout=3,
|
||||
creationflags=_NO_WINDOW
|
||||
)
|
||||
else:
|
||||
subprocess.run(
|
||||
["ping", "-c", "1", "-W", "1", str(ip)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
timeout=3
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _ping_sweep(network, progress_cb=None):
|
||||
"""Ping all IPs in network concurrently to populate ARP table.
|
||||
Calls progress_cb(done, total) after each ping completes if provided.
|
||||
"""
|
||||
net = ipaddress.ip_network(network, strict=False)
|
||||
|
||||
# Only ping sweep for /24 or smaller to avoid flooding
|
||||
if net.num_addresses > 256:
|
||||
return
|
||||
|
||||
is_win = sys.platform == "win32"
|
||||
hosts = list(net.hosts())
|
||||
total = len(hosts)
|
||||
done_count = [0]
|
||||
|
||||
def _ping_and_track(ip):
|
||||
_ping_one(ip, is_win)
|
||||
done_count[0] += 1
|
||||
if progress_cb:
|
||||
progress_cb(done_count[0], total)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=50) as executor:
|
||||
futures = [executor.submit(_ping_and_track, ip) for ip in hosts]
|
||||
for f in as_completed(futures):
|
||||
pass
|
||||
|
||||
|
||||
def _scan_with_arp_table(network):
|
||||
"""Fallback: read ARP table using system 'arp -a' (no root needed).
|
||||
Supports both macOS and Windows output formats.
|
||||
"""
|
||||
# Ping sweep first to populate ARP table with active devices
|
||||
_ping_sweep(network)
|
||||
|
||||
# Brief pause to let the OS finalize ARP cache entries
|
||||
time.sleep(1)
|
||||
|
||||
try:
|
||||
output = subprocess.check_output(
|
||||
["arp", "-a"], text=True, creationflags=_NO_WINDOW
|
||||
)
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
devices = []
|
||||
net = ipaddress.ip_network(network, strict=False)
|
||||
|
||||
if sys.platform == "win32":
|
||||
# Windows format:
|
||||
# 192.168.4.1 cc-2d-21-a5-85-b0 dynamic
|
||||
pattern = re.compile(
|
||||
r"(\d+\.\d+\.\d+\.\d+)\s+"
|
||||
r"([0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-"
|
||||
r"[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2})\s+"
|
||||
r"(dynamic|static)",
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
for line in output.splitlines():
|
||||
m = pattern.search(line)
|
||||
if m:
|
||||
ip_str = m.group(1)
|
||||
# Convert Windows MAC format (cc-2d-21-a5-85-b0) to standard (cc:2d:21:a5:85:b0)
|
||||
mac = m.group(2).replace("-", ":")
|
||||
if mac.upper() != "FF:FF:FF:FF:FF:FF":
|
||||
try:
|
||||
if ipaddress.ip_address(ip_str) in net:
|
||||
devices.append({"ip": ip_str, "mac": mac})
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
# macOS/Linux format:
|
||||
# ? (192.168.1.1) at aa:bb:cc:dd:ee:ff on en0
|
||||
pattern = re.compile(
|
||||
r"\((\d+\.\d+\.\d+\.\d+)\)\s+at\s+([0-9a-fA-F:]+)"
|
||||
)
|
||||
|
||||
for line in output.splitlines():
|
||||
m = pattern.search(line)
|
||||
if m:
|
||||
ip_str, mac = m.group(1), m.group(2)
|
||||
if mac.lower() != "(incomplete)" and mac != "ff:ff:ff:ff:ff:ff":
|
||||
try:
|
||||
if ipaddress.ip_address(ip_str) in net:
|
||||
devices.append({"ip": ip_str, "mac": mac})
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
return devices
|
||||
|
||||
|
||||
def scan_network(network, progress_cb=None, stage_cb=None):
|
||||
"""Scan network: ping sweep first, then merge scapy ARP + arp table."""
|
||||
# Phase 1: Ping sweep — wake up devices and populate ARP cache
|
||||
if stage_cb:
|
||||
stage_cb("ping")
|
||||
_ping_sweep(network, progress_cb)
|
||||
time.sleep(1)
|
||||
|
||||
# Collect results from both methods and merge by IP
|
||||
seen = {} # ip -> device dict
|
||||
|
||||
# Phase 2: ARP table (populated by ping sweep above)
|
||||
if stage_cb:
|
||||
stage_cb("arp")
|
||||
try:
|
||||
output = subprocess.check_output(
|
||||
["arp", "-a"], text=True, creationflags=_NO_WINDOW
|
||||
)
|
||||
net = ipaddress.ip_network(network, strict=False)
|
||||
if sys.platform == "win32":
|
||||
pattern = re.compile(
|
||||
r"(\d+\.\d+\.\d+\.\d+)\s+"
|
||||
r"([0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-"
|
||||
r"[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2})\s+"
|
||||
r"(dynamic|static)",
|
||||
re.IGNORECASE
|
||||
)
|
||||
for line in output.splitlines():
|
||||
m = pattern.search(line)
|
||||
if m:
|
||||
ip_str = m.group(1)
|
||||
mac = m.group(2).replace("-", ":")
|
||||
if mac.upper() != "FF:FF:FF:FF:FF:FF":
|
||||
try:
|
||||
if ipaddress.ip_address(ip_str) in net:
|
||||
seen[ip_str] = {"ip": ip_str, "mac": mac}
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
pattern = re.compile(
|
||||
r"\((\d+\.\d+\.\d+\.\d+)\)\s+at\s+([0-9a-fA-F:]+)"
|
||||
)
|
||||
for line in output.splitlines():
|
||||
m = pattern.search(line)
|
||||
if m:
|
||||
ip_str, mac = m.group(1), m.group(2)
|
||||
if mac.lower() not in ("(incomplete)", "ff:ff:ff:ff:ff:ff"):
|
||||
try:
|
||||
if ipaddress.ip_address(ip_str) in net:
|
||||
seen[ip_str] = {"ip": ip_str, "mac": mac}
|
||||
except ValueError:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Phase 3: scapy ARP scan (if Npcap available) — fills in any gaps
|
||||
if stage_cb:
|
||||
stage_cb("scapy")
|
||||
try:
|
||||
import io, os
|
||||
_stderr = sys.stderr
|
||||
sys.stderr = io.StringIO()
|
||||
try:
|
||||
from scapy.all import ARP, Ether, srp
|
||||
finally:
|
||||
sys.stderr = _stderr
|
||||
|
||||
arp = ARP(pdst=str(network))
|
||||
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
|
||||
result = srp(ether / arp, timeout=2, verbose=0)[0]
|
||||
for sent, received in result:
|
||||
ip = received.psrc
|
||||
if ip not in seen:
|
||||
seen[ip] = {"ip": ip, "mac": received.hwsrc}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return sorted(seen.values(), key=lambda d: ipaddress.ip_address(d["ip"]))
|
||||
Reference in New Issue
Block a user