adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-07 09:23:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update login, path and req response types.

Browse Source 
- Simplified KISS modem setup instructions in README.md by removing unnecessary details.
- Refactored ConfigManager in config_manager.py to improve code clarity and efficiency, including changes to the save_to_file method and live_update_daemon method.
- Updated logging and error handling for better debugging and maintenance.
- Adjusted method signatures for consistency and clarity across the ConfigManager class.
- Modified device_version in frame_server.py to use FIRMWARE_VER_CODE from pyMC_core for better version management.
- Enhanced login.py and protocol_request.py with additional payload type handling and logging improvements.
- Cleaned up auth_endpoints.py for better readability and consistency in response formatting.
This commit is contained in:
agessaman
2026-02-21 21:19:22 -08:00
parent e9841adeff
commit f1a944222f
6 changed files with 352 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -6
View File
@@ -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://<repeater-ip>: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:
repeater/companion/frame_server.py
+1 -1
View File
@@ -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,
repeater/config_manager.py
+70 -75
View File
@@ -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 []
}
repeater/handler_helpers/login.py
+7 -3
View File
@@ -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:
repeater/handler_helpers/protocol_request.py
+1
View File
@@ -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,
repeater/web/auth_endpoints.py
+272 -259
View File
@@ -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')