mirror of
https://github.com/fr33n0w/rmap-rnode-web-flasher
synced 2026-08-12 18:08:52 -04:00
200 lines
8.5 KiB
Python
200 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
server.py - server autocontenuto per il RNode Flasher.
|
|
- Serve i file statici (URL relativi → portabile PC/VPS, zero hardcoding).
|
|
- /releases?src=<key> → elenco release (tag, data, asset) via GitHub API
|
|
- /firmware?file=<zip>&src=<key>&tag=<tag> → scarica lo zip da GitHub LATO SERVER
|
|
(in RAM, mai su disco) e lo passa al browser. Senza tag = ultima release.
|
|
|
|
Uso: python server.py [porta] (default 8000). Identico in locale e su VPS.
|
|
"""
|
|
import http.server
|
|
import socketserver
|
|
import urllib.request
|
|
import urllib.parse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import shutil
|
|
import threading
|
|
|
|
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
CACHE_DIR = os.path.join(ROOT, "fw_cache") # firmware scaricati una volta e riusati (basso uso RAM)
|
|
CHUNK = 65536
|
|
|
|
# sorgenti firmware: chiave -> repo GitHub (owner/repo). Estendibile a piacere.
|
|
SOURCES = {
|
|
"official": "markqvist/RNode_Firmware",
|
|
"ce": "liberatedsystems/RNode_Firmware_CE",
|
|
"tn": "attermann/microReticulum_Firmware", # microReticulum / Transport Node (experimental)
|
|
"rtnode": "jrl290/RTNode-HeltecV4", # RTNode v4 standalone (Heltec V3/V4)
|
|
"t1000e": "idan2025/Rnode_Firmware", # Seeed SenseCAP T1000-E (nRF52 DFU)
|
|
}
|
|
STATS_FILE = os.path.join(ROOT, "stats.json") # contatori visite + flash (persistiti)
|
|
_stats_lock = threading.Lock()
|
|
NAME_RE = re.compile(r"^[A-Za-z0-9._-]+\.(zip|bin)$") # .zip (RNode) o .bin (RTNode); no traversal
|
|
TAG_RE = re.compile(r"^[A-Za-z0-9._-]+$")
|
|
|
|
|
|
class Handler(http.server.SimpleHTTPRequestHandler):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, directory=ROOT, **kwargs)
|
|
|
|
def log_message(self, fmt, *args):
|
|
sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))
|
|
|
|
def _send(self, code, body, ctype):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _fetch(self, url, accept=None):
|
|
headers = {"User-Agent": "rnode-flasher"}
|
|
if accept:
|
|
headers["Accept"] = accept
|
|
req = urllib.request.Request(url, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
return resp.read()
|
|
|
|
# contatori visite/flash, esclusivi del flasher. ?hit=visit | ?hit=flash incrementa.
|
|
def stats(self, q):
|
|
hit = (q.get("hit") or [""])[0]
|
|
with _stats_lock:
|
|
try:
|
|
with open(STATS_FILE, "r") as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
data = {"visits": 0, "flashed": 0}
|
|
if hit == "visit":
|
|
data["visits"] = int(data.get("visits", 0)) + 1
|
|
elif hit == "flash":
|
|
data["flashed"] = int(data.get("flashed", 0)) + 1
|
|
if hit in ("visit", "flash"):
|
|
try:
|
|
with open(STATS_FILE, "w") as f:
|
|
json.dump(data, f)
|
|
except Exception as e:
|
|
sys.stderr.write("[stats] write failed: %s\n" % e)
|
|
out = {"visits": int(data.get("visits", 0)), "flashed": int(data.get("flashed", 0))}
|
|
self._send(200, json.dumps(out).encode(), "application/json")
|
|
|
|
def do_GET(self):
|
|
parsed = urllib.parse.urlparse(self.path)
|
|
q = urllib.parse.parse_qs(parsed.query)
|
|
if parsed.path == "/stats":
|
|
return self.stats(q)
|
|
if parsed.path == "/releases":
|
|
return self.list_releases(q)
|
|
if parsed.path == "/firmware":
|
|
return self.proxy_firmware(q)
|
|
return super().do_GET()
|
|
|
|
def list_releases(self, q):
|
|
src = (q.get("src") or ["official"])[0]
|
|
if src not in SOURCES:
|
|
self.send_error(400, "bad src"); return
|
|
api = "https://api.github.com/repos/%s/releases?per_page=50" % SOURCES[src]
|
|
try:
|
|
raw = self._fetch(api, accept="application/vnd.github+json")
|
|
data = json.loads(raw)
|
|
except Exception as e:
|
|
self.send_error(502, "releases fetch failed: %s" % e); return
|
|
out = [{
|
|
"tag": rel.get("tag_name"),
|
|
"name": rel.get("name"),
|
|
"prerelease": rel.get("prerelease", False),
|
|
"published_at": rel.get("published_at"),
|
|
"assets": [a.get("name") for a in rel.get("assets", [])],
|
|
} for rel in data]
|
|
self._send(200, json.dumps(out).encode(), "application/json")
|
|
|
|
def proxy_firmware(self, q):
|
|
name = (q.get("file") or [""])[0]
|
|
src = (q.get("src") or ["official"])[0]
|
|
tag = (q.get("tag") or [""])[0]
|
|
if not NAME_RE.match(name) or src not in SOURCES:
|
|
self.send_error(400, "bad request"); return
|
|
if tag and not TAG_RE.match(tag):
|
|
self.send_error(400, "bad tag"); return
|
|
|
|
repo = SOURCES[src]
|
|
if tag:
|
|
url = "https://github.com/%s/releases/download/%s/%s" % (repo, tag, name)
|
|
else:
|
|
url = "https://github.com/%s/releases/latest/download/%s" % (repo, name)
|
|
|
|
sys.stderr.write("[firmware] %s\n" % url)
|
|
|
|
# Versione specifica (tag) -> cache su disco riusabile (l'asset di una release è immutabile).
|
|
# Senza tag ("latest") -> stream diretto, niente cache (potrebbe cambiare).
|
|
if tag:
|
|
cache_name = "%s__%s__%s" % (src, tag, name) # tutti caratteri già validati
|
|
cache_path = os.path.join(CACHE_DIR, cache_name)
|
|
if not os.path.exists(cache_path):
|
|
tmp = cache_path + ".part"
|
|
try:
|
|
os.makedirs(CACHE_DIR, exist_ok=True) # dentro il try: permessi/disco -> 502 pulito
|
|
req = urllib.request.Request(url, headers={"User-Agent": "rnode-flasher"})
|
|
with urllib.request.urlopen(req, timeout=120) as resp, open(tmp, "wb") as f:
|
|
shutil.copyfileobj(resp, f, CHUNK) # GitHub -> disco a chunk (RAM minima)
|
|
os.replace(tmp, cache_path)
|
|
except Exception as e:
|
|
try:
|
|
os.remove(tmp)
|
|
except OSError:
|
|
pass
|
|
self.send_error(502, "firmware fetch failed: %s" % e); return
|
|
else:
|
|
sys.stderr.write("[firmware] cache hit: %s\n" % cache_name)
|
|
try:
|
|
size = os.path.getsize(cache_path)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/octet-stream")
|
|
self.send_header("Content-Length", str(size))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
with open(cache_path, "rb") as f:
|
|
shutil.copyfileobj(f, self.wfile, CHUNK) # disco -> browser a chunk (RAM minima)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
return
|
|
|
|
# latest: stream diretto GitHub -> browser, senza bufferare tutto in RAM
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "rnode-flasher"})
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/octet-stream")
|
|
cl = resp.headers.get("Content-Length")
|
|
if cl:
|
|
self.send_header("Content-Length", cl)
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
shutil.copyfileobj(resp, self.wfile, CHUNK)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
except Exception as e:
|
|
self.send_error(502, "firmware fetch failed: %s" % e)
|
|
|
|
|
|
class ThreadingServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
|
|
daemon_threads = True
|
|
allow_reuse_address = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
with ThreadingServer(("0.0.0.0", PORT), Handler) as httpd:
|
|
print("RNode Flasher su http://localhost:%d/ (Ctrl+C per fermare)" % PORT)
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nstop.")
|