Compare commits
17
Commits
v1.0
...
251b1f6261
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
251b1f6261 | ||
|
|
f5cddb335a | ||
|
|
af5ff8039b | ||
|
|
c0b9e58aa8 | ||
|
|
12279fe29d | ||
|
|
7536f3ca45 | ||
|
|
be2685aa15 | ||
|
|
ff3bd16855 | ||
|
|
4fa7f8f855 | ||
|
|
cd39833f4b | ||
|
|
4effc08a25 | ||
|
|
b2098bae7d | ||
|
|
7aff767f2a | ||
|
|
27df97ec43 | ||
|
|
acbf4c0ce9 | ||
|
|
5f6dc5bcdb | ||
|
|
2bf39ecb9d |
@@ -3,4 +3,5 @@ __pycache__/
|
||||
*.pyc
|
||||
|
||||
.worktrees/
|
||||
.openchamber/
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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'` | `1–14` | `HT20` | `radio0` |
|
||||
| 5GHz | `BAND_5G = '5'` | `36–177` | `VHT80` | `radio1` |
|
||||
| 6GHz | `BAND_6G = '6'` | `181–233` (step 4) | `HE80` | `radio1` |
|
||||
|
||||
- `channel_band(ch)` classifies **2.4GHz (1–14) first, then 5GHz (36–177)**, 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 + (ch−1)·5`; 5 → `5180 + (ch−36)·5`;
|
||||
6 → `5955 + (ch−1)·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 (1–14):** 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 (36–177) / 6GHz (181–233):** `_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 (1–11), a `5 GHz` optgroup (36–165, 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
Executable
+186
@@ -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"
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
@@ -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
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user