17 Commits
Author SHA1 Message Date
bzuccaro 251b1f6261 feat: recon surveys with OUI/vendor lookup and report views
Reconnaissance surveys (GPSD-tethered scan recording), OUI vendor
lookup for client/AP tables, survey/report pages, and the matching
test_recon.py suite. Co-authored with the recon-feature agent whose
work was finished in this checkout.
2026-08-18 08:35:22 -05:00
bzuccaro f5cddb335a docs: correct 5GHz picker range and HEAD reference 2026-08-18 08:31:02 -05:00
bzuccaro af5ff8039b fix: gate channel_band 6GHz to 181-233, reject open-6GHz, self-heal load-time 6GHz hints, harden hop/BSSID handling 2026-08-18 08:30:45 -05:00
bzuccaro c0b9e58aa8 docs: record on-device verification results for 5/6GHz AP 2026-08-18 08:24:19 -05:00
bzuccaro 12279fe29d fix: write explicit disabled=0 for radio1 AP ifaces 2026-08-18 08:13:41 -05:00
bzuccaro 7536f3ca45 fix: self-heal radio1 AP after stock daemon wireless writes 2026-08-18 08:08:50 -05:00
bzuccaro be2685aa15 docs: 5GHz/6GHz rogue AP feature and coexistence notes 2026-08-18 07:30:06 -05:00
bzuccaro ff3bd16855 feat: band-aware channel pickers for Open AP and Evil WPA 2026-08-18 07:26:55 -05:00
bzuccaro 4fa7f8f855 fix: reject mixed 2.4GHz and radio1 AP requests with clear error 2026-08-18 07:23:25 -05:00
bzuccaro cd39833f4b test: update proxy tests for get_ap channel shape and radio1 cleanup 2026-08-17 23:29:47 -05:00
bzuccaro 4effc08a25 feat: radio1 5GHz/6GHz rogue AP via UCI with hop pause 2026-08-17 23:28:54 -05:00
bzuccaro b2098bae7d fix: fall back to radio channel when AP iface lacks channel 2026-08-17 23:21:09 -05:00
bzuccaro 7aff767f2a feat: read radio1 AP state in wifi get_ap 2026-08-17 23:17:57 -05:00
bzuccaro 27df97ec43 test: pin CHANNEL_BANDS consistency and DFS coverage 2026-08-17 23:15:01 -05:00
bzuccaro acbf4c0ce9 feat: channel/band mapping helpers for radio1 AP 2026-08-17 23:01:16 -05:00
bzuccaro 5f6dc5bcdb release: Mark VIII 1.1
Wireless client mode (connect to WiFi as client):
- settings/wifi/client API: state, scan, connect, disconnect, route
- Internet Connection topbar dialog and functional Settings > Networking card
- routing toggle syncing UCI flag and daemon state
- fix trailing-slash hash routes (View not available)
- hidden-SSID filtering, encryption classification (Open/WPA2/WPA3/mixed)
- tests for client state, scan parsing, connect/disconnect, routing
2026-08-17 22:28:41 -05:00
bzuccaroandfactory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> 2bf39ecb9d fix: repair recon scanning and add macOS deployment
Start scans via the Pager's native /api/pineap/recon/new so history
appends instead of rotating. Enforce finite durations (1-86400s,
default 30s), remove the unsupported Continuous mode, and refuse the
unsafe manual stop that left recon.db locked.

Rework scan list and detail reads into single-pass aggregate SQL,
shorten lock retries, and serve cached results during short exclusive
lock windows. Fall back to immutable read-only access when the
firmware leaves a stale lock after native completion.

Frontend auto-follows new scans, queues a single in-flight detail
refresh, keeps previous tables visible while a scan starts, and shows
completion toasts and daemon error details.

Add scripts/deploy.sh for macOS/Linux (zip packaging, scp/ssh install,
atomic payload replacement, service restart, portal refresh) with
README instructions, plus regression coverage for native start,
safe stop semantics, aggregate queries, and stale-lock fallback.

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
2026-08-17 17:46:54 -05:00
18 changed files with 4097 additions and 153 deletions
+1
View File
@@ -3,4 +3,5 @@ __pycache__/
*.pyc
.worktrees/
.openchamber/
+22 -3
View File
@@ -8,22 +8,41 @@ Recon (scans from `recon.db`), Handshakes/Loot, Payloads (embedded stock Pager
Portal), Logs, Settings (hostname/NTP/password/prefs), and a bottom-docked xterm
terminal.
- 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,
6GHz requires WPA3. While a radio1 AP is enabled the stock monitor-hopping
(`wlan1mon`) is paused and resumed on disable; 2.4GHz PineAP is untouched.
## Requirements
- WiFi Pineapple Pager, firmware `Pineapple Pager 24.10.1`
- `python3` on the device (present on current firmware)
- Windows dev box with Python 3.11 (`winget install Python.Python.3.11`)
- Python 3.11 on the development machine
## Install (sideload)
macOS/Linux:
```bash
# Recommended: key authentication
./scripts/deploy.sh --ssh-key "$HOME/.ssh/pager_key"
# Password authentication requires sshpass
brew install hudochenkov/sshpass/sshpass
./scripts/deploy.sh --password '<device-password>'
```
Windows:
```powershell
# deploy.ps1 needs either an SSH key or sshpass for password auth:
& .\scripts\deploy.ps1 -SshKey "$HOME\.ssh\pager_key"
# or set up a key and add it: ssh-copy-id root@172.16.52.1
```
This builds `build\pager-webui\payload-<b64>.zip`, uploads it, extracts it to
`/root/payloads/user/remote_access/pager-webui/`, and refreshes the portal index.
The deployment scripts build `build/pager-webui/payload-<b64>.zip`, upload it,
extract it to `/root/payloads/user/remote_access/pager-webui/`, and refresh the
portal index.
Then on the Pager menu, run **Mark VIII**:
- **Yes** to "Run as background service?" -> procd service (respawns on crash,
+217
View File
@@ -0,0 +1,217 @@
# Radio1 5GHz/6GHz Rogue AP — Design Record
- **Date:** 2026-08-17
- **Status:** Implemented and verified on-device (see §9)
- **Owner:** Hak5 WiFi Pineapple Pager expansion project
- **Scope:** Mark VIII WebUI (`http://172.16.52.1:8080/`) running a rogue AP on
the Pager's second radio (`radio1` = MT7921U Wi-Fi 6E) on 5GHz and 6GHz, with
band-aware channel pickers, without breaking the stock Pager UI's control of
its own interfaces.
## 1. Goal
Today Mark VIII's PineAP Open AP and Evil WPA pages configure only `wlan0open`
/ `wlan0wpa` on `radio0` (2.4GHz) via the stock daemon's `set_ap` endpoint. The
Pager carries a second radio — an MT7921U Wi-Fi 6E on internal USB — that is
otherwise only used by the stock daemon's hopping monitor `wlan1mon`. This
feature lets Mark VIII run an Open AP or Evil WPA (WPA2-PSK / WPA3-SAE /
WPA3-OWE) on `radio1` at 5GHz and 6GHz, configured directly via UCI, with a
UI that picks channels by band and enforces WPA3 on 6GHz.
## 2. Hardware findings
Verified against the committed implementation (`server.py`, `views.js`) and the
plan's on-device groundwork:
- **`radio0`** — MT7628, **2.4GHz only**. The stock PineAP daemon owns
`wlan0open`, `wlan0wpa`, `wlan0cli`, `wlan0mon`, `wlan0mgmt`. This feature
does not change that ownership.
- **`radio1`** — MT7921U Wi-Fi 6E on internal USB (`phy1`), capable of
2.4/5/6GHz. The stock daemon owns `wlan1mon`, a daemon-managed hopping
monitor interface: `pineapd.wlan1mon` has `bands='2,5,6'` and `hop='1'`.
- Because `radio1` has a single channel shared by all its virtual interfaces,
a `wlan1` AP and the hopping `wlan1mon` cannot both be active with an
operator-chosen channel — hence the hop pause/resume mechanism (§6).
- UCI state (`wireless.radio1`): stock defaults are `band='5g'`,
`channel='auto'`, `htmode='VHT80'`, `country` set for the device region.
## 3. Band / channel model
`server.py` defines the authoritative mapping (helpers added near
`_uci_wifi_iface`):
| Band | `BAND_*` | Channels | `band_htmode` | `band_radio` |
|---|---|---|---|---|
| 2.4GHz | `BAND_2G = '2.4'` | `114` | `HT20` | `radio0` |
| 5GHz | `BAND_5G = '5'` | `36177` | `VHT80` | `radio1` |
| 6GHz | `BAND_6G = '6'` | `181233` (step 4) | `HE80` | `radio1` |
- `channel_band(ch)` classifies **2.4GHz (114) first, then 5GHz (36177)**, and
only then 6GHz (`1 ≤ ch ≤ 233` and `(ch 1) % 4 == 0`). Because 2.4 and 5GHz
take precedence, **the only reachable 6GHz channels are `181, 185, …, 233`**
(the low 6E channels `1,5,…,177` are swallowed by the 2.4/5GHz ranges). The
UI's 6GHz group therefore offers exactly `181..233 step 4`.
- `CHANNEL_BANDS` mirrors this: `range(1, 15)`, `range(36, 178)`,
`range(181, 234, 4)` — held consistent by tests.
- `DFS_CHANNELS` = `52..64` and `100..144` (step 4). DFS channels are surfaced
to the operator in the UI with a `(DFS)` marker; this feature does not attempt
radar-CAC handling on-device.
- `channel_freq(band, ch)`: 2.4 → `2412 + (ch1)·5`; 5 → `5180 + (ch36)·5`;
6 → `5955 + (ch1)·5`.
- `band_htmode` maps 2.4/5/6 → `HT20` / `VHT80` / `HE80` (written to
`wireless.radio1.htmode`). `band_radio` maps 2.4 → `radio0`, 5/6 → `radio1`.
## 4. API behavior
### 4.1 `h_pineap_wifi_get_ap` (POST `/api/pineap/wifi/get_ap`)
- If `wlan1open` **or** `wlan1wpa` exists in `wireless`, state is read from
`radio1` (`wlan1open`/`wlan1wpa`); otherwise from `radio0`
(`wlan0open`/`wlan0wpa`) — the 2.4GHz response shape is unchanged.
- `open.channel` and `wpa.channel` are reported **per interface**, falling back
to the owning radio's channel when the iface section has no channel option
(regression-tested). `wpa.enctype` is normalized to `psk2` / `sae` / `owe`.
- A new `radio1` object reports the raw `wireless.radio1` state:
`{band, channel, htmode, country}`, with `band` normalized from `2g`/`5g`/`6g`
to `'2.4'`/`'5'`/`'6'` (default `'5'`), and `channel` left as the raw string
(`'auto'` or a channel number) — the UI converts; `'auto'` is not int-coerced.
### 4.2 `h_pineap_wifi_set_ap` (POST `/api/pineap/wifi/set_ap`)
`channel` is now accepted in both `open` and `wpa` payloads. Behavior matrix:
- **2.4GHz channels (114):** exactly today's path — daemon
`PUT /api/settings/wifi/set_ap` for `wlan0open`/`wlan0wpa` followed by
`_apply_open_radio` (which persists `radio0.channel`/`country`). If a
`wlan1*` AP exists it is removed first (`_remove_radio1_ap`, §6) so a 2.4GHz
save tears down a stale radio1 AP.
- **5GHz (36177) / 6GHz (181233):** `_apply_radio1_ap` (below).
- **Mixed request:** a request carrying both a 2.4GHz object and a 5/6GHz
object returns `400 'cannot configure 2.4GHz and radio1 APs in one request'`
(radio1 is one physical radio — one band/channel per request).
- **Disable:** a radio1 request with no active 5/6GHz object (or a 2.4GHz save)
runs `_remove_radio1_ap()` + `wifi reload` and returns `200`.
`_apply_radio1_ap(openap, wpa)` (one of the two is active):
1. Validates the band — a radio1 AP requires a 5GHz or 6GHz channel
(`ValueError` → HTTP 400).
2. **6GHz requires WPA3:** when the active AP is a WPA AP on 6GHz, `enctype`
must be `sae` or `owe`; `psk2` is rejected with HTTP 400
(`'6GHz requires WPA3 (sae or owe)'`). An **open** 6GHz AP is accepted by
the backend, but the UI warns that real clients generally won't associate to
an open 6GHz network.
3. Deletes any existing `wlan1open`/`wlan1wpa` (idempotent), then writes UCI:
- `wireless.radio1.band` = `5g`/`6g`, `wireless.radio1.channel`,
`wireless.radio1.htmode` (`band_htmode`), `wireless.radio1.country`
(when supplied);
- a `wifi-iface` section `wlan1open` (encryption `none`, optional BSSID) or
`wlan1wpa` (`encryption` + `key`) on `device=radio1`, `mode=ap`, with the
interface-level `channel`/`hidden`/`ssid`.
4. `uci commit wireless`, `_pause_hop()` (§6), then `wifi reload`.
## 5. Coexistence rules — "don't break stock"
- **Mark VIII owns:** `wireless.wlan1open`, `wireless.wlan1wpa`, and
`wireless.radio1.{channel,band,htmode,country}`. These are new sections /
values it creates and tears down.
- **Stock owns (never modified by Mark VIII):** `wireless.wlan0open`,
`wlan0wpa`, `wlan0mgmt`, `wlan0cli`, `wlan0mon`, `wireless.wlan1mon`, and all
`pineapd.*` UCI values (bands configuration included).
- **The only stock-owned value this feature writes is
`pineapd.wlan1mon.hop`** — and it is always restored to its prior value
(`_resume_hop` sets it back to `1` only if it was `0`; `_pause_hop` sets it
to `0` only if it was not already `0`).
- The 2.4GHz daemon path (`PUT /api/settings/wifi/set_ap` for `wlan0open` /
`wlan0wpa`) is byte-for-byte unchanged.
- **Known risk (accepted):** the stock daemon's `set_ap`/pager-UI writes may
rewrite `wireless` wholesale and drop the `wlan1*` sections. Mitigation is
UCI-commit persistence, hop restore on disable, and on-device verification
(§9). If clobbering is observed, the deferred fix is a reconcile-on-load step
in `get_ap` that re-applies a saved radio1 AP from `PINEAP_STATE_FILE`.
## 6. Hop pause / resume
`wlan1mon` is the stock daemon's channel-hopping monitor. With a radio1 AP
active, hopping would fight the AP's fixed channel, so it is paused while the
AP is enabled:
- `_read_hop()` reads the value via `uci get pineapd.wlan1mon.hop` (a leaf
read — not `_uci_wifi_iface`, which forces the `wireless.` prefix).
- `_pause_hop()`: if `hop != '0'`, set `pineapd.wlan1mon.hop=0`, `uci commit
pineapd`, reload `/etc/init.d/pineapd`.
- `_resume_hop()`: if `hop == '0'`, set it back to `1`, commit, reload.
- `_apply_radio1_ap` calls `_pause_hop()` before `wifi reload`;
`_remove_radio1_ap` calls `_resume_hop()` after resetting
`radio1.channel=auto` / `radio1.band=5g`. Every code path that pauses hopping
also restores it.
## 7. Frontend (`www/js/views.js`)
- `BAND_GROUPS` drives the channel pickers shared by the Open AP and Evil WPA
views: a `2.4 GHz` optgroup (111), a `5 GHz` optgroup (36165, DFS channels
`52..64` / `100..144` labelled `(DFS)`), and a `6 GHz (WPA3/OWE only)`
optgroup (`181..233` step 4).
- `chanFreq`/`chanLabel` render `Channel N (… MHz)` (+` (DFS)`),
`chanSelect` builds the optgroups and restores a stored value when in range,
`bandOfChannel` mirrors `channel_band`.
- **Open AP:** channel select + a hint that appears on 6GHz ("…most devices
will not associate to an open 6 GHz network.").
- **Evil WPA:** a channel select added to the config card; selecting a 6GHz
channel disables the `psk2` option and switches to `sae`, with a
"6 GHz requires WPA3 (SAE or OWE)." hint. Save payloads for both views
include `channel` (Open AP also `country`).
## 8. Automated verification
- `tests/test_pineap_bands.py` covers the channel/band helpers
(`ChannelBandTest`, `ChannelBandsConsistencyTest`, `ChannelFreqTest`,
`BandAuxTest` incl. DFS marker), `get_ap` (`GetApRadio1Test`,
`GetApRadio1AbsentTest`, `GetApRadioChannelFallbackTest`) and `set_ap`
(`SetApRadio1Test`: 5GHz open writes `radio1` sections + hop pause; 6GHz
WPA3-SAE accepted; 6GHz `psk2` rejected; disable removes the radio1 AP and
restores hop; 2.4GHz still uses the daemon path; 2.4GHz save removes a stale
radio1 AP; mixed 2.4GHz + radio1 rejected).
- **All 14 test modules pass at HEAD (`af5ff80`)**, run per-module in separate
processes per the repo convention (`test_auth`, `test_core`, `test_loot`,
`test_misc`, `test_pineap_bands`, `test_pineap_clients`,
`test_pineap_enterprise`, `test_pineap_modes`, `test_pineap_pool`,
`test_pineap_proxy`, `test_pineap_settings`, `test_recon`, `test_status`,
`test_ws`).
## 9. On-device verification
Run against the user's Pager at `172.16.52.1` (Pineapple Pager 24.10.1). All
checks passed:
- **2.4GHz unchanged:** Open AP save (channel 1) leaves `wlan0open`/`radio0`
intact and `pineapd.wlan1mon` untouched (`bands=2,5,6`, `hop=1`).
- **5GHz Evil WPA (WPA3-SAE, channel 36, VHT80):** `radio1.band=5g`,
`channel=36`, `htmode=VHT80`; `wlan1wpa` (netdev named `wlan1wpa` via
`option ifname`) comes up beaconing `Test5G` / WPA3 SAE (CCMP);
`pineapd.wlan1mon.hop=0`. `get_ap` reports `wpa.enabled=true` (after the
`disabled=0` fix).
- **Stock Pager UI coexistence:** the stock daemon's own `set_ap` (what the
pager UI uses to change the 2.4GHz Evil WPA) tears down the radio1 AP
netdev; the `wlan1wpa` UCI section survives and `get_ap` self-heals it with a
`wifi reload` (`/sys/class/net/<iface>` missing check). Under rapid reload
churn the `mt7921u` driver can transiently return EBUSY; a later reload
succeeds. The pager UI itself is unaffected.
- **Handshake capture:** `Examine` on channel 36 returns success; handshake
logging (`loghandshake`/`logpartialhandshake`) confirmed on. A live WPA
handshake file requires a physical client (not exercised).
- **Reboot persistence:** `wlan1wpa` UCI, `disabled=0`, `hop=0`, the procd
Mark VIII service, and the AP itself all survive reboot.
- **Disable path:** removing the 5GHz AP deletes `wlan1wpa`, resets
`radio1.channel=auto`/`band=5g`, restores `hop=1`; `wlan1mon` hopping
resumes (observed 6GHz ch13 → ch221 in 30s).
- **5GHz Open AP:** `wlan1open` (channel 44, open) brings up `Test5GOpen`
with `hop=0`.
- **6GHz AP:** `radio1.band=6g`, `htmode=HE80`, WPA3 SAE on channel 181
(6.855 GHz) comes up.
- **Cleanup:** disable restores the 2.4GHz baseline (`pager-open`, channel 1,
`hop=1`).
Known follow-ups: the Clients tab lists only `wlan0*` interfaces, so 5GHz AP
clients are not yet shown; live-client handshake capture is untested without a
physical client.
@@ -8,7 +8,7 @@
"title": "Mark VIII",
"author": "c4ch3c4d3",
"description": "Mark VII-style web management UI for the WiFi Pineapple Pager",
"version": "1.0",
"version": "1.1",
"category": "remote_access",
"tags": ["remote-access", "web-interface", "device-management", "pineap"],
"firmware": "Pineapple Pager 24.10.1"
@@ -2,7 +2,7 @@
# Title: Mark VIII
# Description: Mark VII-style web management UI for the WiFi Pineapple Pager
# Author: c4ch3c4d3
# Version: 1.0
# Version: 1.1
# Category: Remote-Access
# Tags: remote-access, web-interface, device-management, pineap
# Firmware: Pineapple Pager 24.10.1
@@ -28,7 +28,7 @@ get_pager_ip() {
}
LOG "cyan" "+---------------------------+"
LOG "cyan" "| Mark VIII v1.0 |"
LOG "cyan" "| Mark VIII v1.1 |"
LOG "cyan" "+---------------------------+"
if ! command -v python3 >/dev/null 2>&1; then
File diff suppressed because it is too large Load Diff
@@ -354,6 +354,69 @@ html.dark .recon-row-selected td { background: #565656; }
th.recon-sorted { color: var(--primary); }
.recon-per { width: auto; }
/* ---- Recon supercharge: dBm bars, chips, pills ---- */
.recon-dbm-cell { display: inline-flex; align-items: center; gap: 8px; white-space: nowrap; }
.recon-dbm-bar { display: inline-block; width: 46px; height: 6px; border-radius: 3px; background: var(--surface-alt); overflow: hidden; vertical-align: middle; }
html.dark .recon-dbm-bar { background: #333; }
.recon-dbm-fill { display: block; height: 100%; border-radius: 3px; }
.recon-dbm-val { font-variant-numeric: tabular-nums; }
.recon-chips-row { display: flex; align-items: center; gap: 6px; flex-wrap: wrap; margin: 4px 0 10px; }
.recon-chips-label { font-size: 11px; text-transform: uppercase; letter-spacing: .06em; color: var(--muted); margin: 0 2px 0 8px; }
.recon-chips-label:first-child { margin-left: 0; }
.recon-chip { border: 1px solid var(--border); background: transparent; color: var(--muted); border-radius: 12px; padding: 3px 11px; font-size: 12px; cursor: pointer; }
.recon-chip:hover { color: var(--text); border-color: var(--primary); }
.recon-chip.active { background: var(--primary); border-color: var(--primary); color: #fff; }
.recon-pill { border: 1px solid var(--border); background: transparent; color: var(--muted); border-radius: 12px; padding: 3px 11px; font-size: 12px; cursor: pointer; display: inline-flex; align-items: center; gap: 5px; max-width: 260px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.recon-pill:hover { color: var(--text); border-color: var(--primary); }
.recon-pill:disabled { opacity: .5; cursor: default; }
.recon-pill.on { background: #e8f5e9; border-color: #a5d6a7; color: #2e7d32; }
html.dark .recon-pill.on { background: #1b3a23; color: #81c784; }
/* ---- Survey view ---- */
.survey-live-bar { display: flex; align-items: center; gap: 12px; flex-wrap: wrap; margin-bottom: 10px; }
.survey-live-status { font-size: 13px; color: var(--text); }
.survey-dur { width: auto; }
.survey-pill { border: 1px solid var(--border); background: transparent; color: var(--muted); border-radius: 12px; padding: 4px 12px; font-size: 12px; cursor: pointer; white-space: nowrap; }
.survey-pill:hover { color: var(--text); border-color: var(--primary); }
.survey-pill.on { background: #e8f5e9; border-color: #a5d6a7; color: #2e7d32; }
html.dark .survey-pill.on { background: #1b3a23; color: #81c784; }
.survey-wigle { font-size: 13px; color: var(--text); margin: 0; }
.survey-rec-card { display: flex; align-items: center; gap: 10px; flex-wrap: wrap; margin-bottom: 10px; }
.survey-rec-name { flex: 1 1 220px; max-width: 340px; }
.survey-rec-status { font-size: 12px; color: var(--muted); }
.survey-rec-status.on { color: #e53935; font-weight: 600; }
.survey-filter-row { margin-top: 10px; }
.survey-search { width: auto; }
.survey-chan-box { margin-top: 4px; }
.survey-chan-band { font-size: 12px; font-weight: 600; color: var(--primary); margin: 10px 0 4px; }
.survey-chan-row { display: flex; align-items: center; gap: 10px; margin: 3px 0; }
.survey-chan-label { width: 46px; font-size: 12px; color: var(--muted); font-variant-numeric: tabular-nums; }
.survey-chan-track { flex: 1; height: 14px; border-radius: 3px; background: var(--surface-alt); overflow: hidden; }
html.dark .survey-chan-track { background: #333; }
.survey-chan-fill { display: block; height: 100%; background: var(--primary); border-radius: 3px; }
.survey-chan-count { width: 30px; font-size: 12px; color: var(--muted); text-align: right; font-variant-numeric: tabular-nums; }
.survey-cmp-hint { font-size: 12px; color: var(--muted); margin: -4px 0 8px; }
.survey-cmp-legend { display: flex; flex-wrap: wrap; gap: 12px; margin-top: 6px; }
.survey-cmp-legend-item { display: inline-flex; align-items: center; gap: 6px; font-size: 12px; }
.survey-cmp-swatch { width: 10px; height: 10px; border-radius: 50%; flex: none; }
.survey-cmp-sig { color: var(--muted); font-variant-numeric: tabular-nums; }
.survey-cmp-check { display: inline-flex; }
.survey-cmp-check input { width: auto; }
.survey-discover { margin-top: 10px; }
.survey-discover-head { display: flex; align-items: center; justify-content: space-between; gap: 10px; }
.survey-discover-title { font-size: 20px; font-weight: 500; }
.survey-discover-readout { font-size: 44px; font-weight: 700; line-height: 1.1; font-variant-numeric: tabular-nums; }
.survey-discover-sub { color: var(--muted); font-size: 13px; margin: 2px 0 8px; word-break: break-all; }
/* ---- Reports view ---- */
.survey-detail-hint { font-size: 12px; color: var(--muted); margin-top: 10px; }
.survey-detail-row { padding: 8px 10px; border: 1px solid var(--border); border-radius: 3px; margin-top: 6px; cursor: pointer; font-size: 13px; }
.survey-detail-row:hover { border-color: var(--primary); }
.survey-detail-row.open { border-color: var(--primary); background: var(--surface-alt); }
.survey-detail-body { padding: 8px 4px; }
.survey-detail-gps { font-size: 12px; color: var(--muted); margin: 6px 0; }
.survey-wigle-warn { color: #ef6c00; font-size: 12px; }
/* ---- Mark VII handshakes table + settings dialog ---- */
.hs-cell-center { text-align: center; }
.hs-ok, .hs-bad, .hs-na { display: inline-flex; vertical-align: middle; }
@@ -474,6 +537,27 @@ html.dark .pineap-infobox.info { background: #10263a; color: #9cc7f0; border-col
.settings-diagnostics { max-height: 520px; white-space: pre-wrap; word-break: break-word; }
.settings-card > .switch { display: flex; margin: 10px 0; color: var(--text); font-size: 13px; }
.settings-card a { color: var(--primary); }
/* ---- WiFi client mode ---- */
.client-status { display: grid; grid-template-columns: repeat(auto-fit, minmax(220px, 1fr)); gap: 4px 20px; }
.client-kv { display: flex; align-items: center; justify-content: space-between; gap: 12px; border-bottom: 1px solid var(--border); padding: 6px 0; font-size: 13px; }
.client-kv > span { color: var(--muted); }
.client-actions { display: flex; gap: 8px; flex-wrap: wrap; margin: 12px 0 4px; }
.client-networks { display: flex; flex-direction: column; gap: 6px; margin-top: 8px; max-height: 320px; overflow-y: auto; }
.client-net-row { display: flex; align-items: center; gap: 12px; padding: 9px 10px; border: 1px solid var(--border); border-radius: 3px; }
.client-net-main { flex: 1; min-width: 0; display: flex; flex-direction: column; }
.client-net-ssid { font-weight: 500; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.client-net-enc { font-size: 11px; color: var(--muted); }
.client-net-signal { color: var(--muted); font-size: 12px; white-space: nowrap; }
.client-connect-form { display: flex; gap: 8px; align-items: center; flex-wrap: wrap; width: 100%; }
.client-connect-form input[type=password] { flex: 1 1 160px; }
.client-routing { display: flex; align-items: center; gap: 8px; margin: 8px 0 0; color: var(--text); font-size: 13px; }
.client-routing.hidden { display: none; }
.client-connect-form .client-routing { margin: 0; }
.client-modal { width: 560px; }
@media (max-width: 700px) {
.client-modal { min-width: 0; width: auto; }
}
@media (max-width: 700px) {
.tabbar { overflow-x: auto; flex-wrap: nowrap; }
.tabbar .tab { flex: 0 0 auto; }
@@ -261,13 +261,13 @@
<script src="js/config.js"></script>
<script src="js/icons.js?v=20260811-6"></script>
<script src="js/api.js?v=20260811-3"></script>
<script src="js/api.js?v=20260817-4"></script>
<script src="js/chart.js"></script>
<script src="js/xterm.min.js"></script>
<script src="js/xterm-addon-fit.min.js"></script>
<script src="js/terminal.js"></script>
<script src="js/pager.js"></script>
<script src="js/views.js?v=20260811-11"></script>
<script src="js/views.js?v=20260817-12"></script>
<script src="js/app.js?v=20260811-9"></script>
</body>
</html>
@@ -23,7 +23,9 @@ const PagerAPI = (() => {
data = await res.text();
}
if (!res.ok) {
const message = data && data.error ? data.error : ('HTTP ' + res.status);
const detail = data && typeof data.detail === 'string' ? data.detail : '';
const message = (data && data.error ? data.error : ('HTTP ' + res.status)) +
(detail ? ': ' + detail : '');
const error = new Error(message);
error.status = res.status;
error.data = data;
@@ -83,7 +83,7 @@ const App = (() => {
function route() {
closeToolbarMenus();
const hash = location.hash || '#/dashboard';
const hash = (location.hash || '#/dashboard').replace(/\/+$/, '');
const name = routes[hash];
if (currentView && currentView.destroy) currentView.destroy();
els.content.innerHTML = '';
@@ -233,7 +233,8 @@ const App = (() => {
location.hash = '#/settings/advanced';
toast('Update controls are available in Advanced settings');
} else if (action === 'internet') {
checkInternet(true);
if (typeof views.openClientModeModal === 'function') views.openClientModeModal();
else checkInternet(true);
} else if (action === 'logout') {
PagerAPI.post('/api/logout')
.then(() => showLogin())
@@ -395,6 +396,8 @@ const App = (() => {
'#/pineap/clients': 'pineap_clients',
'#/pineap/filtering': 'pineap_filtering',
'#/recon': 'recon',
'#/recon/survey': 'recon_survey',
'#/recon/reports': 'recon_reports',
'#/recon/handshakes': 'recon_handshakes',
'#/logging': 'logging',
'#/logging/system': 'logging_system',
@@ -410,8 +413,9 @@ const App = (() => {
'#/settings/help': 'settings_help'
};
return { init, route, toast, showLogin, wsUrl: (p) => WS_BASE + p, terminalWs: TERMINAL_WS,
pagerScreenWs: PAGER_SCREEN_WS, pagerKeysWs: PAGER_KEYS_WS, apiBase: API_BASE,
return { init, route, toast, showLogin, checkInternet, wsUrl: (p) => WS_BASE + p,
terminalWs: TERMINAL_WS, pagerScreenWs: PAGER_SCREEN_WS,
pagerKeysWs: PAGER_KEYS_WS, apiBase: API_BASE,
keyOf, railItems, go: (h) => { location.hash = h; },
get key() { return keyOf(location.hash); } };
})();
@@ -10,7 +10,10 @@ const MiniChart = (() => {
ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
const w = canvas.clientWidth, h = 140;
ctx.clearRect(0, 0, w, h);
const max = Math.max(o.max || 10, ...series.map((s) => Math.max(...s.points, 0)), 1);
const oMin = o.min == null ? 0 : o.min;
const max = Math.max(o.max || 10, ...series.map((s) => Math.max(...s.points, oMin)), oMin + 1);
const min = Math.min(oMin, ...series.map((s) => Math.min(...s.points, oMin)));
const span = Math.max(max - min, 1);
const pad = 8;
ctx.strokeStyle = o.grid || '#e0e0e0';
ctx.lineWidth = 1;
@@ -28,14 +31,14 @@ const MiniChart = (() => {
pts.forEach((v, i) => {
if (v == null) { started = false; return; }
const x = pad + (w - pad * 2) * i / Math.max(pts.length - 1, 1);
const y = h - pad - (h - pad * 2) * (v / max);
const y = h - pad - (h - pad * 2) * ((v - min) / span);
if (!started) { ctx.moveTo(x, y); started = true; } else ctx.lineTo(x, y);
});
ctx.stroke();
const last = pts[pts.length - 1];
if (last != null) {
const x = pad + (w - pad * 2) * (pts.length - 1) / Math.max(pts.length - 1, 1);
const y = h - pad - (h - pad * 2) * (last / max);
const y = h - pad - (h - pad * 2) * ((last - min) / span);
ctx.fillStyle = s.color || '#1976d2';
ctx.beginPath(); ctx.arc(x, y, 3, 0, Math.PI * 2); ctx.fill();
}
@@ -30,5 +30,11 @@ window.PineappleIcons = {
help: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,2A10,10 0 1,0 22,12A10,10 0 0,0 12,2M13,19H11V17H13V19M15.07,11.25L14.17,12.17C13.45,12.9 13,13.5 13,15H11V14.5C11,13.4 11.45,12.4 12.17,11.67L13.41,10.41C13.78,10.05 14,9.55 14,9A2,2 0 0,0 10,9H8A4,4 0 0,1 16,9C16,9.88 15.64,10.68 15.07,11.25Z"/></svg>',
update: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M21,10.12H14.22L16.96,7.3C14.23,4.6 9.81,4.5 7.08,7.2A6.85,6.85 0 0,0 7.08,17C9.81,19.7 14.23,19.7 16.96,17C18.32,15.65 19,14.08 19,12.1H21C21,14.08 20.18,16.4 18.36,18.2C14.85,21.7 9.15,21.7 5.64,18.2C2.14,14.72 2.14,9.05 5.64,5.57C9.15,2.08 14.85,2.08 18.36,5.57L21,2.88V10.12M12.5,8V12.25L16,14.33L15.28,15.54L11,13V8H12.5Z"/></svg>',
logout: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M14.08,15.59L16.67,13H7V11H16.67L14.08,8.41L15.5,7L20.5,12L15.5,17L14.08,15.59M5,3H13A2,2 0 0,1 15,5V8H13V5H5V19H13V16H15V19A2,2 0 0,1 13,21H5A2,2 0 0,1 3,19V5A2,2 0 0,1 5,3Z"/></svg>',
reboot: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M13,3H11V13H13V3M17.83,5.17L16.42,6.58A7,7 0 1,1 7.58,6.58L6.17,5.17A9,9 0 1,0 17.83,5.17Z"/></svg>'
reboot: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M13,3H11V13H13V3M17.83,5.17L16.42,6.58A7,7 0 1,1 7.58,6.58L6.17,5.17A9,9 0 1,0 17.83,5.17Z"/></svg>',
table_chart: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M12,20H9V4H12V20M19,20H16V10H19V20M5,20H2V14H5V20Z"/></svg>',
description: '<svg viewBox="0 0 24 24" fill="currentColor"><path d="M14,2H6C4.9,2 4,2.9 4,4V20C4,21.1 4.9,22 6,22H18C19.1,22 20,21.1 20,20V8L14,2M18,20H6V4H13V9H18V20M16,18H8V16H16V18M16,14H8V12H16V14Z"/></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>',
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>'
};
File diff suppressed because it is too large Load Diff
+186
View File
@@ -0,0 +1,186 @@
#!/usr/bin/env bash
set -euo pipefail
PAGER_HOST="172.16.52.1"
PAGER_USER="root"
PASSWORD=""
SSH_KEY=""
BUILD_DIR=""
PORTAL_REFRESH=true
usage() {
cat <<'EOF'
Usage: scripts/deploy.sh [options]
Options:
--host HOST Pager address (default: 172.16.52.1)
--user USER SSH user (default: root)
--password PASSWORD SSH/device password (requires sshpass)
--ssh-key PATH SSH private key
--build-dir PATH Build output directory (default: <repo>/build)
--no-portal-refresh Skip the best-effort portal refresh
-h, --help Show this help
If neither --password nor --ssh-key is supplied, ssh/scp prompt normally.
EOF
}
while (($#)); do
case "$1" in
--host) PAGER_HOST="${2:?missing value for --host}"; shift 2 ;;
--user) PAGER_USER="${2:?missing value for --user}"; shift 2 ;;
--password) PASSWORD="${2:?missing value for --password}"; shift 2 ;;
--ssh-key) SSH_KEY="${2:?missing value for --ssh-key}"; shift 2 ;;
--build-dir) BUILD_DIR="${2:?missing value for --build-dir}"; shift 2 ;;
--no-portal-refresh) PORTAL_REFRESH=false; shift ;;
-h|--help) usage; exit 0 ;;
*) printf 'Unknown option: %s\n' "$1" >&2; usage >&2; exit 2 ;;
esac
done
for command in python3 zip scp ssh; do
command -v "$command" >/dev/null || {
printf 'Required command not found: %s\n' "$command" >&2
exit 1
}
done
if [[ -n "$PASSWORD" ]] && ! command -v sshpass >/dev/null; then
printf 'Password deployment requires sshpass (brew install hudochenkov/sshpass/sshpass).\n' >&2
exit 1
fi
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
PAYLOAD_KEY="pager-webui"
PAYLOAD_CATEGORY="remote_access"
PAYLOAD_DIR="$ROOT/payload/user/$PAYLOAD_CATEGORY/$PAYLOAD_KEY"
BUILD_DIR="${BUILD_DIR:-$ROOT/build}"
OUT_DIR="$BUILD_DIR/$PAYLOAD_KEY"
STAGE="$OUT_DIR/stage"
[[ -d "$PAYLOAD_DIR" ]] || {
printf 'Payload directory not found: %s\n' "$PAYLOAD_DIR" >&2
exit 1
}
mkdir -p "$OUT_DIR"
rm -rf "$STAGE"
mkdir -p "$STAGE/user/$PAYLOAD_CATEGORY"
cp -R "$PAYLOAD_DIR" "$STAGE/user/$PAYLOAD_CATEGORY/$PAYLOAD_KEY"
find "$STAGE" \( -type d -name __pycache__ -o -type f -name '*.pyc' \) -prune -exec rm -rf {} +
B64_KEY="$(python3 -c 'import base64; print(base64.urlsafe_b64encode(b"pager-webui").decode().rstrip("="))')"
ZIP_NAME="payload-$B64_KEY.zip"
ZIP_PATH="$OUT_DIR/$ZIP_NAME"
MANIFEST_PATH="$OUT_DIR/_hak5_manifest.json"
rm -f "$ZIP_PATH"
(
cd "$STAGE"
zip -q -r "$ZIP_PATH" user
)
HASH="$(python3 -c 'import hashlib, sys; print(hashlib.sha256(open(sys.argv[1], "rb").read()).hexdigest())' "$ZIP_PATH")"
python3 - "$PAYLOAD_DIR/_hak5_manifest.json" "$MANIFEST_PATH" "$HASH" "$ZIP_NAME" <<'PY'
import json
import sys
import time
source, destination, digest, zip_name = sys.argv[1:]
with open(source, encoding='utf-8') as handle:
manifest = json.load(handle)
manifest['time'] = int(time.time())
manifest['last_hash'] = digest
manifest['zip'] = zip_name
with open(destination, 'w', encoding='ascii') as handle:
json.dump(manifest, handle, indent=2)
handle.write('\n')
PY
printf 'Built: %s\n' "$ZIP_PATH"
TARGET="$PAGER_USER@$PAGER_HOST"
run_scp() {
if [[ -n "$PASSWORD" && -n "$SSH_KEY" ]]; then
SSHPASS="$PASSWORD" sshpass -e scp -i "$SSH_KEY" "$@"
elif [[ -n "$PASSWORD" ]]; then
SSHPASS="$PASSWORD" sshpass -e scp "$@"
elif [[ -n "$SSH_KEY" ]]; then
scp -i "$SSH_KEY" "$@"
else
scp "$@"
fi
}
run_ssh() {
if [[ -n "$PASSWORD" && -n "$SSH_KEY" ]]; then
SSHPASS="$PASSWORD" sshpass -e ssh -i "$SSH_KEY" "$@"
elif [[ -n "$PASSWORD" ]]; then
SSHPASS="$PASSWORD" sshpass -e ssh "$@"
elif [[ -n "$SSH_KEY" ]]; then
ssh -i "$SSH_KEY" "$@"
else
ssh "$@"
fi
}
run_scp "$ZIP_PATH" "$MANIFEST_PATH" "$TARGET:/tmp/"
REMOTE_PAYLOAD_DIR="user/$PAYLOAD_CATEGORY/$PAYLOAD_KEY"
LEGACY_PAYLOAD_DIR="user/general/$PAYLOAD_KEY"
REMOTE_COMMAND="set -e
cd /root/payloads
stage='.pager-webui.deploy.\$\$'
backup='.pager-webui.backup.\$\$'
trap 'rm -rf \"\$stage\" \"\$backup\"' EXIT
mkdir -p \"\$stage\"
cd \"\$stage\"
unzip -q '/tmp/$ZIP_NAME'
new=\"\$PWD/$REMOTE_PAYLOAD_DIR\"
[ -f \"\$new/server.py\" ] && [ -f \"\$new/payload.sh\" ] && [ -d \"\$new/www\" ]
cp /tmp/_hak5_manifest.json \"\$new/_hak5_manifest.json\"
chmod +x \"\$new/payload.sh\" \"\$new/pagerwebui.init\"
chmod -R 755 \"\$new/www\"
cd /root/payloads
if [ -d '$REMOTE_PAYLOAD_DIR' ]; then
mkdir -p \"\$(dirname \"\$backup\")\"
mv '$REMOTE_PAYLOAD_DIR' \"\$backup\"
fi
if mv \"\$new\" '$REMOTE_PAYLOAD_DIR'; then
rm -rf \"\$backup\" '$LEGACY_PAYLOAD_DIR'
else
[ ! -d \"\$backup\" ] || mv \"\$backup\" '$REMOTE_PAYLOAD_DIR'
exit 1
fi
rm -f '/tmp/$ZIP_NAME' /tmp/_hak5_manifest.json
if [ -x /etc/init.d/pagerwebui ] && /etc/init.d/pagerwebui running >/dev/null 2>&1; then
/etc/init.d/pagerwebui restart
fi
echo EXTRACT_OK"
run_ssh "$TARGET" "$REMOTE_COMMAND"
printf 'Installed to /root/payloads/%s/\n' "$REMOTE_PAYLOAD_DIR"
if $PORTAL_REFRESH && [[ -n "$PASSWORD" ]]; then
PASSWORD_B64="$(printf '%s' "$PASSWORD" | base64)"
PORTAL_OK=false
for attempt in 1 2 3; do
if printf '%s\n' "$PASSWORD_B64" | run_ssh "$TARGET" 'read -r password_b64
password=$(printf "%s" "$password_b64" | base64 -d)
login_body=$(printf "%s" "$password" | python3 -c '"'"'import json, sys; print(json.dumps({"username": "root", "password": sys.stdin.read()}))'"'"')
token=$(curl -sS -X POST http://127.0.0.1:1471/api/login -H "Content-Type: application/json" -d "$login_body" |
python3 -c '"'"'import json, sys; print(json.load(sys.stdin).get("token", ""))'"'"')
[ -n "$token" ] &&
curl -fsS -X POST http://127.0.0.1:1471/api/payloads/portal/refresh -H "Authorization: Bearer $token" >/dev/null'; then
PORTAL_OK=true
break
fi
sleep 2
done
if $PORTAL_OK; then
printf 'Portal refreshed.\n'
else
printf 'Warning: portal refresh failed; payload installation is complete.\n' >&2
fi
elif $PORTAL_REFRESH; then
printf 'Skipping portal refresh without --password; payload installation is complete.\n'
fi
printf 'Deploy complete. Browse http://%s:8080/\n' "$PAGER_HOST"
+267
View File
@@ -41,6 +41,26 @@ class PayloadsProxyTest(unittest.TestCase):
server.h_payloads_remove(type('C', (), {'args': (), 'body': {'key': 'nautilus'}})())
self.assertTrue(any(m == 'POST' and '/api/payloads/portal/nautilus/remove' in p for m, p in calls))
def test_install_surfaces_daemon_error_detail(self):
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (
500, {'error': 'network error: Get "https://downloads.hak5.org/.../download": dial tcp: lookup downloads.hak5.org on [::1]:53: server misbehaving'})
server.current_token = lambda: 'tok'
status, payload = server.h_payloads_install(type('C', (), {
'args': (), 'body': {'key': 'recon~client~recon_reporter'}})())
self.assertEqual(status, 500)
self.assertIn('downloads.hak5.org', payload['detail'])
def test_install_unwraps_raw_json_daemon_error(self):
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (
500, '{"error":"network error: Get \\"https://downloads.hak5.org/...\\": dial tcp: lookup downloads.hak5.org on [::1]:53: server misbehaving"}\n')
server.current_token = lambda: 'tok'
status, payload = server.h_payloads_install(type('C', (), {
'args': (), 'body': {'key': 'recon~client~recon_reporter'}})())
self.assertEqual(status, 500)
self.assertIn('network error', payload['detail'])
self.assertIn('downloads.hak5.org', payload['detail'])
self.assertNotIn('{', payload['detail'])
def test_installed_inventory_flattens_firmware_records(self):
old = server._payload_daemon
server._payload_daemon = lambda method, path, body=None: (200, [{
@@ -275,5 +295,252 @@ class SettingsTest(unittest.TestCase):
self.assertNotIn('password', payload)
class WifiClientModeTest(unittest.TestCase):
@staticmethod
def fake_device_run(calls, responses):
def run(args, timeout=20, input_data=None):
calls.append((list(args), timeout))
key = (args[0], args[1])
return responses.get(key, (0, '', ''))
return run
def test_client_state_parses_disabled(self):
calls = []
run = self.fake_device_run(calls, {
('uci', 'show'): (0,
"wireless.wlan0cli=wifi-iface\n"
"wireless.wlan0cli.ssid='OldNet'\n"
"wireless.wlan0cli.disabled='1'\n"
"wireless.wlan0cli.routed='0'\n", ''),
('iw', 'dev'): (0, '', ''),
('ip', '-4'): (0, '', ''),
})
old = server.device_run
server.device_run = run
try:
status, payload = server.h_settings_wifi_client(type('C', (), {})())
finally:
server.device_run = old
self.assertEqual(status, 200)
self.assertFalse(payload['enabled'])
self.assertFalse(payload['connected'])
self.assertEqual(payload['ssid'], 'OldNet')
self.assertFalse(payload['routed'])
self.assertEqual(payload['ip'], '')
def test_client_state_parses_connected(self):
calls = []
run = self.fake_device_run(calls, {
('uci', 'show'): (0,
"wireless.wlan0cli=wifi-iface\n"
"wireless.wlan0cli.ssid='All RPH Guest WIFI'\n"
"wireless.wlan0cli.disabled='0'\n"
"wireless.wlan0cli.routed='0'\n", ''),
('iw', 'dev'): (0,
"Connected to 02:18:4a:a7:6a:d9 (on wlan0cli)\n"
"\tSSID: All RPH Guest WIFI\n"
"\tfreq: 2462\n"
"\tsignal: -57 dBm\n", ''),
('ip', '-4'): (0,
"6: wlan0cli: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
" inet 10.10.10.5/24 brd 10.10.10.255 scope global wlan0cli\n", ''),
})
old = server.device_run
server.device_run = run
try:
status, payload = server.h_settings_wifi_client(type('C', (), {})())
finally:
server.device_run = old
self.assertEqual(status, 200)
self.assertTrue(payload['enabled'])
self.assertTrue(payload['connected'])
self.assertEqual(payload['connected_ssid'], 'All RPH Guest WIFI')
self.assertEqual(payload['ip'], '10.10.10.5')
self.assertEqual(payload['signal'], -57)
self.assertEqual(payload['freq'], 2462)
def test_scan_parses_networks_and_deduplicates(self):
sample = (
"BSS 02:18:4a:a7:6a:d2(on wlan0)\n"
"\tfreq: 2462.0\n"
"\tsignal: -63.00 dBm\n"
"\tSSID: Riverwalk Plaza Staff\n"
"\tRSN:\t * Version: 1\n"
"\t\t * Group cipher: TKIP\n"
"\t\t * Pairwise ciphers: CCMP TKIP\n"
"\t\t * Authentication suites: PSK\n"
"BSS 02:18:4a:a7:6a:d9(on wlan0)\n"
"\tfreq: 2462.0\n"
"\tsignal: -61.00 dBm\n"
"\tSSID: All RPH Guest WIFI\n"
"\tRSN:\t * Version: 1\n"
"\t\t * Group cipher: CCMP\n"
"\t\t * Pairwise ciphers: CCMP\n"
"\t\t * Authentication suites: SAE\n"
"BSS ea:cb:bc:8e:c5:0e(on wlan0)\n"
"\tfreq: 2462.0\n"
"\tsignal: -50.00 dBm\n"
"\tSSID: OpenGuest\n"
"BSS c6:cb:bc:8e:c5:0e(on wlan0)\n"
"\tfreq: 2462.0\n"
"\tsignal: -55.00 dBm\n"
"\tSSID: \\x00\\x00\\x00\\x00\n"
"BSS 42:18:4a:a7:6a:d2(on wlan0)\n"
"\tfreq: 2462.0\n"
"\tsignal: -65.00 dBm\n"
"\tSSID: Riverwalk Plaza Staff\n"
"\tWPA:\t * Version: 1\n"
"\t\t * Group cipher: TKIP\n"
"\t\t * Authentication suites: PSK\n")
networks = server._parse_wifi_scan(sample)
by_ssid = {n['ssid']: n for n in networks}
self.assertIn('Riverwalk Plaza Staff', by_ssid)
self.assertIn('All RPH Guest WIFI', by_ssid)
self.assertIn('OpenGuest', by_ssid)
self.assertEqual(by_ssid['Riverwalk Plaza Staff']['encryption'], 'WPA2')
self.assertEqual(by_ssid['All RPH Guest WIFI']['encryption'], 'WPA3')
self.assertEqual(by_ssid['OpenGuest']['encryption'], 'Open')
self.assertEqual(by_ssid['OpenGuest']['channel'], 11)
# hidden SSID entries are omitted
self.assertNotIn('', by_ssid)
# strongest BSS per SSID wins and results are signal-sorted (strongest first)
self.assertEqual(networks[0]['ssid'], 'OpenGuest')
self.assertEqual(networks[0]['signal'], -50)
self.assertGreater(networks[0]['signal'], networks[1]['signal'])
def test_scan_reports_device_failure(self):
old = server.device_run
server.device_run = lambda args, timeout=20: (1, '', 'scan not supported')
try:
status, payload = server.h_settings_wifi_client_scan(type('C', (), {})())
finally:
server.device_run = old
self.assertEqual(status, 502)
self.assertIn('scan', payload['error'])
def test_connect_requires_ssid(self):
class H:
command = 'POST'
old = server.device_run
server.device_run = lambda args, timeout=20: (0, '', '')
try:
status, payload = server.h_settings_wifi_client_connect(
type('C', (), {'h': H(), 'body': {'encryption': 'open'}})())
finally:
server.device_run = old
self.assertEqual(status, 400)
self.assertIn('SSID', payload['error'])
def test_connect_rejects_short_password(self):
class H:
command = 'POST'
calls = []
run = self.fake_device_run(calls, {})
old = server.device_run
server.device_run = run
try:
status, payload = server.h_settings_wifi_client_connect(
type('C', (), {'h': H(), 'body': {'ssid': 'X', 'encryption': 'wpa2', 'password': 'short'}})())
finally:
server.device_run = old
self.assertEqual(status, 400)
self.assertIn('password', payload['error'])
self.assertEqual(calls, [])
def test_connect_writes_uci_and_reloads(self):
class H:
command = 'POST'
calls = []
run = self.fake_device_run(calls, {})
daemon_calls = []
old_run = server.device_run
old_sock = server.daemon_sock_call
server.device_run = run
server.daemon_sock_call = lambda m, p, body=None, timeout=10: (
daemon_calls.append((m, p, body)) or (200, {'success': True}))
try:
status, payload = server.h_settings_wifi_client_connect(
type('C', (), {'h': H(), 'body': {
'ssid': 'All RPH Guest WIFI', 'encryption': 'wpa2wpa3',
'password': 'Missions1', 'routed': True}})())
finally:
server.device_run = old_run
server.daemon_sock_call = old_sock
self.assertEqual(status, 200)
sets = [args for args, _t in calls if args[:2] == ['uci', 'set']]
expected = {
'wireless.wlan0cli.ssid=All RPH Guest WIFI',
'wireless.wlan0cli.encryption=sae-mixed',
'wireless.wlan0cli.disabled=0',
'wireless.wlan0cli.routed=1',
'wireless.wlan0cli.key=Missions1',
}
got = {args[2] for args in sets}
self.assertTrue(expected <= got)
self.assertIn((['uci', 'commit', 'wireless'], 20), calls)
self.assertIn((['wifi', 'reload'], 45), calls)
self.assertEqual(daemon_calls,
[('PUT', '/api/settings/wifi/set_client_route', {'routed': True})])
def test_connect_open_clears_key(self):
class H:
command = 'POST'
calls = []
run = self.fake_device_run(calls, {})
old_run = server.device_run
old_sock = server.daemon_sock_call
server.device_run = run
server.daemon_sock_call = lambda m, p, body=None, timeout=10: (200, {'success': True})
try:
status, payload = server.h_settings_wifi_client_connect(
type('C', (), {'h': H(), 'body': {'ssid': 'OpenGuest', 'encryption': 'open'}})())
finally:
server.device_run = old_run
server.daemon_sock_call = old_sock
self.assertEqual(status, 200)
self.assertIn((['uci', 'delete', 'wireless.wlan0cli.key'], 20), calls)
self.assertNotIn((['uci', 'set', 'wireless.wlan0cli.key='], 20), calls)
def test_route_toggle_writes_uci_and_syncs_daemon(self):
class H:
command = 'POST'
calls = []
run = self.fake_device_run(calls, {})
daemon_calls = []
old_run = server.device_run
old_sock = server.daemon_sock_call
server.device_run = run
server.daemon_sock_call = lambda m, p, body=None, timeout=10: (
daemon_calls.append((m, p, body)) or (200, {'success': True}))
try:
status, payload = server.h_settings_wifi_client_route(
type('C', (), {'h': H(), 'body': {'routed': True}})())
finally:
server.device_run = old_run
server.daemon_sock_call = old_sock
self.assertEqual(status, 200)
self.assertIn((['uci', 'set', 'wireless.wlan0cli.routed=1'], 20), calls)
self.assertIn((['uci', 'commit', 'wireless'], 20), calls)
self.assertEqual(daemon_calls,
[('PUT', '/api/settings/wifi/set_client_route', {'routed': True})])
def test_disconnect_disables_and_reloads(self):
class H:
command = 'POST'
calls = []
run = self.fake_device_run(calls, {})
old = server.device_run
server.device_run = run
try:
status, payload = server.h_settings_wifi_client_disconnect(
type('C', (), {'h': H(), 'body': {}})())
finally:
server.device_run = old
self.assertEqual(status, 200)
self.assertIn((['uci', 'set', 'wireless.wlan0cli.disabled=1'], 20), calls)
self.assertIn((['uci', 'commit', 'wireless'], 20), calls)
self.assertIn((['wifi', 'reload'], 45), calls)
if __name__ == '__main__':
unittest.main()
+363
View File
@@ -0,0 +1,363 @@
import os
import sys
import unittest
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)
class ChannelBandTest(unittest.TestCase):
def test_2g_channels(self):
for ch in (1, 6, 11, 14):
self.assertEqual(server.channel_band(ch), server.BAND_2G)
def test_5g_channels(self):
for ch in (36, 48, 100, 149, 165, 177):
self.assertEqual(server.channel_band(ch), server.BAND_5G)
def test_6g_channels(self):
for ch in (181, 189, 197, 205, 213, 225, 233):
self.assertEqual(server.channel_band(ch), server.BAND_6G)
def test_invalid(self):
for ch in (0, 15, 17, 21, 33, 35, 178, 234, None, 'x'):
self.assertIsNone(server.channel_band(ch))
class ChannelBandsConsistencyTest(unittest.TestCase):
def test_lists_match_channel_band(self):
for band, channels in server.CHANNEL_BANDS.items():
for ch in channels:
self.assertEqual(server.channel_band(ch), band, '%s should be %s' % (ch, band))
def test_no_out_of_list_channels(self):
for ch in list(range(0, 235)):
band = server.channel_band(ch)
if band is not None:
self.assertIn(ch, server.CHANNEL_BANDS[band], '%s should be listed for %s' % (ch, band))
def test_boundaries(self):
self.assertEqual(server.CHANNEL_BANDS[server.BAND_2G][-1], 14)
self.assertEqual(server.CHANNEL_BANDS[server.BAND_5G][-1], 177)
self.assertEqual(server.CHANNEL_BANDS[server.BAND_6G][0], 181)
self.assertEqual(server.CHANNEL_BANDS[server.BAND_6G][-1], 233)
class ChannelFreqTest(unittest.TestCase):
def test_freqs(self):
self.assertEqual(server.channel_freq(server.BAND_2G, 1), 2412)
self.assertEqual(server.channel_freq(server.BAND_2G, 11), 2462)
self.assertEqual(server.channel_freq(server.BAND_5G, 36), 5180)
self.assertEqual(server.channel_freq(server.BAND_5G, 165), 5825)
self.assertEqual(server.channel_freq(server.BAND_6G, 1), 5955)
self.assertEqual(server.channel_freq(server.BAND_6G, 233), 7115)
class BandAuxTest(unittest.TestCase):
def test_htmode(self):
self.assertEqual(server.band_htmode(server.BAND_2G), 'HT20')
self.assertEqual(server.band_htmode(server.BAND_5G), 'VHT80')
self.assertEqual(server.band_htmode(server.BAND_6G), 'HE80')
def test_radio(self):
self.assertEqual(server.band_radio(server.BAND_2G), 'radio0')
self.assertEqual(server.band_radio(server.BAND_5G), 'radio1')
self.assertEqual(server.band_radio(server.BAND_6G), 'radio1')
def test_dfs_marker(self):
for ch in (52, 64, 100, 144):
self.assertIn(ch, server.DFS_CHANNELS)
for ch in (36, 48, 149):
self.assertNotIn(ch, server.DFS_CHANNELS)
self.assertEqual(set(server.DFS_CHANNELS),
set(range(52, 65, 4)) | set(range(100, 145, 4)))
def ctx(body=None):
return type('C', (), {'body': body, 'args': (), 'query': {}})()
class GetApRadio1Test(unittest.TestCase):
def _uci(self, section):
table = {
'radio0': {'type': 'wifi-device', 'band': '2g', 'channel': '11',
'htmode': 'HT20', 'country': 'US'},
'radio1': {'type': 'wifi-device', 'band': '5g', 'channel': 'auto',
'htmode': 'VHT80', 'country': 'US'},
'wlan0open': {'device': 'radio0', 'mode': 'ap', 'ssid': 'pager-open',
'disabled': '0', 'hidden': '0', 'encryption': 'none',
'channel': '11'},
'wlan0wpa': {'device': 'radio0', 'mode': 'ap', 'ssid': 'Service',
'disabled': '0', 'hidden': '0', 'encryption': 'psk2',
'channel': '1', 'key': 'testpass123'},
'wlan1open': {'device': 'radio1', 'mode': 'ap', 'ssid': 'CorpGuest',
'disabled': '0', 'hidden': '0', 'encryption': 'none',
'channel': '36'},
'wlan1wpa': {'device': 'radio1', 'mode': 'ap', 'ssid': 'Corp',
'disabled': '0', 'hidden': '0', 'encryption': 'sae',
'channel': '1', 'key': 'secret123'},
}
return dict(table.get(section, {}))
def setUp(self):
server._uci_wifi_iface = lambda name: self._uci(name)
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
404, {'error': 'not found'})
def test_open_reports_radio1_when_present(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['open']['ssid'], 'CorpGuest')
self.assertEqual(payload['open']['channel'], 36)
self.assertEqual(payload['open']['country'], 'US')
def test_wpa_reports_radio1_when_present(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['wpa']['ssid'], 'Corp')
self.assertEqual(payload['wpa']['enctype'], 'sae')
self.assertEqual(payload['wpa']['channel'], 1)
def test_radio1_info(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['radio1']['band'], server.BAND_5G)
self.assertEqual(payload['radio1']['channel'], 'auto')
class GetApRadio1AbsentTest(unittest.TestCase):
"""Regression: no radio1 AP sections -> today's 2.4GHz behavior."""
def _uci(self, section):
table = {
'radio0': {'type': 'wifi-device', 'band': '2g', 'channel': '11',
'htmode': 'HT20', 'country': 'US'},
'radio1': {'type': 'wifi-device', 'band': '5g', 'channel': 'auto',
'htmode': 'VHT80', 'country': 'US'},
'wlan0open': {'device': 'radio0', 'mode': 'ap', 'ssid': 'pager-open',
'disabled': '0', 'hidden': '0', 'encryption': 'none',
'channel': '11'},
'wlan0wpa': {'device': 'radio0', 'mode': 'ap', 'ssid': 'Service',
'disabled': '0', 'hidden': '0', 'encryption': 'psk2',
'channel': '1', 'key': 'testpass123'},
}
return dict(table.get(section, {}))
def setUp(self):
server._uci_wifi_iface = lambda name: self._uci(name)
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
404, {'error': 'not found'})
def test_open_uses_wlan0open(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['open']['ssid'], 'pager-open')
self.assertEqual(payload['open']['channel'], 11)
def test_wpa_uses_wlan0wpa(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['wpa']['ssid'], 'Service')
self.assertEqual(payload['wpa']['channel'], 1)
class SetApRadio1Test(unittest.TestCase):
def setUp(self):
self.uci = {}
self.runs = []
def fake_uci_get(section):
return self.uci.get(section)
def fake_run(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.uci[k] = v
if a[:2] == ['uci', 'get']:
return (0, self.uci.get(a[2], '') + '\n', '')
return (0, '', '')
server._uci_wifi_iface = fake_uci_get
server._uci_section = fake_uci_get
server.device_run = fake_run
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'success': True})
def test_5g_open_writes_radio1_sections(self):
server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
'channel': 36, 'country': 'US'}}))
self.assertEqual(self.uci['wireless.radio1.band'], '5g')
self.assertEqual(self.uci['wireless.radio1.channel'], '36')
self.assertEqual(self.uci['wireless.radio1.htmode'], 'VHT80')
self.assertEqual(self.uci['wireless.wlan1open'], 'wifi-iface')
self.assertEqual(self.uci['wireless.wlan1open.device'], 'radio1')
self.assertEqual(self.uci['wireless.wlan1open.disabled'], '0')
self.assertEqual(self.uci['wireless.wlan1open.ssid'], 'CorpGuest')
self.assertEqual(self.uci['wireless.wlan1open.encryption'], 'none')
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '0')
self.assertIn(['wifi', 'reload'], [r[0] for r in self.runs])
self.assertIn(['/etc/init.d/pineapd', 'reload'], [r[0] for r in self.runs])
def test_6g_wpa_sae(self):
server.h_pineap_wifi_set_ap(ctx({'wpa': {
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'sae',
'hidden': False, 'enabled': True, 'channel': 181}}))
self.assertEqual(self.uci['wireless.radio1.band'], '6g')
self.assertEqual(self.uci['wireless.radio1.htmode'], 'HE80')
self.assertEqual(self.uci['wireless.wlan1wpa.encryption'], 'sae')
def test_6g_rejects_psk2(self):
status, payload = server.h_pineap_wifi_set_ap(ctx({'wpa': {
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'psk2',
'hidden': False, 'enabled': True, 'channel': 181}}))
self.assertEqual(status, 400)
def test_6g_open_rejected(self):
status, payload = server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
'channel': 181, 'country': 'US'}}))
self.assertEqual(status, 400)
self.assertIn('6GHz', payload['error'])
def test_6g_disable_removes_radio1(self):
self.uci['pineapd.wlan1mon.hop'] = '0'
self.uci['wlan1wpa'] = {'device': 'radio1'}
status, payload = server.h_pineap_wifi_set_ap(ctx({'wpa': {
'ssid': 'Corp', 'passphrase': 'secret123', 'enctype': 'sae',
'hidden': False, 'enabled': False, 'channel': 181}}))
self.assertEqual(status, 200)
self.assertEqual(self.uci['wireless.radio1.channel'], 'auto')
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
def test_5g_open_rejects_bad_bssid(self):
status, payload = server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'CorpGuest', 'bssid': 'not-a-mac', 'hidden': False,
'enabled': True, 'channel': 36, 'country': 'US'}}))
self.assertEqual(status, 400)
def test_hop_read_failure_still_pauses(self):
def fake_run(args, timeout=20, input_data=None):
self.runs.append((list(args), input_data))
a = list(args)
if a[:2] == ['uci', 'get']:
return (1, '', '')
if a[:2] == ['uci', 'set']:
k, _, v = a[2].partition('=')
self.uci[k] = v
return (0, '', '')
server.device_run = fake_run
server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
'channel': 36, 'country': 'US'}}))
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '0')
def test_2g_still_uses_daemon_path(self):
calls = []
def fake_sock(method, path, body=None, timeout=10):
if method == 'GET':
return 200, {'loghandshake': True}
calls.append((method, path, body))
return 200, {'success': True}
server.daemon_sock_call = fake_sock
server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'pager-open', 'hidden': False, 'enabled': True,
'channel': 11, 'country': 'US'}}))
put = [c for c in calls if c[0] == 'PUT']
self.assertEqual(len(put), 1)
self.assertEqual(put[0][1], '/api/settings/wifi/set_ap')
self.assertEqual(put[0][2]['configs'][0]['interface'], 'wlan0open')
def test_2g_removes_existing_radio1(self):
self.uci['pineapd.wlan1mon.hop'] = '0'
self.uci['wlan1open'] = {'device': 'radio1'}
server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'pager-open', 'hidden': False, 'enabled': True,
'channel': 11, 'country': 'US'}}))
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
self.assertEqual(self.uci.get('wireless.radio1.channel'), 'auto')
def test_mixed_24g_and_radio1_rejected(self):
status, payload = server.h_pineap_wifi_set_ap(ctx({
'open': {'ssid': 'CorpGuest', 'hidden': False, 'enabled': True,
'channel': 36, 'country': 'US'},
'wpa': {'ssid': 'Office', 'passphrase': 'secret123', 'enctype': 'psk2',
'hidden': False, 'enabled': True, 'channel': 6}}))
self.assertEqual(status, 400)
self.assertIn('2.4GHz', payload['error'])
class GetApRadioChannelFallbackTest(unittest.TestCase):
"""Regression: iface without a channel option inherits the radio channel."""
def _uci(self, section):
table = {
'radio0': {'type': 'wifi-device', 'band': '2g', 'channel': '11',
'htmode': 'HT20', 'country': 'US'},
'radio1': {'type': 'wifi-device', 'band': '5g', 'channel': 'auto',
'htmode': 'VHT80', 'country': 'US'},
'wlan0open': {'device': 'radio0', 'mode': 'ap', 'ssid': 'pager-open',
'disabled': '0', 'hidden': '0', 'encryption': 'none'},
'wlan0wpa': {'device': 'radio0', 'mode': 'ap', 'ssid': 'Service',
'disabled': '0', 'hidden': '0', 'encryption': 'psk2',
'key': 'testpass123'},
}
return dict(table.get(section, {}))
def setUp(self):
server._uci_wifi_iface = lambda name: self._uci(name)
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
404, {'error': 'not found'})
def test_open_falls_back_to_radio_channel(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['open']['channel'], 11)
def test_wpa_falls_back_to_radio_channel(self):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['wpa']['channel'], 11)
class GetApReconcileTest(unittest.TestCase):
def setUp(self):
server._last_reconcile = 0.0
self.uci = {'wlan1wpa': {'device': 'radio1', 'mode': 'ap', 'ssid': 'Corp',
'disabled': '0', 'encryption': 'sae', 'channel': '36'}}
self.runs = []
server._uci_wifi_iface = lambda name: dict(self.uci.get(name, {}))
server._uci_section = lambda name: {}
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
404, {'error': 'not found'})
server.device_run = lambda args, timeout=20, input_data=None: (
self.runs.append(list(args)) or (0, '', ''))
def test_missing_netdev_triggers_wifi_reload(self):
server.os.path.exists = lambda p: False
server.h_pineap_wifi_get_ap(ctx())
self.assertIn(['wifi', 'reload'], self.runs)
def test_present_netdev_skips_reload(self):
server.os.path.exists = lambda p: True
server.h_pineap_wifi_get_ap(ctx())
self.assertNotIn(['wifi', 'reload'], self.runs)
def test_disabled_section_skips_reload(self):
self.uci['wlan1wpa']['disabled'] = '1'
server.os.path.exists = lambda p: False
server.h_pineap_wifi_get_ap(ctx())
self.assertNotIn(['wifi', 'reload'], self.runs)
if __name__ == '__main__':
unittest.main()
+2 -2
View File
@@ -129,7 +129,7 @@ class PineapProxyTest(unittest.TestCase):
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['wpa'], {'ssid': 'Evil1', 'passphrase': 'sekret', 'enctype': 'psk2',
'hidden': False, 'enabled': True})
'hidden': False, 'enabled': True, 'channel': 6})
self.assertEqual(payload['open']['enabled'], False)
self.assertEqual(payload['open']['ssid'], 'pager-open')
self.assertEqual(payload['open']['bssid'], 'DE:AD:BE:EF:00:01')
@@ -150,7 +150,7 @@ class PineapProxyTest(unittest.TestCase):
def fake_run(args):
run_calls.append(args)
if args[0] == 'uci' and args[1] == 'show':
if args[0] == 'uci' and args[1] == 'show' and args[2] == 'wireless.radio0':
return 0, "wireless.radio0.channel='1'\n", ''
return 0, '', ''
+542 -20
View File
@@ -182,9 +182,8 @@ class DaemonSockTest(unittest.TestCase):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
server.h_recon_stop(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
self.assertEqual(calls[1], ('POST', '/api/pineap/log/recon/stop', {}))
self.assertEqual(calls, [
('POST', '/api/pineap/recon/new', {'scan_time': 30})])
def test_start_forwards_scan_time(self):
calls = []
@@ -192,19 +191,36 @@ class DaemonSockTest(unittest.TestCase):
ctx = type('C', (), {'args': (), 'body': {'scan_time': 60}})()
status, data = server.h_recon_start(ctx)
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {'scan_time': 60}))
self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 60}))
def test_start_defaults_empty_body(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 30}))
def test_start_rejects_invalid_scan_time(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body))
ctx = type('C', (), {'args': (), 'body': {'scan_time': 'forever'}})()
status, data = server.h_recon_start(ctx)
self.assertEqual(status, 400)
self.assertIn('scan_time', data['error'])
self.assertEqual(calls, [])
def test_start_reports_native_failure(self):
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
server.daemon_sock_call = lambda m, p, body=None: (500, {'error': 'no radio'})
status, data = server.h_recon_start(
type('C', (), {'args': (), 'body': {'scan_time': 30}})())
self.assertEqual(status, 502)
self.assertEqual(data['error'], 'native recon scan failed')
self.assertEqual(data['detail'], {'error': 'no radio'})
self.assertFalse(server._recon_scan_state['active'])
class ReconScanStateTest(unittest.TestCase):
"""The daemon ignores scan_time and scans continuously until 'stop'. The webui
must track the requested duration itself so timed scans actually end and the
toggle can reflect real scan state."""
"""The webui mirrors the duration of the Pager's native timed scan."""
def setUp(self):
self.db = make_db()
@@ -248,26 +264,33 @@ class ReconScanStateTest(unittest.TestCase):
self.assertFalse(data['scanning'])
self.assertEqual(data['scan_remaining'], 0)
def test_continuous_scan_has_no_remaining(self):
def test_zero_duration_is_rejected(self):
server.time.time = lambda: 1000.0
self._start(scan_time=0)
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
status, data = self._start(scan_time=0)
self.assertEqual(status, 400)
self.assertFalse(server._recon_scan_state['active'])
def test_default_start_is_continuous(self):
def test_default_start_uses_thirty_seconds(self):
server.time.time = lambda: 1000.0
self._start()
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
self.assertEqual(data['scan_remaining'], 30)
def test_stop_clears_scanning(self):
def test_stop_rejects_active_native_scan(self):
server.time.time = lambda: 1000.0
self._start(scan_time=30)
self._stop()
status, data = self._stop()
self.assertEqual(status, 409)
self.assertIn('finish automatically', data['error'])
self.assertEqual(data['scan_remaining'], 30)
status, data = self._status()
self.assertFalse(data['scanning'])
self.assertTrue(data['scanning'])
def test_stop_is_idempotent_when_inactive(self):
status, data = self._stop()
self.assertEqual(status, 200)
self.assertEqual(data, {'ok': True})
def test_start_failure_does_not_mark_scanning(self):
server.time.time = lambda: 1000.0
@@ -277,13 +300,13 @@ class ReconScanStateTest(unittest.TestCase):
self.assertFalse(data['scanning'])
def test_watchdog_stops_expired_timed_scan(self):
calls = []
server.time.time = lambda: 1000.0
self._start(scan_time=10)
server.time.time = lambda: 1012.0
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True})
server._recon_watchdog_tick()
self.assertEqual(calls, [('POST', '/api/pineap/log/recon/stop')])
self.assertEqual(calls, [])
self.assertFalse(server._recon_scan_state['active'])
def test_watchdog_leaves_active_scan_alone(self):
@@ -454,6 +477,21 @@ class CliFallbackTest(unittest.TestCase):
server.RECON_DB = '/nonexistent.db'
self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_'), 'casaalicia\x00.\ufffd_')
def test_completed_recon_lock_uses_immutable_read(self):
calls = []
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
def locked_then_read(args, timeout=20):
calls.append(args)
if len(calls) == 1:
return 5, '', 'Error: database is locked'
return 0, '[{"id": 2}]', ''
server.device_run = locked_then_read
rows = server._db_rows(self.db, 'SELECT MAX(id) AS id FROM scan')
self.assertEqual(rows, [{'id': 2}])
self.assertEqual(calls[1][-2], 'file:%s?immutable=1' % self.db)
def make_hs_db():
db = make_db()
@@ -597,3 +635,487 @@ class HandshakeRoutesTest(unittest.TestCase):
self.assertEqual(data['files'], [])
self.assertEqual(data['handshakes'], [])
self.assertEqual(os.listdir(self.dir), ['.hidden'])
def make_survey_db():
"""Single-scan recon DB so the newest scan carries the AP data."""
fd, db = tempfile.mkstemp(suffix='.db')
os.close(fd)
conn = sqlite3.connect(db)
conn.executescript(SCHEMA)
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u1', 1786466531, 'pager')")
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (1, 1, 'AE77C0EB3141', 1786466531, -71, 2412, 5)")
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (2, 1, 'C89E43648080', 1786466532, -76, 5745, 9)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (10, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466532, -76, 5745, 149, 0x400400108)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (11, 2, 1, 8, '506F9A010000', X'', 1, 1786466532, -64, 5745, 149, 0)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (12, 1, 1, 4, NULL, X'5A6E6574', NULL, 1786466531, -40, 2412, NULL, NULL)")
conn.execute("INSERT INTO handshake (hash, scan, stahash, aphash, time) VALUES (20, 1, 1, 2, 1786466600)")
conn.commit()
conn.close()
return db
class OuiVendorTest(unittest.TestCase):
def test_oui_prefix_forms(self):
self.assertEqual(server._oui_prefix('C8:9E:43:64:80:80'), 'C89E43')
self.assertEqual(server._oui_prefix('C89E43648080'), 'C89E43')
self.assertEqual(server._oui_prefix('c8:9e:43:64:80:80'), 'C89E43')
self.assertIsNone(server._oui_prefix(None))
self.assertIsNone(server._oui_prefix(''))
self.assertIsNone(server._oui_prefix('XX:YY:ZZ:00:00:00'))
def test_oui_vendor_lookup(self):
self.assertEqual(server.oui_vendor('B8:27:EB:00:00:00'), 'Raspberry Pi')
self.assertEqual(server.oui_vendor('10:BF:48:00:00:00'), 'Apple')
self.assertEqual(server.oui_vendor('14:CC:20:00:00:00'), 'TP-Link')
self.assertEqual(server.oui_vendor('FC:63:3E:00:00:00'), 'Google')
def test_oui_vendor_unknown_and_local(self):
self.assertEqual(server.oui_vendor('C8:9E:43:64:80:80'), 'Unknown')
self.assertEqual(server.oui_vendor('AE:77:C0:EB:31:41'), 'Local')
self.assertEqual(server.oui_vendor(None), 'Unknown')
self.assertEqual(server.oui_vendor('--'), 'Unknown')
def test_band_of_frequencies(self):
self.assertEqual(server.band_of(2412), '2.4')
self.assertEqual(server.band_of(5200), '5')
self.assertEqual(server.band_of(6180), '6')
self.assertEqual(server.band_of(0), '--')
self.assertEqual(server.band_of(None), '--')
def test_curated_table_has_no_garbage_keys(self):
for key in server.OUI_VENDORS:
self.assertRegex(key, r'^[0-9A-F]{6}$')
self.assertNotIn('349A...', server.OUI_VENDORS)
class ReconEnrichmentTest(unittest.TestCase):
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
def tearDown(self):
os.unlink(self.db)
def test_scan_detail_enriches_aps(self):
data = server.recon_scan_data(1)
aps = {a['bssid']: a for a in data['aps']}
a = aps['C8:9E:43:64:80:80']
self.assertEqual(a['band'], '5')
self.assertEqual(a['vendor'], 'Unknown')
self.assertEqual(a['first_seen'], 1786466532)
self.assertEqual(a['last_seen'], 1786466532)
hidden = aps['50:6F:9A:01:00:00']
self.assertEqual(hidden['band'], '5')
self.assertEqual(hidden['vendor'], 'Unknown')
def test_scan_detail_unassociated_count(self):
data = server.recon_scan_data(1)
self.assertEqual(data['unassociated'], 1)
def test_scan_detail_bounded_mode_counts_unassociated(self):
data = server.recon_scan_data(1, _limit=1)
self.assertEqual(data['unassociated'], 1)
self.assertEqual(len(data['aps']), 2)
self.assertLessEqual(len(data['clients']), 1)
self.assertEqual(data['scan']['id'], 1)
def test_first_last_seen_span_multiple_rows(self):
conn = sqlite3.connect(self.db)
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (30, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466540, -80, 5745, 149, 0x400400108)")
conn.commit()
conn.close()
data = server.recon_scan_data(1)
a = [a for a in data['aps'] if a['bssid'] == 'C8:9E:43:64:80:80'][0]
self.assertEqual(a['first_seen'], 1786466532)
self.assertEqual(a['last_seen'], 1786466540)
def test_band_for_24ghz_ap(self):
conn = sqlite3.connect(self.db)
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (3, 1, 'FC633E000001', 1786466533, -60, 2412, 4)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (31, 3, 1, 8, 'FC633E000001', X'4E6574776F726B', 0, 1786466533, -60, 2412, 6, 0x08)")
conn.commit()
conn.close()
data = server.recon_scan_data(1)
a = [a for a in data['aps'] if a['bssid'] == 'FC:63:3E:00:00:01'][0]
self.assertEqual(a['band'], '2.4')
self.assertEqual(a['vendor'], 'Google')
class ReconReportTest(unittest.TestCase):
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
def tearDown(self):
os.unlink(self.db)
def _ctx(self, args=()):
return type('C', (), {'args': args, 'body': {}})()
def test_csv_download_contains_aps_and_unassociated(self):
status, payload = server.h_recon_scan_download_csv(self._ctx(('1',)))
self.assertEqual(status, 200)
self.assertEqual(payload.ctype, 'text/csv')
self.assertEqual(payload.filename, 'scan-1.csv')
text = payload.data.decode('utf-8')
self.assertIn('Anderson-5', text)
self.assertIn('unassociated,1', text)
self.assertIn('C8:9E:43:64:80:80', text)
def test_html_download_contains_stats(self):
status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
self.assertEqual(status, 200)
self.assertEqual(payload.ctype, 'text/html')
self.assertEqual(payload.filename, 'scan-1.html')
text = payload.data.decode('utf-8')
self.assertIn('Scan #1', text)
self.assertIn('Anderson-5', text)
self.assertIn('WPA3 WPA2', text)
self.assertIn('Unassociated', text)
def test_download_404_for_missing_scan(self):
status, payload = server.h_recon_scan_download_csv(self._ctx(('999',)))
self.assertEqual(status, 404)
status, payload = server.h_recon_scan_download_html(self._ctx(('999',)))
self.assertEqual(status, 404)
def test_download_503_when_db_unavailable(self):
with mock.patch.object(server, 'recon_scan_data',
side_effect=RuntimeError('sqlite read failed: locked')), \
mock.patch.object(server.time, 'sleep'):
status, payload = server.h_recon_scan_download_csv(self._ctx(('1',)))
self.assertEqual(status, 503)
status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
self.assertEqual(status, 503)
class GpsTest(unittest.TestCase):
def setUp(self):
server._gps_cache.update({'updated': 0, 'data': None})
@unittest.skipIf(os.name == 'nt', 'symlinks are not reliably available on Windows')
def test_serial_candidates_detect_bypath_targets(self):
d = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, d)
self.addCleanup(setattr, server, 'SERIAL_DIR', server.SERIAL_DIR)
server.SERIAL_DIR = d
os.symlink('/dev/ttyACM0', os.path.join(d, '1.3_1-1.3:1.0'))
os.symlink('/dev/ttyACM1', os.path.join(d, '1.3_1-1.3:1.2'))
with open(os.path.join(d, 'not-a-serial'), 'w') as f:
f.write('x')
candidates = server._gps_serial_candidates()
names = [name for name, _ in candidates]
self.assertEqual(names, ['1.3_1-1.3:1.0', '1.3_1-1.3:1.2'])
def test_gps_status_passthrough(self):
with mock.patch.object(server, '_gps_status_data_nocache',
return_value={'present': True, 'wigle': True}):
status, data = server.h_recon_gps(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertTrue(data['present'])
self.assertTrue(data['wigle'])
def test_configure_binds_preferred_device_and_locks(self):
candidates = [('1.2_1-1.2:1.0', '/dev/1.2'), ('1.3_2-1.3:1.0', '/dev/1.3')]
with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \
mock.patch.object(server, '_uci_gps_get', return_value='1.3_2-1.3:1.0'), \
mock.patch.object(server, '_uci_gps_set') as uci_set, \
mock.patch.object(server, '_gpsd_restart'), \
mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \
mock.patch.object(server, '_gps_from_gpspipe',
return_value={'fix': 3, 'lat': 37.7, 'lon': -122.4, 'satellites': 8}), \
mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}):
status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertTrue(data['lock'])
self.assertEqual(data['tried'], ['1.3_2-1.3:1.0'])
uci_set.assert_called_once_with('1.3_2-1.3:1.0')
def test_configure_no_candidates_errors(self):
with mock.patch.object(server, '_gps_serial_candidates', return_value=[]):
status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertIn('error', data)
def test_configure_fallback_binds_first_with_note(self):
candidates = [('1.2_1-1.2:1.0', '/dev/1.2'), ('1.3_2-1.3:1.0', '/dev/1.3')]
with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \
mock.patch.object(server, '_uci_gps_get', return_value=None), \
mock.patch.object(server, '_uci_gps_set'), \
mock.patch.object(server, '_gpsd_restart'), \
mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \
mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \
mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}):
status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['device'], '1.2_1-1.2:1.0')
self.assertIn('waiting for a fix', data['note'])
def test_configure_tries_at_most_three_candidates(self):
candidates = [(str(i), '/dev/%d' % i) for i in range(5)]
with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \
mock.patch.object(server, '_uci_gps_get', return_value=None), \
mock.patch.object(server, '_uci_gps_set'), \
mock.patch.object(server, '_gpsd_restart'), \
mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \
mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \
mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}):
status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['tried'], ['0', '1', '2'])
class WigleTest(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
self._orig = server.WIGLE_DIR
server.WIGLE_DIR = self.dir
def tearDown(self):
server.WIGLE_DIR = self._orig
shutil.rmtree(self.dir)
def _ctx(self, args=(), body=None):
return type('C', (), {'args': args, 'body': body or {}})()
def _write(self, name, content):
with open(os.path.join(self.dir, name), 'w') as f:
f.write(content)
def test_file_rows_count_excludes_header(self):
self._write('a.csv', 'header\nr1\nr2\n')
self._write('b.csv', 'onlyheader\n')
status, data = server.h_recon_wigle_files(self._ctx())
self.assertEqual(status, 200)
files = {f['name']: f for f in data['files']}
self.assertEqual(files['a.csv']['rows'], 2)
self.assertEqual(files['b.csv']['rows'], 0)
self.assertEqual(files['a.csv']['size'], len('header\nr1\nr2\n'))
def test_file_rows_count_ignores_wigle_meta_and_header(self):
meta = 'WigleWifi-1.6,appRelease=0.0.0,model=pineapplepager,release=0.0.0\n'
header = 'MAC,SSID,AuthMode,FirstSeen,Channel,Frequency,RSSI,CurrentLatitude,CurrentLongitude\n'
self._write('empty.csv', meta + header)
self._write('full.csv', meta + header + 'AA:BB:CC:DD:EE:FF,test,0,,1,2412,-60,37.7,-122.4\n')
status, data = server.h_recon_wigle_files(self._ctx())
files = {f['name']: f for f in data['files']}
self.assertEqual(files['empty.csv']['rows'], 0)
self.assertEqual(files['full.csv']['rows'], 1)
def test_file_download(self):
self._write('wigle-1.csv', 'lat,lon\n37.7,-122.4\n')
status, payload = server.h_recon_wigle_file(self._ctx(('wigle-1.csv',)))
self.assertEqual(status, 200)
self.assertEqual(payload.filename, 'wigle-1.csv')
self.assertIn(b'37.7', payload.data)
def test_file_download_404_and_traversal(self):
status, payload = server.h_recon_wigle_file(self._ctx(('missing.csv',)))
self.assertEqual(status, 404)
status, payload = server.h_recon_wigle_file(self._ctx(('..%2F..%2Fetc%2Fpasswd',)))
self.assertEqual(status, 404)
def test_toggle_enable_and_disable(self):
with mock.patch.object(server, '_wigle_set', return_value=(200, {'ok': True})), \
mock.patch.object(server, 'hak5') as hak5, \
mock.patch.object(server, 'wigle_files_data',
return_value={'files': [{'name': 'w.csv'}]}):
status, data = server.h_recon_wigle(self._ctx(body={'enable': True}))
self.assertEqual(status, 200)
self.assertTrue(data['wigle'])
self.assertEqual(data['filename'], 'w.csv')
hak5.assert_called_once_with('WIGLE_START', timeout=10)
status, data = server.h_recon_wigle(self._ctx(body={'enable': False}))
self.assertEqual(status, 200)
self.assertFalse(data['wigle'])
hak5.assert_called_with('WIGLE_STOP', timeout=10)
class SurveyTest(unittest.TestCase):
def setUp(self):
self.db = make_survey_db()
server.RECON_DB = self.db
self.dir = tempfile.mkdtemp()
self._orig = {
'SURVEY_DIR': server.SURVEY_DIR,
'SURVEY_MAX_SAMPLES': server.SURVEY_MAX_SAMPLES,
'SURVEY_SAMPLE_INTERVAL': server.SURVEY_SAMPLE_INTERVAL,
}
server.SURVEY_DIR = self.dir
server.SURVEY_MAX_SAMPLES = 3
server.SURVEY_SAMPLE_INTERVAL = 0.0
server._survey_state = {'active': False, 'id': None, 'name': None, 'path': None,
'started': 0, 'samples': 0, 'last_sample': 0}
server._gps_cache.update({'updated': 0, 'data': None})
def tearDown(self):
server.SURVEY_DIR = self._orig['SURVEY_DIR']
server.SURVEY_MAX_SAMPLES = self._orig['SURVEY_MAX_SAMPLES']
server.SURVEY_SAMPLE_INTERVAL = self._orig['SURVEY_SAMPLE_INTERVAL']
server._survey_state = {'active': False, 'id': None, 'name': None, 'path': None,
'started': 0, 'samples': 0, 'last_sample': 0}
server._gps_cache.update({'updated': 0, 'data': None})
shutil.rmtree(self.dir)
os.unlink(self.db)
def _ctx(self, args=(), body=None):
return type('C', (), {'args': args, 'body': body or {}})()
def test_start_creates_meta_file(self):
status, data = server.h_recon_survey_start(self._ctx(body={'name': 'Kitchen Walk'}))
self.assertEqual(status, 200)
self.assertTrue(data['ok'])
path = os.path.join(self.dir, data['id'] + '.jsonl')
self.assertTrue(os.path.isfile(path))
with open(path) as f:
first = f.readline()
self.assertIn('"meta"', first)
self.assertIn('Kitchen Walk', first)
def test_start_rejects_duplicate(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
status, data = server.h_recon_survey_start(self._ctx(body={'name': 'B'}))
self.assertEqual(status, 409)
def test_sample_via_watchdog_and_cap(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
server._recon_watchdog_tick()
server._recon_watchdog_tick()
self.assertEqual(server._survey_state['samples'], 2)
status, data = server.h_recon_survey_stop(self._ctx())
self.assertEqual(status, 200)
self.assertEqual(data['samples'], 2)
self.assertFalse(server._survey_state['active'])
def test_sample_cap_stops_recording(self):
server.SURVEY_MAX_SAMPLES = 2
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
for _ in range(4):
server._recon_watchdog_tick()
self.assertFalse(server._survey_state['active'])
self.assertEqual(server._survey_state['samples'], 2)
def test_live_reports_scan_unassociated_and_recording(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
server._recon_watchdog_tick()
status, data = server.h_recon_survey_live(self._ctx())
self.assertEqual(status, 200)
self.assertEqual(data['scan']['id'], 1)
self.assertEqual(len(data['aps']), 2)
self.assertEqual(data['unassociated'], 1)
self.assertTrue(data['recording']['active'])
self.assertEqual(data['recording']['samples'], 1)
self.assertIn('wigle', data['gps'])
def test_live_recording_none_when_stopped(self):
status, data = server.h_recon_survey_live(self._ctx())
self.assertEqual(status, 200)
self.assertIsNone(data['recording'])
def test_surveys_list_counts_samples(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
server._recon_watchdog_tick()
server.h_recon_survey_stop(self._ctx())
status, data = server.h_recon_surveys(self._ctx())
self.assertEqual(status, 200)
self.assertEqual(len(data['surveys']), 1)
survey = data['surveys'][0]
self.assertEqual(survey['name'], 'A')
self.assertEqual(survey['samples'], 1)
self.assertGreater(survey['size'], 0)
def test_detail_aggregates_signal(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
server._recon_watchdog_tick()
server._recon_watchdog_tick()
sid = server._survey_state['id']
server.h_recon_survey_stop(self._ctx())
status, data = server.h_recon_survey_detail(self._ctx((sid,)))
self.assertEqual(status, 200)
aps = {a['bssid']: a for a in data['aps']}
a = aps['C8:9E:43:64:80:80']
self.assertEqual(a['min'], -76)
self.assertEqual(a['max'], -76)
self.assertEqual(a['avg'], -76)
self.assertEqual(a['samples'], 2)
self.assertEqual(a['band'], '5')
self.assertEqual(a['channel'], 149)
self.assertEqual(a['first_seen'], a['last_seen'])
self.assertEqual(data['gps_fixes'], 0)
def test_downloads_all_formats(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
server._recon_watchdog_tick()
sid = server._survey_state['id']
server.h_recon_survey_stop(self._ctx())
for fmt, ctype in [('json', 'application/json'), ('csv', 'text/csv'), ('html', 'text/html')]:
status, payload = server.h_recon_survey_download(self._ctx((sid, fmt)))
self.assertEqual(status, 200)
self.assertEqual(payload.ctype, ctype)
self.assertEqual(payload.filename, 'survey-%s.%s' % (sid, fmt))
status, payload = server.h_recon_survey_download(self._ctx((sid, 'csv')))
self.assertIn('C8:9E:43:64:80:80', payload.data.decode('utf-8'))
def test_delete_removes_file_then_404(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
sid = server._survey_state['id']
server.h_recon_survey_stop(self._ctx())
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
self.assertEqual(status, 200)
self.assertFalse(os.listdir(self.dir))
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
self.assertEqual(status, 404)
def test_delete_blocks_active_survey(self):
server.h_recon_survey_start(self._ctx(body={'name': 'A'}))
sid = server._survey_state['id']
status, data = server.h_recon_survey_delete(self._ctx((sid,)))
self.assertEqual(status, 409)
def test_detail_404_missing(self):
status, data = server.h_recon_survey_detail(self._ctx(('nope',)))
self.assertEqual(status, 404)
class ReconRoutesTest(unittest.TestCase):
def test_new_routes_registered(self):
expected = [
('GET', '/api/recon/scans/1/download/csv', 'h_recon_scan_download_csv'),
('GET', '/api/recon/scans/1/download/html', 'h_recon_scan_download_html'),
('GET', '/api/recon/gps', 'h_recon_gps'),
('POST', '/api/recon/gps/configure', 'h_recon_gps_configure'),
('POST', '/api/recon/wigle', 'h_recon_wigle'),
('GET', '/api/recon/wigle/files', 'h_recon_wigle_files'),
('GET', '/api/recon/wigle/files/x.csv', 'h_recon_wigle_file'),
('GET', '/api/recon/survey/live', 'h_recon_survey_live'),
('POST', '/api/recon/survey/start', 'h_recon_survey_start'),
('POST', '/api/recon/survey/stop', 'h_recon_survey_stop'),
('GET', '/api/recon/surveys', 'h_recon_surveys'),
('GET', '/api/recon/surveys/20260818-120000-A', 'h_recon_survey_detail'),
('GET', '/api/recon/surveys/20260818-120000-A/download/csv', 'h_recon_survey_download'),
('GET', '/api/recon/surveys/20260818-120000-A/download/json', 'h_recon_survey_download'),
('GET', '/api/recon/surveys/20260818-120000-A/download/html', 'h_recon_survey_download'),
('DELETE', '/api/recon/surveys/20260818-120000-A', 'h_recon_survey_delete'),
]
for method, path, handler in expected:
h, args = server.ROUTER.dispatch(method, path)
self.assertIsNotNone(h, '%s %s' % (method, path))
self.assertEqual(h.__name__, handler, '%s %s' % (method, path))
def test_survey_download_route_captures_format(self):
h, args = server.ROUTER.dispatch('GET', '/api/recon/surveys/abc/download/csv')
self.assertEqual(args, ('abc', 'csv'))
def test_original_recon_routes_unchanged(self):
for path in ['/api/recon/start', '/api/recon/status', '/api/recon/scans',
'/api/recon/events']:
method = 'GET' if path.endswith(('status', 'scans', 'events')) else 'POST'
h, args = server.ROUTER.dispatch(method, path)
self.assertIsNotNone(h, path)