diff --git a/A_pwmenu.py b/A_pwmenu.py index 181f6c0..2ed3553 100644 --- a/A_pwmenu.py +++ b/A_pwmenu.py @@ -17,6 +17,7 @@ import email.utils import html import hashlib +import hmac import shutil import tempfile import ast @@ -25,7 +26,7 @@ import urllib.parse import pwnagotchi.plugins as plugins import pwnagotchi.ui.fonts as fonts -from flask import render_template_string, send_file, make_response, has_request_context +from flask import current_app, send_file, make_response, has_request_context from flask import request as flask_request from pwnagotchi.ui.components import LabeledValue from pwnagotchi.ui.view import BLACK @@ -49,7 +50,7 @@ def get_local_time(timestamp, tz_offset): class A_pwmenu(plugins.Plugin): __author__ = 'NewFPV' - __version__ = '1.3.8' + __version__ = '1.3.9' __license__ = 'GPL3' __description__ = 'Ultimate Password Manager' @@ -120,6 +121,9 @@ def __init__(self): self.quickdic_thread = None self.quickdic_display_status = '' self.quickdic_display_until = 0 + self.web_template_lock = threading.Lock() + self.web_template = None + self.web_template_environment = None def _module_enabled(self, name, default=True): return self._option_bool(f'module_{name}_enabled', default) @@ -159,6 +163,8 @@ def on_loaded(self): self.options.setdefault('ohc_proxy_port', 10809) self.options.setdefault('import_max_bytes', 2097152) self.options.setdefault('archive_memory_limit', 2097152) + self.options.setdefault('web_gzip_level', 1) + self.options.setdefault('web_notification_duration_ms', 2600) self.options.setdefault('hcxpcapngtool_timeout', 90) self.options.setdefault('password_verify_timeout', 45) self.options.setdefault('wpa_sec_api_url', 'https://wpa-sec.stanev.org') @@ -962,18 +968,41 @@ def _render_page(self, notification=None, notif_type=None, active_tab='cracked') self._module_enabled('wpa_sec') and self.options.get('wpa_sec_key') ) - html = render_template_string(self._get_html(), + notification_duration_ms = max( + 250, + min( + self._option_int('web_notification_duration_ms', 2600), + 60000, + ), + ) + html = self._render_cached_template( groups=groups, cracked=cracked, notif=notification, ntype=notif_type, tab=active_tab, stats=stats, ach=ach['badges'], token=tok, show_wpa=show_wpa, map_points=map_points, gps_status=gps_status, no_gps_networks=no_gps_networks, ohc_status=ohc_status, wpa_status=wpa_status, pot_health=pot_health, cleanup_report=cleanup_report, - whitelist=whitelist + whitelist=whitelist, + notification_duration_ms=notification_duration_ms, ) return self._html_response(html) + def _render_cached_template(self, **context): + """Compile the large Jinja template once per Flask environment.""" + app = current_app._get_current_object() + app.update_template_context(context) + environment = app.jinja_env + with self.web_template_lock: + if ( + self.web_template is None + or self.web_template_environment is not environment + ): + self.web_template = environment.from_string(self._get_html()) + self.web_template_environment = environment + template = self.web_template + return template.render(context) + def _html_response(self, html): """Return the UI efficiently, especially over low-bandwidth Bluetooth PAN.""" body = html.encode('utf-8') if isinstance(html, str) else bytes(html) @@ -987,7 +1016,11 @@ def _html_response(self, html): and flask_request.accept_encodings['gzip'] > 0 ) if accepts_gzip and len(body) >= 1024: - compressed = gzip.compress(body, compresslevel=6) + gzip_level = max( + 1, + min(self._option_int('web_gzip_level', 1), 9), + ) + compressed = gzip.compress(body, compresslevel=gzip_level) if len(compressed) < len(body): r.set_data(compressed) r.headers["Content-Encoding"] = "gzip" @@ -1327,7 +1360,7 @@ def _download_wpa_results(self, key): response = requests.get( api_url, cookies={'key': str(key)}, - headers={'User-Agent': 'PWMenu/1.3.8'}, + headers={'User-Agent': 'PWMenu/1.3.9'}, timeout=(10, 30), ) response.raise_for_status() @@ -1365,6 +1398,16 @@ def _handle_ohc_cluster_upload(self, req): if not names: return "No files selected", True queued = self._queue_ohc_files(names, force=True) + if not queued: + reasons = [] + for name in names: + record = self._ohc_file_record(name) + message = str(record.get('message') or '').strip() + if message and message not in reasons: + reasons.append(message) + if reasons: + return "Nothing queued: " + "; ".join(reasons[:3]), True + return "Nothing queued: no unresolved usable capture was selected", True self._start_ohc_upload_thread() if time.time() < self._ohc_retry_at(): return f"Queued {queued} file(s). {self._ohc_backoff_message()}", False @@ -2629,7 +2672,7 @@ def _upload_path_to_wpa(self, path, key): headers={ 'User-Agent': ( 'Mozilla/5.0 (X11; Linux aarch64) ' - 'PWMenu/1.3.8' + 'PWMenu/1.3.9' ) }, timeout=(10, 30) @@ -2858,14 +2901,14 @@ def _save_data(self): pass def _option_bool(self, key, default=True): - v = self.options.get(key, default) + v = getattr(self, 'options', {}).get(key, default) if isinstance(v, str): return v.strip().lower() not in ('0', 'false', 'no', 'off') return bool(v) def _option_int(self, key, default): try: - return int(self.options.get(key, default)) + return int(getattr(self, 'options', {}).get(key, default)) except (TypeError, ValueError): return default @@ -3146,6 +3189,19 @@ def _gps_status(self): def _ohc_status(self): with self.data_lock: pending = len(self.data.get('ohc_pending_files', {}) or {}) + records = dict(self.data.get('ohc_files', {}) or {}) + uncrackable = 0 + for name, record in records.items(): + try: + hash_count = int((record or {}).get('hashes', 0) or 0) + except (TypeError, ValueError): + hash_count = 0 + if ( + str((record or {}).get('status') or '') == 'invalid' + and hash_count <= 0 + and self._find_handshake_path(name) + ): + uncrackable += 1 retry_in = max(0, int(self._ohc_retry_at() - time.time())) return { 'enabled': ( @@ -3156,6 +3212,7 @@ def _ohc_status(self): 'face': self.ohc_upload_face if self.ohc_uploading else '0__0', 'last': self.ohc_last_result, 'pending': pending, + 'uncrackable': uncrackable, 'retry_in': retry_in } @@ -3494,17 +3551,36 @@ def _build_map_points(self, groups): bucket['count'] = sum(max(1, int(x.get('captures', 1) or 1)) for x in bucket['members']) bucket['essid'] = f"{bucket['count']} networks" else: - c = m.copy() - c['id'] = m.get('id') or f"{m['essid']}-{m['lat']}-{m['lon']}" - c['members'] = [m] - c['count'] = max(1, int(m.get('captures', 1) or 1)) + c = { + 'id': m.get('id') or f"{m['essid']}-{m['lat']}-{m['lon']}", + 'essid': m.get('essid', ''), + 'bssid': '', + 'lat': m.get('lat'), + 'lon': m.get('lon'), + 'is_cracked': bool(m.get('is_cracked')), + 'gps_stale': bool(m.get('gps_stale')), + 'status': m.get('status', 'handshake'), + 'count': max(1, int(m.get('captures', 1) or 1)), + 'members': [m], + } clusters.append(c) - for c in clusters: - if c.get('count', 1) == 1: - c.update(c['members'][0]) - c['count'] = max(1, int(c.get('captures', 1) or 1)) - return clusters + compact_clusters = [] + for cluster in clusters: + members = cluster.get('members', []) + if len(members) == 1: + # A singleton already carries all display fields. Do not send + # a second complete copy of it in `members` on every page load. + single = members[0].copy() + single['members'] = [] + single['count'] = max( + 1, + int(single.get('captures', 1) or 1), + ) + compact_clusters.append(single) + else: + compact_clusters.append(cluster) + return compact_clusters def _build_no_gps_networks(self, groups): items = [] @@ -3546,16 +3622,19 @@ def _update_achievements(self, groups, cracked): def _update_achievements_locked(self, groups, cracked): curr_cracked = len(cracked) curr_captured = sum(len(g['files']) for g in groups) + changed = False if curr_cracked > self.data['history_cracked']: diff = curr_cracked - self.data['history_cracked'] self.data['xp'] += diff * 500 self.data['history_cracked'] = curr_cracked + changed = True if curr_captured > self.data['history_captured']: diff = curr_captured - self.data['history_captured'] self.data['xp'] += diff * 50 self.data['history_captured'] = curr_captured + changed = True lvl_map = [ (0, 'Script Kiddie'), (1000, 'Neophyte'), (2500, 'Hacker'), @@ -3599,13 +3678,15 @@ def _update_achievements_locked(self, groups, cracked): self.data['xp'] += 1000 ul = True pct = 100 + changed = True badge_info = b.copy() badge_info['unlocked'] = ul badge_info['progress'] = pct my_badges.append(badge_info) - self._save_data() + if changed: + self._save_data() lvl_p = 100 if next_xp > prev_xp: @@ -3710,6 +3791,162 @@ def _run_aircrack_password_check(self, path, essid, bssid, passwords): except FileNotFoundError: pass + def _verify_pmkid_hash_password(self, hash_line, password): + """Verify one WPA*01 PMKID record locally without spawning a process.""" + parts = str(hash_line or '').strip().lstrip('$').split('*') + if len(parts) < 6 or parts[0] != 'WPA' or parts[1] != '01': + return None + try: + expected_pmkid = bytes.fromhex(parts[2]) + ap_mac = bytes.fromhex(parts[3]) + station_mac = bytes.fromhex(parts[4]) + essid_bytes = bytes.fromhex(parts[5]) + except (TypeError, ValueError): + return None + if ( + len(expected_pmkid) != 16 + or len(ap_mac) != 6 + or len(station_mac) != 6 + or not essid_bytes + ): + return None + try: + password_bytes = str(password).encode('utf-8') + pmk = hashlib.pbkdf2_hmac( + 'sha1', + password_bytes, + essid_bytes, + 4096, + dklen=32, + ) + calculated = hmac.new( + pmk, + b'PMK Name' + ap_mac + station_mac, + hashlib.sha1, + ).digest()[:16] + return hmac.compare_digest(calculated, expected_pmkid) + except (TypeError, ValueError, UnicodeError): + return None + + def _run_hcxpmk_password_check(self, path, essid, bssid, passwords): + """Extract and verify PMKID hashes without exposing passwords in argv.""" + converter = ( + shutil.which('hcxpcapngtool') + or ( + '/usr/bin/hcxpcapngtool' + if os.path.isfile('/usr/bin/hcxpcapngtool') + else '' + ) + ) + if not converter: + return False, False, 'PMKID extraction is unavailable' + + candidates = [ + str(password) for password in passwords + if str(password) + and not any( + char in str(password) for char in ('\x00', '\r', '\n') + ) + ] + if not candidates: + return False, False, 'No valid password candidate was supplied' + + output_path = None + try: + fd, output_path = tempfile.mkstemp( + prefix='pwmenu-verify-', suffix='.22000' + ) + os.close(fd) + os.remove(output_path) + hashes = [] + conversion = None + conversion_commands = ( + [converter, '-o', output_path, path], + [converter, '--all', '-o', output_path, path], + ) + for conversion_command in conversion_commands: + try: + os.remove(output_path) + except FileNotFoundError: + pass + with self.capture_analysis_lock: + conversion = subprocess.run( + conversion_command, + check=False, + capture_output=True, + text=True, + errors='replace', + timeout=self._option_int( + 'hcxpcapngtool_timeout', 90 + ), + ) + if os.path.isfile(output_path): + with open( + output_path, + 'r', + encoding='utf-8', + errors='ignore', + ) as handle: + hashes = list(dict.fromkeys( + line.strip() for line in handle if line.strip() + )) + if hashes: + break + + compact_bssid = self._compact_bssid(bssid) + normalized_essid = self._normalized_essid_key(essid) + matching_hashes = [] + for hash_line in hashes: + hash_essid, hash_bssid = self._wpa_hash_network(hash_line) + if compact_bssid: + matched = hash_bssid == compact_bssid + else: + matched = ( + bool(normalized_essid) + and self._normalized_essid_key(hash_essid) + == normalized_essid + ) + if matched: + matching_hashes.append(hash_line) + + if not matching_hashes: + detail = ( + (conversion.stdout or '') + '\n' + + (conversion.stderr or '') + ).strip() + return ( + False, + False, + 'No matching PMKID/EAPOL hash was extracted' + + (f": {detail[-300:]}" if detail else ''), + ) + + tested = False + for hash_line in matching_hashes: + for password in candidates: + confirmed = self._verify_pmkid_hash_password( + hash_line, password + ) + if confirmed is True: + return ( + True, + True, + 'Password confirmed against PMKID', + ) + if confirmed is False: + tested = True + if tested: + return False, True, 'Password was not confirmed by PMKID' + return False, False, 'No verifiable PMKID hash was extracted' + except (OSError, subprocess.SubprocessError) as error: + return False, False, str(error) + finally: + if output_path: + try: + os.remove(output_path) + except FileNotFoundError: + pass + def _verify_manual_password(self, essid, bssid, password): paths = self._matching_capture_paths(essid, bssid) if not paths: @@ -3720,23 +3957,58 @@ def _verify_manual_password(self, essid, bssid, password): 'the password was not saved', ) - inconclusive = False + conclusive_failure = False + inconclusive_details = [] for path in paths[:3]: - verified, conclusive, detail = self._run_aircrack_password_check( - path, essid, bssid, [password] + verified, aircrack_conclusive, aircrack_detail = ( + self._run_aircrack_password_check( + path, essid, bssid, [password] + ) ) if verified: return ( True, f'Password verified against {os.path.basename(path)}', ) - if not conclusive: - inconclusive = True - logging.warning( - f"[A_pwmenu] Manual password verification was inconclusive " - f"for {os.path.basename(path)}: {detail}" + + verified, hcx_conclusive, hcx_detail = ( + self._run_hcxpmk_password_check( + path, essid, bssid, [password] + ) + ) + if verified: + return ( + True, + f'Password verified against {os.path.basename(path)}', + ) + + if aircrack_conclusive or hcx_conclusive: + conclusive_failure = True + continue + detail = '; '.join( + value for value in (aircrack_detail, hcx_detail) if value + ) + inconclusive_details.append(detail) + logging.warning( + f"[A_pwmenu] Manual password verification was inconclusive " + f"for {os.path.basename(path)}: {detail}" + ) + + if conclusive_failure: + return False, 'Password does not match the captured handshake' + if inconclusive_details: + combined = ' '.join(inconclusive_details) + if re.search( + r'no matching pmkid|no hashes|no eapol|contained no eapol', + combined, + re.IGNORECASE, + ): + return ( + False, + 'Password cannot be verified because this capture ' + 'contains no usable WPA/PMKID hash. Recapture the ' + 'access point', ) - if inconclusive: return ( False, 'The password could not be verified conclusively and was not saved', @@ -3752,6 +4024,11 @@ def _add_manual_password(self, essid, bssid, pwd): name, compact_bssid, password ) if not verified: + logging.warning( + "[A_pwmenu] Manual password rejected for %s: %s", + self._format_bssid(compact_bssid) or name, + message, + ) return False, message mac = self._colon_bssid(compact_bssid) or "00:00:00:00:00:00" @@ -3767,7 +4044,15 @@ def _add_manual_password(self, essid, bssid, pwd): with self.data_lock: self.data['xp'] += 200 self._save_data() + logging.info( + "[A_pwmenu] Manual password saved for %s", + self._format_bssid(compact_bssid) or name, + ) return True, message + '; password saved' + logging.info( + "[A_pwmenu] Manual password was already saved for %s", + self._format_bssid(compact_bssid) or name, + ) return True, message + '; password was already saved' except (OSError, ValueError) as error: logging.error(f"[A_pwmenu] Could not add manual password: {error}") @@ -4137,6 +4422,14 @@ def _is_empty_pcap(self, path): except OSError: return False + def _quality_is_uncrackable(self, quality): + if not isinstance(quality, dict) or quality.get('grade') != 'Partial': + return False + try: + return int(quality.get('hashes', 0) or 0) <= 0 + except (TypeError, ValueError): + return True + def _capture_cleanup_report(self): entries = [] for directory in self.handshake_dirs: @@ -4147,15 +4440,31 @@ def _capture_cleanup_report(self): empty = self._is_empty_pcap(path) quality = self._quality_file_record(name, path) unusable = quality.get('grade') == 'Unusable' - if not empty and not unusable: + uncrackable = self._quality_is_uncrackable(quality) + if not empty and not unusable and not uncrackable: continue - reason = 'Empty PCAP header' if empty else (quality.get('summary') or 'No usable WPA/PMKID material') + if empty: + reason = 'Empty PCAP header' + category = 'empty' + elif uncrackable: + reason = ( + quality.get('summary') + or 'Incomplete exchange; no usable WPA/PMKID hash' + ) + category = 'uncrackable' + else: + reason = ( + quality.get('summary') + or 'No usable WPA/PMKID material' + ) + category = 'unusable' entries.append({ 'name': name, 'path': path, 'signature': self._ohc_file_signature(path), 'reason': reason, 'empty': empty, + 'category': category, }) fingerprint = json.dumps( [(entry['path'], entry['signature'], entry['reason']) for entry in entries], @@ -4165,7 +4474,14 @@ def _capture_cleanup_report(self): return { 'count': len(entries), 'empty_count': len([entry for entry in entries if entry['empty']]), - 'unusable_count': len([entry for entry in entries if not entry['empty']]), + 'uncrackable_count': len([ + entry for entry in entries + if entry['category'] == 'uncrackable' + ]), + 'unusable_count': len([ + entry for entry in entries + if entry['category'] == 'unusable' + ]), 'display_files': entries[:12], 'more': max(0, len(entries) - 12), 'token': hashlib.sha256(fingerprint).hexdigest(), @@ -4176,7 +4492,11 @@ def _clean_capture_candidates(self, report_token): report = self._capture_cleanup_report() total = report['count'] if not total: - return 0, total, 'No empty or unusable capture files to remove' + return ( + 0, + total, + 'No empty, uncrackable, or unusable capture files to remove', + ) if not report_token: return 0, total, 'Review the current cleanup report and confirm again.' if not re.fullmatch(r'[0-9a-f]{64}', report_token) or report_token != report['token']: @@ -4188,7 +4508,12 @@ def _clean_capture_candidates(self, report_token): if self._ohc_file_signature(path) != entry['signature']: continue quality = self._quality_file_record(entry['name'], path) - if not self._is_empty_pcap(path) and quality.get('grade') != 'Unusable': + still_uncrackable = self._quality_is_uncrackable(quality) + if ( + not self._is_empty_pcap(path) + and quality.get('grade') != 'Unusable' + and not still_uncrackable + ): continue try: os.remove(path) @@ -4211,7 +4536,11 @@ def _clean_capture_candidates(self, report_token): except OSError as error: logging.warning(f"[A_pwmenu] Could not remove capture {path}: {error}") logging.info(f"[A_pwmenu] Capture cleanup removed {deleted}/{total} file(s)") - return deleted, total, f"Removed {deleted}/{total} empty or unusable capture files" + return ( + deleted, + total, + f"Removed {deleted}/{total} empty, uncrackable, or unusable capture files", + ) def _process_import(self, content, name): self._ensure_file(self.potfile_ohc) @@ -5997,7 +6326,7 @@ def _get_html(self):
-
{{ g.essid }} {% if g.count > 1 %}{{ g.count }}{% endif %}{% if g.map_count > 0 %}MAP{% elif g.gps_count > 0 %}GPS{% endif %}
+
{{ g.essid }} {% if g.count > 1 %}{{ g.count }}{% endif %}{% if g.map_count > 0 %}MAP{% elif g.gps_count > 0 %}GPS{% endif %}
{{ g.last_seen }}
@@ -6022,7 +6351,7 @@ def _get_html(self):
{% for f in g.files %} -
+
{{ f.filename }}
@@ -6033,7 +6362,7 @@ def _get_html(self):
- + {% if show_wpa %}{% endif %} 22000 @@ -6144,6 +6473,9 @@ def _get_html(self): Persistent queue: {{ ohc_status.pending }} file(s) {% if ohc_status.retry_in > 0 %} • retry in {{ ohc_status.retry_in }}s{% endif %}
+ {% if ohc_status.uncrackable %} +
{{ ohc_status.uncrackable }} capture(s) cannot be sent because they contain no extractable WPA/PMKID hash. Review Capture Cleanup below.
+ {% endif %}
Queues one best unresolved PCAP per BSSID. Local history and the last imported OHC export are excluded before upload.
@@ -6262,13 +6594,13 @@ def _get_html(self):
{{ cleanup_report.count }} cleanup candidate(s)
-
{{ cleanup_report.empty_count }} empty header(s) and {{ cleanup_report.unusable_count }} analyzed unusable capture(s). Every file and signature is checked again immediately before deletion.
+
{{ cleanup_report.empty_count }} empty header(s), {{ cleanup_report.uncrackable_count }} incomplete capture(s) without a usable WPA/PMKID hash, and {{ cleanup_report.unusable_count }} analyzed unusable capture(s). Every file and signature is checked again immediately before deletion.
{% if cleanup_report.display_files %}
    {% for item in cleanup_report.display_files %}
  • {{ item.name }}{{ item.reason }}
    • {% endfor %} {% if cleanup_report.more %}
  • + {{ cleanup_report.more }} more file(s)
    • {% endif %}
-
+ @@ -6289,6 +6621,7 @@ def _get_html(self): let noGpsNetworks = {{ no_gps_networks|tojson }}; const gpsStatus = {{ gps_status|tojson }}; const whitelistedNetworks = new Set({{ whitelist|tojson }}); + const notificationDurationMs = {{ notification_duration_ms }}; let gpsWatchId = null; let selectedMapPoint = null; let userLocation = null; @@ -6309,7 +6642,7 @@ def _get_html(self): startPhoneGps(false); updateGpsStatusDot(); renderMap(); - if(document.getElementById('nt')) setTimeout(() => document.getElementById('nt').style.display='none', 3000); + if(document.getElementById('nt')) setTimeout(() => document.getElementById('nt').style.display='none', notificationDurationMs); function post(u, d) { const f = document.createElement('form'); f.method='POST'; f.action='/plugins/A_pwmenu/'+u; @@ -6843,6 +7176,37 @@ def _get_html(self): saveHandshakeMapPoint(coordinates.lat, coordinates.lon); } + function applyMapPlacementToHandshake(placement) { + if(!placement) return; + document.querySelectorAll('#v-handshakes .handshake-card').forEach(card => { + const cardBssid = String(card.dataset.bssid || '').replace(/[^0-9a-f]/gi, '').toLowerCase(); + const placedBssid = String(placement.bssid || '').replace(/[^0-9a-f]/gi, '').toLowerCase(); + const sameNetwork = cardBssid && placedBssid + ? cardBssid === placedBssid + : String(card.dataset.essid || '') === String(placement.essid || ''); + if(!sameNetwork) return; + + const title = card.querySelector('.tit'); + if(title) { + let badge = title.querySelector('[data-location-badge]'); + if(!badge) { + badge = document.createElement('span'); + badge.className = 'badge location-badge'; + badge.dataset.locationBadge = ''; + title.appendChild(document.createTextNode(' ')); + title.appendChild(badge); + } + badge.textContent = 'MAP'; + } + + card.querySelectorAll('.capture-row[data-filename]').forEach(row => { + if(String(row.dataset.filename || '') !== String(placement.filename || '')) return; + const label = row.querySelector('.map-capture-action span'); + if(label) label.textContent = 'Move'; + }); + }); + } + async function saveHandshakeMapPoint(lat, lon) { if(!mapPlacement || mapPlacement.saving) return; const latitude = Number(lat); @@ -6855,7 +7219,7 @@ def _get_html(self): mapPlacement.saving = true; const placement = Object.assign({}, mapPlacement); const placementName = placement.essid || placement.filename; - showToast(`Adding ${placementName} to the map...`, false, 6000); + showToast(`Adding ${placementName} to the map...`); try { const result = await postAsync('capture-map-set', { filename:placement.filename, @@ -6865,10 +7229,11 @@ def _get_html(self): if(!result.ok) throw new Error(result.message || 'Location was rejected'); if(result.map_points) mapPoints = result.map_points; if(result.no_gps_networks) noGpsNetworks = result.no_gps_networks; + applyMapPlacementToHandshake(placement); cancelMapPlacement(false); const point = findMapMemberById(result.point_id || placement.filename); if(point) showMapPoint(point); - showToast(`Added ${placementName} to the map`, false, 5000); + showToast(`Added ${placementName} to the map`); } catch(error) { if(mapPlacement) mapPlacement.saving = false; showToast(error.message || 'Could not save handshake location', true); @@ -6903,7 +7268,7 @@ def _get_html(self): showToast('Copied'); } - function showToast(text, isError, duration = 2600) { + function showToast(text, isError, duration = notificationDurationMs) { const toast = document.getElementById('mapToast'); if(!toast) return; toast.textContent = text || 'Done'; diff --git a/CHANGELOG.md b/CHANGELOG.md index 44a3823..bb0fb5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,71 @@ All notable changes to A_pwmenu are documented here. +## 1.3.9 — 2026-07-29 + +### Web UI performance and behavior + +- Cached the compiled Jinja template per Flask environment instead of parsing + and compiling the approximately 400 KB source template on every request. +- Stopped rewriting `.a_pwmenu_data.json` and its backup during ordinary page + views when achievement state did not change. +- Changed the default HTML compression level from gzip 6 to gzip 1 to avoid + multi-second compression stalls on Raspberry Pi while retaining a compressed + response. Added `web_gzip_level` with a validated range of 1–9. +- Removed the duplicated nested copy of every singleton map point from the page + payload. Cluster members remain complete when a real multi-network cluster + exists. +- Added `web_notification_duration_ms`, clamped to 250–60000 milliseconds, and + applied it consistently to server notifications and transient Web UI toasts. +- Confirmed on the development Pwnagotchi that repeated compressed page + responses dropped from roughly six seconds to approximately 1.3–1.6 seconds; + actual results depend on capture count, storage, and transport. + +### Manual password verification + +- Kept `aircrack-ng` verification for captures containing a usable EAPOL + exchange. +- Added PMKID-only verification: `hcxpcapngtool` extracts the exact WPA*01 + record, then PWMenu derives the PMK with PBKDF2-HMAC-SHA1, calculates the + PMKID, and compares it in constant time inside the plugin. +- Avoided placing manually entered passwords in a spawned verifier command + line. Incorrect candidates remain rejected and are never written to a + potfile. +- Added a precise rejection for incomplete PCAPs: + `Password cannot be verified because this capture contains no usable + WPA/PMKID hash. Recapture the access point.` +- Added success, duplicate, rejection, and inconclusive verification logging + without logging the submitted password. + +### Capture quality, Cleanup, and OHC + +- Kept the diagnostic quality distinction between `Partial` (EAPOL material + exists) and `Unusable` (no WPA/PMKID material), while making Hashcat + suitability explicit: only captures that produce a mode 22000 hash are + crackable. +- Listed `Partial` captures with zero extractable WPA/PMKID hashes as + **uncrackable** Capture Cleanup candidates. Deletion still requires the + owner's browser confirmation and revalidates the report token, file signature, + and current quality immediately before removing anything. +- Required local WPA/PMKID extraction before OHC submission. Zero-hash PCAPs are + not sent to the API, are marked with `No usable WPA or PMKID hash found`, and + are counted in the OHC status panel with a pointer to Capture Cleanup. +- Made per-capture OHC actions return the stored exclusion reason when nothing + can be queued instead of reporting a misleading upload start for zero files. +- Preserved the existing BSSID-first deduplication, last imported OHC export, + live task reconciliation, one-best-capture selection, and persistent retry + queue for captures that pass the local suitability gate. + +### Map consistency and validation + +- Updated the matching Handshakes card immediately after successful manual map + placement: the location badge becomes **MAP** and the action becomes **Move** + without a full page reload. +- Added regression coverage for template caching, gzip configuration, compact + map payloads, immediate map DOM updates, PMKID success and mismatch, missing + hash errors, uncrackable cleanup, and zero-queue OHC reasons. The v1.3.9 suite + contains 59 passing tests. + ## 1.3.8 — 2026-07-29 - Parsed WPA-sec result files with compact 12-hex BSSID and station fields, preserving exact AP identity instead of creating `Name-only credential` entries. diff --git a/README.md b/README.md index 2443fc4..65d6afc 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,26 @@ Bluetooth PAN connection. - Places the Delete action beside the exact PCAP filename so the selected file is unambiguous. +### Quality meanings + +| Grade | Local result | Hashcat suitability | +|---|---|---| +| **Excellent** | A usable hash plus an authorized EAPOL exchange or written PMKID | Ready | +| **Usable** | At least one valid WPA/PMKID mode 22000 hash | Ready | +| **Partial** | EAPOL material exists, but no usable hash was extracted | Uncrackable as captured | +| **Unusable** | No WPA/PMKID material, or only an empty PCAP header | Uncrackable | + +Raw M1/M2/M3/M4 counts are diagnostic. Frames can belong to incompatible +association attempts or carry mismatched nonces and replay counters. PWMenu uses +successful mode 22000 extraction—not the number of EAPOL frames—as the final +Hashcat suitability test. + +`Partial` captures with zero hashes are excluded from Hashcat and cloud work and +listed as **uncrackable** in Capture Cleanup. The owner sees the exact filename +and reason, confirms the complete candidate count, and the plugin rechecks the +report token, file signature, and current quality before deletion. Cleanup is +never automatic. + ### Uncracked export `Download All Uncracked APs` does not trust an ESSID or filename match alone. @@ -75,10 +95,19 @@ PWMenu combines credentials from: - integrated QuickDic `.cracked` files; - manually entered passwords. -Manual add and edit actions are verified with `aircrack-ng` against a matching -capture before anything is written. An incorrect password, verification timeout, -or missing matching capture is rejected. Successful add, update, and delete -actions refresh the affected UI without reloading the page. +Manual add and edit actions are verified against a matching capture before +anything is written. PWMenu uses `aircrack-ng` for EAPOL handshakes and falls +back to `hcxpcapngtool` plus an internal cryptographic verifier for PMKID-only +captures. An incorrect password, verification timeout, or missing matching +capture is rejected. +Successful add, update, and delete actions refresh the affected UI without +reloading the page. + +If no usable hash exists, PWMenu does not guess or blindly store the value. It +returns: + +> Password cannot be verified because this capture contains no usable +> WPA/PMKID hash. Recapture the access point. The TXT export is sorted UTF-8 TSV with a byte-order mark and Windows-compatible CRLF lines. It contains the columns `ESSID`, `BSSID`, `PASSWORD`, and `SOURCE`; @@ -96,7 +125,9 @@ Location can come from: Choose **Map** or **Move** beside a concrete handshake to place it manually. PWMenu opens the Map workspace with a fixed pin in the center: move the map under the pin, then confirm or cancel using the bottom controls. A capture can -also be attached directly to an existing point or cluster. +also be attached directly to an existing point or cluster. The Handshakes card +changes to **MAP**, and its action changes to **Move**, immediately after the +server confirms the new coordinates; a page reload is not required. Coordinates set by the user are labeled **Map**. Coordinates measured through PwnDroid, browser geolocation, or GPSD are labeled **GPS**. Manual placement, @@ -135,8 +166,11 @@ main.plugins.wpa-sec-list.enabled = false ## OnlineHashCrack -PWMenu converts captures to mode 22000, maintains a durable upload queue, submits -in batches, downloads results, and preserves server backoff across restarts. +PWMenu first converts each capture locally to mode 22000. A PCAP that produces +zero hashes is not sent to OHC: it is marked with the extraction reason, counted +as uncrackable, and exposed to confirmation-bound Capture Cleanup. Valid hashes +enter the durable upload queue, are submitted in batches, and retain server +backoff across restarts. Before work enters the queue, it is compared with: @@ -146,7 +180,9 @@ Before work enters the queue, it is compared with: Already known tasks are recorded instead of submitted again. `Send all missing to OHC` scans unresolved captures, selects one best PCAP per BSSID, and queues -only work that is absent from all three sources. +only locally extractable work that is absent from all three sources. A +per-capture action that queues zero files now returns its stored exclusion +reason instead of reporting a misleading upload start. ```toml main.plugins.A_pwmenu.module_ohc_enabled = true @@ -270,7 +306,7 @@ sudo cp /usr/local/share/pwnagotchi/custom-plugins/A_pwmenu.py \ /root/A_pwmenu.py.backup 2>/dev/null || true sudo wget -O /usr/local/share/pwnagotchi/custom-plugins/A_pwmenu.py \ - https://raw.githubusercontent.com/newfpv/pwmenu/v1.3.8/A_pwmenu.py + https://raw.githubusercontent.com/newfpv/pwmenu/v1.3.9/A_pwmenu.py sudo chown root:root /usr/local/share/pwnagotchi/custom-plugins/A_pwmenu.py sudo chmod 644 /usr/local/share/pwnagotchi/custom-plugins/A_pwmenu.py @@ -301,6 +337,16 @@ http://:8080/plugins/A_pwmenu/ See [`config.example.toml`](./config.example.toml) for every module switch and configuration option. +Web UI timing and compression can be adjusted independently: + +```toml +# 1 is the recommended fast setting for Raspberry Pi; allowed range is 1-9. +main.plugins.A_pwmenu.web_gzip_level = 1 + +# How long server messages and toast notifications remain visible. +main.plugins.A_pwmenu.web_notification_duration_ms = 2600 +``` + ## Independent module switches Each subsystem can be disabled without disabling the complete plugin: @@ -323,7 +369,8 @@ main.plugins.A_pwmenu.module_quickdic_enabled = true - The Python 3.11 environment used by Pwnagotchi. - `requests`. - `websockets` when PwnDroid is enabled. -- `hcxpcapngtool` for quality analysis and mode 22000 conversion. +- `hcxpcapngtool` for quality analysis, mode 22000 conversion, and PMKID + extraction for password verification. - `aircrack-ng` for manual password verification and integrated QuickDic. - Xray only when OHC VLESS routing is enabled. diff --git a/config.example.toml b/config.example.toml index 7232b61..8857e8c 100644 --- a/config.example.toml +++ b/config.example.toml @@ -99,6 +99,11 @@ main.plugins.A_pwmenu.time_sync_interval = 1800 main.plugins.A_pwmenu.import_max_bytes = 2097152 main.plugins.A_pwmenu.archive_memory_limit = 2097152 +# Web UI behavior. Level 1 compression is much faster on a Raspberry Pi while +# still reducing the page size. Notification duration is in milliseconds. +main.plugins.A_pwmenu.web_gzip_level = 1 +main.plugins.A_pwmenu.web_notification_duration_ms = 2600 + # After migration, keep the legacy standalone plugins disabled to avoid duplicate # screen elements, uploads, downloads, and dictionary work. main.plugins.wpa-sec.enabled = false diff --git a/tests/test_capture_quality.py b/tests/test_capture_quality.py index 5be1aef..ab7b500 100644 --- a/tests/test_capture_quality.py +++ b/tests/test_capture_quality.py @@ -111,6 +111,35 @@ def test_empty_cleanup_requires_current_report_token(self): self.assertEqual((deleted, total), (1, 1)) self.assertFalse(os.path.exists(empty_path)) + def test_partial_capture_without_hash_is_cleanup_candidate(self): + path = os.path.join( + self.tempdir.name, "Partial_aabbccddeeff.pcap" + ) + with open(path, "wb") as handle: + handle.write(b"partial capture frames") + name = os.path.basename(path) + self.plugin.data["capture_quality"][name] = { + "grade": "Partial", + "hashes": 0, + "summary": "Incomplete EAPOL exchange (M1/M2/M3/M4: 4/0/0/0)", + "signature": self.plugin._ohc_file_signature(path), + } + + report = self.plugin._capture_cleanup_report() + + self.assertEqual(report["count"], 1) + self.assertEqual(report["uncrackable_count"], 1) + self.assertEqual( + report["entries"][0]["category"], + "uncrackable", + ) + + deleted, total, _ = self.plugin._clean_capture_candidates( + report["token"] + ) + self.assertEqual((deleted, total), (1, 1)) + self.assertFalse(os.path.exists(path)) + def test_later_usable_capture_archives_weak_capture_for_same_bssid(self): old_path = os.path.join(self.tempdir.name, "Old_aabbccddeeff.pcap") new_path = os.path.join(self.tempdir.name, "New_aabbccddeeff.pcap") @@ -607,7 +636,11 @@ def test_web_template_renders_quality_cleanup_and_branding(self): "detail": "", }, no_gps_networks=[], - ohc_status={"pending": 0, "retry_in": 0}, + ohc_status={ + "pending": 0, + "retry_in": 0, + "uncrackable": 0, + }, pot_health={ "ok": True, "credentials": 0, @@ -619,12 +652,14 @@ def test_web_template_renders_quality_cleanup_and_branding(self): cleanup_report={ "count": 0, "empty_count": 0, + "uncrackable_count": 0, "unusable_count": 0, "display_files": [], "more": 0, "token": "0" * 64, }, whitelist=[], + notification_duration_ms=2600, ) self.assertIn("function qualityStatusBlock", page) @@ -639,12 +674,14 @@ def test_web_template_renders_quality_cleanup_and_branding(self): self.assertIn("async function updateWhitelistAsync", page) self.assertIn("async function runMapAction", page) self.assertIn("function placeHandshakeOnMap", page) + self.assertIn("function applyMapPlacementToHandshake", page) self.assertIn("function confirmMapPlacement", page) self.assertIn("function mapItemCoordinates", page) self.assertIn('id="mapPlacementTarget"', page) self.assertNotIn('id="mapPlacementBanner"', page) self.assertIn("capture-file-delete", page) self.assertIn("capture-file-name", page) + self.assertIn("const notificationDurationMs = 2600", page) self.assertIn("Map", page) self.assertIn("capture-map-set", page) self.assertNotIn("map-point-add", page) diff --git a/tests/test_pwmenu_features.py b/tests/test_pwmenu_features.py index 5394d5f..9f778a3 100644 --- a/tests/test_pwmenu_features.py +++ b/tests/test_pwmenu_features.py @@ -1,3 +1,5 @@ +import hashlib +import hmac import os import sys import tempfile @@ -103,6 +105,140 @@ def test_manual_password_rejection_does_not_create_potfile(self): self.assertIn("No matching capture", message) self.assertFalse(os.path.exists(self.plugin.potfile_manual)) + def test_manual_password_falls_back_to_hcxtools_for_pmkid(self): + capture = os.path.join( + self.tempdir.name, "Cafe_aabbccddeeff.pcap" + ) + with open(capture, "wb") as handle: + handle.write(b"pmkid capture") + + with ( + mock.patch.object( + self.plugin, "_matching_capture_paths", return_value=[capture] + ), + mock.patch.object( + self.plugin, + "_run_aircrack_password_check", + return_value=(False, False, "no EAPOL data"), + ), + mock.patch.object( + self.plugin, + "_run_hcxpmk_password_check", + return_value=(True, True, "PMKID confirmed"), + ) as hcx_check, + ): + ok, message = self.plugin._add_manual_password( + "Cafe", "aa:bb:cc:dd:ee:ff", "correcthorse" + ) + + self.assertTrue(ok) + self.assertIn("verified", message) + hcx_check.assert_called_once_with( + capture, + "Cafe", + "aabbccddeeff", + ["correcthorse"], + ) + + def test_manual_password_explains_when_capture_has_no_hash(self): + capture = os.path.join( + self.tempdir.name, "Cafe_aabbccddeeff.pcap" + ) + with open(capture, "wb") as handle: + handle.write(b"incomplete capture") + + with ( + mock.patch.object( + self.plugin, "_matching_capture_paths", return_value=[capture] + ), + mock.patch.object( + self.plugin, + "_run_aircrack_password_check", + return_value=(False, False, "Packets contained no EAPOL data"), + ), + mock.patch.object( + self.plugin, + "_run_hcxpmk_password_check", + return_value=( + False, + False, + "No matching PMKID/EAPOL hash was extracted", + ), + ), + ): + verified, message = self.plugin._verify_manual_password( + "Cafe", + "aabbccddeeff", + "correcthorse", + ) + + self.assertFalse(verified) + self.assertEqual( + message, + "Password cannot be verified because this capture contains no " + "usable WPA/PMKID hash. Recapture the access point", + ) + + def test_pmkid_verification_is_local_and_password_not_in_subprocess(self): + capture = os.path.join( + self.tempdir.name, "Cafe_aabbccddeeff.pcap" + ) + with open(capture, "wb") as handle: + handle.write(b"pmkid capture") + + tool_calls = [] + ap_mac = bytes.fromhex("aabbccddeeff") + station_mac = bytes.fromhex("112233445566") + essid_bytes = b"Cafe" + pmk = hashlib.pbkdf2_hmac( + "sha1", + b"correcthorse", + essid_bytes, + 4096, + dklen=32, + ) + pmkid = hmac.new( + pmk, + b"PMK Name" + ap_mac + station_mac, + hashlib.sha1, + ).digest()[:16].hex() + hash_line = ( + f"WPA*01*{pmkid}*" + "aabbccddeeff*112233445566*43616665" + ) + + def run_tool(args, **kwargs): + tool_calls.append((args, kwargs)) + with open(args[2], "w", encoding="utf-8") as handle: + handle.write(hash_line + "\n") + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + with ( + mock.patch( + "A_pwmenu.shutil.which", + side_effect=lambda name: f"/fake/{name}", + ), + mock.patch("A_pwmenu.subprocess.run", side_effect=run_tool), + ): + verified, conclusive, _ = self.plugin._run_hcxpmk_password_check( + capture, + "Cafe", + "aa:bb:cc:dd:ee:ff", + ["correcthorse"], + ) + + self.assertTrue(verified) + self.assertTrue(conclusive) + self.assertEqual(len(tool_calls), 1) + argv, call_options = tool_calls[0] + self.assertNotIn("correcthorse", argv) + self.assertNotIn("input", call_options) + self.assertFalse( + self.plugin._verify_pmkid_hash_password( + hash_line, "definitelywrong" + ) + ) + def test_capture_map_location_is_persisted_for_exact_handshake(self): filename = "Cafe_aabbccddeeff.pcap" capture = os.path.join(self.tempdir.name, filename) @@ -131,6 +267,26 @@ def test_capture_map_location_is_persisted_for_exact_handshake(self): self.assertFalse(changed) self.assertEqual(53.9, resolved["lat"]) + def test_single_map_point_does_not_duplicate_member_payload(self): + points = self.plugin._build_map_points([ + { + "essid": "Cafe", + "is_cracked": False, + "files": [{ + "filename": "Cafe_aabbccddeeff.pcap", + "bssid": "aa:bb:cc:dd:ee:ff", + "lat": 53.9, + "lon": 27.56, + "gps_source": "manual-map", + "quality": {}, + }], + } + ]) + + self.assertEqual(len(points), 1) + self.assertEqual(points[0]["source"], "manual-map") + self.assertEqual(points[0]["members"], []) + def test_capture_map_location_rejects_invalid_target_or_coordinates(self): filename = "Cafe_aabbccddeeff.pcap" capture = os.path.join(self.tempdir.name, filename) @@ -171,6 +327,35 @@ def test_password_txt_is_utf8_tsv_and_preserves_colons(self): self.assertTrue(text.startswith("ESSID\tBSSID\tPASSWORD\tSOURCE\r\n")) self.assertIn("Cafe\tAA:BB:CC:DD:EE:FF\tpass:word\tManual", text) + def test_single_ohc_action_reports_why_nothing_was_queued(self): + request = types.SimpleNamespace( + form={"filenames": "Partial_aabbccddeeff.pcap"} + ) + with ( + mock.patch.object( + self.plugin, + "_queue_ohc_files", + return_value=0, + ), + mock.patch.object( + self.plugin, + "_ohc_file_record", + return_value={ + "status": "invalid", + "message": "No usable WPA or PMKID hash found", + }, + ), + ): + message, is_error = self.plugin._handle_ohc_cluster_upload( + request + ) + + self.assertTrue(is_error) + self.assertEqual( + message, + "Nothing queued: No usable WPA or PMKID hash found", + ) + def test_wpa_upload_uses_service_cookie(self): capture = os.path.join( self.tempdir.name, "Cafe_aabbccddeeff.pcap" diff --git a/tests/test_web_transport.py b/tests/test_web_transport.py index a5c2a96..c216d16 100644 --- a/tests/test_web_transport.py +++ b/tests/test_web_transport.py @@ -2,6 +2,7 @@ import sys import types import unittest +from unittest import mock from flask import Flask @@ -42,9 +43,17 @@ def test_large_html_is_gzipped_when_browser_accepts_it(self): "/plugins/A_pwmenu/", headers={"Accept-Encoding": "gzip, deflate"}, ): - response = self.plugin._html_response(self.html) + with mock.patch( + "A_pwmenu.gzip.compress", + wraps=gzip.compress, + ) as compress: + response = self.plugin._html_response(self.html) self.assertEqual(response.headers["Content-Encoding"], "gzip") + compress.assert_called_once_with( + self.html.encode("utf-8"), + compresslevel=1, + ) self.assertEqual(response.headers["Vary"], "Accept-Encoding") self.assertEqual(gzip.decompress(response.get_data()).decode("utf-8"), self.html) self.assertLess(len(response.get_data()), len(self.html.encode("utf-8")) // 4) @@ -60,6 +69,24 @@ def test_identity_response_remains_plain_utf8_html(self): self.assertEqual(response.mimetype, "text/html") self.assertEqual(response.get_data().decode("utf-8"), self.html) + def test_large_jinja_template_is_compiled_once(self): + with self.app.test_request_context("/plugins/A_pwmenu/"): + with mock.patch.object( + self.plugin, + "_get_html", + return_value="PWMenu {{ marker }}", + ) as get_html: + self.assertEqual( + self.plugin._render_cached_template(marker="first"), + "PWMenu first", + ) + self.assertEqual( + self.plugin._render_cached_template(marker="second"), + "PWMenu second", + ) + + get_html.assert_called_once_with() + if __name__ == "__main__": unittest.main()