fix: ADR-080 P0 security + CI remediation from QE analysis

Address all 5 P0 issues from QE analysis (55/100 score):

- P0-1: Rate limiter bypass — validate X-Forwarded-For against trusted proxy list
- P0-2: Exception detail leak — generic 500 messages, exception_type gated by dev mode
- P0-3: WebSocket JWT in URL (CWE-598) — first-message auth pattern replaces query param
- P0-4: Rust tests not in CI — add rust-tests job gating docker-build and notify
- P0-5: WebSocket path mismatch — use WS_PATH constant instead of hardcoded /ws/sensing

Includes ADR-080 remediation plan and 9 QE reports (4,914 lines).
Firmware validated on ESP32-S3 (COM8): CSI collecting, calibration OK.

Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv
2026-04-06 16:12:13 -04:00
parent b5e924cd72
commit 924c32547e
17 changed files with 5169 additions and 68 deletions
+9 -9
View File
@@ -137,7 +137,7 @@ async def get_current_pose_estimation(
logger.error(f"Error in pose estimation: {e}")
raise HTTPException(
status_code=500,
detail=f"Pose estimation failed: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -174,7 +174,7 @@ async def analyze_pose_data(
logger.error(f"Error in pose analysis: {e}")
raise HTTPException(
status_code=500,
detail=f"Pose analysis failed: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -208,7 +208,7 @@ async def get_zone_occupancy(
logger.error(f"Error getting zone occupancy: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get zone occupancy: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -232,7 +232,7 @@ async def get_zones_summary(
logger.error(f"Error getting zones summary: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get zones summary: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -285,7 +285,7 @@ async def get_historical_data(
logger.error(f"Error getting historical data: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get historical data: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -313,7 +313,7 @@ async def get_detected_activities(
logger.error(f"Error getting activities: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get activities: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -357,7 +357,7 @@ async def calibrate_pose_system(
logger.error(f"Error starting calibration: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to start calibration: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -383,7 +383,7 @@ async def get_calibration_status(
logger.error(f"Error getting calibration status: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get calibration status: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -416,5 +416,5 @@ async def get_pose_statistics(
logger.error(f"Error getting statistics: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get statistics: {str(e)}"
detail="An internal error occurred. Please try again later."
)
+89 -31
View File
@@ -2,6 +2,7 @@
WebSocket streaming API endpoints
"""
import asyncio
import json
import logging
from typing import Dict, List, Optional, Any
@@ -71,26 +72,55 @@ async def websocket_pose_stream(
zone_ids: Optional[str] = Query(None, description="Comma-separated zone IDs"),
min_confidence: float = Query(0.5, ge=0.0, le=1.0),
max_fps: int = Query(30, ge=1, le=60),
token: Optional[str] = Query(None, description="Authentication token")
):
"""WebSocket endpoint for real-time pose data streaming."""
client_id = None
try:
# Accept WebSocket connection
await websocket.accept()
# Check authentication if enabled
# First-message authentication (CWE-598 fix: no JWT in URL)
from src.config.settings import get_settings
settings = get_settings()
if settings.enable_authentication and not token:
await websocket.send_json({
"type": "error",
"message": "Authentication token required"
})
await websocket.close(code=1008)
return
if settings.enable_authentication:
try:
raw = await asyncio.wait_for(websocket.receive_text(), timeout=10.0)
auth_msg = json.loads(raw)
if auth_msg.get("type") != "auth" or not auth_msg.get("token"):
await websocket.send_json({
"type": "error",
"message": "First message must be {\"type\": \"auth\", \"token\": \"<jwt>\"}"
})
await websocket.close(code=1008)
return
# Verify the token
from src.middleware.auth import get_auth_middleware
auth_middleware = get_auth_middleware(settings)
try:
auth_middleware.token_manager.verify_token(auth_msg["token"])
except Exception:
await websocket.send_json({
"type": "error",
"message": "Invalid or expired authentication token"
})
await websocket.close(code=1008)
return
except asyncio.TimeoutError:
await websocket.send_json({
"type": "error",
"message": "Authentication timeout: no auth message received within 10 seconds"
})
await websocket.close(code=1008)
return
except (json.JSONDecodeError, Exception) as e:
await websocket.send_json({
"type": "error",
"message": "Invalid authentication message format"
})
await websocket.close(code=1008)
return
# Parse zone IDs
zone_list = None
@@ -157,25 +187,53 @@ async def websocket_events_stream(
websocket: WebSocket,
event_types: Optional[str] = Query(None, description="Comma-separated event types"),
zone_ids: Optional[str] = Query(None, description="Comma-separated zone IDs"),
token: Optional[str] = Query(None, description="Authentication token")
):
"""WebSocket endpoint for real-time event streaming."""
client_id = None
try:
await websocket.accept()
# Check authentication if enabled
# First-message authentication (CWE-598 fix: no JWT in URL)
from src.config.settings import get_settings
settings = get_settings()
if settings.enable_authentication and not token:
await websocket.send_json({
"type": "error",
"message": "Authentication token required"
})
await websocket.close(code=1008)
return
if settings.enable_authentication:
try:
raw = await asyncio.wait_for(websocket.receive_text(), timeout=10.0)
auth_msg = json.loads(raw)
if auth_msg.get("type") != "auth" or not auth_msg.get("token"):
await websocket.send_json({
"type": "error",
"message": "First message must be {\"type\": \"auth\", \"token\": \"<jwt>\"}"
})
await websocket.close(code=1008)
return
from src.middleware.auth import get_auth_middleware
auth_middleware = get_auth_middleware(settings)
try:
auth_middleware.token_manager.verify_token(auth_msg["token"])
except Exception:
await websocket.send_json({
"type": "error",
"message": "Invalid or expired authentication token"
})
await websocket.close(code=1008)
return
except asyncio.TimeoutError:
await websocket.send_json({
"type": "error",
"message": "Authentication timeout: no auth message received within 10 seconds"
})
await websocket.close(code=1008)
return
except (json.JSONDecodeError, Exception) as e:
await websocket.send_json({
"type": "error",
"message": "Invalid authentication message format"
})
await websocket.close(code=1008)
return
# Parse parameters
event_list = None
@@ -294,7 +352,7 @@ async def get_stream_status(
logger.error(f"Error getting stream status: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get stream status: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -324,7 +382,7 @@ async def start_streaming(
logger.error(f"Error starting streaming: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to start streaming: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -349,7 +407,7 @@ async def stop_streaming(
logger.error(f"Error stopping streaming: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to stop streaming: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -371,7 +429,7 @@ async def get_connected_clients(
logger.error(f"Error getting connected clients: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get connected clients: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -403,7 +461,7 @@ async def disconnect_client(
logger.error(f"Error disconnecting client: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to disconnect client: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -442,7 +500,7 @@ async def broadcast_message(
logger.error(f"Error broadcasting message: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to broadcast message: {str(e)}"
detail="An internal error occurred. Please try again later."
)
@@ -461,5 +519,5 @@ async def get_streaming_metrics():
logger.error(f"Error getting streaming metrics: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get streaming metrics: {str(e)}"
detail="An internal error occurred. Please try again later."
)
+1 -7
View File
@@ -237,13 +237,7 @@ class AuthenticationMiddleware:
"""Authenticate the request and return user info."""
# Try to get token from Authorization header
authorization = request.headers.get("Authorization")
if not authorization:
# For WebSocket connections, try to get token from query parameters
if request.url.path.startswith("/ws"):
token = request.query_params.get("token")
if token:
authorization = f"Bearer {token}"
if not authorization:
if self._requires_auth(request):
raise AuthenticationError("Missing authorization header")
+3 -4
View File
@@ -202,11 +202,10 @@ class ErrorHandler:
)
# Determine error details
details = {
"exception_type": type(exc).__name__,
}
details = {}
if self.include_traceback:
details["exception_type"] = type(exc).__name__
details["traceback"] = traceback.format_exception(
type(exc), exc, exc.__traceback__
)
+25 -13
View File
@@ -5,7 +5,7 @@ Rate limiting middleware for WiFi-DensePose API
import asyncio
import logging
import time
from typing import Dict, Any, Optional, Callable, Tuple
from typing import Dict, Any, Optional, Callable, Set, Tuple
from datetime import datetime, timedelta
from collections import defaultdict, deque
from dataclasses import dataclass
@@ -128,6 +128,11 @@ class RateLimiter:
self.authenticated_limit = settings.rate_limit_authenticated_requests
self.window_size = settings.rate_limit_window
# Trusted proxy IPs — only trust X-Forwarded-For/X-Real-IP from these
self.trusted_proxies: Set[str] = set(
getattr(settings, "trusted_proxies", [])
)
# Storage for rate limit data
self._sliding_windows: Dict[str, SlidingWindowCounter] = {}
self._token_buckets: Dict[str, TokenBucket] = {}
@@ -196,18 +201,25 @@ class RateLimiter:
return f"ip:{client_ip}"
def _get_client_ip(self, request: Request) -> str:
"""Get client IP address."""
# Check for forwarded headers
forwarded_for = request.headers.get("X-Forwarded-For")
if forwarded_for:
return forwarded_for.split(",")[0].strip()
real_ip = request.headers.get("X-Real-IP")
if real_ip:
return real_ip
# Fall back to direct connection
return request.client.host if request.client else "unknown"
"""Get client IP address.
Only trusts X-Forwarded-For / X-Real-IP when the direct connection
originates from a known trusted proxy. This prevents clients from
spoofing forwarded headers to bypass rate limiting.
"""
connection_ip = request.client.host if request.client else "unknown"
# Only honour forwarded headers from trusted proxies
if connection_ip in self.trusted_proxies:
forwarded_for = request.headers.get("X-Forwarded-For")
if forwarded_for:
return forwarded_for.split(",")[0].strip()
real_ip = request.headers.get("X-Real-IP")
if real_ip:
return real_ip
return connection_ip
def _get_rate_limit(self, request: Request) -> int:
"""Get rate limit for request."""