chore: secure dashboard proxy endpoint and decouple lan checks

dashboard/backend/config.py

@@ -46,3 +46,17 @@ CHAT_SUMMARY_DB = os.environ.get("CHAT_SUMMARY_DB", "/app/data/chat-summarizer/c
 
 # OpenClaw
 OPENCLAW_GATEWAY_TOKEN = os.environ.get("OPENCLAW_GATEWAY_TOKEN", "")
+
+# Networking configs
+LAN_IP_RANGES = os.environ.get("LAN_IP_RANGES", "192.168.31.0/24,100.64.0.0/10").split(",")
+
+def is_lan_ip(ip_str: str) -> bool:
+    import ipaddress
+    try:
+        ip_obj = ipaddress.ip_address(ip_str)
+        for subnet_str in LAN_IP_RANGES:
+            if ip_obj in ipaddress.ip_network(subnet_str.strip()):
+                return True
+        return False
+    except ValueError:
+        return False

dashboard/backend/main.py

@@ -41,7 +41,7 @@ async def security_headers(request: Request, call_next):
     response.headers["X-Frame-Options"] = "DENY"
     response.headers["X-XSS-Protection"] = "1; mode=block"
     response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
-    response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self' wss://nas.jimmygan.com wss://nas.jimmygan.com:8443; frame-ancestors 'none'"
+    response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self' wss: ws:; frame-ancestors 'none'"
     return response
 
 # Audit logger
@@ -66,6 +66,8 @@ async def audit_log(request: Request, call_next):
         if auth_header.startswith("Bearer "):
             payload = jwt.decode(auth_header[7:], SECRET_KEY, algorithms=[ALGORITHM])
             user = payload.get("sub", "-")
+    except jwt.PyJWTError:
+        user = "Invalid/Malformed Token"
     except Exception:
         pass
     _audit_logger.info(
@@ -116,8 +118,13 @@ async def health():
 
 @app.get("/api/client-ip")
 async def client_ip(request: Request):
-    ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() or request.client.host
-    lan = ip.startswith("192.168.31.") or (ip.startswith("100.") and _router_reachable())
+    ip = request.client.host
+    if ip in ("127.0.0.1", "::1"):
+        forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+        if forwarded:
+            ip = forwarded
+    import config
+    lan = config.is_lan_ip(ip) if not ip.startswith("100.") else config.is_lan_ip(ip) and _router_reachable()
     return {"lan": lan}
 
 def _router_reachable():

dashboard/backend/routers/auth.py

@@ -113,6 +113,11 @@ async def login(creds: LoginRequest, request: Request):
 async def proxy_auth(request: Request):
     """Auto-login when Authelia has already authenticated the user via forward-auth.
     Caddy sets Remote-User header after successful Authelia 2FA verification."""
+    # Prevent IP spoofing: ensure proxy traffic is from localhost (Caddy)
+    if request.client.host not in ("127.0.0.1", "::1"):
+        logger.warning(f"Rejected proxy auth attempt from unauthorized IP: {request.client.host}")
+        raise HTTPException(status_code=403, detail="Unauthorized proxy source")
+
     remote_user = request.headers.get("Remote-User")
     if not remote_user:
         raise HTTPException(status_code=401, detail="No proxy auth header")
@@ -199,13 +204,25 @@ async def change_password(req: ChangePasswordRequest, request: Request, current_
 _challenges = {}
 
 def _store_challenge(challenge: bytes):
-    _challenges["current"] = (challenge, time.time())
+    _challenges[challenge] = time.time()
+    now = time.time()
+    expired = [k for k, v in _challenges.items() if now - v > 300]
+    for k in expired:
+        _challenges.pop(k, None)
 
-def _get_challenge() -> bytes:
-    entry = _challenges.pop("current", None)
-    if not entry or time.time() - entry[1] > 300:
-        raise HTTPException(status_code=400, detail="Challenge expired")
-    return entry[0]
+def _get_challenge(client_data_b64: str) -> bytes:
+    if not client_data_b64:
+        raise HTTPException(status_code=400, detail="Missing clientDataJSON")
+    try:
+        data = json.loads(base64url_to_bytes(client_data_b64).decode('utf-8'))
+        chal_bytes = base64url_to_bytes(data.get("challenge", ""))
+    except Exception:
+        raise HTTPException(status_code=400, detail="Invalid clientDataJSON")
+    
+    entry = _challenges.pop(chal_bytes, None)
+    if not entry or time.time() - entry > 300:
+        raise HTTPException(status_code=400, detail="Challenge expired or invalid")
+    return chal_bytes
 
 @router.post("/passkey/register/options")
 async def passkey_register_options(current_user: str = Depends(auth.get_current_user)):
@@ -225,7 +242,7 @@ async def passkey_register_options(current_user: str = Depends(auth.get_current_
 @router.post("/passkey/register/verify")
 async def passkey_register_verify(request: Request, current_user: str = Depends(auth.get_current_user)):
     body = await request.json()
-    challenge = _get_challenge()
+    challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
     try:
         verification = verify_registration_response(
             credential=body,
@@ -263,7 +280,7 @@ async def passkey_login_options(request: Request):
 @limiter.limit("10/minute")
 async def passkey_login_verify(request: Request):
     body = await request.json()
-    challenge = _get_challenge()
+    challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
     creds = auth.load_passkey_credentials()
     cred_id_b64 = body.get("id", "")
     matched = next((c for c in creds if c["credential_id"] == cred_id_b64), None)

dashboard/backend/routers/system.py

@@ -116,7 +116,12 @@ def _classify(method, path, status, user):
 
 @router.get("/openclaw-token")
 def openclaw_token(request: Request):
-    ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() or request.client.host
-    if not (ip.startswith("192.168.31.") or ip.startswith("100.")):
+    ip = request.client.host
+    if ip in ("127.0.0.1", "::1"):
+        forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+        if forwarded:
+            ip = forwarded
+    import config
+    if not config.is_lan_ip(ip):
         raise HTTPException(status_code=403, detail="Access denied")
     return {"token": OPENCLAW_GATEWAY_TOKEN}