diff --git a/modem_gui/.gitignore b/modem_gui/.gitignore new file mode 100644 index 0000000..79f17db --- /dev/null +++ b/modem_gui/.gitignore @@ -0,0 +1,20 @@ +# Python bytecode +__pycache__/ +*.py[cod] +*.pyo + +# Virtual environments +.venv/ +venv/ +env/ + +# Distribution / packaging +dist/ +build/ +*.spec + +# pytest +.pytest_cache/ + +# PyInstaller / cx_Freeze +*.exe diff --git a/modem_gui/README.md b/modem_gui/README.md new file mode 100644 index 0000000..72f7885 --- /dev/null +++ b/modem_gui/README.md @@ -0,0 +1,229 @@ +# Modem Manager GUI + +A lightweight standalone desktop application for managing a wireless modem or +LTE/5G router via its HTTP API. Built with Python 3.10+ and PySide6. + +--- + +## Features + +- Connect to any router that exposes an XML-over-HTTP REST API (e.g. Huawei + HiLink series, Brovi, Alcatel, ZTE, and most unlocked USB dongles). +- Log in to obtain a Session ID (SID) and persist it locally for future runs. +- View real-time device status (connection state, signal, IP address, etc.). +- Send raw AT commands and inspect responses — useful for diagnostics and + validating that your SID is still active (send `AT`, expect `OK`). + +--- + +## Project Structure + +``` +modem_gui/ +├── main.py # Entry point — run this to launch the app +├── main_window.py # PySide6 main window and UI logic +├── router_client.py # HTTP API client (login, status, AT commands) +├── sid_store.py # Persistent SID storage (~/.modem_gui/sid.json) +├── requirements.txt # Python dependencies +└── tests/ + ├── test_router_client.py + └── test_sid_store.py +``` + +--- + +## Prerequisites + +| Requirement | Minimum version | +|---|---| +| Python | 3.10 | +| pip | 23.x | + +--- + +## Setup + +### 1. Create and activate a virtual environment + +**Windows (PowerShell)** +```powershell +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +``` + +**macOS / Linux** +```bash +python3 -m venv .venv +source .venv/bin/activate +``` + +### 2. Install dependencies + +```bash +pip install -r requirements.txt +``` + +### 3. Run the application + +```bash +python main.py +``` + +--- + +## Usage + +1. **Enter Host** — type the router's IP address (usually `192.168.8.1`). +2. **Login** — click *Login…* and enter the router admin username and password. + The SID is saved automatically for future sessions. +3. **Apply SID** — if you already have a SID, paste it in the *SID* field and + click *Apply SID* (useful when you share a SID from another tool). +4. **Refresh Status** — click *Refresh Status* to populate the device status + table. +5. **AT Commands** — type a command (e.g. `AT`, `AT+CSQ`, `ATI`) in the + *AT Command* field and click *Send* (or press Enter). A response of `OK` + for the bare `AT` command confirms the SID is valid. + +--- + +## Running Tests + +Tests do **not** require PySide6 or a real router — they mock all HTTP calls. + +```bash +# from the modem_gui/ directory +python -m pytest tests/ -v +``` + +--- + +## Packaging for Windows + +### Option A — PyInstaller (single-file `.exe`) + +```powershell +pip install pyinstaller +pyinstaller --onefile --windowed --name ModemManager main.py +``` + +The executable will appear in `dist\ModemManager.exe`. + +> **Tip:** Add `--icon your_icon.ico` to set a custom window icon. + +### Option B — cx_Freeze (MSI installer) + +```powershell +pip install cx_Freeze +``` + +Create `setup_cx.py`: + +```python +from cx_Freeze import setup, Executable + +setup( + name="ModemManager", + version="0.1.0", + executables=[Executable("main.py", base="Win32GUI", target_name="ModemManager.exe")], +) +``` + +```powershell +python setup_cx.py bdist_msi +``` + +The installer will appear in the `dist\` folder. + +--- + +## Configuration & Data Storage + +The SID and last-used host are stored in: + +| Platform | Path | +|---|---| +| Windows | `%USERPROFILE%\.modem_gui\sid.json` | +| macOS / Linux | `~/.modem_gui/sid.json` | + +The file is created automatically on first save. On POSIX systems the file +permissions are set to `0600` (owner read/write only). The file contains +**only the SID**, not your password. + +To reset the stored session, delete the file or call *Login…* to obtain a new +SID. + +--- + +## Troubleshooting + +### "Not authorised" / API error 125003 + +Your SID has expired. Click *Login…* to obtain a fresh one. + +### No response from router + +- Confirm the IP address is correct (try opening `http://192.168.8.1` in a + browser). +- Ensure the application and router are on the same network. +- Check that the router's HTTP API is enabled in its settings. + +### AT command returns empty string + +Some routers restrict AT command access by firmware version. Try `AT+CSQ` or +`ATI` as alternatives to verify connectivity. + +--- + +## Git Clone Authentication Troubleshooting + +If you receive authentication errors when cloning this repository (`git clone` +hangs or returns `403 Forbidden` / `remote: Repository not found`), use one of +the two recommended methods below. + +### Option 1 — Personal Access Token (PAT) via HTTPS + +1. Go to **GitHub → Settings → Developer settings → Personal access tokens → + Fine-grained tokens** (or classic tokens). +2. Create a token with at least **repo → Contents: Read** scope. +3. Clone using your PAT in place of the password: + +```bash +git clone https://:@github.com/deilert00/Goald.git +``` + +Or configure the credential helper once so you are not prompted repeatedly: + +```bash +git config --global credential.helper store +git clone https://github.com/deilert00/Goald.git +# Enter your GitHub username and PAT when prompted — they are saved locally. +``` + +> ⚠️ Store PATs in a password manager, not in plain text files or shell history. + +### Option 2 — SSH key + +1. Generate an SSH key pair (skip if you already have one): + +```bash +ssh-keygen -t ed25519 -C "your@email.com" +``` + +2. Add the public key to GitHub: **Settings → SSH and GPG keys → New SSH key** + (paste the contents of `~/.ssh/id_ed25519.pub`). + +3. Clone via SSH: + +```bash +git clone git@github.com:deilert00/Goald.git +``` + +4. Test your connection: + +```bash +ssh -T git@github.com +# Expected: Hi ! You've successfully authenticated… +``` + +SSH is the recommended method for developer workstations because it avoids +token rotation and works seamlessly with `git push`. diff --git a/modem_gui/main.py b/modem_gui/main.py new file mode 100644 index 0000000..49c14b6 --- /dev/null +++ b/modem_gui/main.py @@ -0,0 +1,43 @@ +"""main.py — Entry point for the Modem Manager GUI application. + +Run with:: + + python main.py + +or, after packaging (see README.md):: + + ModemManager.exe +""" + +from __future__ import annotations + +import logging +import sys + +from PySide6.QtWidgets import QApplication + +from main_window import MainWindow + + +def _configure_logging() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + + +def main() -> None: + _configure_logging() + app = QApplication(sys.argv) + app.setApplicationName("Modem Manager") + app.setApplicationVersion("0.1.0") + + window = MainWindow() + window.show() + + sys.exit(app.exec()) + + +if __name__ == "__main__": + main() diff --git a/modem_gui/main_window.py b/modem_gui/main_window.py new file mode 100644 index 0000000..3d05b30 --- /dev/null +++ b/modem_gui/main_window.py @@ -0,0 +1,364 @@ +"""main_window.py — PySide6 main window for the modem management GUI. + +Layout +------ +┌─────────────────────────────────────────────────────┐ +│ Connection ─────────────────────────────────────── │ +│ Host: [192.168.8.1 ___] SID: [____________] [Login]│ +│ │ +│ Status ────────────────────────────────────────── │ +│ [Refresh] │ +│ ┌──────────────────────────────────────────────┐ │ +│ │ (status key/value table) │ │ +│ └──────────────────────────────────────────────┘ │ +│ │ +│ AT Command ────────────────────────────────────── │ +│ Command: [AT__________________________] [Send] │ +│ ┌──────────────────────────────────────────────┐ │ +│ │ (response / error output) │ │ +│ └──────────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────┘ +""" + +from __future__ import annotations + +import logging +from typing import Callable, Optional + +from PySide6.QtCore import Qt, QThread, Signal +from PySide6.QtWidgets import ( + QApplication, + QDialog, + QFormLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QLineEdit, + QMainWindow, + QMessageBox, + QPushButton, + QTableWidget, + QTableWidgetItem, + QTextEdit, + QVBoxLayout, + QWidget, +) + +from router_client import RouterAPIError, RouterClient +from sid_store import SIDStore +import requests + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Background workers (keeps the GUI responsive during HTTP calls) +# --------------------------------------------------------------------------- + + +class _Worker(QThread): + """Generic one-shot worker thread. + + Runs *task* in a background thread and emits either ``result`` or + ``error`` when done. + """ + + result: Signal = Signal(object) + error: Signal = Signal(str) + + def __init__(self, task: Callable[[], object]) -> None: + super().__init__() + self._task = task + + def run(self) -> None: + try: + self.result.emit(self._task()) + except (RouterAPIError, requests.RequestException, OSError) as exc: + self.error.emit(str(exc)) + except Exception as exc: # noqa: BLE001 — surface unexpected errors to UI, never swallow + self.error.emit(f"Unexpected error: {exc}") + + +# --------------------------------------------------------------------------- +# Main window +# --------------------------------------------------------------------------- + + +class MainWindow(QMainWindow): + """Primary application window.""" + + _WINDOW_TITLE = "Modem Manager" + _MIN_WIDTH = 640 + _MIN_HEIGHT = 520 + + def __init__(self) -> None: + super().__init__() + self._client: Optional[RouterClient] = None + self._sid_store = SIDStore() + self._workers: list[_Worker] = [] # keep references alive + + self.setWindowTitle(self._WINDOW_TITLE) + self.setMinimumSize(self._MIN_WIDTH, self._MIN_HEIGHT) + + central = QWidget() + self.setCentralWidget(central) + layout = QVBoxLayout(central) + layout.setContentsMargins(12, 12, 12, 12) + layout.setSpacing(12) + + layout.addWidget(self._build_connection_group()) + layout.addWidget(self._build_status_group()) + layout.addWidget(self._build_at_group()) + + self._restore_saved_state() + + # ------------------------------------------------------------------ + # UI construction helpers + # ------------------------------------------------------------------ + + def _build_connection_group(self) -> QGroupBox: + box = QGroupBox("Connection") + form = QFormLayout(box) + + self._host_edit = QLineEdit() + self._host_edit.setPlaceholderText("192.168.8.1") + self._host_edit.setToolTip("Router IP address or hostname") + form.addRow("Host:", self._host_edit) + + sid_row = QHBoxLayout() + self._sid_edit = QLineEdit() + self._sid_edit.setPlaceholderText("Paste existing SID or leave blank") + self._sid_edit.setEchoMode(QLineEdit.EchoMode.Password) + self._sid_edit.setToolTip("Session ID (obtained after login)") + sid_row.addWidget(self._sid_edit) + + self._login_btn = QPushButton("Login…") + self._login_btn.setToolTip("Open login dialog to obtain a new SID") + self._login_btn.clicked.connect(self._on_login) + sid_row.addWidget(self._login_btn) + + self._apply_btn = QPushButton("Apply SID") + self._apply_btn.setToolTip("Use the SID entered above without re-logging in") + self._apply_btn.clicked.connect(self._on_apply_sid) + sid_row.addWidget(self._apply_btn) + + form.addRow("SID:", sid_row) + + self._status_label = QLabel("Not connected") + self._status_label.setAlignment(Qt.AlignmentFlag.AlignRight) + form.addRow("", self._status_label) + + return box + + def _build_status_group(self) -> QGroupBox: + box = QGroupBox("Device Status") + vbox = QVBoxLayout(box) + + self._refresh_btn = QPushButton("Refresh Status") + self._refresh_btn.clicked.connect(self._on_refresh_status) + vbox.addWidget(self._refresh_btn) + + self._status_table = QTableWidget(0, 2) + self._status_table.setHorizontalHeaderLabels(["Field", "Value"]) + self._status_table.horizontalHeader().setStretchLastSection(True) + self._status_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) + self._status_table.setAlternatingRowColors(True) + vbox.addWidget(self._status_table) + + return box + + def _build_at_group(self) -> QGroupBox: + box = QGroupBox("AT Command") + vbox = QVBoxLayout(box) + + cmd_row = QHBoxLayout() + self._at_edit = QLineEdit() + self._at_edit.setPlaceholderText("AT") + self._at_edit.setToolTip("Enter an AT command (e.g. AT, AT+CSQ)") + self._at_edit.returnPressed.connect(self._on_send_at) + cmd_row.addWidget(self._at_edit) + + self._send_btn = QPushButton("Send") + self._send_btn.clicked.connect(self._on_send_at) + cmd_row.addWidget(self._send_btn) + + vbox.addLayout(cmd_row) + + self._at_output = QTextEdit() + self._at_output.setReadOnly(True) + self._at_output.setPlaceholderText("Response will appear here…") + self._at_output.setFixedHeight(120) + vbox.addWidget(self._at_output) + + return box + + # ------------------------------------------------------------------ + # State helpers + # ------------------------------------------------------------------ + + def _restore_saved_state(self) -> None: + saved_host = self._sid_store.load_host() + saved_sid = self._sid_store.load() + if saved_host: + self._host_edit.setText(saved_host) + if saved_sid: + self._sid_edit.setText(saved_sid) + self._apply_client(self._host_edit.text() or saved_host or "", saved_sid) + + def _apply_client(self, host: str, sid: str) -> None: + if not host: + return + self._client = RouterClient(host) + self._client.sid = sid + self._status_label.setText(f"SID set for {host}") + logger.debug("Client initialised for host=%s", host) + + # ------------------------------------------------------------------ + # Slot handlers + # ------------------------------------------------------------------ + + def _on_login(self) -> None: + host = self._host_edit.text().strip() + if not host: + QMessageBox.warning(self, "Missing Host", "Please enter the router's IP address or hostname.") + return + + dialog = _LoginDialog(self) + if dialog.exec() != dialog.DialogCode.Accepted: + return + + username, password = dialog.credentials() + client = RouterClient(host) + + def task() -> str: + return client.login(username, password) + + worker = _Worker(task) + worker.result.connect(lambda sid: self._on_login_done(client, host, sid)) + worker.error.connect(self._on_error) + self._workers.append(worker) + self._set_busy(True) + worker.start() + + def _on_login_done(self, client: RouterClient, host: str, sid: str) -> None: + self._set_busy(False) + self._client = client + self._sid_edit.setText(sid) + self._sid_store.save(sid, host) + self._status_label.setText(f"Logged in to {host}") + logger.info("Login successful, SID saved") + + def _on_apply_sid(self) -> None: + host = self._host_edit.text().strip() + sid = self._sid_edit.text().strip() + if not host: + QMessageBox.warning(self, "Missing Host", "Please enter the router's IP address or hostname.") + return + if not sid: + QMessageBox.warning(self, "Missing SID", "Please enter a Session ID.") + return + self._apply_client(host, sid) + self._sid_store.save(sid, host) + + def _on_refresh_status(self) -> None: + if not self._client: + QMessageBox.information(self, "Not Connected", "Set a host and SID first.") + return + + def task() -> dict[str, str]: + return self._client.get_status() # type: ignore[union-attr] + + worker = _Worker(task) + worker.result.connect(self._populate_status_table) + worker.error.connect(self._on_error) + self._workers.append(worker) + self._set_busy(True) + worker.start() + + def _populate_status_table(self, data: dict[str, str]) -> None: + self._set_busy(False) + self._status_table.setRowCount(0) + for key, value in data.items(): + row = self._status_table.rowCount() + self._status_table.insertRow(row) + self._status_table.setItem(row, 0, QTableWidgetItem(key)) + self._status_table.setItem(row, 1, QTableWidgetItem(value)) + + def _on_send_at(self) -> None: + if not self._client: + QMessageBox.information(self, "Not Connected", "Set a host and SID first.") + return + + command = self._at_edit.text().strip() or "AT" + + def task() -> str: + return self._client.send_at_command(command) # type: ignore[union-attr] + + worker = _Worker(task) + worker.result.connect(self._show_at_result) + worker.error.connect(self._on_at_error) + self._workers.append(worker) + self._set_busy(True) + worker.start() + + def _show_at_result(self, response: str) -> None: + self._set_busy(False) + self._at_output.setStyleSheet("") + self._at_output.setPlainText(response) + + def _on_at_error(self, message: str) -> None: + self._set_busy(False) + self._at_output.setStyleSheet("color: red;") + self._at_output.setPlainText(f"ERROR: {message}") + logger.error("AT command error: %s", message) + + def _on_error(self, message: str) -> None: + self._set_busy(False) + QMessageBox.critical(self, "Error", message) + self._status_label.setText("Error — check connection") + logger.error("Router error: %s", message) + + def _set_busy(self, busy: bool) -> None: + self._login_btn.setEnabled(not busy) + self._apply_btn.setEnabled(not busy) + self._refresh_btn.setEnabled(not busy) + self._send_btn.setEnabled(not busy) + if busy: + self._status_label.setText("Working…") + + +# --------------------------------------------------------------------------- +# Login dialog +# --------------------------------------------------------------------------- + + +class _LoginDialog(QDialog): + """Simple username/password dialog.""" + + def __init__(self, parent: Optional[QWidget] = None) -> None: + super().__init__(parent) + self.setWindowTitle("Login to Router") + self.setModal(True) + self.setFixedWidth(320) + + form = QFormLayout(self) + self._user_edit = QLineEdit() + self._user_edit.setText("admin") + form.addRow("Username:", self._user_edit) + + self._pass_edit = QLineEdit() + self._pass_edit.setEchoMode(QLineEdit.EchoMode.Password) + form.addRow("Password:", self._pass_edit) + + btn_row = QHBoxLayout() + ok_btn = QPushButton("Login") + ok_btn.setDefault(True) + ok_btn.clicked.connect(self.accept) + cancel_btn = QPushButton("Cancel") + cancel_btn.clicked.connect(self.reject) + btn_row.addWidget(ok_btn) + btn_row.addWidget(cancel_btn) + form.addRow(btn_row) + + def credentials(self) -> tuple[str, str]: + return self._user_edit.text(), self._pass_edit.text() diff --git a/modem_gui/requirements.txt b/modem_gui/requirements.txt new file mode 100644 index 0000000..2e8d7dd --- /dev/null +++ b/modem_gui/requirements.txt @@ -0,0 +1,2 @@ +PySide6==6.7.3 +requests==2.32.3 diff --git a/modem_gui/router_client.py b/modem_gui/router_client.py new file mode 100644 index 0000000..93fc780 --- /dev/null +++ b/modem_gui/router_client.py @@ -0,0 +1,217 @@ +"""router_client.py — HTTP API client for wireless modem management. + +Communicates with the router's REST API at ``http:///api``. +Authentication is handled via a Session-ID (SID) cookie/header that is +obtained by logging in and must be refreshed when it expires. + +Typical usage:: + + client = RouterClient("192.168.8.1") + client.login("admin", "password") # stores SID internally + print(client.get_status()) # fetch device status + print(client.send_at_command("AT")) # validate SID via AT command + client.logout() +""" + +from __future__ import annotations + +import logging +import re +import xml.etree.ElementTree as ET +from typing import Optional + +import requests + +logger = logging.getLogger(__name__) + +# Default timeout for every HTTP request (seconds) +_DEFAULT_TIMEOUT = 10 + + +class RouterAPIError(Exception): + """Raised when the router API returns an error response.""" + + def __init__(self, code: str, message: str = "") -> None: + self.code = code + super().__init__(f"Router API error {code}: {message}" if message else f"Router API error {code}") + + +class RouterClient: + """Thin HTTP client for a wireless modem's REST API. + + The API is modelled after the common Huawei HiLink interface exposed by + many LTE/5G dongles and routers, but the class can be adapted to any + router that uses XML-over-HTTP with SID cookies. + + Args: + host: IP address or hostname of the router (no scheme, no trailing slash). + timeout: Request timeout in seconds. + """ + + def __init__(self, host: str, timeout: int = _DEFAULT_TIMEOUT) -> None: + self.host = host.rstrip("/") + self.timeout = timeout + self._session = requests.Session() + self._sid: Optional[str] = None + + # ------------------------------------------------------------------ + # Public helpers + # ------------------------------------------------------------------ + + @property + def base_url(self) -> str: + return f"http://{self.host}/api" + + @property + def sid(self) -> Optional[str]: + """Current session ID, or *None* if not authenticated.""" + return self._sid + + @sid.setter + def sid(self, value: Optional[str]) -> None: + self._sid = value + if value: + self._session.headers.update({"Cookie": f"SessionID={value}"}) + else: + self._session.headers.pop("Cookie", None) + + # ------------------------------------------------------------------ + # Authentication + # ------------------------------------------------------------------ + + def login(self, username: str, password: str) -> str: + """Authenticate with the router and store the returned SID. + + Returns: + The new SID string. + + Raises: + RouterAPIError: If the router returns an error code. + requests.RequestException: On network failure. + """ + body = ( + "" + "" + f"{_xml_escape(username)}" + f"{_xml_escape(password)}" + "4" + "" + ) + root = self._post("user/login", body) + sid = root.findtext("SesInfo") or root.findtext("SessionID") or "" + if not sid: + raise RouterAPIError("no-sid", "Login succeeded but no SID was returned") + self.sid = sid + logger.debug("Login successful, SID=%s…", sid[:8]) + return sid + + def logout(self) -> None: + """Terminate the current session on the router.""" + try: + self._post( + "user/logout", + "1", + ) + except (RouterAPIError, requests.RequestException): + pass # best-effort logout; clear SID regardless of network outcome + finally: + self.sid = None + logger.debug("Logged out") + + # ------------------------------------------------------------------ + # Status + # ------------------------------------------------------------------ + + def get_status(self) -> dict[str, str]: + """Fetch device status summary. + + Returns: + A dictionary with keys such as ``ConnectionStatus``, + ``SignalIcon``, ``CurrentNetworkType``, ``WanIPAddress``, etc. + """ + root = self._get("monitoring/status") + return {child.tag: (child.text or "") for child in root} + + def get_device_info(self) -> dict[str, str]: + """Fetch static device information (model, firmware, IMEI, …).""" + root = self._get("device/information") + return {child.tag: (child.text or "") for child in root} + + # ------------------------------------------------------------------ + # AT commands + # ------------------------------------------------------------------ + + def send_at_command(self, command: str) -> str: + """Send a raw AT command to the modem and return the response text. + + The plain ``AT`` command is useful for validating that the SID is + still valid — a healthy modem returns ``OK``. + + Args: + command: AT command string, e.g. ``"AT"`` or ``"AT+CSQ"``. + + Returns: + The modem's response text (e.g. ``"OK"`` or ``"+CSQ: 18,0\\r\\nOK"``). + + Raises: + RouterAPIError: On API-level errors. + requests.RequestException: On network failures. + """ + body = ( + "" + "" + f"0" + f"{_xml_escape(command)}\r\n" + "" + ) + root = self._post("device/at-execute", body) + response = root.findtext("AT") or "" + logger.debug("AT command %r → %r", command, response) + return response.strip() + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _get(self, endpoint: str) -> ET.Element: + url = f"{self.base_url}/{endpoint}" + resp = self._session.get(url, timeout=self.timeout) + resp.raise_for_status() + return self._parse_response(resp.text) + + def _post(self, endpoint: str, body: str) -> ET.Element: + url = f"{self.base_url}/{endpoint}" + headers = {"Content-Type": "text/xml; charset=UTF-8"} + resp = self._session.post(url, data=body.encode("utf-8"), headers=headers, timeout=self.timeout) + resp.raise_for_status() + return self._parse_response(resp.text) + + @staticmethod + def _parse_response(text: str) -> ET.Element: + """Parse an XML response; raise RouterAPIError for ```` roots.""" + try: + root = ET.fromstring(text) + except ET.ParseError as exc: + raise RouterAPIError("parse-error", str(exc)) from exc + + if root.tag == "error": + code = root.findtext("code") or "unknown" + message = root.findtext("message") or "" + raise RouterAPIError(code, message) + + return root + + +# ------------------------------------------------------------------ +# Helpers +# ------------------------------------------------------------------ + +def _xml_escape(value: str) -> str: + """Minimal XML character escaping for values inserted into XML bodies.""" + return ( + value.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + .replace("'", "'") + ) diff --git a/modem_gui/sid_store.py b/modem_gui/sid_store.py new file mode 100644 index 0000000..f6b5c6d --- /dev/null +++ b/modem_gui/sid_store.py @@ -0,0 +1,100 @@ +"""sid_store.py — Persistent storage for the router Session ID (SID). + +The SID is saved to a small JSON file in the user's home directory so that +it survives application restarts. The store deliberately contains only the +SID (not the password) to minimise the risk of credential exposure. + +Typical usage:: + + store = SIDStore() + store.save("abc123def456") + print(store.load()) # "abc123def456" + store.clear() + print(store.load()) # None +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +# Default storage location: ~/.modem_gui/sid.json +_DEFAULT_STORE_DIR = Path.home() / ".modem_gui" +_DEFAULT_STORE_FILE = _DEFAULT_STORE_DIR / "sid.json" + +_KEY_SID = "sid" +_KEY_HOST = "host" + + +class SIDStore: + """Read/write the session ID and last-used host to a local JSON file. + + Args: + path: Override the default storage path (useful for testing). + """ + + def __init__(self, path: Optional[Path] = None) -> None: + self._path = path or _DEFAULT_STORE_FILE + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def save(self, sid: str, host: str = "") -> None: + """Persist *sid* (and optionally *host*) to disk. + + Creates parent directories if they do not exist. + + Args: + sid: The session ID string returned by the router. + host: Router hostname/IP to remember alongside the SID. + """ + self._path.parent.mkdir(parents=True, exist_ok=True) + data: dict[str, str] = {_KEY_SID: sid} + if host: + data[_KEY_HOST] = host + try: + self._path.write_text(json.dumps(data), encoding="utf-8") + # Restrict file permissions on POSIX systems (owner read/write only) + if os.name == "posix": + os.chmod(self._path, 0o600) + except OSError as exc: + logger.warning("Could not write SID store at %s: %s", self._path, exc) + + def load(self) -> Optional[str]: + """Return the stored SID, or *None* if nothing is saved.""" + data = self._read() + return data.get(_KEY_SID) or None # treat empty-string SID as absent + + def load_host(self) -> Optional[str]: + """Return the stored host, or *None* if nothing is saved.""" + data = self._read() + return data.get(_KEY_HOST) or None # treat empty-string host as absent + + def clear(self) -> None: + """Delete the stored SID and host.""" + try: + self._path.unlink(missing_ok=True) + except OSError as exc: + logger.warning("Could not clear SID store at %s: %s", self._path, exc) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _read(self) -> dict[str, str]: + try: + text = self._path.read_text(encoding="utf-8") + data = json.loads(text) + if isinstance(data, dict): + return {str(k): str(v) for k, v in data.items()} + except FileNotFoundError: + pass + except (json.JSONDecodeError, OSError) as exc: + logger.warning("Could not read SID store at %s: %s", self._path, exc) + return {} diff --git a/modem_gui/tests/test_router_client.py b/modem_gui/tests/test_router_client.py new file mode 100644 index 0000000..6764bf8 --- /dev/null +++ b/modem_gui/tests/test_router_client.py @@ -0,0 +1,195 @@ +"""tests/test_router_client.py — Unit tests for RouterClient and helpers. + +Run with:: + + cd modem_gui + python -m pytest tests/ -v +""" + +from __future__ import annotations + +import sys +import os +import unittest +from unittest.mock import MagicMock, patch +import xml.etree.ElementTree as ET + +# Ensure the modem_gui package is importable when running from repo root. +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +from router_client import RouterAPIError, RouterClient, _xml_escape + + +class TestXmlEscape(unittest.TestCase): + def test_plain_string_unchanged(self) -> None: + self.assertEqual(_xml_escape("hello"), "hello") + + def test_ampersand(self) -> None: + self.assertEqual(_xml_escape("a&b"), "a&b") + + def test_lt_gt(self) -> None: + self.assertEqual(_xml_escape(""), "<tag>") + + def test_quotes(self) -> None: + self.assertEqual(_xml_escape('"it\'s"'), ""it's"") + + def test_combined(self) -> None: + self.assertEqual(_xml_escape(''), "<a b="c&d">") + + +class TestRouterAPIError(unittest.TestCase): + def test_message_format_with_detail(self) -> None: + err = RouterAPIError("125001", "Wrong password") + self.assertIn("125001", str(err)) + self.assertIn("Wrong password", str(err)) + + def test_message_format_without_detail(self) -> None: + err = RouterAPIError("999") + self.assertIn("999", str(err)) + + +class TestRouterClientProperties(unittest.TestCase): + def setUp(self) -> None: + self.client = RouterClient("192.168.8.1") + + def test_base_url(self) -> None: + self.assertEqual(self.client.base_url, "http://192.168.8.1/api") + + def test_base_url_strips_trailing_slash(self) -> None: + client = RouterClient("192.168.8.1/") + self.assertEqual(client.base_url, "http://192.168.8.1/api") + + def test_sid_initially_none(self) -> None: + self.assertIsNone(self.client.sid) + + def test_set_sid_updates_header(self) -> None: + self.client.sid = "abc123" + self.assertIn("abc123", self.client._session.headers.get("Cookie", "")) + + def test_clear_sid_removes_header(self) -> None: + self.client.sid = "abc123" + self.client.sid = None + self.assertNotIn("Cookie", self.client._session.headers) + + +class TestParseResponse(unittest.TestCase): + def test_parses_valid_xml(self) -> None: + xml = "902" + root = RouterClient._parse_response(xml) + self.assertEqual(root.tag, "response") + self.assertEqual(root.findtext("Status"), "902") + + def test_raises_on_error_root(self) -> None: + xml = "125003Too many attempts" + with self.assertRaises(RouterAPIError) as ctx: + RouterClient._parse_response(xml) + self.assertEqual(ctx.exception.code, "125003") + self.assertIn("Too many attempts", str(ctx.exception)) + + def test_raises_on_invalid_xml(self) -> None: + with self.assertRaises(RouterAPIError) as ctx: + RouterClient._parse_response("not xml at all") + self.assertEqual(ctx.exception.code, "parse-error") + + def test_error_without_message(self) -> None: + xml = "999" + with self.assertRaises(RouterAPIError) as ctx: + RouterClient._parse_response(xml) + self.assertEqual(ctx.exception.code, "999") + + +class TestLoginLogout(unittest.TestCase): + def setUp(self) -> None: + self.client = RouterClient("192.168.8.1") + + def _mock_post_response(self, xml_text: str) -> MagicMock: + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.text = xml_text + mock_resp.raise_for_status = MagicMock() + return mock_resp + + def test_login_stores_sid(self) -> None: + xml = "sid_value_xyz" + self.client._session.post = MagicMock(return_value=self._mock_post_response(xml)) + result = self.client.login("admin", "pass") + self.assertEqual(result, "sid_value_xyz") + self.assertEqual(self.client.sid, "sid_value_xyz") + + def test_login_raises_on_missing_sid(self) -> None: + xml = "1" + self.client._session.post = MagicMock(return_value=self._mock_post_response(xml)) + with self.assertRaises(RouterAPIError): + self.client.login("admin", "wrong") + + def test_logout_clears_sid(self) -> None: + self.client.sid = "some_sid" + xml = "1" + self.client._session.post = MagicMock(return_value=self._mock_post_response(xml)) + self.client.logout() + self.assertIsNone(self.client.sid) + + def test_logout_is_safe_when_not_logged_in(self) -> None: + import requests as req_mod + self.client._session.post = MagicMock(side_effect=req_mod.exceptions.ConnectionError("network error")) + self.client.logout() # must not raise + self.assertIsNone(self.client.sid) + + +class TestGetStatus(unittest.TestCase): + def setUp(self) -> None: + self.client = RouterClient("192.168.8.1") + self.client.sid = "test_sid" + + def test_returns_dict(self) -> None: + xml = ( + "" + "901" + "10.0.0.1" + "" + ) + mock_resp = MagicMock() + mock_resp.text = xml + mock_resp.raise_for_status = MagicMock() + self.client._session.get = MagicMock(return_value=mock_resp) + + status = self.client.get_status() + self.assertEqual(status["ConnectionStatus"], "901") + self.assertEqual(status["WanIPAddress"], "10.0.0.1") + + +class TestSendAtCommand(unittest.TestCase): + def setUp(self) -> None: + self.client = RouterClient("192.168.8.1") + self.client.sid = "test_sid" + + def _mock_post(self, at_response: str) -> MagicMock: + xml = f"{at_response}" + mock_resp = MagicMock() + mock_resp.text = xml + mock_resp.raise_for_status = MagicMock() + return mock_resp + + def test_at_returns_ok(self) -> None: + self.client._session.post = MagicMock(return_value=self._mock_post("\r\nOK\r\n")) + result = self.client.send_at_command("AT") + self.assertEqual(result, "OK") + + def test_at_command_with_response(self) -> None: + self.client._session.post = MagicMock(return_value=self._mock_post("+CSQ: 18,0\r\nOK\r\n")) + result = self.client.send_at_command("AT+CSQ") + self.assertIn("CSQ", result) + self.assertIn("OK", result) + + def test_at_raises_on_api_error(self) -> None: + error_xml = "125002Not authorized" + mock_resp = MagicMock() + mock_resp.text = error_xml + mock_resp.raise_for_status = MagicMock() + self.client._session.post = MagicMock(return_value=mock_resp) + with self.assertRaises(RouterAPIError): + self.client.send_at_command("AT") + + +if __name__ == "__main__": + unittest.main() diff --git a/modem_gui/tests/test_sid_store.py b/modem_gui/tests/test_sid_store.py new file mode 100644 index 0000000..c2803dc --- /dev/null +++ b/modem_gui/tests/test_sid_store.py @@ -0,0 +1,96 @@ +"""tests/test_sid_store.py — Unit tests for SIDStore. + +Run with:: + + cd modem_gui + python -m pytest tests/ -v +""" + +from __future__ import annotations + +import sys +import os +import tempfile +import unittest +from pathlib import Path + +# Ensure the modem_gui package is importable when running from repo root. +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +from sid_store import SIDStore + + +class TestSIDStore(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self._store_path = Path(self._tmp.name) / "test_sid.json" + self.store = SIDStore(path=self._store_path) + + def tearDown(self) -> None: + self._tmp.cleanup() + + # ------------------------------------------------------------------ + # save / load + # ------------------------------------------------------------------ + + def test_save_and_load_sid(self) -> None: + self.store.save("abc123") + self.assertEqual(self.store.load(), "abc123") + + def test_save_and_load_host(self) -> None: + self.store.save("sid_val", host="192.168.8.1") + self.assertEqual(self.store.load_host(), "192.168.8.1") + + def test_load_returns_none_when_not_saved(self) -> None: + self.assertIsNone(self.store.load()) + + def test_load_host_returns_none_when_not_saved(self) -> None: + self.assertIsNone(self.store.load_host()) + + def test_save_overwrites_previous(self) -> None: + self.store.save("first_sid") + self.store.save("second_sid") + self.assertEqual(self.store.load(), "second_sid") + + # ------------------------------------------------------------------ + # clear + # ------------------------------------------------------------------ + + def test_clear_removes_sid(self) -> None: + self.store.save("to_delete") + self.store.clear() + self.assertIsNone(self.store.load()) + + def test_clear_is_safe_when_nothing_saved(self) -> None: + self.store.clear() # must not raise + + # ------------------------------------------------------------------ + # robustness + # ------------------------------------------------------------------ + + def test_load_returns_none_on_corrupt_file(self) -> None: + self._store_path.parent.mkdir(parents=True, exist_ok=True) + self._store_path.write_text("not valid json", encoding="utf-8") + self.assertIsNone(self.store.load()) + + def test_load_returns_none_on_wrong_json_type(self) -> None: + self._store_path.parent.mkdir(parents=True, exist_ok=True) + self._store_path.write_text("[1, 2, 3]", encoding="utf-8") + self.assertIsNone(self.store.load()) + + def test_save_creates_parent_directories(self) -> None: + deep_path = Path(self._tmp.name) / "a" / "b" / "c" / "sid.json" + store = SIDStore(path=deep_path) + store.save("deep_sid") + self.assertEqual(store.load(), "deep_sid") + + def test_sid_not_exposed_as_host(self) -> None: + self.store.save("my_sid", host="router.local") + # load() must return the SID, not the host + self.assertEqual(self.store.load(), "my_sid") + # load_host() must return the host, not the SID + self.assertEqual(self.store.load_host(), "router.local") + + +if __name__ == "__main__": + unittest.main()