Add optional multi-agent controller enhancements (push ingest, Elastic, UI fixes)

.gitignore

@@ -11,3 +11,6 @@ cur/
 tmp/
 __pycache__/
 *.DS_Store
+controller/.venv/
+controller/state/controller.db
+controller.db

controller/README.md

@@ -0,0 +1,107 @@
+# Sparrow Multi-Agent Controller (Proof of Concept)
+
+This directory contains an HTTP controller that aggregates multiple `sparrowwifiagent` instances, proxies scan commands, and stores their responses in SQLite.  The intent is to keep all remote agents untouched and provide a single place to trigger Wi-Fi/Falcon/Bluetooth scans and visualize their results via a simple web UI.
+
+## Features
+- Register remote agents (hostname/IP, port, descriptive metadata, reported capabilities)
+- Trigger scans (basic Wi-Fi, Falcon advanced, Bluetooth discovery) against one or more agents
+- Persist raw responses in SQLite for table/map rendering and historical queries
+- Provide REST endpoints plus a WebSocket stream for the JavaScript front end
+- Leaflet-based map that aggregates Wi-Fi/Falcon observations (plus Bluetooth devices) from every agent in real time
+- Falcon monitor-mode controls (start/stop monitor mode, start/stop scans, view live status/polling indicator)
+- Optional continuous scans per agent/interface with controller-managed scheduling
+- Bluetooth results tab with live map markers plus per-agent Falcon panels for networks/clients and inline actions (deauth/capture)
+- Spectrum/HackRF modal that can launch 2.4 GHz/5 GHz sweeps or snapshots from the browser
+- Leave hooks that can later forward the aggregated data into Elastic
+- Optional ingest endpoint so agents can push Wi‑Fi/Falcon/Bluetooth results directly into the controller (pull remains available)
+
+## Running the controller
+```bash
+cd controller
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+uvicorn app.main:app --reload
+```
+
+The service stores its SQLite database in `controller/state/controller.db` by default.  Set `CONTROLLER_DB_URL` to override (e.g., point to PostgreSQL) and `CONTROLLER_ELASTIC_URL` once you are ready to forward events to Elastic.
+
+## Development notes
+- The API is documented through the built-in FastAPI schema.  Once uvicorn is running, browse to `http://localhost:8000/docs`.
+- The placeholder UI lives in `controller/frontend` and is served statically from `/`.
+- Future exporter hooks can subscribe to the internal event bus located in `app/events.py`.
+
+## Architecture overview
+
+The controller is a fairly small FastAPI service.  Each registered agent row stores:
+
+- Base URL + API key (if supplied) used by `AgentClient`
+- Cached interface metadata (`agent.interfaces`) and monitor map (`agent.monitor_map`)
+- Reported capabilities (wifi/falcon/bluetooth), GPS snapshot, and free-form description
+
+Every 30 seconds the UI refreshes `/api/agents` to keep the interface list, monitor map, and capability badges in sync with what agents report over `/wireless/interfaces`.  The WebSocket `/ws/scans` is used for incremental scan updates so the table/map refresh without a full page reload.
+
+## Falcon workflow
+
+1. Choose an agent in the “Selected Agent” dropdown.  The detail drawer loads `/api/agents/<id>`, `/status`, and (if a Falcon scan is running) `/falcon/<id>/scan/results` so the tab shows the last snapshot without rerunning a scan.
+2. Enter the managed interface (e.g., `wlan0`) and click **Start Monitor**.  The controller forwards the call to `/falcon/startmonmode/<iface>` and caches the returned alias (typically `wlan0mon`).  The dropdowns are updated immediately and a warning prevents stopping monitor mode while Falcon scans are still running.
+3. Select the monitor alias in the **Monitor Interface** dropdown and click **Start Scan**.  The UI kicks off `/falcon/<id>/scan/start` and enters auto-poll mode: results refresh every 5 seconds while the scan is running.  A “Auto-refreshing results…” badge appears at the top of the Falcon panel so it is obvious when polling is active.
+4. Click **Stop Scan** to call `/falcon/<id>/scan/stop`.  Polling ceases, the buttons re-enable, and the monitor stop button becomes available again.  The controller also re-fetches `/api/agents` to keep the monitor map synchronized.
+
+The Falcon log (console in the bottom of the tab) records every REST call the UI makes to simplify troubleshooting.  Each entry includes the agent ID, interface, and raw JSON response from the agent so discrepancies can be diagnosed quickly.
+
+## Spectrum / HackRF controls
+
+Open the **Spectrum** modal (top right button) to manage HackRF sweeps remotely:
+
+1. Pick an agent that has a HackRF connected from the dropdown.
+2. Click **Start 2.4 GHz** or **Start 5 GHz** to launch a continuous sweep via `/spectrum/<id>/start?band=24|5`.  The controller now hits the legacy agent endpoints (`/spectrum/scanstart24`, `/scanstart5`, `/scanstop`, `/scanstatus`) so both Ubertooth and HackRF agents behave as expected.
+3. While a sweep is running the chart polls `/spectrum/<id>/channels` every 2 seconds and draws the reported power buckets.  Use **Stop Scan** to halt the sweep, or **Snapshot** buttons to capture a single sweep without leaving a long-running process on the agent.
+
+The X-axis intentionally mirrors whatever the agent returns (which can be raw Hz buckets or Wi‑Fi channels) so you can correlate with the textual data returned by `/spectrum/scanstatus`.  For finer resolution adjust `self.binWidth`, `minFreq`, and `maxFreq` in `sparrowhackrf.py` at the agent.
+
+Note: The spectrum modal is currently geared toward HackRF responses. Ubertooth 2.4 GHz specan data is not rendered in the chart yet even though the legacy endpoints are called.
+
+## Screenshots
+
+![Dashboard](images/sparrow-dashboard.png)
+
+![Falcon](images/sparrow-falcon.png)
+
+![Spectrum 5 GHz](images/sparrow-spectrum-5g.png)
+
+## Documentation for Falcon controls
+
+- Use the **Falcon Monitor & Scan Control** section to toggle monitor mode per agent and start/stop dedicated Falcon scans.
+- Monitor-mode commands simply forward to the agent's `/falcon/startmonmode` and `/falcon/stopmonmode` endpoints; supply the managed interface (e.g., `wlan0`) when entering monitor mode and the resulting monitor interface (typically `wlan0mon`) when launching scans.
+- The **Refresh Status** button queries `/falcon/scanrunning/<iface>` for the selected agent and interface and appends the response to the Falcon log pane.
+- Polling only occurs when a scan is confirmed to be running; deleting an agent or receiving a 404 automatically tears down the poller to avoid hammering offline devices.
+
+## Outstanding issues / TODO
+
+1. **Monitor map persistence:** The “Monitor Map” JSON panel in the detail drawer still reflects whatever aliases the agent last reported.  If an agent fails to remove `wlan0mon` from its `/wireless/interfaces` payload we will show a stale alias after refresh.  We should teach the controller to prune aliases whose monitor interface disappears.
+2. **Falcon alias tracking with multiple interfaces:** We currently track one `monitor_map` per managed interface, but richer UI cues would help when an agent has multiple radios.  Consider showing both managed and alias names in the dropdown so operators can see which pairings exist.
+3. **Spectrum axis formatting:** The X-axis now mirrors raw agent keys; we should revisit a GHz-normalized plot once we have consistent frequency values (possibly by normalizing on the agent before sending channel data).
+4. **UI hint for lingering Falcon results:** When a Falcon scan is stopped the tab continues to show the last snapshot (by design).  Adding a timestamp or badge noting “Last update at …” would make it obvious when the data is stale.
+5. **Agent deletion safety:** Deleting an agent with continuous scans still requires manual cleanup on the agent side.  Long term we should teach the controller to call `/scans/continuous/stop` whenever an agent row disappears.
+
+## Mock agents for testing
+You can spin up mock agents that simulate Wi-Fi, Falcon, and Bluetooth data so the controller can be exercised without any radios.
+
+```
+cd controller
+source .venv/bin/activate
+python tests/mock_agent.py --name mock-east --port 9001 --lat 40.7128 --lon -74.0060
+# In another terminal:
+python tests/mock_agent.py --name mock-west --port 9002 --lat 34.0522 --lon -118.2437
+```
+
+With the controller running, register each mock agent via the UI (URL `http://localhost:9001`, etc.) or run the automated workflow script:
+
+```
+python tests/demo_workflow.py --controller http://localhost:8000 \\
+    --agent mock-east http://localhost:9001 \\
+    --agent mock-west http://localhost:9002
+```
+
+The script registers each agent, toggles monitor mode on `wlan0`, and launches both Wi-Fi and Falcon scans so that the dashboard receives live map data and WebSocket events.

controller/app/agent_client.py

@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+import time
+import gzip
+import json
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from .models import Agent, ScanType
+
+
+class AgentHTTPError(RuntimeError):
+    pass
+
+
+class AgentClient:
+    def __init__(self, agent: Agent):
+        self.agent = agent
+        self.base_url = agent.base_url.rstrip('/')
+        self.headers = {}
+        if agent.api_key:
+            self.headers["X-API-Key"] = agent.api_key
+
+    def _get_response(self, path: str, **kwargs) -> httpx.Response:
+        url = f"{self.base_url}{path}"
+        timeout = kwargs.pop("timeout", 60.0)
+        try:
+            response = httpx.get(url, headers=self.headers, timeout=timeout, **kwargs)
+        except httpx.RequestError as exc:
+            raise AgentHTTPError(f"Unable to reach {self.agent.name} at {url}: {exc}") from exc
+        if response.status_code >= 400:
+            raise AgentHTTPError(f"Agent {self.agent.name} returned {response.status_code}: {response.text}")
+        return response
+
+    def _get(self, path: str, **kwargs) -> Dict[str, Any]:
+        response = self._get_response(path, **kwargs)
+        return response.json()
+
+    def _post(self, path: str, data: Dict[str, Any] | None = None, **kwargs) -> Dict[str, Any]:
+        url = f"{self.base_url}{path}"
+        try:
+            response = httpx.post(url, json=data or {}, headers=self.headers, timeout=kwargs.get("timeout", 60.0))
+        except httpx.RequestError as exc:
+            raise AgentHTTPError(f"Unable to reach {self.agent.name} at {url}: {exc}") from exc
+        if response.status_code >= 400:
+            raise AgentHTTPError(f"Agent {self.agent.name} returned {response.status_code}: {response.text}")
+        return response.json()
+
+    def _normalize_interfaces(self, payload: Any) -> Dict[str, Any]:
+        if isinstance(payload, dict):
+            interfaces = payload.get('interfaces')
+            if isinstance(interfaces, dict):
+                return {'interfaces': interfaces}
+            if isinstance(interfaces, list):
+                return {'interfaces': {name: {} for name in interfaces if isinstance(name, str)}}
+        elif isinstance(payload, list):
+            return {'interfaces': {name: {} for name in payload if isinstance(name, str)}}    
+        return {'interfaces': {}}
+
+    def get_interfaces(self) -> Dict[str, Any]:
+        payload = self._get('/wireless/interfaces')
+        return self._normalize_interfaces(payload)
+
+    def wifi_scan(self, interface: str, channels: Optional[List[int]] = None, progress_cb=None) -> Dict[str, Any]:
+        if progress_cb:
+            progress_cb({'stage': 'running', 'message': f'Scanning Wi-Fi on {interface}'})
+        chan_str = ''
+        if channels:
+            chan_str = '?Frequencies=' + ','.join(str(ch) for ch in channels)
+        path = f"/wireless/networks/{interface}{chan_str}"
+        result = self._get(path)
+        if progress_cb:
+            progress_cb({'stage': 'collected', 'networks': len(result.get('networks', []))})
+        return result
+
+    def falcon_scan(
+        self,
+        interface: str,
+        channels: Optional[List[int]] = None,
+        progress_cb=None,
+        poll_interval: float = 2.0,
+        timeout: float = 90.0,
+    ) -> Dict[str, Any]:
+        # Start capture if not already running
+        self._get(f"/falcon/startscan/{interface}")
+        if channels:
+            # The Falcon agent determines channels via config; channel hints are stored for UI only
+            pass
+        start_time = time.time()
+        result = None
+        while True:
+            running_resp = self._get(f"/falcon/scanrunning/{interface}")
+            is_running = running_resp.get('errcode') == 0
+            snapshot = self._get('/falcon/getscanresults')
+            result = snapshot
+            if progress_cb:
+                progress_cb(
+                    {
+                        'stage': 'running' if is_running else 'finalizing',
+                        'running': is_running,
+                        'status': running_resp,
+                        'snapshot': snapshot,
+                    }
+                )
+            if not is_running:
+                break
+            if (time.time() - start_time) > timeout:
+                raise TimeoutError("Falcon scan timed out")
+            time.sleep(poll_interval)
+
+        return result
+
+    def bluetooth_discovery(self, active: bool = True, duration: float = 5.0, progress_cb=None) -> Dict[str, Any]:
+        if progress_cb:
+            progress_cb({'stage': 'running', 'message': f"Bluetooth discovery ({'active' if active else 'passive'})"})
+        if active:
+            self._get('/bluetooth/discoverystarta')
+        else:
+            self._get('/bluetooth/discoverystartp')
+        time.sleep(duration)
+        status = self._get('/bluetooth/discoverystatus')
+        if progress_cb:
+            progress_cb({'stage': 'collected', 'devices': len(status.get('devices', [])) if isinstance(status, dict) else None})
+        return status
+
+    def bluetooth_clear(self) -> Dict[str, Any]:
+        return self._get('/bluetooth/discoveryclear')
+
+    def bluetooth_stop(self) -> Dict[str, Any]:
+        return self._get('/bluetooth/discoverystop')
+
+    def bluetooth_running(self) -> Dict[str, Any]:
+        return self._get('/bluetooth/running')
+
+    def falcon_start_monitor(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/startmonmode/{interface}')
+
+    def falcon_stop_monitor(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/stopmonmode/{interface}')
+
+    def falcon_scan_running(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/scanrunning/{interface}')
+
+    def falcon_start_scan(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/startscan/{interface}')
+
+    def falcon_stop_scan(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/stopscan/{interface}')
+
+    def falcon_get_results(self) -> Dict[str, Any]:
+        return self._get('/falcon/getscanresults')
+
+    def falcon_deauth(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return self._post('/falcon/deauth', payload)
+
+    def falcon_stop_deauth(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return self._post('/falcon/stopdeauth', payload)
+
+    def falcon_stop_all_deauths(self, interface: str) -> Dict[str, Any]:
+        return self._get(f'/falcon/stopalldeauths/{interface}')
+
+    def falcon_get_deauths(self) -> Dict[str, Any]:
+        return self._get('/falcon/getalldeauths')
+
+    def falcon_start_crack(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return self._post('/falcon/startcrack', payload)
+
+    def gps_status(self) -> Dict[str, Any]:
+        return self._get('/gps/status')
+
+    def hackrf_status(self) -> Dict[str, Any]:
+        return self._get('/spectrum/hackrfstatus')
+
+    def hackrf_start(self, band: str) -> Dict[str, Any]:
+        if band == '24':
+            return self._get('/spectrum/scanstart24')
+        if band == '5':
+            return self._get('/spectrum/scanstart5')
+        raise ValueError('Band must be "24" or "5"')
+
+    def hackrf_stop(self) -> Dict[str, Any]:
+        return self._get('/spectrum/scanstop')
+
+    def hackrf_channel_data(self) -> Dict[str, Any]:
+        response = self._get_response('/spectrum/scanstatus')
+        content = response.content
+        if response.headers.get('Content-Encoding', '').lower() == 'gzip':
+            try:
+                content = gzip.decompress(content)
+            except OSError:
+                # Some agents set the header but return plain JSON; fall back gracefully
+                pass
+        try:
+            return json.loads(content.decode('utf-8'))
+        except Exception as exc:
+            raise AgentHTTPError(f"Invalid spectrum response from {self.agent.name}: {exc}") from exc
+
+
+def execute_scan(
+    agent: Agent,
+    scan_type: ScanType,
+    *,
+    interface: str | None,
+    channels: List[int] | None,
+    extras: Dict[str, Any] | None,
+    progress_cb=None,
+) -> Dict[str, Any]:
+    client = AgentClient(agent)
+    if scan_type == ScanType.WIFI:
+        if not interface:
+            raise ValueError('Wi-Fi scans require an interface name')
+        return client.wifi_scan(interface, channels, progress_cb=progress_cb)
+    if scan_type == ScanType.FALCON:
+        if not interface:
+            raise ValueError('Falcon scans require an interface name')
+        poll_interval = float(extras.get('poll_interval', 2.0)) if extras else 2.0
+        timeout = float(extras.get('timeout', 90.0)) if extras else 90.0
+        return client.falcon_scan(
+            interface,
+            channels,
+            progress_cb=progress_cb,
+            poll_interval=poll_interval,
+            timeout=timeout,
+        )
+    if scan_type == ScanType.BLUETOOTH:
+        duration = float(extras.get('duration', 5.0)) if extras else 5.0
+        active = bool(extras.get('active', True)) if extras else True
+        return client.bluetooth_discovery(active=active, duration=duration, progress_cb=progress_cb)
+    raise ValueError(f'Unsupported scan type: {scan_type}')

controller/app/config.py

@@ -0,0 +1,24 @@
+import os
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    controller_db_url: str = os.environ.get(
+        "CONTROLLER_DB_URL",
+        "sqlite:///" + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "state", "controller.db")),
+    )
+    controller_base_url: str = os.environ.get("CONTROLLER_BASE_URL", "http://localhost:8000")
+    elastic_url: str | None = os.environ.get("CONTROLLER_ELASTIC_URL")
+    elastic_index_wifi: str = os.environ.get("CONTROLLER_ELASTIC_INDEX_WIFI", "sparrowwifi")
+    elastic_index_bluetooth: str = os.environ.get("CONTROLLER_ELASTIC_INDEX_BT", "sparrowbt")
+    elastic_timeout_seconds: float = float(os.environ.get("CONTROLLER_ELASTIC_TIMEOUT", "5.0"))
+
+    class Config:
+        env_prefix = "CONTROLLER_"
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    return Settings()