import re import string import time import sys import os import requests import gradio as gr from threading import Lock from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import asyncio import aiohttp import multiprocessing as mp # ═══════════════════════════════════════════════════ # ⚙️ ULTRA-FAST CONFIGURATION # ═══════════════════════════════════════════════════ TIMEOUT = 1.5 TARGET_TIME = 30 MIN_WORKERS = 500 MAX_WORKERS = 5000 BATCH_SIZE = 500 checked_lock = Lock() # ═══════════════════════════════════════════════════ # 🎨 UTILITY FUNCTIONS # ═══════════════════════════════════════════════════ def format_size(size_str): try: size = int(size_str) if size >= 1_000_000_000: return f"{size/1_000_000_000:.2f} GB" elif size >= 1_000_000: return f"{size/1_000_000:.2f} MB" elif size >= 1_000: return f"{size/1_000:.2f} KB" else: return f"{size} B" except: return size_str def format_time(seconds): if seconds < 60: return f"{seconds:.1f}s" elif seconds < 3600: mins = int(seconds // 60) secs = seconds % 60 return f"{mins}m {secs:.0f}s" else: hours = int(seconds // 3600) mins = int((seconds % 3600) // 60) return f"{hours}h {mins}m" def calculate_optimal_workers(total_requests): requests_per_second_per_worker = 100 total_needed_per_second = total_requests / TARGET_TIME optimal_workers = total_needed_per_second / requests_per_second_per_worker optimal_workers *= 1.2 optimal_workers = max(MIN_WORKERS, min(optimal_workers, MAX_WORKERS)) return int(optimal_workers) def parse_firmware_url(url): pattern = re.compile( r"^(https?://.+?/5G/UDO/)" r"DIN\d+/" r"([^_/]+)_(\d+)_S/" r"([A-Za-z])(.*)$" ) match = pattern.match(url.strip()) if match: return { 'base_prefix': match.group(1), 'model': match.group(2), 'current_version': int(match.group(3)), 'suffix': match.group(5), 'url': url.strip() } return None # ═══════════════════════════════════════════════════ # ⚡ ASYNC SCANNER # ═══════════════════════════════════════════════════ async def async_check_url(session, url, semaphore): async with semaphore: try: async with session.head( url, timeout=aiohttp.ClientTimeout(total=TIMEOUT), allow_redirects=True ) as r: if r.status in (403, 405): async with session.get( url, timeout=aiohttp.ClientTimeout(total=TIMEOUT) ) as r2: size = r2.headers.get("Content-Length", "Unknown") return r2.status == 200, url, size else: size = r.headers.get("Content-Length", "Unknown") return r.status == 200, url, size except: return False, url, "Error" async def batch_scan_urls(urls, max_concurrent=1000): semaphore = asyncio.Semaphore(max_concurrent) connector = aiohttp.TCPConnector( limit=max_concurrent, limit_per_host=max_concurrent, ttl_dns_cache=300, force_close=False, enable_cleanup_closed=True ) timeout = aiohttp.ClientTimeout(total=TIMEOUT, connect=0.5) async with aiohttp.ClientSession( connector=connector, timeout=timeout, headers={'Connection': 'keep-alive'} ) as session: tasks = [async_check_url(session, url, semaphore) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) return results def generate_urls(version, din_start, din_end, base_url, suffix): urls = [] for din_num in range(din_start, din_end + 1): din = f"DIN{din_num:03d}" for letter in string.ascii_lowercase + string.ascii_uppercase: filename = f"{letter}{suffix}" url = base_url.format(din=din, ver=version, file=filename) urls.append((url, version)) return urls # ═══════════════════════════════════════════════════ # 🤖 FIRMWARE SCANNER # ═══════════════════════════════════════════════════ class FirmwareScanner: def __init__(self): self.current_hits = [] self.scanning = False self.stop_requested = False def stop_scan(self): self.stop_requested = True self.scanning = False return "## Scan Stopped" def start_scan(self, firmware_url, version_input, din_input, stop_on_first): self.scanning = True self.stop_requested = False self.current_hits = [] if not firmware_url or not firmware_url.strip(): yield "## Error\n\nPlease enter a firmware URL.", "", "", "" return firmware_info = parse_firmware_url(firmware_url) if not firmware_info: yield "## Invalid URL\n\nPlease check the URL format.", "", "", "" return if not version_input or not version_input.strip(): yield "## Error\n\nPlease enter version or range.", "", "", "" return if not din_input or not din_input.strip(): yield "## Error\n\nPlease enter DIN or range.", "", "", "" return try: if '-' in version_input: parts = version_input.split('-') ver_start = int(parts[0].strip()) ver_end = int(parts[1].strip()) else: ver_start = ver_end = int(version_input.strip()) except: yield "## Invalid Version\n\nEnter a number or range.", "", "", "" return try: if '-' in din_input: parts = din_input.split('-') din_start = int(parts[0].strip()) din_end = int(parts[1].strip()) else: din_start = din_end = int(din_input.strip()) except: yield "## Invalid DIN\n\nEnter a number or range.", "", "", "" return total_versions = ver_end - ver_start + 1 din_count = din_end - din_start + 1 total_requests = total_versions * din_count * 52 optimal_workers = calculate_optimal_workers(total_requests) base_url = firmware_info['base_prefix'] + "{din}/" + firmware_info['model'] + "_{ver}_S/{file}" suffix = firmware_info['suffix'] model = firmware_info['model'] all_hits = [] overall_start = time.time() total_checked = 0 yield ( f"## Scanning Started\n\n" f"Model: {model}\n" f"Versions: {ver_start}-{ver_end}\n" f"DINs: {din_start}-{din_end}\n" f"Total URLs: {total_requests:,}\n" f"Workers: {optimal_workers}", "", f"## Config\n\n" f"| Setting | Value |\n" f"|---------|-------|\n" f"| Model | {model} |\n" f"| Versions | {ver_start}-{ver_end} |\n" f"| DINs | {din_start}-{din_end} |\n" f"| Total | {total_requests:,} |\n" f"| Workers | {optimal_workers} |", "" ) for version in range(ver_start, ver_end + 1): if self.stop_requested: break urls = generate_urls(version, din_start, din_end, base_url, suffix) total_urls = len(urls) should_stop = stop_on_first and ((ver_start == ver_end) or (din_start == din_end)) batch_size = min(BATCH_SIZE, total_urls) version_hits = [] for i in range(0, total_urls, batch_size): if self.stop_requested: break batch = urls[i:i+batch_size] batch_urls = [url for url, _ in batch] try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) results = loop.run_until_complete( batch_scan_urls(batch_urls, optimal_workers) ) loop.close() except: results = [] with ThreadPoolExecutor(max_workers=optimal_workers) as executor: def check_sync(url): try: r = requests.head(url, timeout=TIMEOUT, allow_redirects=True) if r.status_code in (403, 405): r = requests.get(url, timeout=TIMEOUT, stream=True) return r.status_code == 200, url, r.headers.get("Content-Length", "Unknown") except: return False, url, "Error" futures = [executor.submit(check_sync, url) for url in batch_urls] results = [f.result() for f in futures] for result in results: if isinstance(result, tuple) and len(result) == 3: ok, url, size = result if ok: hit = (version, url, size) version_hits.append(hit) all_hits.append(hit) with open("found_200_urls.txt", "a", encoding="utf-8") as f: f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]\n") f.write(f"Version: {version}\nURL: {url}\nSize: {size}\n{'─'*50}\n") if should_stop: self.stop_requested = True break total_checked += len(batch) elapsed = time.time() - overall_start rps = total_checked / elapsed if elapsed > 0 else 0 progress = (total_checked / total_requests) * 100 eta = (total_requests - total_checked) / rps if rps > 0 else 0 bar_width = 30 filled = int(bar_width * min(progress, 100) / 100) bar = '█' * filled + '░' * (bar_width - filled) progress_html = f""" ## Scanning v{version} ### {bar} {progress:.1f}% - Checked: {total_checked:,} / {total_requests:,} - Speed: {rps:,.0f} req/s - ETA: {format_time(eta)} - Elapsed: {format_time(elapsed)} - Hits: {len(all_hits)} """ stats_html = f""" ## Analytics | Metric | Value | |--------|-------| | Model | {model} | | Version | {version} | | DIN | {din_start:03d}-{din_end:03d} | | Progress | {progress:.1f}% | | Speed | {rps:,.0f} req/s | | Workers | {optimal_workers} | | Hits | {len(all_hits)} | | Elapsed | {format_time(elapsed)} | """ results_html = "" if version_hits: results_html = "## FIRMWARE FOUND!\n\n" for j, (ver, url, size) in enumerate(version_hits[-5:], 1): results_html += f""" ### Hit #{len(all_hits) - len(version_hits) + j} - Version: {ver} - Size: {format_size(size)} - URL: `{url}` """ yield progress_html, results_html, stats_html, "" if should_stop and version_hits: break if should_stop and version_hits: break self.scanning = False overall_elapsed = time.time() - overall_start summary_html = f""" ## Scan Complete! ### Performance - Duration: {format_time(overall_elapsed)} - Checked: {total_checked:,} - Speed: {int(total_checked/overall_elapsed) if overall_elapsed > 0 else 0:,} req/s - Hits: {len(all_hits)} ### Discovered Firmware """ if all_hits: seen = set() unique_hits = [] for hit in all_hits: if hit[1] not in seen: seen.add(hit[1]) unique_hits.append(hit) for j, (ver, url, size) in enumerate(unique_hits[:20], 1): summary_html += f""" **#{j}** v{ver} | {format_size(size)} `{url}` """ if len(unique_hits) > 20: summary_html += f"\n*...and {len(unique_hits) - 20} more*" summary_html += "\nSaved to `found_200_urls.txt`" else: summary_html += "\nNo firmware found." yield progress_html, results_html, stats_html, summary_html # ═══════════════════════════════════════════════════ # 🚀 GPU FUNCTION FOR HUGGING FACE # ═══════════════════════════════════════════════════ import spaces @spaces.GPU def gpu_scanner(input_text): """GPU-accelerated scanner for Hugging Face Spaces""" return input_text # Initialize GPU on startup gpu_scanner("init") # ═══════════════════════════════════════════════════ # 🎨 GRADIO UI # ═══════════════════════════════════════════════════ def create_ui(): scanner = FirmwareScanner() with gr.Blocks(title="FOTA Scanner") as app: gr.HTML("""

FOTA Firmware Scanner

30-Second Completion • 5,000+ Workers • Auto-Save

""") firmware_url = gr.Textbox( label="Firmware URL", placeholder="Enter firmware URL here...", lines=2, ) with gr.Row(): version_input = gr.Textbox( label="Version / Range", placeholder="Enter version or range", scale=1, ) din_input = gr.Textbox( label="DIN / Range", placeholder="Enter DIN or range", scale=1, ) stop_on_first = gr.Checkbox( label="Stop on first hit", value=False, ) with gr.Row(): scan_btn = gr.Button("START SCAN", variant="primary", size="lg") stop_btn = gr.Button("STOP", variant="stop", size="lg") progress_display = gr.Markdown("## Ready\n\nFill all fields and click START SCAN.") with gr.Tabs(): with gr.Tab("Results"): results_display = gr.Markdown("") with gr.Tab("Analytics"): stats_display = gr.Markdown("") with gr.Tab("Summary"): summary_display = gr.Markdown("") scan_event = scan_btn.click( fn=scanner.start_scan, inputs=[firmware_url, version_input, din_input, stop_on_first], outputs=[progress_display, results_display, stats_display, summary_display] ) stop_btn.click( fn=scanner.stop_scan, inputs=[], outputs=[progress_display], cancels=[scan_event] ) return app # ═══════════════════════════════════════════════════ # 🚀 LAUNCH # ═══════════════════════════════════════════════════ if __name__ == "__main__": mp.set_start_method('spawn', force=True) app = create_ui() app.launch( server_name="0.0.0.0", server_port=7860, share=False, debug=False )