windyboy
f7e54692a9
- Add core agent architecture with Command + Skill pattern - Implement Claude API integration for content analysis - Add Obsidian REST API integration for vault operations - Create conversational interface (v2.0) with natural language processing - Add comprehensive configuration management and validation - Include project documentation and developer guides - Set up testing framework with unit, integration, and property tests - Add Kiro specs for Claude API configuration and code quality improvements - Configure project steering files for development guidelines
657 lines
24 KiB
Python
657 lines
24 KiB
Python
"""
|
|
API response validation and sanitization utilities
|
|
Provides comprehensive validation for API responses to handle malformed data gracefully
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any, Dict, List, Optional, Union, Tuple, Callable
|
|
|
|
try:
|
|
from .error_handling import ValidationError, APIError
|
|
except ImportError:
|
|
from error_handling import ValidationError, APIError
|
|
|
|
|
|
class APIResponseValidator:
|
|
"""Comprehensive API response validation and sanitization"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
# Response size limits
|
|
self.max_response_size = 10 * 1024 * 1024 # 10MB
|
|
self.max_json_depth = 32
|
|
self.max_array_length = 10000
|
|
self.max_string_length = 1000000 # 1MB for individual strings
|
|
|
|
# Content type patterns
|
|
self.json_content_types = [
|
|
'application/json',
|
|
'application/vnd.api+json',
|
|
'text/json'
|
|
]
|
|
|
|
self.text_content_types = [
|
|
'text/plain',
|
|
'text/html',
|
|
'text/markdown',
|
|
'text/xml'
|
|
]
|
|
|
|
def validate_http_response(
|
|
self,
|
|
response: Any,
|
|
expected_status_codes: Optional[List[int]] = None,
|
|
expected_content_type: Optional[str] = None,
|
|
max_size: Optional[int] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate HTTP response object (aiohttp.ClientResponse or similar)
|
|
|
|
Args:
|
|
response: HTTP response object
|
|
expected_status_codes: List of acceptable status codes
|
|
expected_content_type: Expected content type
|
|
max_size: Maximum response size in bytes
|
|
|
|
Returns:
|
|
Dictionary with validation results
|
|
|
|
Raises:
|
|
APIError: If response validation fails
|
|
"""
|
|
validation_result = {
|
|
'valid': True,
|
|
'status_code': None,
|
|
'content_type': None,
|
|
'content_length': None,
|
|
'warnings': []
|
|
}
|
|
|
|
try:
|
|
# Check if response object has expected attributes
|
|
if not hasattr(response, 'status'):
|
|
raise APIError(
|
|
message="Response object missing 'status' attribute",
|
|
api_name="unknown"
|
|
)
|
|
|
|
validation_result['status_code'] = response.status
|
|
|
|
# Validate status code
|
|
if expected_status_codes and response.status not in expected_status_codes:
|
|
raise APIError(
|
|
message=f"Unexpected status code: {response.status} (expected: {expected_status_codes})",
|
|
api_name="unknown",
|
|
status_code=response.status
|
|
)
|
|
|
|
# Check content type if available
|
|
if hasattr(response, 'headers') and 'content-type' in response.headers:
|
|
content_type = response.headers['content-type'].split(';')[0].strip().lower()
|
|
validation_result['content_type'] = content_type
|
|
|
|
if expected_content_type and not content_type.startswith(expected_content_type.lower()):
|
|
validation_result['warnings'].append(
|
|
f"Unexpected content type: {content_type} (expected: {expected_content_type})"
|
|
)
|
|
|
|
# Check content length if available
|
|
if hasattr(response, 'headers') and 'content-length' in response.headers:
|
|
try:
|
|
content_length = int(response.headers['content-length'])
|
|
validation_result['content_length'] = content_length
|
|
|
|
max_allowed = max_size or self.max_response_size
|
|
if content_length > max_allowed:
|
|
raise APIError(
|
|
message=f"Response too large: {content_length} bytes (max: {max_allowed})",
|
|
api_name="unknown"
|
|
)
|
|
except ValueError:
|
|
validation_result['warnings'].append("Invalid content-length header")
|
|
|
|
return validation_result
|
|
|
|
except Exception as e:
|
|
if isinstance(e, APIError):
|
|
raise
|
|
else:
|
|
raise APIError(
|
|
message=f"Response validation error: {str(e)}",
|
|
api_name="unknown",
|
|
cause=e
|
|
)
|
|
|
|
def validate_json_response(
|
|
self,
|
|
json_data: Any,
|
|
schema: Optional[Dict[str, Any]] = None,
|
|
api_name: str = "unknown"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate JSON response data with optional schema validation
|
|
|
|
Args:
|
|
json_data: Parsed JSON data to validate
|
|
schema: Optional schema definition for validation
|
|
api_name: Name of the API for error context
|
|
|
|
Returns:
|
|
Validated and sanitized JSON data
|
|
|
|
Raises:
|
|
APIError: If validation fails
|
|
"""
|
|
try:
|
|
# Basic structure validation
|
|
self._validate_json_structure(json_data, api_name)
|
|
|
|
# Schema validation if provided
|
|
if schema:
|
|
self._validate_json_schema(json_data, schema, api_name)
|
|
|
|
# Sanitize the data
|
|
sanitized_data = self._sanitize_json_data(json_data)
|
|
|
|
return sanitized_data
|
|
|
|
except Exception as e:
|
|
if isinstance(e, (APIError, ValidationError)):
|
|
raise
|
|
else:
|
|
raise APIError(
|
|
message=f"JSON validation error: {str(e)}",
|
|
api_name=api_name,
|
|
cause=e
|
|
)
|
|
|
|
def validate_text_response(
|
|
self,
|
|
text_data: str,
|
|
max_length: Optional[int] = None,
|
|
allowed_patterns: Optional[List[str]] = None,
|
|
forbidden_patterns: Optional[List[str]] = None,
|
|
api_name: str = "unknown"
|
|
) -> str:
|
|
"""
|
|
Validate text response with content checks
|
|
|
|
Args:
|
|
text_data: Text response to validate
|
|
max_length: Maximum allowed text length
|
|
allowed_patterns: List of regex patterns that must be present
|
|
forbidden_patterns: List of regex patterns that must not be present
|
|
api_name: Name of the API for error context
|
|
|
|
Returns:
|
|
Validated and sanitized text
|
|
|
|
Raises:
|
|
APIError: If validation fails
|
|
"""
|
|
if not isinstance(text_data, str):
|
|
raise APIError(
|
|
message="Response data must be a string",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Length validation
|
|
max_len = max_length or self.max_string_length
|
|
if len(text_data) > max_len:
|
|
raise APIError(
|
|
message=f"Response text too long: {len(text_data)} characters (max: {max_len})",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Pattern validation
|
|
if allowed_patterns:
|
|
for pattern in allowed_patterns:
|
|
if not re.search(pattern, text_data, re.IGNORECASE | re.DOTALL):
|
|
raise APIError(
|
|
message=f"Response missing required pattern: {pattern}",
|
|
api_name=api_name
|
|
)
|
|
|
|
if forbidden_patterns:
|
|
for pattern in forbidden_patterns:
|
|
if re.search(pattern, text_data, re.IGNORECASE | re.DOTALL):
|
|
raise APIError(
|
|
message=f"Response contains forbidden pattern: {pattern}",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Sanitize the text
|
|
sanitized_text = self._sanitize_text_data(text_data)
|
|
|
|
return sanitized_text
|
|
|
|
def parse_and_validate_json(
|
|
self,
|
|
response_text: str,
|
|
schema: Optional[Dict[str, Any]] = None,
|
|
api_name: str = "unknown"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Parse JSON response text and validate the result
|
|
|
|
Args:
|
|
response_text: Raw response text to parse
|
|
schema: Optional schema for validation
|
|
api_name: Name of the API for error context
|
|
|
|
Returns:
|
|
Parsed and validated JSON data
|
|
|
|
Raises:
|
|
APIError: If parsing or validation fails
|
|
"""
|
|
# Basic text validation first
|
|
if not isinstance(response_text, str):
|
|
raise APIError(
|
|
message="Response must be a string",
|
|
api_name=api_name
|
|
)
|
|
|
|
if len(response_text) > self.max_response_size:
|
|
raise APIError(
|
|
message=f"Response too large: {len(response_text)} bytes",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Try to parse JSON
|
|
try:
|
|
json_data = json.loads(response_text)
|
|
except json.JSONDecodeError as e:
|
|
# Try to extract JSON from response if it's embedded
|
|
json_data = self._extract_json_from_text(response_text, api_name)
|
|
if json_data is None:
|
|
raise APIError(
|
|
message=f"Invalid JSON response: {str(e)}",
|
|
api_name=api_name,
|
|
response_data=response_text[:500], # First 500 chars for debugging
|
|
cause=e
|
|
)
|
|
|
|
# Validate the parsed JSON
|
|
return self.validate_json_response(json_data, schema, api_name)
|
|
|
|
def validate_obsidian_api_response(
|
|
self,
|
|
response_data: Any,
|
|
operation: str = "unknown"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate Obsidian API response with operation-specific checks
|
|
|
|
Args:
|
|
response_data: Response data to validate
|
|
operation: Type of operation (read, write, list, etc.)
|
|
|
|
Returns:
|
|
Validated response data
|
|
|
|
Raises:
|
|
APIError: If validation fails
|
|
"""
|
|
api_name = "obsidian"
|
|
|
|
if operation == "read":
|
|
# For read operations, expect text content
|
|
if not isinstance(response_data, str):
|
|
raise APIError(
|
|
message="Obsidian read response must be text",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Validate as text with reasonable limits
|
|
return {
|
|
'content': self.validate_text_response(
|
|
response_data,
|
|
max_length=10 * 1024 * 1024, # 10MB for note content
|
|
api_name=api_name
|
|
),
|
|
'length': len(response_data)
|
|
}
|
|
|
|
elif operation == "write":
|
|
# Write operations might return status info
|
|
if isinstance(response_data, str):
|
|
# Simple text response
|
|
return {'message': response_data}
|
|
elif isinstance(response_data, dict):
|
|
# Structured response
|
|
return self.validate_json_response(response_data, api_name=api_name)
|
|
else:
|
|
# Assume success if no specific response
|
|
return {'success': True}
|
|
|
|
elif operation == "list":
|
|
# List operations should return array or object with files
|
|
if isinstance(response_data, list):
|
|
# Validate as array of file info
|
|
validated_files = []
|
|
for item in response_data:
|
|
if isinstance(item, str):
|
|
# Simple filename
|
|
validated_files.append(self._sanitize_filename(item))
|
|
elif isinstance(item, dict):
|
|
# File info object
|
|
validated_files.append(self._validate_file_info(item, api_name))
|
|
else:
|
|
self.logger.warning(f"Unexpected file list item type: {type(item)}")
|
|
|
|
return {'files': validated_files, 'count': len(validated_files)}
|
|
|
|
elif isinstance(response_data, dict):
|
|
# Object with file list
|
|
return self.validate_json_response(response_data, api_name=api_name)
|
|
|
|
else:
|
|
raise APIError(
|
|
message="Obsidian list response must be array or object",
|
|
api_name=api_name
|
|
)
|
|
|
|
else:
|
|
# Generic validation for unknown operations
|
|
if isinstance(response_data, str):
|
|
return {'content': self.validate_text_response(response_data, api_name=api_name)}
|
|
elif isinstance(response_data, (dict, list)):
|
|
return self.validate_json_response(response_data, api_name=api_name)
|
|
else:
|
|
return {'data': str(response_data)}
|
|
|
|
def validate_claude_api_response(
|
|
self,
|
|
response_data: Any,
|
|
operation: str = "unknown"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate Claude API response with operation-specific checks
|
|
|
|
Args:
|
|
response_data: Response data to validate
|
|
operation: Type of operation (analyze, transform, etc.)
|
|
|
|
Returns:
|
|
Validated response data
|
|
|
|
Raises:
|
|
APIError: If validation fails
|
|
"""
|
|
api_name = "claude"
|
|
|
|
# Claude API typically returns structured objects
|
|
if not isinstance(response_data, dict):
|
|
raise APIError(
|
|
message="Claude API response must be an object",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Validate basic structure
|
|
validated_response = self.validate_json_response(response_data, api_name=api_name)
|
|
|
|
# Check for required fields based on operation
|
|
if operation in ["analyze", "transform"]:
|
|
# Expect content field
|
|
if 'content' not in validated_response:
|
|
raise APIError(
|
|
message="Claude response missing 'content' field",
|
|
api_name=api_name
|
|
)
|
|
|
|
# Validate content structure
|
|
content = validated_response['content']
|
|
if isinstance(content, list) and len(content) > 0:
|
|
# Check first content item
|
|
first_item = content[0]
|
|
if isinstance(first_item, dict) and 'text' in first_item:
|
|
# Validate the text content
|
|
text_content = first_item['text']
|
|
if isinstance(text_content, str):
|
|
validated_response['content'][0]['text'] = self.validate_text_response(
|
|
text_content,
|
|
max_length=1000000, # 1MB for AI responses
|
|
api_name=api_name
|
|
)
|
|
|
|
return validated_response
|
|
|
|
def _validate_json_structure(self, data: Any, api_name: str, depth: int = 0) -> None:
|
|
"""Recursively validate JSON structure"""
|
|
if depth > self.max_json_depth:
|
|
raise APIError(
|
|
message=f"JSON structure too deep (max depth: {self.max_json_depth})",
|
|
api_name=api_name
|
|
)
|
|
|
|
if isinstance(data, dict):
|
|
if len(data) > 1000: # Reasonable limit for object keys
|
|
raise APIError(
|
|
message=f"JSON object has too many keys: {len(data)}",
|
|
api_name=api_name
|
|
)
|
|
|
|
for key, value in data.items():
|
|
if not isinstance(key, str):
|
|
raise APIError(
|
|
message=f"JSON object key must be string, got {type(key)}",
|
|
api_name=api_name
|
|
)
|
|
|
|
if len(key) > 1000: # Reasonable key length limit
|
|
raise APIError(
|
|
message=f"JSON object key too long: {len(key)} characters",
|
|
api_name=api_name
|
|
)
|
|
|
|
self._validate_json_structure(value, api_name, depth + 1)
|
|
|
|
elif isinstance(data, list):
|
|
if len(data) > self.max_array_length:
|
|
raise APIError(
|
|
message=f"JSON array too long: {len(data)} items (max: {self.max_array_length})",
|
|
api_name=api_name
|
|
)
|
|
|
|
for item in data:
|
|
self._validate_json_structure(item, api_name, depth + 1)
|
|
|
|
elif isinstance(data, str):
|
|
if len(data) > self.max_string_length:
|
|
raise APIError(
|
|
message=f"JSON string too long: {len(data)} characters (max: {self.max_string_length})",
|
|
api_name=api_name
|
|
)
|
|
|
|
def _validate_json_schema(self, data: Any, schema: Dict[str, Any], api_name: str) -> None:
|
|
"""Basic JSON schema validation"""
|
|
# This is a simplified schema validator
|
|
# For production use, consider using jsonschema library
|
|
|
|
if 'type' in schema:
|
|
expected_type = schema['type']
|
|
|
|
type_mapping = {
|
|
'object': dict,
|
|
'array': list,
|
|
'string': str,
|
|
'number': (int, float),
|
|
'integer': int,
|
|
'boolean': bool,
|
|
'null': type(None)
|
|
}
|
|
|
|
if expected_type in type_mapping:
|
|
expected_python_type = type_mapping[expected_type]
|
|
if not isinstance(data, expected_python_type):
|
|
raise APIError(
|
|
message=f"Expected {expected_type}, got {type(data).__name__}",
|
|
api_name=api_name
|
|
)
|
|
|
|
if isinstance(data, dict) and 'properties' in schema:
|
|
# Validate object properties
|
|
for prop_name, prop_schema in schema['properties'].items():
|
|
if prop_name in data:
|
|
self._validate_json_schema(data[prop_name], prop_schema, api_name)
|
|
|
|
# Check required properties
|
|
if 'required' in schema:
|
|
for required_prop in schema['required']:
|
|
if required_prop not in data:
|
|
raise APIError(
|
|
message=f"Missing required property: {required_prop}",
|
|
api_name=api_name
|
|
)
|
|
|
|
elif isinstance(data, list) and 'items' in schema:
|
|
# Validate array items
|
|
item_schema = schema['items']
|
|
for item in data:
|
|
self._validate_json_schema(item, item_schema, api_name)
|
|
|
|
def _sanitize_json_data(self, data: Any) -> Any:
|
|
"""Sanitize JSON data by removing/replacing problematic content"""
|
|
if isinstance(data, dict):
|
|
sanitized = {}
|
|
for key, value in data.items():
|
|
# Sanitize key
|
|
clean_key = self._sanitize_string(str(key))
|
|
# Recursively sanitize value
|
|
sanitized[clean_key] = self._sanitize_json_data(value)
|
|
return sanitized
|
|
|
|
elif isinstance(data, list):
|
|
return [self._sanitize_json_data(item) for item in data]
|
|
|
|
elif isinstance(data, str):
|
|
return self._sanitize_string(data)
|
|
|
|
else:
|
|
# Numbers, booleans, null - return as-is
|
|
return data
|
|
|
|
def _sanitize_text_data(self, text: str) -> str:
|
|
"""Sanitize text data"""
|
|
return self._sanitize_string(text)
|
|
|
|
def _sanitize_string(self, text: str) -> str:
|
|
"""Sanitize string content"""
|
|
if not isinstance(text, str):
|
|
return str(text)
|
|
|
|
# Remove null bytes
|
|
sanitized = text.replace('\x00', '')
|
|
|
|
# Remove other control characters except common whitespace
|
|
sanitized = ''.join(char for char in sanitized if ord(char) >= 32 or char in '\t\n\r')
|
|
|
|
# Limit length
|
|
if len(sanitized) > self.max_string_length:
|
|
sanitized = sanitized[:self.max_string_length] + '...[truncated]'
|
|
|
|
return sanitized
|
|
|
|
def _sanitize_filename(self, filename: str) -> str:
|
|
"""Sanitize filename from API response"""
|
|
if not isinstance(filename, str):
|
|
filename = str(filename)
|
|
|
|
# Remove dangerous characters
|
|
sanitized = re.sub(r'[<>:"|?*\x00-\x1f]', '', filename)
|
|
|
|
# Remove path separators
|
|
sanitized = sanitized.replace('/', '').replace('\\', '')
|
|
|
|
# Limit length
|
|
if len(sanitized) > 255:
|
|
sanitized = sanitized[:255]
|
|
|
|
return sanitized
|
|
|
|
def _validate_file_info(self, file_info: Dict[str, Any], api_name: str) -> Dict[str, Any]:
|
|
"""Validate file information object"""
|
|
validated = {}
|
|
|
|
# Common file info fields
|
|
if 'name' in file_info:
|
|
validated['name'] = self._sanitize_filename(str(file_info['name']))
|
|
|
|
if 'path' in file_info:
|
|
validated['path'] = self._sanitize_string(str(file_info['path']))
|
|
|
|
if 'size' in file_info:
|
|
try:
|
|
validated['size'] = int(file_info['size'])
|
|
except (ValueError, TypeError):
|
|
self.logger.warning(f"Invalid file size: {file_info['size']}")
|
|
|
|
if 'modified' in file_info:
|
|
validated['modified'] = self._sanitize_string(str(file_info['modified']))
|
|
|
|
if 'type' in file_info:
|
|
validated['type'] = self._sanitize_string(str(file_info['type']))
|
|
|
|
return validated
|
|
|
|
def _extract_json_from_text(self, text: str, api_name: str) -> Optional[Dict[str, Any]]:
|
|
"""Try to extract JSON from text response (e.g., if wrapped in markdown)"""
|
|
# Look for JSON blocks in markdown
|
|
json_patterns = [
|
|
r'```json\s*\n(.*?)\n```', # Markdown JSON block
|
|
r'```\s*\n(\{.*?\})\n```', # Generic code block with JSON
|
|
r'(\{.*\})', # Any JSON-like structure
|
|
]
|
|
|
|
for pattern in json_patterns:
|
|
matches = re.findall(pattern, text, re.DOTALL | re.IGNORECASE)
|
|
for match in matches:
|
|
try:
|
|
return json.loads(match.strip())
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
return None
|
|
|
|
|
|
# Global validator instance
|
|
api_response_validator = APIResponseValidator()
|
|
|
|
|
|
def validate_api_response(
|
|
response_data: Any,
|
|
api_name: str,
|
|
operation: str = "unknown",
|
|
schema: Optional[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Convenience function for validating API responses
|
|
|
|
Args:
|
|
response_data: Response data to validate
|
|
api_name: Name of the API
|
|
operation: Type of operation
|
|
schema: Optional schema for validation
|
|
|
|
Returns:
|
|
Validated response data
|
|
|
|
Raises:
|
|
APIError: If validation fails
|
|
"""
|
|
if api_name.lower() == "obsidian":
|
|
return api_response_validator.validate_obsidian_api_response(response_data, operation)
|
|
elif api_name.lower() == "claude":
|
|
return api_response_validator.validate_claude_api_response(response_data, operation)
|
|
else:
|
|
# Generic validation
|
|
if isinstance(response_data, str):
|
|
return {'content': api_response_validator.validate_text_response(response_data, api_name=api_name)}
|
|
elif isinstance(response_data, (dict, list)):
|
|
return api_response_validator.validate_json_response(response_data, schema, api_name)
|
|
else:
|
|
return {'data': str(response_data)} |