Address code review findings
All checks were successful
Build and Push Docker Image / build (push) Successful in 8s
All checks were successful
Build and Push Docker Image / build (push) Successful in 8s
- Fix infinite recursion on re-auth by adding attempt counter - Add endpoint validation to unifi_api_call (block path traversal, require /api/ prefix) - Clean up redundant SSL context creation - Add safe port parsing with fallback and warning log
This commit is contained in:
30
server.py
30
server.py
@@ -32,12 +32,22 @@ logging.basicConfig(
|
||||
)
|
||||
logger = logging.getLogger("unifi-mcp-light")
|
||||
|
||||
|
||||
def _parse_port(value: str, default: int = 443) -> int:
|
||||
"""Safely parse port from environment variable."""
|
||||
try:
|
||||
return int(value)
|
||||
except (ValueError, TypeError):
|
||||
logger.warning(f"Invalid UNIFI_PORT '{value}', using default {default}")
|
||||
return default
|
||||
|
||||
|
||||
# Configuration from environment
|
||||
CONFIG = {
|
||||
"host": os.getenv("UNIFI_HOST", ""),
|
||||
"username": os.getenv("UNIFI_USERNAME", ""),
|
||||
"password": os.getenv("UNIFI_PASSWORD", ""),
|
||||
"port": int(os.getenv("UNIFI_PORT", "443")),
|
||||
"port": _parse_port(os.getenv("UNIFI_PORT", "443")),
|
||||
"site": os.getenv("UNIFI_SITE", "default"),
|
||||
"verify_ssl": os.getenv("UNIFI_VERIFY_SSL", "false").lower() == "true",
|
||||
"allow_writes": os.getenv("UNIFI_ALLOW_WRITES", "false").lower() == "true",
|
||||
@@ -324,15 +334,27 @@ async def unifi_api_call(
|
||||
params: "{}"
|
||||
|
||||
# Block a client (requires UNIFI_ALLOW_WRITES=true)
|
||||
endpoint: "/api/s/default/cmd/stamgr"
|
||||
method: "POST"
|
||||
params: '{"cmd": "block-sta", "mac": "aa:bb:cc:dd:ee:ff"}'
|
||||
endpoint: \"/api/s/default/cmd/stamgr\"
|
||||
method: \"POST\"
|
||||
params: '{\"cmd\": \"block-sta\", \"mac\": \"aa:bb:cc:dd:ee:ff\"}'
|
||||
"""
|
||||
try:
|
||||
# Validate endpoint to prevent path traversal attacks
|
||||
if ".." in endpoint:
|
||||
return json.dumps({"error": "Invalid endpoint: path traversal not allowed"})
|
||||
if not endpoint.startswith("/api/"):
|
||||
return json.dumps(
|
||||
{
|
||||
"error": "Invalid endpoint: must start with /api/",
|
||||
"hint": "Example: /api/s/default/stat/sta",
|
||||
}
|
||||
)
|
||||
|
||||
# Parse params JSON
|
||||
try:
|
||||
payload = json.loads(params) if params and params != "{}" else None
|
||||
except json.JSONDecodeError as e:
|
||||
logger.warning(f"Invalid params JSON in unifi_api_call: {e}")
|
||||
return json.dumps({"error": f"Invalid params JSON: {e}"})
|
||||
|
||||
client = await get_client()
|
||||
|
||||
Reference in New Issue
Block a user