disallowedfor html&script

Signed-off-by: NAYANAR <[email protected]>

Author: NAYANAR0502

mcpgateway/main.py

@@ -287,50 +287,78 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
 
 
 # Global exceptions handlers
-@app.exception_handler(RequestValidationError)
-async def validation_exception_handler(_request: Request, exc: RequestValidationError):
-    """Handle Pydantic validation errors globally.
-
-    Args:
-        _request: The FastAPI request object that triggered the validation error.
-        exc: The Pydantic ValidationError exception containing validation failure details.
-
-    Returns:
-        JSONResponse: A 422 Unprocessable Entity response with formatted validation error details.
-    """
-    return JSONResponse(status_code=422, content=ErrorFormatter.format_validation_error(exc))
 
 
 # Register exception handler for custom ValidationError
 @app.exception_handler(ValidationError)
-async def content_validation_exception_handler(_request: Request, exc: ValidationError):
-    """Handle content security validation errors with a plain message and no traceback.
+async def content_validation_exception_handler(request: Request, exc: ValidationError):
+    """Handle content security validation errors with a clean message format.
 
     Args:
-        _request: The FastAPI request object that triggered validation error.
+        request: The FastAPI request object that triggered validation error.
         exc: The ValidationError exception containing failure details.
 
     Returns:
-        PlainTextResponse: Plain text error message with 400 status code.
+        JSONResponse: Clean error message with 400 status code.
     """
-    return PlainTextResponse(f"mcpgateway.services.content_security.ValidationError: {exc}", status_code=400)
+    # Determine the operation type from the request path
+    path = request.url.path
+    if "/resources" in path:
+        operation = "register resource"
+    elif "/prompts" in path:
+        operation = "create prompt"
+    elif "/tools" in path:
+        operation = "create tool"
+    else:
+        operation = "process request"
+    
+    # Extract the actual error message and clean it up
+    error_msg = str(exc)
+    if "Content contains HTML tags that may cause display issues" in error_msg:
+        error_msg = "Resource content contains disallowed HTML tags"
+    
+    return JSONResponse(
+        status_code=400,
+        content={"detail": f"Failed to {operation}: {error_msg}"}
+    )
 
 
 @app.exception_handler(RequestValidationError)
-async def request_validation_exception_handler(_request: Request, exc: RequestValidationError):
+async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
     """Handle FastAPI request validation errors (automatic request parsing).
 
     This handles ValidationErrors that occur during FastAPI's automatic request
     parsing before the request reaches your endpoint.
 
     Args:
-        _request: The FastAPI request object that triggered validation error.
+        request: The FastAPI request object that triggered validation error.
         exc: The RequestValidationError exception containing failure details.
 
     Returns:
         JSONResponse: A 422 Unprocessable Entity response with error details.
     """
-    if _request.url.path.startswith("/tools"):
+    # Check if this is a resource creation request with content validation error
+    if request.url.path.startswith("/resources") and request.method == "POST":
+        for error in exc.errors():
+            msg = error.get("msg", "")
+            loc = error.get("loc", [])
+            # Debug logging
+            logger.debug(f"Validation error - loc: {loc}, msg: {msg}")
+            if len(loc) >= 2 and loc[-2:] == ["body", "content"] and "HTML tags" in msg:
+                # Provide user-friendly message for HTML content validation
+                return JSONResponse(
+                    status_code=400,
+                    content={"detail": "Failed to register resource: Resource content contains disallowed HTML tags"}
+                )
+            elif len(loc) >= 2 and loc[-2:] == ["body", "content"] and "Content contains" in msg:
+                # Extract the actual error message after "Value error, "
+                clean_msg = msg.replace("Value error, ", "") if "Value error, " in msg else msg
+                return JSONResponse(
+                    status_code=400,
+                    content={"detail": f"Failed to register resource: {clean_msg}"}
+                )
+    
+    if request.url.path.startswith("/tools"):
         error_details = []
 
         for error in exc.errors():
@@ -348,7 +376,7 @@ async def request_validation_exception_handler(_request: Request, exc: RequestVa
 
         response_content = {"detail": error_details}
         return JSONResponse(status_code=422, content=response_content)
-    return await fastapi_default_validation_handler(_request, exc)
+    return await fastapi_default_validation_handler(request, exc)
 
 
 @app.exception_handler(IntegrityError)
@@ -1586,15 +1614,15 @@ async def create_resource(
         )
     except SecurityError as e:
         logger.warning(f"Security violation in resource creation by user {user}: {str(e)}")
-        raise HTTPException(status_code=400, detail="Content failed security validation")
+        raise HTTPException(status_code=400, detail=f"Failed to register resource: {str(e)}")
     except ValidationError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        raise HTTPException(status_code=400, detail=f"Failed to register resource: {str(e)}")
     except ResourceURIConflictError as e:
         raise HTTPException(status_code=409, detail=str(e))
     except ResourceError as e:
         if "Rate limit" in str(e):
             raise HTTPException(status_code=429, detail=str(e))
-        raise HTTPException(status_code=400, detail=str(e))
+        raise HTTPException(status_code=400, detail=f"Failed to register resource: {str(e)}")
     except IntegrityError as e:
         logger.error(f"Integrity error while creating resource: {e}")
         raise HTTPException(status_code=409, detail=ErrorFormatter.format_database_error(e))

mcpgateway/schemas.py

@@ -1111,11 +1111,12 @@ def validate_mime_type(cls, v: Optional[str]) -> Optional[str]:
 
     @field_validator("content")
     @classmethod
-    def validate_content(cls, v: Optional[Union[str, bytes]]) -> Optional[Union[str, bytes]]:
+    def validate_content(cls, v: Optional[Union[str, bytes]], info: ValidationInfo) -> Optional[Union[str, bytes]]:
         """Validate content size and safety
 
         Args:
             v (Union[str, bytes]): Value to validate
+            info (ValidationInfo): Validation context containing other field values
 
         Returns:
             Union[str, bytes]: Value if validated as safe
@@ -1137,12 +1138,18 @@ def validate_content(cls, v: Optional[Union[str, bytes]]) -> Optional[Union[str,
         else:
             text = v
 
-        # Skip HTML validation in test environment
+        # Get MIME type from validation context
+        mime_type = (info.data.get("mime_type") or "").lower() if info.data else ""
+        
+        # Always block HTML content regardless of MIME type (except in tests)
         if not os.environ.get("PYTEST_CURRENT_TEST") and re.search(SecurityValidator.DANGEROUS_HTML_PATTERN, text, re.IGNORECASE):
-            raise ValueError("Content contains HTML tags that may cause display issues")
-
-        # if re.search(SecurityValidator.DANGEROUS_HTML_PATTERN, text, re.IGNORECASE):
-        #     raise ValueError("Content contains HTML tags that may cause display issues")
+            # Check for specific dangerous tags
+            if "<script" in text.lower():
+                raise ValueError("Resource content contains disallowed script tags")
+            elif "<html" in text.lower():
+                raise ValueError("Resource content contains disallowed HTML tags")
+            else:
+                raise ValueError("Resource content contains disallowed HTML tags")
 
         return v
 
@@ -1225,11 +1232,12 @@ def validate_mime_type(cls, v: Optional[str]) -> Optional[str]:
 
     @field_validator("content")
     @classmethod
-    def validate_content(cls, v: Optional[Union[str, bytes]]) -> Optional[Union[str, bytes]]:
+    def validate_content(cls, v: Optional[Union[str, bytes]], info: ValidationInfo) -> Optional[Union[str, bytes]]:
         """Validate content size and safety
 
         Args:
             v (Union[str, bytes]): Value to validate
+            info (ValidationInfo): Validation context containing other field values
 
         Returns:
             Union[str, bytes]: Value if validated as safe
@@ -1250,9 +1258,19 @@ def validate_content(cls, v: Optional[Union[str, bytes]]) -> Optional[Union[str,
                 raise ValueError("Content must be UTF-8 decodable")
         else:
             text = v
-        # Skip HTML validation in test environment
+            
+        # Get MIME type from validation context
+        mime_type = (info.data.get("mime_type") or "").lower() if info.data else ""
+        
+        # Always block HTML content regardless of MIME type (except in tests)
         if not os.environ.get("PYTEST_CURRENT_TEST") and re.search(SecurityValidator.DANGEROUS_HTML_PATTERN, text, re.IGNORECASE):
-            raise ValueError("Content contains HTML tags that may cause display issues")
+            # Check for specific dangerous tags
+            if "<script" in text.lower():
+                raise ValueError("Resource content contains disallowed script tags")
+            elif "<html" in text.lower():
+                raise ValueError("Resource content contains disallowed HTML tags")
+            else:
+                raise ValueError("Resource content contains disallowed HTML tags")
 
         return v
 

mcpgateway/services/content_security.py

@@ -171,7 +171,13 @@ async def _validate_content(self, content: str, _mime_type: str, context: str) -
                 for pattern in self.dangerous_patterns:
                     if pattern.search(content_lower):
                         self.security_violations["dangerous_pattern"] += 1
-                        raise SecurityError(f"{context.capitalize()} content contains potentially dangerous pattern: {pattern.pattern}")
+                        # Check for specific script tags
+                        if "<script" in content_lower:
+                            raise SecurityError("Resource content contains disallowed script tags")
+                        elif "<html" in content_lower:
+                            raise SecurityError("Resource content contains disallowed HTML tags")
+                        else:
+                            raise SecurityError(f"Resource content contains disallowed {pattern.pattern} pattern")
         # Check for excessive whitespace (potential padding attack, only for text)
         if isinstance(content, str) and len(content) > 1000:  # Only check larger content
             whitespace_ratio = sum(1 for c in content if c.isspace()) / len(content)

mcpgateway/services/resource_service.py

@@ -272,7 +272,7 @@ async def register_resource(
                     lowered_content = resource.content.lower()
                     for tag in disallowed_tags:
                         if tag in lowered_content:
-                            raise ResourceError(f"Resource content contains disallowed {tag} tag")
+                            raise ResourceError(f"Resource content contains disallowed {tag.replace('<', '')} tags")
                 validated_content, _ = await content_security.validate_resource_content(content=resource.content, uri=resource.uri, mime_type=resource.mime_type)
                 resource.content = validated_content
                 if not resource.mime_type:
@@ -289,7 +289,7 @@ async def register_resource(
 
             # --- Enforce disallowed MIME types ---
             if mime_type in DISALLOWED_MIME_TYPES:
-                raise ResourceError(f"MIME type '{mime_type}' is not allowed")
+                raise ResourceError(f"Resource content contains disallowed MIME type '{mime_type}'")
 
             # Determine content storage
             is_text = mime_type and mime_type.startswith("text/") or isinstance(resource.content, str)

test_resource_security.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""
+Simple test script to verify resource security validation.
+This tests the specific scenarios mentioned in the user's curl commands.
+"""
+
+import asyncio
+import sys
+import os
+
+# Add the project root to Python path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from mcpgateway.services.resource_service import ResourceService, ResourceError
+from mcpgateway.schemas import ResourceCreate
+from mcpgateway.config import settings
+from unittest.mock import MagicMock
+
+async def test_script_injection():
+    """Test that script tags are blocked."""
+    print("Testing script injection...")
+    
+    service = ResourceService()
+    
+    # Mock database session
+    db = MagicMock()
+    
+    # Test 1: Script injection
+    try:
+        resource = ResourceCreate(
+            uri="test://script",
+            name="Script",
+            content="<script>alert(1)</script>"
+        )
+        
+        result = await service.register_resource(db, resource)
+        print("❌ Script injection was NOT blocked!")
+        return False
+    except ResourceError as e:
+        if "disallowed script tags" in str(e):
+            print("✅ Script injection correctly blocked:", str(e))
+            return True
+        else:
+            print("❌ Script injection blocked but with wrong message:", str(e))
+            return False
+    except Exception as e:
+        print("❌ Unexpected error:", str(e))
+        return False
+
+async def test_html_mime_type():
+    """Test that HTML MIME type is blocked."""
+    print("Testing HTML MIME type...")
+    
+    service = ResourceService()
+    
+    # Mock database session
+    db = MagicMock()
+    
+    # Test 2: HTML MIME type
+    try:
+        resource = ResourceCreate(
+            uri="test.html",
+            name="HTML",
+            content="<html>test</html>",
+            mime_type="text/html"
+        )
+        
+        result = await service.register_resource(db, resource)
+        print("❌ HTML MIME type was NOT blocked!")
+        return False
+    except ResourceError as e:
+        if "disallowed MIME type" in str(e) and "text/html" in str(e):
+            print("✅ HTML MIME type correctly blocked:", str(e))
+            return True
+        else:
+            print("❌ HTML MIME type blocked but with wrong message:", str(e))
+            return False
+    except Exception as e:
+        print("❌ Unexpected error:", str(e))
+        return False
+
+async def test_valid_content():
+    """Test that valid content is allowed."""
+    print("Testing valid content...")
+    
+    service = ResourceService()
+    
+    # Mock database session and its methods
+    db = MagicMock()
+    db.add = MagicMock()
+    db.commit = MagicMock()
+    db.refresh = MagicMock()
+    
+    # Mock the resource object that would be created
+    mock_resource = MagicMock()
+    mock_resource.id = 1
+    mock_resource.uri = "test://valid"
+    mock_resource.name = "Valid"
+    mock_resource.content = "This is valid content"
+    mock_resource.metrics = []
+    mock_resource.tags = []
+    
+    # Make refresh set the mock resource
+    def refresh_side_effect(resource):
+        resource.id = 1
+        resource.metrics = []
+        resource.tags = []
+    
+    db.refresh.side_effect = refresh_side_effect
+    
+    try:
+        resource = ResourceCreate(
+            uri="test://valid",
+            name="Valid",
+            content="This is valid content"
+        )
+        
+        result = await service.register_resource(db, resource)
+        print("✅ Valid content correctly allowed")
+        return True
+    except Exception as e:
+        print("❌ Valid content was blocked:", str(e))
+        return False
+
+async def main():
+    """Run all tests."""
+    print("Running resource security validation tests...\n")
+    
+    # Enable content validation patterns for testing
+    settings.content_validate_patterns = True
+    
+    results = []
+    
+    # Test script injection
+    results.append(await test_script_injection())
+    print()
+    
+    # Test HTML MIME type
+    results.append(await test_html_mime_type())
+    print()
+    
+    # Test valid content
+    results.append(await test_valid_content())
+    print()
+    
+    # Summary
+    passed = sum(results)
+    total = len(results)
+    
+    print(f"Results: {passed}/{total} tests passed")
+    
+    if passed == total:
+        print("✅ All security validation tests passed!")
+        return 0
+    else:
+        print("❌ Some security validation tests failed!")
+        return 1
+
+if __name__ == "__main__":
+    exit_code = asyncio.run(main())
+    sys.exit(exit_code)