Fix security vulnerabilities and add validation

- Fix XSS vulnerability by using data attributes instead of inline
  onclick handlers in node_tags.html template
- Fix URL injection by using urlencode for all redirect URL parameters
- Add validation to reject moves where source and destination nodes
  are the same (returns 400 Bad Request)
- Add error handling for response.json() calls that may fail
- Add missing test coverage for update endpoint error scenarios
This commit is contained in:
Claude
2026-01-11 11:51:57 +00:00
parent 367f838371
commit d8a0f2abb8
5 changed files with 197 additions and 59 deletions
+7
View File
@@ -144,6 +144,13 @@ async def move_node_tag(
data: NodeTagMove,
) -> NodeTagRead:
"""Move a node tag to a different node."""
# Check if source and destination are the same
if public_key == data.new_public_key:
raise HTTPException(
status_code=400,
detail="Source and destination nodes are the same",
)
# Find source node
source_query = select(Node).where(Node.public_key == public_key)
source_node = session.execute(source_query).scalar_one_or_none()
+73 -37
View File
@@ -1,13 +1,40 @@
"""Admin page routes."""
import logging
from typing import Optional
from typing import Any, Optional
from urllib.parse import urlencode
from fastapi import APIRouter, Form, HTTPException, Query, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from httpx import Response
from meshcore_hub.web.app import get_network_context, get_templates
def _build_redirect_url(
public_key: str,
message: Optional[str] = None,
error: Optional[str] = None,
) -> str:
"""Build a properly encoded redirect URL with optional message/error."""
params: dict[str, str] = {"public_key": public_key}
if message:
params["message"] = message
if error:
params["error"] = error
return f"/a/node-tags?{urlencode(params)}"
def _get_error_detail(response: Response) -> str:
"""Safely extract error detail from response JSON."""
try:
data: Any = response.json()
detail: str = data.get("detail", "Unknown error")
return detail
except Exception:
return "Unknown error"
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/a", tags=["admin"])
@@ -116,8 +143,6 @@ async def admin_create_node_tag(
"""Create a new node tag."""
_check_admin_enabled(request)
redirect_url = f"/a/node-tags?public_key={public_key}"
try:
response = await request.app.state.http_client.post(
f"/api/v1/nodes/{public_key}/tags",
@@ -128,17 +153,22 @@ async def admin_create_node_tag(
},
)
if response.status_code == 201:
redirect_url += f"&message=Tag '{key}' created successfully"
redirect_url = _build_redirect_url(
public_key, message=f"Tag '{key}' created successfully"
)
elif response.status_code == 409:
redirect_url += f"&error=Tag '{key}' already exists"
redirect_url = _build_redirect_url(
public_key, error=f"Tag '{key}' already exists"
)
elif response.status_code == 404:
redirect_url += "&error=Node not found"
redirect_url = _build_redirect_url(public_key, error="Node not found")
else:
detail = response.json().get("detail", "Unknown error")
redirect_url += f"&error={detail}"
redirect_url = _build_redirect_url(
public_key, error=_get_error_detail(response)
)
except Exception as e:
logger.exception("Failed to create tag: %s", e)
redirect_url += "&error=Failed to create tag"
redirect_url = _build_redirect_url(public_key, error="Failed to create tag")
return RedirectResponse(url=redirect_url, status_code=303)
@@ -154,8 +184,6 @@ async def admin_update_node_tag(
"""Update an existing node tag."""
_check_admin_enabled(request)
redirect_url = f"/a/node-tags?public_key={public_key}"
try:
response = await request.app.state.http_client.put(
f"/api/v1/nodes/{public_key}/tags/{key}",
@@ -165,15 +193,20 @@ async def admin_update_node_tag(
},
)
if response.status_code == 200:
redirect_url += f"&message=Tag '{key}' updated successfully"
redirect_url = _build_redirect_url(
public_key, message=f"Tag '{key}' updated successfully"
)
elif response.status_code == 404:
redirect_url += f"&error=Tag '{key}' not found"
redirect_url = _build_redirect_url(
public_key, error=f"Tag '{key}' not found"
)
else:
detail = response.json().get("detail", "Unknown error")
redirect_url += f"&error={detail}"
redirect_url = _build_redirect_url(
public_key, error=_get_error_detail(response)
)
except Exception as e:
logger.exception("Failed to update tag: %s", e)
redirect_url += "&error=Failed to update tag"
redirect_url = _build_redirect_url(public_key, error="Failed to update tag")
return RedirectResponse(url=redirect_url, status_code=303)
@@ -188,32 +221,32 @@ async def admin_move_node_tag(
"""Move a node tag to a different node."""
_check_admin_enabled(request)
# Redirect to the destination node after move
redirect_url = f"/a/node-tags?public_key={new_public_key}"
try:
response = await request.app.state.http_client.put(
f"/api/v1/nodes/{public_key}/tags/{key}/move",
json={"new_public_key": new_public_key},
)
if response.status_code == 200:
redirect_url += f"&message=Tag '{key}' moved successfully"
# Redirect to the destination node after successful move
redirect_url = _build_redirect_url(
new_public_key, message=f"Tag '{key}' moved successfully"
)
elif response.status_code == 404:
# Stay on source node if not found
redirect_url = f"/a/node-tags?public_key={public_key}"
detail = response.json().get("detail", "Not found")
redirect_url += f"&error={detail}"
redirect_url = _build_redirect_url(
public_key, error=_get_error_detail(response)
)
elif response.status_code == 409:
redirect_url = f"/a/node-tags?public_key={public_key}"
redirect_url += f"&error=Tag '{key}' already exists on destination node"
redirect_url = _build_redirect_url(
public_key, error=f"Tag '{key}' already exists on destination node"
)
else:
redirect_url = f"/a/node-tags?public_key={public_key}"
detail = response.json().get("detail", "Unknown error")
redirect_url += f"&error={detail}"
redirect_url = _build_redirect_url(
public_key, error=_get_error_detail(response)
)
except Exception as e:
logger.exception("Failed to move tag: %s", e)
redirect_url = f"/a/node-tags?public_key={public_key}"
redirect_url += "&error=Failed to move tag"
redirect_url = _build_redirect_url(public_key, error="Failed to move tag")
return RedirectResponse(url=redirect_url, status_code=303)
@@ -227,21 +260,24 @@ async def admin_delete_node_tag(
"""Delete a node tag."""
_check_admin_enabled(request)
redirect_url = f"/a/node-tags?public_key={public_key}"
try:
response = await request.app.state.http_client.delete(
f"/api/v1/nodes/{public_key}/tags/{key}",
)
if response.status_code == 204:
redirect_url += f"&message=Tag '{key}' deleted successfully"
redirect_url = _build_redirect_url(
public_key, message=f"Tag '{key}' deleted successfully"
)
elif response.status_code == 404:
redirect_url += f"&error=Tag '{key}' not found"
redirect_url = _build_redirect_url(
public_key, error=f"Tag '{key}' not found"
)
else:
detail = response.json().get("detail", "Unknown error")
redirect_url += f"&error={detail}"
redirect_url = _build_redirect_url(
public_key, error=_get_error_detail(response)
)
except Exception as e:
logger.exception("Failed to delete tag: %s", e)
redirect_url += "&error=Failed to delete tag"
redirect_url = _build_redirect_url(public_key, error="Failed to delete tag")
return RedirectResponse(url=redirect_url, status_code=303)
@@ -95,7 +95,7 @@
</thead>
<tbody>
{% for tag in tags %}
<tr>
<tr data-tag-key="{{ tag.key }}" data-tag-value="{{ tag.value or '' }}" data-tag-type="{{ tag.value_type }}">
<td class="font-mono font-semibold">{{ tag.key }}</td>
<td class="max-w-xs truncate" title="{{ tag.value or '' }}">{{ tag.value or '-' }}</td>
<td>
@@ -104,13 +104,13 @@
<td class="text-sm opacity-70">{{ tag.updated_at[:10] if tag.updated_at else '-' }}</td>
<td>
<div class="flex gap-1">
<button class="btn btn-ghost btn-xs" onclick="openEditModal('{{ tag.key }}', '{{ tag.value|e if tag.value else '' }}', '{{ tag.value_type }}')">
<button class="btn btn-ghost btn-xs btn-edit">
Edit
</button>
<button class="btn btn-ghost btn-xs" onclick="openMoveModal('{{ tag.key }}')">
<button class="btn btn-ghost btn-xs btn-move">
Move
</button>
<button class="btn btn-ghost btn-xs text-error" onclick="openDeleteModal('{{ tag.key }}')">
<button class="btn btn-ghost btn-xs text-error btn-delete">
Delete
</button>
</div>
@@ -316,25 +316,45 @@
{% block extra_scripts %}
<script>
function openEditModal(key, value, valueType) {
document.getElementById('editKey').value = key;
document.getElementById('editKeyDisplay').value = key;
document.getElementById('editValue').value = value;
document.getElementById('editValueType').value = valueType;
editModal.showModal();
}
// Use event delegation to handle button clicks safely
document.addEventListener('DOMContentLoaded', function() {
// Edit button handler
document.querySelectorAll('.btn-edit').forEach(function(btn) {
btn.addEventListener('click', function() {
var row = this.closest('tr');
var key = row.dataset.tagKey;
var value = row.dataset.tagValue;
var valueType = row.dataset.tagType;
document.getElementById('editKey').value = key;
document.getElementById('editKeyDisplay').value = key;
document.getElementById('editValue').value = value;
document.getElementById('editValueType').value = valueType;
editModal.showModal();
});
});
function openMoveModal(key) {
document.getElementById('moveKey').value = key;
document.getElementById('moveKeyDisplay').value = key;
document.getElementById('moveDestination').selectedIndex = 0;
moveModal.showModal();
}
// Move button handler
document.querySelectorAll('.btn-move').forEach(function(btn) {
btn.addEventListener('click', function() {
var row = this.closest('tr');
var key = row.dataset.tagKey;
document.getElementById('moveKey').value = key;
document.getElementById('moveKeyDisplay').value = key;
document.getElementById('moveDestination').selectedIndex = 0;
moveModal.showModal();
});
});
function openDeleteModal(key) {
document.getElementById('deleteKey').value = key;
document.getElementById('deleteKeyDisplay').textContent = key;
deleteModal.showModal();
}
// Delete button handler
document.querySelectorAll('.btn-delete').forEach(function(btn) {
btn.addEventListener('click', function() {
var row = this.closest('tr');
var key = row.dataset.tagKey;
document.getElementById('deleteKey').value = key;
document.getElementById('deleteKeyDisplay').textContent = key;
deleteModal.showModal();
});
});
});
</script>
{% endblock %}
+33
View File
@@ -288,3 +288,36 @@ class TestMoveNodeTag:
headers={"Authorization": "Bearer test-read-key"},
)
assert response.status_code == 403
def test_move_node_tag_same_node(self, client_no_auth, api_db_session):
"""Test moving a tag to the same node returns 400."""
from datetime import datetime, timezone
from meshcore_hub.common.models import Node, NodeTag
# Create node with 64-char public key
full_key = "abc123def456abc123def456abc123deabc123def456abc123def456abc123de"
node = Node(
public_key=full_key,
name="Test Node 64",
adv_type="REPEATER",
first_seen=datetime.now(timezone.utc),
)
api_db_session.add(node)
api_db_session.commit()
# Create tag
tag = NodeTag(
node_id=node.id,
key="test_tag",
value="test_value",
)
api_db_session.add(tag)
api_db_session.commit()
response = client_no_auth.put(
f"/api/v1/nodes/{full_key}/tags/test_tag/move",
json={"new_public_key": full_key},
)
assert response.status_code == 400
assert "same" in response.json()["detail"].lower()
+42
View File
@@ -298,6 +298,48 @@ class TestAdminUpdateTag:
assert "message=" in response.headers["location"]
assert "updated" in response.headers["location"]
def test_update_tag_not_found(
self, admin_app, mock_http_client_admin: MockHttpClient
):
"""Test updating a non-existent tag returns error."""
# Set up 404 response for this specific tag
mock_http_client_admin.set_response(
"PUT",
"/api/v1/nodes/abc123def456abc123def456abc123de/tags/nonexistent",
404,
{"detail": "Tag not found"},
)
admin_app.state.http_client = mock_http_client_admin
client = TestClient(admin_app, raise_server_exceptions=True)
response = client.post(
"/a/node-tags/update",
data={
"public_key": "abc123def456abc123def456abc123de",
"key": "nonexistent",
"value": "value",
"value_type": "string",
},
follow_redirects=False,
)
assert response.status_code == 303
assert "error=" in response.headers["location"]
assert "not+found" in response.headers["location"].lower()
def test_update_tag_disabled(self, admin_client_disabled):
"""Test updating tag when admin is disabled."""
response = admin_client_disabled.post(
"/a/node-tags/update",
data={
"public_key": "abc123def456abc123def456abc123de",
"key": "environment",
"value": "staging",
"value_type": "string",
},
follow_redirects=False,
)
assert response.status_code == 404
class TestAdminMoveTag:
"""Tests for moving node tags."""