diff --git a/README.md b/README.md index 8f2a5a4..3ea3f44 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ The repeater daemon runs continuously as a background process, forwarding LoRa p The repeater supports two radio backends: - **SX1262 (SPI)** — Direct connection to LoRa modules (HATs, etc.) as listed below. -- **KISS modem** — Serial TNC using the KISS protocol. Requires a pyMC_core build with KISS support (e.g. [agessaman/pyMC_core (dev)](https://github.com/agessaman/pyMC_core/tree/dev)). Set `radio_type: kiss` in config and configure `kiss.port` and `kiss.baud_rate`. The setup script (`./setup-radio-config.sh`) offers a "KISS modem" option when configuring the repeater. +- **KISS modem** — Serial TNC using the KISS protocol. Set `radio_type: kiss` in config and configure `kiss.port` and `kiss.baud_rate`. The following SX1262 hardware is currently supported out-of-the-box: @@ -164,11 +164,6 @@ http://:8000 pip install -e . ``` -On **macOS** (or when using only the KISS modem), the base install is enough. On **Raspberry Pi** with SX1262 hardware, install with the optional hardware extra so SPI/spidev is available: -```bash -pip install -e .[hardware] -``` - ## Configuration The configuration file is created and configured during installation at: diff --git a/repeater/companion/frame_server.py b/repeater/companion/frame_server.py index ba051f1..a019702 100644 --- a/repeater/companion/frame_server.py +++ b/repeater/companion/frame_server.py @@ -43,7 +43,7 @@ class CompanionFrameServer(_BaseFrameServer): port=port, bind_address=bind_address, device_model="pyMC-Repeater-Companion", - device_version="1.0.0", + device_version=None, # use FIRMWARE_VER_CODE from pyMC_core build_date="13 Feb 2026", local_hash=local_hash, stats_getter=stats_getter, diff --git a/repeater/config_manager.py b/repeater/config_manager.py index a30003c..b4c13f8 100644 --- a/repeater/config_manager.py +++ b/repeater/config_manager.py @@ -1,21 +1,18 @@ -from __future__ import annotations - import logging import os -from typing import Any, Dict, List, Optional - import yaml +from typing import Optional, Dict, Any, List logger = logging.getLogger("ConfigManager") class ConfigManager: """Manages configuration persistence and live updates to the daemon.""" - + def __init__(self, config_path: str, config: dict, daemon_instance=None): """ Initialize ConfigManager. - + Args: config_path: Path to the YAML config file config: Reference to the config dictionary @@ -24,105 +21,100 @@ class ConfigManager: self.config_path = config_path self.config = config self.daemon = daemon_instance - - def save_to_file(self) -> tuple[bool, str]: + + def save_to_file(self) -> bool: """ Save current config to YAML file. - + Returns: - (True, "") if successful, (False, error_message) otherwise + True if successful, False otherwise """ try: - dirpath = os.path.dirname(self.config_path) - if dirpath: - os.makedirs(dirpath, exist_ok=True) - with open(self.config_path, "w") as f: + os.makedirs(os.path.dirname(self.config_path), exist_ok=True) + with open(self.config_path, 'w') as f: # Use safe_dump with explicit width to prevent line wrapping # Setting width to a very large number prevents truncation of long strings like identity keys yaml.safe_dump( - self.config, - f, - default_flow_style=False, - indent=2, + self.config, + f, + default_flow_style=False, + indent=2, width=1000000, # Very large width to prevent any line wrapping sort_keys=False, - allow_unicode=True, + allow_unicode=True ) logger.info(f"Configuration saved to {self.config_path}") - return True, "" + return True except Exception as e: - msg = f"Failed to save config to {self.config_path}: {e}" - logger.error(msg, exc_info=True) - return False, str(e) - + logger.error(f"Failed to save config to {self.config_path}: {e}", exc_info=True) + return False + def live_update_daemon(self, sections: Optional[List[str]] = None) -> bool: """ Apply configuration changes to the running daemon's in-memory config. - + Args: sections: List of config sections to update (e.g., ['repeater', 'delays']). If None, updates all common sections. - + Returns: True if live update was successful, False otherwise """ - if not self.daemon or not hasattr(self.daemon, "config"): + if not self.daemon or not hasattr(self.daemon, 'config'): logger.warning("Daemon not available for live update") return False - + try: daemon_config = self.daemon.config - + # Default sections to update if not specified if sections is None: - sections = ["repeater", "delays", "radio", "acl", "identities"] - + sections = ['repeater', 'delays', 'radio', 'acl', 'identities'] + # Update each section for section in sections: if section in self.config: if section not in daemon_config: daemon_config[section] = {} - + # Deep copy the section to avoid reference issues if isinstance(self.config[section], dict): daemon_config[section].update(self.config[section]) else: daemon_config[section] = self.config[section] - + logger.debug(f"Live updated daemon config section: {section}") - + logger.info(f"Live updated daemon config sections: {', '.join(sections)}") - + # Also reload runtime config in RepeaterHandler if delays or repeater sections changed - if self.daemon and hasattr(self.daemon, "repeater_handler"): - if any(s in ["delays", "repeater"] for s in sections): - if hasattr(self.daemon.repeater_handler, "reload_runtime_config"): + if self.daemon and hasattr(self.daemon, 'repeater_handler'): + if any(s in ['delays', 'repeater'] for s in sections): + if hasattr(self.daemon.repeater_handler, 'reload_runtime_config'): self.daemon.repeater_handler.reload_runtime_config() logger.info("Reloaded RepeaterHandler runtime config") - + return True - + except Exception as e: logger.error(f"Failed to live update daemon config: {e}", exc_info=True) return False - - def update_and_save( - self, - updates: Dict[str, Any], - live_update: bool = True, - live_update_sections: Optional[List[str]] = None, - ) -> Dict[str, Any]: + + def update_and_save(self, + updates: Dict[str, Any], + live_update: bool = True, + live_update_sections: Optional[List[str]] = None) -> Dict[str, Any]: """ Apply updates to config, save to file, and optionally live update daemon. - + This is the main method that should be used by both mesh_cli and api_endpoints. - + Args: updates: Dictionary of config updates in nested format. Example: {"repeater": {"node_name": "NewName"}, "delays": {"tx_delay_factor": 1.5}} live_update: Whether to apply changes to running daemon immediately live_update_sections: Specific sections to live update. If None, auto-detects from updates. - + Returns: Dict with keys: - success: bool - Whether operation succeeded @@ -130,59 +122,62 @@ class ConfigManager: - live_updated: bool - Whether daemon was live updated - error: str (optional) - Error message if failed """ - result = {"success": False, "saved": False, "live_updated": False} - + result = { + "success": False, + "saved": False, + "live_updated": False + } + try: # Apply updates to config for section, values in updates.items(): if section not in self.config: self.config[section] = {} - + if isinstance(values, dict): self.config[section].update(values) else: self.config[section] = values - + # Save to file - saved, err = self.save_to_file() - result["saved"] = saved - + result["saved"] = self.save_to_file() + if not result["saved"]: - result["error"] = err or "Failed to save config to file" + result["error"] = "Failed to save config to file" return result - + # Live update daemon if requested if live_update: # Auto-detect sections if not specified if live_update_sections is None: live_update_sections = list(updates.keys()) - + result["live_updated"] = self.live_update_daemon(live_update_sections) - + result["success"] = result["saved"] return result - + except Exception as e: logger.error(f"Error in update_and_save: {e}", exc_info=True) result["error"] = str(e) return result - + def update_nested(self, path: str, value: Any, live_update: bool = True) -> Dict[str, Any]: """ Update a nested config value using dot notation. - + Convenience method for simple updates like "repeater.node_name" = "NewName" - + Args: path: Dot-separated path to config value (e.g., "repeater.node_name") value: Value to set live_update: Whether to apply changes to running daemon - + Returns: Result dict from update_and_save """ - parts = path.split(".") - + parts = path.split('.') + if len(parts) == 1: # Top-level key updates = {parts[0]: value} @@ -201,26 +196,26 @@ class ConfigManager: current[part] = {} current = current[part] current[parts[-1]] = value - + # Determine which section to live update section = parts[0] - + return self.update_and_save( updates=updates, live_update=live_update, - live_update_sections=[section] if live_update else None, + live_update_sections=[section] if live_update else None ) - + def get_status(self) -> Dict[str, Any]: """ Get status information about the ConfigManager. - + Returns: Dict with config file path, existence, daemon availability """ return { "config_path": self.config_path, "config_exists": os.path.exists(self.config_path), - "daemon_available": self.daemon is not None and hasattr(self.daemon, "config"), - "config_sections": list(self.config.keys()) if self.config else [], + "daemon_available": self.daemon is not None and hasattr(self.daemon, 'config'), + "config_sections": list(self.config.keys()) if self.config else [] } diff --git a/repeater/handler_helpers/login.py b/repeater/handler_helpers/login.py index 5912866..3e9924b 100644 --- a/repeater/handler_helpers/login.py +++ b/repeater/handler_helpers/login.py @@ -8,6 +8,7 @@ import asyncio import logging from pymc_core.node.handlers.login_server import LoginServerHandler +from pymc_core.protocol.constants import PAYLOAD_TYPE_ANON_REQ logger = logging.getLogger("LoginHelper") @@ -125,9 +126,12 @@ class LoginHelper: packet.mark_do_not_retransmit() return True else: - logger.debug( - f"No login handler registered for hash 0x{dest_hash:02X}, allowing forward" - ) + # ANON_REQ to other nodes (e.g. owner-info to firmware) is normal; skip log to avoid spam + ptype = getattr(packet, "get_payload_type", lambda: None)() + if ptype != PAYLOAD_TYPE_ANON_REQ: + logger.debug( + f"No login handler registered for hash 0x{dest_hash:02X}, allowing forward" + ) return False except Exception as e: diff --git a/repeater/handler_helpers/protocol_request.py b/repeater/handler_helpers/protocol_request.py index 5490a65..ac42f69 100644 --- a/repeater/handler_helpers/protocol_request.py +++ b/repeater/handler_helpers/protocol_request.py @@ -12,6 +12,7 @@ import time from pymc_core.node.handlers.protocol_request import ( REQ_TYPE_GET_ACCESS_LIST, REQ_TYPE_GET_NEIGHBOURS, + REQ_TYPE_GET_OWNER_INFO, REQ_TYPE_GET_STATUS, REQ_TYPE_GET_TELEMETRY_DATA, SERVER_RESPONSE_DELAY_MS, diff --git a/repeater/web/auth_endpoints.py b/repeater/web/auth_endpoints.py index 4f9e1ed..2ceb572 100644 --- a/repeater/web/auth_endpoints.py +++ b/repeater/web/auth_endpoints.py @@ -1,11 +1,8 @@ """ Authentication endpoints for login and token management """ - -import logging - import cherrypy - +import logging from .auth.middleware import require_auth logger = logging.getLogger(__name__) @@ -27,101 +24,123 @@ class TokensAPIEndpoint: @require_auth def index(self): # Handle CORS preflight - if cherrypy.request.method == "OPTIONS": + if cherrypy.request.method == 'OPTIONS': return {} - + # Get token manager from cherrypy config - token_manager = cherrypy.config.get("token_manager") + token_manager = cherrypy.config.get('token_manager') if not token_manager: cherrypy.response.status = 500 - return {"success": False, "error": "Token manager not available"} - - if cherrypy.request.method == "GET": + return {'success': False, 'error': 'Token manager not available'} + + if cherrypy.request.method == 'GET': try: tokens = token_manager.list_tokens() - return {"success": True, "tokens": tokens} + return { + 'success': True, + 'tokens': tokens + } except Exception as e: logger.error(f"Token list error: {e}") cherrypy.response.status = 500 - return {"success": False, "error": "Failed to list tokens"} - - elif cherrypy.request.method == "POST": + return { + 'success': False, + 'error': 'Failed to list tokens' + } + + elif cherrypy.request.method == 'POST': try: import json - - body = cherrypy.request.body.read().decode("utf-8") + body = cherrypy.request.body.read().decode('utf-8') data = json.loads(body) if body else {} - name = data.get("name", "").strip() - + name = data.get('name', '').strip() + if not name: cherrypy.response.status = 400 - return {"success": False, "error": "Token name is required"} - + return { + 'success': False, + 'error': 'Token name is required' + } + # Create the token token_id, plaintext_token = token_manager.create_token(name) - - logger.info( - f"Generated API token '{name}' (ID: {token_id}) by user {cherrypy.request.user['username']}" - ) - + + logger.info(f"Generated API token '{name}' (ID: {token_id}) by user {cherrypy.request.user['username']}") + return { - "success": True, - "token": plaintext_token, - "token_id": token_id, - "name": name, - "warning": "Save this token securely - it will not be shown again", + 'success': True, + 'token': plaintext_token, + 'token_id': token_id, + 'name': name, + 'warning': 'Save this token securely - it will not be shown again' } - + except Exception as e: logger.error(f"Token generation error: {e}") cherrypy.response.status = 500 - return {"success": False, "error": "Failed to generate token"} + return { + 'success': False, + 'error': 'Failed to generate token' + } else: raise cherrypy.HTTPError(405, "Method not allowed") - + @cherrypy.expose @cherrypy.tools.json_out() @require_auth def default(self, token_id=None): # Handle CORS preflight - if cherrypy.request.method == "OPTIONS": + if cherrypy.request.method == 'OPTIONS': return {} - + # Get token manager from cherrypy config - token_manager = cherrypy.config.get("token_manager") + token_manager = cherrypy.config.get('token_manager') if not token_manager: cherrypy.response.status = 500 - return {"success": False, "error": "Token manager not available"} - - if cherrypy.request.method == "DELETE": + return {'success': False, 'error': 'Token manager not available'} + + if cherrypy.request.method == 'DELETE': try: if not token_id: cherrypy.response.status = 400 - return {"success": False, "error": "Token ID is required"} - + return { + 'success': False, + 'error': 'Token ID is required' + } + # Convert to int try: token_id_int = int(token_id) except ValueError: cherrypy.response.status = 400 - return {"success": False, "error": "Invalid token ID"} - + return { + 'success': False, + 'error': 'Invalid token ID' + } + # Revoke the token success = token_manager.revoke_token(token_id_int) - + if success: - logger.info( - f"Revoked API token ID {token_id_int} by user {cherrypy.request.user['username']}" - ) - return {"success": True, "message": "Token revoked successfully"} + logger.info(f"Revoked API token ID {token_id_int} by user {cherrypy.request.user['username']}") + return { + 'success': True, + 'message': 'Token revoked successfully' + } else: cherrypy.response.status = 404 - return {"success": False, "error": "Token not found"} - + return { + 'success': False, + 'error': 'Token not found' + } + except Exception as e: logger.error(f"Token revocation error: {e}") cherrypy.response.status = 500 - return {"success": False, "error": "Failed to revoke token"} + return { + 'success': False, + 'error': 'Failed to revoke token' + } else: raise cherrypy.HTTPError(405, "Method not allowed") @@ -137,314 +156,308 @@ class AuthEndpoints: @cherrypy.expose def login(self, **kwargs): - cherrypy.response.headers["Content-Type"] = "application/json" - + cherrypy.response.headers['Content-Type'] = 'application/json' + # Handle CORS preflight - if cherrypy.request.method == "OPTIONS": - cherrypy.response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS" - cherrypy.response.headers["Access-Control-Allow-Headers"] = ( - "Content-Type, Authorization, X-API-Key" - ) - return b"" - - if cherrypy.request.method != "POST": + if cherrypy.request.method == 'OPTIONS': + cherrypy.response.headers['Access-Control-Allow-Methods'] = 'POST, OPTIONS' + cherrypy.response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-API-Key' + return b'' + + if cherrypy.request.method != 'POST': raise cherrypy.HTTPError(405, "Method not allowed") - + try: # Parse JSON body manually since we can't use json_in decorator with OPTIONS import json - - body = cherrypy.request.body.read().decode("utf-8") + body = cherrypy.request.body.read().decode('utf-8') data = json.loads(body) if body else {} - - username = data.get("username", "").strip() - password = data.get("password", "") - client_id = data.get("client_id", "").strip() - + + username = data.get('username', '').strip() + password = data.get('password', '') + client_id = data.get('client_id', '').strip() + if not username or not password or not client_id: - return json.dumps( - { - "success": False, - "error": "Missing required fields: username, password, client_id", - } - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Missing required fields: username, password, client_id' + }).encode('utf-8') + # Validate credentials against config # Check if username is 'admin' and password matches config - repeater_config = self.config.get("repeater", {}) - security_config = repeater_config.get("security", {}) - config_password = security_config.get("admin_password", "") - + repeater_config = self.config.get('repeater', {}) + security_config = repeater_config.get('security', {}) + config_password = security_config.get('admin_password', '') + # Don't allow login with empty or unconfigured password if not config_password: logger.warning(f"Login attempt rejected - password not configured") - return json.dumps( - { - "success": False, - "error": "System not configured. Please complete setup wizard.", - } - ).encode("utf-8") - - if username == "admin" and password == config_password: + return json.dumps({ + 'success': False, + 'error': 'System not configured. Please complete setup wizard.' + }).encode('utf-8') + + if username == 'admin' and password == config_password: # Create JWT token token = self.jwt_handler.create_jwt(username, client_id) - - logger.info( - f"Successful login for user '{username}' from client '{client_id[:8]}...'" - ) - - return json.dumps( - { - "success": True, - "token": token, - "expires_in": self.jwt_handler.expiry_minutes * 60, - "username": username, - } - ).encode("utf-8") + + logger.info(f"Successful login for user '{username}' from client '{client_id[:8]}...'") + + return json.dumps({ + 'success': True, + 'token': token, + 'expires_in': self.jwt_handler.expiry_minutes * 60, + 'username': username + }).encode('utf-8') else: logger.warning(f"Failed login attempt for user '{username}'") - + # Don't reveal which part was wrong - return json.dumps( - {"success": False, "error": "Invalid username or password"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Invalid username or password' + }).encode('utf-8') + except Exception as e: logger.error(f"Login error: {e}") - return json.dumps({"success": False, "error": "Internal server error"}).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Internal server error' + }).encode('utf-8') + @cherrypy.expose @cherrypy.tools.json_out() @require_auth def verify(self): - if cherrypy.request.method != "GET": + if cherrypy.request.method != 'GET': raise cherrypy.HTTPError(405, "Method not allowed") - - return {"success": True, "authenticated": True, "user": cherrypy.request.user} - + + return { + 'success': True, + 'authenticated': True, + 'user': cherrypy.request.user + } + @cherrypy.expose def refresh(self, **kwargs): - cherrypy.response.headers["Content-Type"] = "application/json" - + cherrypy.response.headers['Content-Type'] = 'application/json' + # Handle CORS preflight - if cherrypy.request.method == "OPTIONS": - cherrypy.response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS" - cherrypy.response.headers["Access-Control-Allow-Headers"] = ( - "Content-Type, Authorization, X-API-Key" - ) - return b"" - - if cherrypy.request.method != "POST": + if cherrypy.request.method == 'OPTIONS': + cherrypy.response.headers['Access-Control-Allow-Methods'] = 'POST, OPTIONS' + cherrypy.response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-API-Key' + return b'' + + if cherrypy.request.method != 'POST': raise cherrypy.HTTPError(405, "Method not allowed") - + try: import json - + # Manual authentication check (can't use @require_auth since we need to handle OPTIONS) - auth_header = cherrypy.request.headers.get("Authorization", "") - api_key = cherrypy.request.headers.get("X-API-Key", "") - - jwt_handler = cherrypy.config.get("jwt_handler") - token_manager = cherrypy.config.get("token_manager") - + auth_header = cherrypy.request.headers.get('Authorization', '') + api_key = cherrypy.request.headers.get('X-API-Key', '') + + jwt_handler = cherrypy.config.get('jwt_handler') + token_manager = cherrypy.config.get('token_manager') + user_info = None - + # Check JWT first - if auth_header.startswith("Bearer "): + if auth_header.startswith('Bearer '): token = auth_header[7:] payload = jwt_handler.verify_jwt(token) if payload: user_info = { - "username": payload["sub"], - "client_id": payload.get("client_id"), - "auth_method": "jwt", + 'username': payload['sub'], + 'client_id': payload.get('client_id'), + 'auth_method': 'jwt' } - + # Check API token if not user_info and api_key: token_data = token_manager.verify_token(api_key) if token_data: user_info = { - "username": "admin", - "token_id": token_data["id"], - "auth_method": "api_token", + 'username': 'admin', + 'token_id': token_data['id'], + 'auth_method': 'api_token' } - + if not user_info: - return json.dumps( - {"success": False, "error": "Unauthorized - Valid JWT or API token required"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Unauthorized - Valid JWT or API token required' + }).encode('utf-8') + # Parse request body - body = cherrypy.request.body.read().decode("utf-8") + body = cherrypy.request.body.read().decode('utf-8') data = json.loads(body) if body else {} - - client_id = data.get("client_id", user_info.get("client_id", "")).strip() - + + client_id = data.get('client_id', user_info.get('client_id', '')).strip() + if not client_id: - return json.dumps({"success": False, "error": "Client ID is required"}).encode( - "utf-8" - ) - + return json.dumps({ + 'success': False, + 'error': 'Client ID is required' + }).encode('utf-8') + # Create new JWT token (refreshes expiry time) - new_token = self.jwt_handler.create_jwt(user_info["username"], client_id) - - logger.info( - f"Token refreshed for user '{user_info['username']}' from client '{client_id[:8]}...'" - ) - - return json.dumps( - { - "success": True, - "token": new_token, - "expires_in": self.jwt_handler.expiry_minutes * 60, - "username": user_info["username"], - } - ).encode("utf-8") - + new_token = self.jwt_handler.create_jwt(user_info['username'], client_id) + + logger.info(f"Token refreshed for user '{user_info['username']}' from client '{client_id[:8]}...'") + + return json.dumps({ + 'success': True, + 'token': new_token, + 'expires_in': self.jwt_handler.expiry_minutes * 60, + 'username': user_info['username'] + }).encode('utf-8') + except Exception as e: logger.error(f"Token refresh error: {e}") - return json.dumps({"success": False, "error": "Failed to refresh token"}).encode( - "utf-8" - ) - + return json.dumps({ + 'success': False, + 'error': 'Failed to refresh token' + }).encode('utf-8') + @cherrypy.expose def change_password(self): import json - - cherrypy.response.headers["Content-Type"] = "application/json" - + + cherrypy.response.headers['Content-Type'] = 'application/json' + # Handle CORS preflight - if cherrypy.request.method == "OPTIONS": - cherrypy.response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS" - cherrypy.response.headers["Access-Control-Allow-Headers"] = ( - "Content-Type, Authorization, X-API-Key" - ) - return b"" - - if cherrypy.request.method != "POST": + if cherrypy.request.method == 'OPTIONS': + cherrypy.response.headers['Access-Control-Allow-Methods'] = 'POST, OPTIONS' + cherrypy.response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-API-Key' + return b'' + + if cherrypy.request.method != 'POST': raise cherrypy.HTTPError(405, "Method not allowed") - + # Require authentication for POST # Get auth handlers from global cherrypy config - jwt_handler = cherrypy.config.get("jwt_handler") - token_manager = cherrypy.config.get("token_manager") - + jwt_handler = cherrypy.config.get('jwt_handler') + token_manager = cherrypy.config.get('token_manager') + if not jwt_handler or not token_manager: logger.error("Auth handlers not configured") raise cherrypy.HTTPError(500, "Authentication not configured") - + # Try JWT authentication first - auth_header = cherrypy.request.headers.get("Authorization", "") + auth_header = cherrypy.request.headers.get('Authorization', '') user = None - - if auth_header.startswith("Bearer "): + + if auth_header.startswith('Bearer '): token = auth_header[7:] # Remove 'Bearer ' prefix payload = jwt_handler.verify_jwt(token) - + if payload: user = { - "username": payload["sub"], - "client_id": payload["client_id"], - "auth_type": "jwt", + 'username': payload['sub'], + 'client_id': payload['client_id'], + 'auth_type': 'jwt' } - + # Try API token authentication if JWT failed if not user: - api_key = cherrypy.request.headers.get("X-API-Key", "") + api_key = cherrypy.request.headers.get('X-API-Key', '') if api_key: token_info = token_manager.verify_token(api_key) - + if token_info: user = { - "username": "api_token", - "token_name": token_info["name"], - "token_id": token_info["id"], - "auth_type": "api_token", + 'username': 'api_token', + 'token_name': token_info['name'], + 'token_id': token_info['id'], + 'auth_type': 'api_token' } - + if not user: cherrypy.response.status = 401 - return json.dumps( - {"success": False, "error": "Unauthorized - Valid JWT or API token required"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Unauthorized - Valid JWT or API token required' + }).encode('utf-8') + try: # Parse JSON body manually - body = cherrypy.request.body.read().decode("utf-8") + body = cherrypy.request.body.read().decode('utf-8') data = json.loads(body) if body else {} - - current_password = data.get("current_password", "") - new_password = data.get("new_password", "") - + + current_password = data.get('current_password', '') + new_password = data.get('new_password', '') + if not current_password or not new_password: cherrypy.response.status = 400 - return json.dumps( - { - "success": False, - "error": "Both current_password and new_password are required", - } - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Both current_password and new_password are required' + }).encode('utf-8') + # Validate new password strength if len(new_password) < 8: cherrypy.response.status = 400 - return json.dumps( - {"success": False, "error": "New password must be at least 8 characters long"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'New password must be at least 8 characters long' + }).encode('utf-8') + # Verify current password - repeater_config = self.config.get("repeater", {}) - security_config = repeater_config.get("security", {}) - config_password = security_config.get("admin_password", "") - + repeater_config = self.config.get('repeater', {}) + security_config = repeater_config.get('security', {}) + config_password = security_config.get('admin_password', '') + if not config_password: cherrypy.response.status = 500 - return json.dumps({"success": False, "error": "System configuration error"}).encode( - "utf-8" - ) - + return json.dumps({ + 'success': False, + 'error': 'System configuration error' + }).encode('utf-8') + if current_password != config_password: cherrypy.response.status = 401 - return json.dumps( - {"success": False, "error": "Current password is incorrect"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Current password is incorrect' + }).encode('utf-8') + # Update password in config - if "repeater" not in self.config: - self.config["repeater"] = {} - if "security" not in self.config["repeater"]: - self.config["repeater"]["security"] = {} - - self.config["repeater"]["security"]["admin_password"] = new_password - + if 'repeater' not in self.config: + self.config['repeater'] = {} + if 'security' not in self.config['repeater']: + self.config['repeater']['security'] = {} + + self.config['repeater']['security']['admin_password'] = new_password + # Save to config file using ConfigManager if self.config_manager: - saved, _ = self.config_manager.save_to_file() - if saved: + if self.config_manager.save_to_file(): logger.info(f"Admin password changed successfully by user {user['username']}") - return json.dumps( - { - "success": True, - "message": "Password changed successfully. Please log in again with your new password.", - } - ).encode("utf-8") + return json.dumps({ + 'success': True, + 'message': 'Password changed successfully. Please log in again with your new password.' + }).encode('utf-8') else: cherrypy.response.status = 500 - return json.dumps( - {"success": False, "error": "Failed to save password to config file"} - ).encode("utf-8") + return json.dumps({ + 'success': False, + 'error': 'Failed to save password to config file' + }).encode('utf-8') else: cherrypy.response.status = 500 - return json.dumps( - {"success": False, "error": "Config manager not available"} - ).encode("utf-8") - + return json.dumps({ + 'success': False, + 'error': 'Config manager not available' + }).encode('utf-8') + except Exception as e: logger.error(f"Password change error: {e}") cherrypy.response.status = 500 - return json.dumps({"success": False, "error": "Failed to change password"}).encode( - "utf-8" - ) + return json.dumps({ + 'success': False, + 'error': 'Failed to change password' + }).encode('utf-8') \ No newline at end of file