fix(portals,capture): round-2 validation fixes, live-verified on Pager
- portals: replace zipfile with struct+zlib ZIP writer in portal download (python3-light has no zipfile; GET /api/portals/<name>/download 500ed) - capture: revive watchdog re-arms the 5 GHz deploy auto-capture if the post-deploy radio settle kills it (was: empty pcap, dead tcpdump) - capture: route GET /api/attacks/capture to status (was unrouted -> 404) New tests/test_validation_fixes2.py covers each fix (TDD); full suite (30 modules) green. Live-verified: download CRC-clean via stock zipfile, capture survived settle window and revived automatically (56 MB pcap), GET status returns proper JSON. Round-2 validation report added at docs/validation/ (8/9 attack types PASS against in-scope networks; enterprise PARTIAL per firmware limits).
This commit is contained in:
c4ch3c4d3
@@ -0,0 +1,85 @@
|
|||||||
|
# Mark-VIII PineAP Attack Validation — Round 2 (post-fix), 2026-08-23
|
||||||
|
|
||||||
|
Full live re-validation of every PineAP attack type exposed by Mark VIII, executed after the
|
||||||
|
fix round in commit `d23ea56`. All attacks were run against authorized, in-scope networks only.
|
||||||
|
|
||||||
|
## Environment
|
||||||
|
|
||||||
|
| Role | Host | Identity |
|
||||||
|
|---|---|---|
|
||||||
|
| Attack platform | WiFi Pineapple Pager 24.10.1, `root@172.16.52.1`, Mark VIII `:8080` | radio0 MAC base `00:13:37:ae:e0:50`, radio1 `00:13:37:ae:8e:7c` |
|
||||||
|
| Victim client | Kali Linux, `bzuccaro@192.168.1.103`, wlan0 `a0:a4:c5:93:f8:05` | NetworkManager + standalone wpa_supplicant |
|
||||||
|
| In-scope targets | `Znet` (WPA2/WPA3-SAE-mixed, 5GHz ch36/44/48/…; PSK provided) and `Znet-Open` (open, 2.4GHz ch6 `B6:8B:A9:17:2A:6E`, ch11 `B6:8B:A9:17:47:33`) | recon.db fresh scans confirmed both |
|
||||||
|
|
||||||
|
Pre-flight: `/api/health` env `pass` (pineapd alive, both monitors up, recon readable).
|
||||||
|
|
||||||
|
## Results summary
|
||||||
|
|
||||||
|
| # | Attack / capability | Endpoint(s) | Result | Proof captured |
|
||||||
|
|---|---|---|---|---|
|
||||||
|
| 1 | Recon | recon.db / scan history | **PASS** | Both SSIDs present with fresh timestamps, correct BSSIDs/channels/crypto |
|
||||||
|
| 2 | Evil Open twin + Evil Portal credential capture | `POST /api/attacks/deploy kind=open` + portals API | **PASS** | Victim associated to spoofed-BSSID twin (`172.16.52.123`); DNS hijack resolved arbitrary domain → `172.16.52.1`; portal served on :80; POSTed creds recorded with MAC/hostname/IP (`evidence/r2_t1_portal_captures.json`) |
|
||||||
|
| 3 | Karma association | (implicit) | **PASS** | Real third-party client `48:e1:e9:4d:98:8a` associated to Znet-Open twin unprompted; later handshakes from `68:9e:19:d1:6e:e1` / `e6:75:7f:45:fd:57` on the WPA twin across multiple probed SSIDs |
|
||||||
|
| 4 | Filters auto-config post-deploy | `GET /api/pineap/filters/{ssid,client}` | **PASS** (BUG 2 fix verified) | After deploy both filters read `deny` + empty with no manual help — twins accept clients unaided |
|
||||||
|
| 5 | Single deauth via API | `POST /api/attacks/deauth` | **PASS** (BUG 1 fix verified) | `ok:true, inject:wlan0mon`; pcap holds **556 deauth frames** incl. directed SA=twin-BSSID → DA=victim (`evidence/r2_t2_deauth.cap`); victim dropped (NM re-associated sub-second) |
|
||||||
|
| 6 | Client kick | `POST /api/pineap/clients/kick` | **PASS** | Deny filter added for victim MAC; victim flapped DISCONNECTED/CONNECTED and could not hold association until filter cleared |
|
||||||
|
| 7 | Evil WPA twin 5GHz + handshake capture | `deploy kind=wpa` + `attacks/capture` | **PASS** | Twin live ch44 (`02:13:37:ae:8e:7c`); full EAPOL 4-way from victim captured on pinned `wlan1mon`; export produced 18-row `.hc22000` incl. victim AND real-client handshakes |
|
||||||
|
| 8 | hc22000 export → crack | `GET /api/attacks/export/hc22000` + hashcat -m 22000 (Kali) | **PASS** | PSK recovered for victim (`021337ae8e7c:a0a4c593f805:Znet`) and real client (`…:e6757f45fd57:Znet`) = exact known PSK (`evidence/r2_t4_cracked.txt`) |
|
||||||
|
| 9 | Bulk deauth | `POST /api/attacks/deauth/bulk` | **PASS** | 3-target batch: 2 valid sent (`sent:2`), malformed target rejected per-index without aborting batch; victim dropped on-air |
|
||||||
|
| 10 | Evil Enterprise PEAP/MSCHAPv2 | `deploy kind=enterprise` | **PARTIAL** | AP verified + ctrl-linked + runtime-bridged; client associates; TLS tunnel up; server issues inner MSCHAPv2 success for `victim@znet.local`. BUT wpa_supplicant rejects the karma wpad's success request ("Invalid authenticator response") so no full CONNECTED; `hostap_basic`/`hostap_chalresp` remain empty (known firmware residual). See §T6 |
|
||||||
|
| 11 | Post-suite hygiene | — | **PASS** | All attacks stopped, zero leftover APs/tcpdump/watchdog churn, health env `pass`, victim restored to real `Znet` (`18:e8:29:b5:a4:2c`) |
|
||||||
|
|
||||||
|
## New defects found this round
|
||||||
|
|
||||||
|
### D1 — MEDIUM: portal download endpoint crashes on device
|
||||||
|
`h_portal_download` (`server.py:5395`) imports `zipfile`, which python3-light does not ship
|
||||||
|
(the import endpoint was converted to struct+zlib in d23ea56 but download was missed):
|
||||||
|
`GET /api/portals/<name>/download` → `{"error": "No module named 'urllib'"}`.
|
||||||
|
Fix: reuse the minimal ZIP writer approach or stream raw files.
|
||||||
|
|
||||||
|
### D2 — MEDIUM: deploy-time auto-capture produces an empty pcap
|
||||||
|
The 5GHz WPA deploy reported `"capture": true` but the auto-started `wlan1mon` capture died
|
||||||
|
during AP bring-up (file stayed at the 24-byte header; no tcpdump process left). A capture
|
||||||
|
started *after* the AP is up works fine (frames flow, monitor inherits phy channel context).
|
||||||
|
Fix: arm the auto-capture after hostapd verify-loop completes, and/or have `_capture_state`
|
||||||
|
detect-and-restart the dead pid (ISSUE 7 stale logic exists but did not fire here).
|
||||||
|
Also: `h_attacks_capture` silently ignores a `channel` body param — either honor it or reject it.
|
||||||
|
|
||||||
|
### D3 — LOW/cosmetic: `GET /api/attacks/capture` with no active capture returns 404
|
||||||
|
`{"error":"not found"}` instead of `{running:false,...}` — UI-hostile shape.
|
||||||
|
|
||||||
|
## T6 detail (Enterprise PARTIAL)
|
||||||
|
|
||||||
|
Repro: deploy `kind=enterprise ssid=Znet enctype=wpa2 channel=44 passphrase=VictimPass123!`
|
||||||
|
→ `verified:true, ctrl_linked:true`. Victim (standalone wpa_supplicant, MAC randomization off):
|
||||||
|
|
||||||
|
- Association OK; outer PEAP TLS tunnel completes (`CTRL-EVENT-EAP-PROPOSED-METHOD method=25`)
|
||||||
|
- Inner MSCHAPv2 exchange runs; client logs `EAP-MSCHAPV2: Received success` — i.e. the
|
||||||
|
standalone hostapd accepted the inner credentials (BUG 3 eap_users grammar fix works)
|
||||||
|
- But every attempt then logs `EAP-MSCHAPV2: Invalid authenticator response in success request`
|
||||||
|
→ supplicant refuses, disconnects, retries forever. Same result with matched and mismatched
|
||||||
|
passwords (server auto-accepts but its AuthResp never verifies) — consistent with the karma-
|
||||||
|
patched wpad issuing success without computing it from the stored secret.
|
||||||
|
- TTLS/PAP could not be differentiated this round (client-side sed failure meant PEAP ran;
|
||||||
|
association-level flapping prevented a clean second attempt).
|
||||||
|
- `hostap_basic` / `hostap_chalresp`: still empty (documented firmware residual — pineapd does
|
||||||
|
not forward from foreign hostapd instances). Enterprise client list DOES record associations.
|
||||||
|
|
||||||
|
Net: enterprise twin captures inner-auth material server-side only as far as hostapd's own
|
||||||
|
logs; portable credential loot remains impossible on 24.10.1 without a Hak5 pineapd change.
|
||||||
|
|
||||||
|
## Evidence index (`evidence/r2_*`)
|
||||||
|
|
||||||
|
| File | Content |
|
||||||
|
|---|---|
|
||||||
|
| `r2_t1_portal_captures.json` | Captured portal credentials (user/pass, victim MAC/hostname/IP) |
|
||||||
|
| `r2_t2_deauth.cap` | wlan0mon pcap, 556 deauth frames (directed at victim) |
|
||||||
|
| `r2_t2_deauth.json`, `r2_t2_kick.json` | API responses proving deauth/kick ok:true |
|
||||||
|
| `r2_t3_deploy_wpa.json` | WPA twin deploy response (`capture:true`, `verified:true`) |
|
||||||
|
| `r2_t3_hc_export.json`, `r2_handshakes.hc22000` | Exported 18 handshake hashes |
|
||||||
|
| `r2_t4_cracked.txt` | hashcat --show output recovering the true PSK (victim + real client) |
|
||||||
|
| `r2_t5_bulk.json` | Bulk deauth batch results (2 sent / 1 rejected) |
|
||||||
|
| `r2_t6_deploy_ent.json` | Enterprise deploy response |
|
||||||
|
|
||||||
|
Copies of key artifacts also live on-device (`/root/loot/**`) and on Kali (`/tmp/r2.hc22000`,
|
||||||
|
removed wordlist).
|
||||||
@@ -3812,6 +3812,8 @@ def _deploy_wpa_open(kind, fields):
|
|||||||
# loot the 4-ways from a pinned monitor capture instead.
|
# loot the 4-ways from a pinned monitor capture instead.
|
||||||
try:
|
try:
|
||||||
capture = _ensure_attack_capture('wlan1mon')
|
capture = _ensure_attack_capture('wlan1mon')
|
||||||
|
if capture:
|
||||||
|
_schedule_capture_revive('wlan1mon')
|
||||||
except Exception:
|
except Exception:
|
||||||
capture = None
|
capture = None
|
||||||
return {'kind': kind, 'ssid': ssid, 'iface': iface, 'band': band,
|
return {'kind': kind, 'ssid': ssid, 'iface': iface, 'band': band,
|
||||||
@@ -4673,6 +4675,47 @@ def _teardown_attack_capture(iface):
|
|||||||
return running
|
return running
|
||||||
|
|
||||||
|
|
||||||
|
CAPTURE_REVIVE_CHECKS = 6
|
||||||
|
CAPTURE_REVIVE_INTERVAL = 10.0
|
||||||
|
|
||||||
|
|
||||||
|
def _capture_revive_once(iface):
|
||||||
|
"""Re-arm the auto-capture if it died (e.g. radios dropped during the
|
||||||
|
post-deploy settle window). No-op while a live capture runs or when the
|
||||||
|
monitor iface itself is gone."""
|
||||||
|
pidfile = '/tmp/mk8_capture_%s.pid' % iface
|
||||||
|
running, _, _ = _capture_state(pidfile, iface)
|
||||||
|
if running:
|
||||||
|
return None
|
||||||
|
if not os.path.exists('/sys/class/net/%s' % iface):
|
||||||
|
return None
|
||||||
|
return _ensure_attack_capture(iface)
|
||||||
|
|
||||||
|
|
||||||
|
def _schedule_capture_revive(iface='wlan1mon',
|
||||||
|
checks=CAPTURE_REVIVE_CHECKS,
|
||||||
|
interval=CAPTURE_REVIVE_INTERVAL):
|
||||||
|
"""Watch a freshly armed capture through the deploy settle window.
|
||||||
|
|
||||||
|
The 5 GHz attack AP bring-up can recreate monitors / reload wifi, which
|
||||||
|
kills the just-started tcpdump; poll a few times and restart it once the
|
||||||
|
radios have converged. Runs in a daemon thread, self-terminates."""
|
||||||
|
def _revive():
|
||||||
|
for _ in range(checks):
|
||||||
|
time.sleep(interval)
|
||||||
|
try:
|
||||||
|
result = _capture_revive_once(iface)
|
||||||
|
except Exception:
|
||||||
|
result = None
|
||||||
|
if result is not None or _capture_state(
|
||||||
|
'/tmp/mk8_capture_%s.pid' % iface, iface)[0]:
|
||||||
|
# Restarted, or the original capture is alive again.
|
||||||
|
return
|
||||||
|
thread = threading.Thread(target=_revive, daemon=True)
|
||||||
|
thread.start()
|
||||||
|
return thread
|
||||||
|
|
||||||
|
|
||||||
def h_attacks_capture(ctx):
|
def h_attacks_capture(ctx):
|
||||||
body = ctx.body or {}
|
body = ctx.body or {}
|
||||||
action = body.get('action') or 'status'
|
action = body.get('action') or 'status'
|
||||||
@@ -4897,6 +4940,41 @@ def _portal_list():
|
|||||||
|
|
||||||
_EOCD_SIG = b'PK\x05\x06'
|
_EOCD_SIG = b'PK\x05\x06'
|
||||||
_CDH_SIG = b'PK\x01\x02'
|
_CDH_SIG = b'PK\x01\x02'
|
||||||
|
_LFH_SIG = b'PK\x03\x04'
|
||||||
|
|
||||||
|
|
||||||
|
def _zip_create(files):
|
||||||
|
"""Minimal ZIP writer built on struct+zlib only.
|
||||||
|
|
||||||
|
python3-light has no zipfile, so portal download serializes stored
|
||||||
|
(uncompressed) entries by hand: local file headers, central directory,
|
||||||
|
end-of-central-directory. Returns archive bytes; names must be plain
|
||||||
|
relative paths."""
|
||||||
|
import struct
|
||||||
|
try:
|
||||||
|
import zlib
|
||||||
|
except ImportError:
|
||||||
|
zlib = None
|
||||||
|
out = bytearray()
|
||||||
|
central = bytearray()
|
||||||
|
for name in sorted(files):
|
||||||
|
data = files[name]
|
||||||
|
nbuf = name.encode('utf-8')
|
||||||
|
crc = zlib.crc32(data) & 0xFFFFFFFF if zlib is not None else 0
|
||||||
|
offset = len(out)
|
||||||
|
out += struct.pack('<4sHHHHHIIIHH', _LFH_SIG, 20, 0, 0, 0, 0,
|
||||||
|
crc, len(data), len(data), len(nbuf), 0)
|
||||||
|
out += nbuf
|
||||||
|
out += data
|
||||||
|
central += struct.pack('<4sHHHHHHIIIHHHHHII', _CDH_SIG, 20, 20, 0, 0,
|
||||||
|
0, 0, crc, len(data), len(data), len(nbuf),
|
||||||
|
0, 0, 0, 0, 0, offset)
|
||||||
|
central += nbuf
|
||||||
|
cd_off = len(out)
|
||||||
|
out += central
|
||||||
|
out += struct.pack('<4sHHHHIIH', _EOCD_SIG, 0, 0, len(files),
|
||||||
|
len(files), len(central), cd_off, 0)
|
||||||
|
return bytes(out)
|
||||||
|
|
||||||
|
|
||||||
def _zip_entries(data_bytes):
|
def _zip_entries(data_bytes):
|
||||||
@@ -5393,23 +5471,23 @@ def h_portal_logs(ctx):
|
|||||||
|
|
||||||
|
|
||||||
def h_portal_download(ctx):
|
def h_portal_download(ctx):
|
||||||
import zipfile
|
|
||||||
import io as _io
|
|
||||||
name = ctx.args[0] if ctx.args else ''
|
name = ctx.args[0] if ctx.args else ''
|
||||||
root = _portal_root(name)
|
root = _portal_root(name)
|
||||||
if not root or not os.path.isdir(root):
|
if not root or not os.path.isdir(root):
|
||||||
return 404, {'error': 'portal not found'}
|
return 404, {'error': 'portal not found'}
|
||||||
buf = _io.BytesIO()
|
files = {}
|
||||||
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
|
for dirpath, _, fnames in os.walk(root):
|
||||||
for dirpath, _, files in os.walk(root):
|
for fn in fnames:
|
||||||
for fn in files:
|
full = os.path.join(dirpath, fn)
|
||||||
full = os.path.join(dirpath, fn)
|
rel = os.path.relpath(full, root)
|
||||||
rel = os.path.relpath(full, root)
|
try:
|
||||||
try:
|
with open(full, 'rb') as f:
|
||||||
zf.write(full, os.path.join(name, rel))
|
files['%s/%s' % (name, rel.replace(os.sep, '/'))] = f.read()
|
||||||
except OSError:
|
except OSError:
|
||||||
continue
|
continue
|
||||||
return 200, Download(buf.getvalue(), 'application/zip',
|
if not files:
|
||||||
|
return 404, {'error': 'portal is empty'}
|
||||||
|
return 200, Download(_zip_create(files), 'application/zip',
|
||||||
'%s.zip' % name)
|
'%s.zip' % name)
|
||||||
|
|
||||||
|
|
||||||
@@ -7909,6 +7987,7 @@ ROUTER.add('POST', r'/api/attacks/deploy', h_attacks_deploy)
|
|||||||
ROUTER.add('POST', r'/api/attacks/stop', h_attacks_stop)
|
ROUTER.add('POST', r'/api/attacks/stop', h_attacks_stop)
|
||||||
ROUTER.add('GET', r'/api/attacks/status', h_attacks_status)
|
ROUTER.add('GET', r'/api/attacks/status', h_attacks_status)
|
||||||
ROUTER.add('POST', r'/api/attacks/capture', h_attacks_capture)
|
ROUTER.add('POST', r'/api/attacks/capture', h_attacks_capture)
|
||||||
|
ROUTER.add('GET', r'/api/attacks/capture', h_attacks_capture)
|
||||||
ROUTER.add('GET', r'/api/attacks/export/hc22000', h_attacks_export_hc22000)
|
ROUTER.add('GET', r'/api/attacks/export/hc22000', h_attacks_export_hc22000)
|
||||||
ROUTER.add('GET', r'/api/attacks/export/hc22000/([^/]+)', h_attacks_download_hc22000)
|
ROUTER.add('GET', r'/api/attacks/export/hc22000/([^/]+)', h_attacks_download_hc22000)
|
||||||
ROUTER.add('POST', r'/api/attacks/deauth', h_attacks_deauth)
|
ROUTER.add('POST', r'/api/attacks/deauth', h_attacks_deauth)
|
||||||
|
|||||||
@@ -0,0 +1,193 @@
|
|||||||
|
"""Regression tests for round-2 validation findings (D1-D3).
|
||||||
|
|
||||||
|
D1 portal download must build the zip with struct+zlib, not zipfile
|
||||||
|
(python3-light has no zipfile; GET /api/portals/<name>/download 500s).
|
||||||
|
D2 the deploy-time auto-capture must survive post-deploy radio settle:
|
||||||
|
a revive pass re-arms the capture when it dies during bring-up.
|
||||||
|
D3 GET /api/attacks/capture must be routed to the status handler instead
|
||||||
|
of 404 (UI and scripts poll status).
|
||||||
|
"""
|
||||||
|
import io
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
import zipfile
|
||||||
|
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload',
|
||||||
|
'user', 'remote_access', 'pager-webui'))
|
||||||
|
import server
|
||||||
|
|
||||||
|
|
||||||
|
def setUpModule():
|
||||||
|
__import__('importlib').reload(server)
|
||||||
|
|
||||||
|
|
||||||
|
def ctx(body=None, args=()):
|
||||||
|
return type('C', (), {'body': body if body is not None else {},
|
||||||
|
'args': args, 'query': {}})()
|
||||||
|
|
||||||
|
|
||||||
|
class BlockZipfile:
|
||||||
|
"""Import hook simulating python3-light: importing zipfile raises."""
|
||||||
|
|
||||||
|
def find_spec(self, fullname, path=None, target=None):
|
||||||
|
if fullname == 'zipfile':
|
||||||
|
raise ImportError("No module named 'zipfile'")
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def without_zipfile(fn):
|
||||||
|
saved = sys.modules.pop('zipfile', None)
|
||||||
|
blocker = BlockZipfile()
|
||||||
|
sys.meta_path.insert(0, blocker)
|
||||||
|
try:
|
||||||
|
return fn()
|
||||||
|
finally:
|
||||||
|
sys.meta_path.remove(blocker)
|
||||||
|
if saved is not None:
|
||||||
|
sys.modules['zipfile'] = saved
|
||||||
|
|
||||||
|
|
||||||
|
class ZipCreateTest(unittest.TestCase):
|
||||||
|
"""D1: _zip_create builds archives readable by _zip_entries."""
|
||||||
|
|
||||||
|
def test_roundtrip_stored_entries(self):
|
||||||
|
files = {'index.php': b'<html>portal</html>', 'sub/a.ep': b'x' * 40}
|
||||||
|
data = server._zip_create(files)
|
||||||
|
self.assertEqual(server._zip_entries(data), files)
|
||||||
|
|
||||||
|
def test_output_is_standard_zip(self):
|
||||||
|
data = server._zip_create({'index.php': b'hello'})
|
||||||
|
with zipfile.ZipFile(io.BytesIO(data)) as zf:
|
||||||
|
self.assertEqual(zf.namelist(), ['index.php'])
|
||||||
|
self.assertEqual(zf.read('index.php'), b'hello')
|
||||||
|
|
||||||
|
def test_no_zipfile_import_needed(self):
|
||||||
|
def build():
|
||||||
|
return server._zip_create({'index.php': b'data'})
|
||||||
|
data = without_zipfile(build)
|
||||||
|
self.assertIn(b'PK\x03\x04', data[:4])
|
||||||
|
|
||||||
|
|
||||||
|
class PortalDownloadWithoutZipfileTest(unittest.TestCase):
|
||||||
|
"""D1: h_portal_download must work where zipfile is absent."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.tmp = tempfile.mkdtemp(prefix='mk8-d1-')
|
||||||
|
portal = os.path.join(self.tmp, 'p1')
|
||||||
|
os.makedirs(portal)
|
||||||
|
with open(os.path.join(portal, 'index.php'), 'w') as f:
|
||||||
|
f.write('<html><form method="post"></form></html>')
|
||||||
|
with open(os.path.join(portal, '.logs'), 'w') as f:
|
||||||
|
f.write('[log line]\n')
|
||||||
|
self.old_dir = server.PORTALS_DIR
|
||||||
|
server.PORTALS_DIR = self.tmp
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
server.PORTALS_DIR = self.old_dir
|
||||||
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
||||||
|
|
||||||
|
def test_download_returns_parseable_zip(self):
|
||||||
|
def run():
|
||||||
|
status, payload = server.h_portal_download(ctx(args=('p1',)))
|
||||||
|
return status, payload
|
||||||
|
result = without_zipfile(run)
|
||||||
|
status, payload = result
|
||||||
|
self.assertEqual(status, 200)
|
||||||
|
entries = server._zip_entries(payload.data)
|
||||||
|
self.assertIn('p1/index.php', entries)
|
||||||
|
self.assertIn(b'<form method="post">', entries['p1/index.php'])
|
||||||
|
|
||||||
|
def test_download_missing_portal_404(self):
|
||||||
|
status, payload = server.h_portal_download(ctx(args=('nope',)))
|
||||||
|
self.assertEqual(status, 404)
|
||||||
|
|
||||||
|
|
||||||
|
class CaptureReviveTest(unittest.TestCase):
|
||||||
|
"""D2: revive pass re-arms a dead auto-capture."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.started = []
|
||||||
|
self.old_state = server._capture_state
|
||||||
|
self.old_ensure = server._ensure_attack_capture
|
||||||
|
self.old_exists = os.path.exists
|
||||||
|
# Pretend the monitor iface exists (tests run off-device).
|
||||||
|
os.path.exists = lambda p: True
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
os.path.exists = self.old_exists
|
||||||
|
server._capture_state = self.old_state
|
||||||
|
server._ensure_attack_capture = self.old_ensure
|
||||||
|
|
||||||
|
def test_revive_restarts_dead_capture(self):
|
||||||
|
states = iter([(False, 123, True)])
|
||||||
|
server._capture_state = lambda pf, iface: next(states)
|
||||||
|
server._ensure_attack_capture = \
|
||||||
|
lambda iface: self.started.append(iface) or {'running': True}
|
||||||
|
server._capture_revive_once('wlan1mon')
|
||||||
|
self.assertEqual(self.started, ['wlan1mon'])
|
||||||
|
|
||||||
|
def test_revive_skips_running_capture(self):
|
||||||
|
server._capture_state = lambda pf, iface: (True, 5, False)
|
||||||
|
|
||||||
|
def boom(iface):
|
||||||
|
raise AssertionError('must not restart a running capture')
|
||||||
|
server._ensure_attack_capture = boom
|
||||||
|
server._capture_revive_once('wlan1mon')
|
||||||
|
|
||||||
|
def test_revive_skips_when_iface_gone(self):
|
||||||
|
server._capture_state = lambda pf, iface: (False, 7, True)
|
||||||
|
os.path.exists = lambda p: 'net/wlan1mon' not in str(p) and \
|
||||||
|
self.old_exists(p)
|
||||||
|
server._ensure_attack_capture = \
|
||||||
|
lambda iface: self.started.append(iface) or None
|
||||||
|
server._capture_revive_once('wlan1mon')
|
||||||
|
self.assertEqual(self.started, [])
|
||||||
|
|
||||||
|
def test_deploy_schedules_revive_for_5g_wpa(self):
|
||||||
|
scheduled = []
|
||||||
|
old_sched = server._schedule_capture_revive
|
||||||
|
old_ensure = server._ensure_attack_capture
|
||||||
|
server._schedule_capture_revive = lambda iface='wlan1mon': \
|
||||||
|
scheduled.append(iface)
|
||||||
|
server._ensure_attack_capture = \
|
||||||
|
lambda iface: {'running': True, 'pid': 1, 'iface': iface}
|
||||||
|
try:
|
||||||
|
import tests.test_attacks as ta
|
||||||
|
harness = ta.AttacksDeployTest('test_deploy_wpa_5g_writes_radio1')
|
||||||
|
harness.setUp()
|
||||||
|
try:
|
||||||
|
status, payload = server.h_attacks_deploy(ctx({
|
||||||
|
'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
|
||||||
|
'enctype': 'psk2', 'hidden': False, 'channel': 36}))
|
||||||
|
self.assertEqual(status, 200)
|
||||||
|
self.assertTrue(payload['verified'])
|
||||||
|
self.assertEqual(scheduled, ['wlan1mon'])
|
||||||
|
finally:
|
||||||
|
harness.tearDown()
|
||||||
|
finally:
|
||||||
|
server._schedule_capture_revive = old_sched
|
||||||
|
server._ensure_attack_capture = old_ensure
|
||||||
|
|
||||||
|
|
||||||
|
class CaptureGetRouteTest(unittest.TestCase):
|
||||||
|
"""D3: GET /api/attacks/capture routes to the capture handler."""
|
||||||
|
|
||||||
|
def test_get_route_registered(self):
|
||||||
|
handler, groups = server.ROUTER.dispatch('GET', '/api/attacks/capture')
|
||||||
|
self.assertIsNotNone(handler,
|
||||||
|
'GET /api/attacks/capture is not routed')
|
||||||
|
self.assertEqual(handler, server.h_attacks_capture)
|
||||||
|
|
||||||
|
def test_status_without_body_reports_stopped_not_error(self):
|
||||||
|
status, payload = server.h_attacks_capture(
|
||||||
|
ctx(body=None))
|
||||||
|
self.assertEqual(status, 200)
|
||||||
|
self.assertFalse(payload.get('running'))
|
||||||
|
self.assertNotIn('error', payload)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user