Patched code sample

import xml.etree.ElementTree as ET
import urllib.request
import urllib.parse
import re
from typing import Dict, Any, Optional, Set
from pathlib import Path

class SecureXMLProcessor:
    """Secure XML processor with comprehensive XXE protection"""
    
    def __init__(self, max_file_size: int = 1024*1024):
        self.max_file_size = max_file_size  # 1MB limit
        self.allowed_tags = {
            'root', 'data', 'item', 'value', 'name', 'content',
            'message', 'timestamp', 'response', 'element'
        }
        self.trusted_domains = {'api.example.com', 'data.mysite.com'}
    
    def parse_xml_document_secure(self, xml_content: str) -> Dict[str, Any]:
        """Securely parse XML document with XXE protection"""
        try:
            # Input validation
            if not xml_content or not isinstance(xml_content, str):
                return {"status": "error", "message": "Invalid XML content"}
            
            # Size limit
            if len(xml_content) > self.max_file_size:
                return {"status": "error", "message": "XML content too large"}
            
            # Check for external entity declarations (XXE prevention)
            if self._contains_external_entities(xml_content):
                return {"status": "error", "message": "External entities not allowed"}
            
            # Create secure XML parser with XXE protection
            parser = ET.XMLParser()
            
            # CRITICAL: Disable external entity processing to prevent XXE
            parser.parser.DefaultHandler = lambda data: None
            parser.parser.ExternalEntityRefHandler = lambda context, base, sysId, notationName: False
            parser.parser.EntityDeclHandler = lambda entityName, is_parameter_entity, value, base, systemId, publicId, notationName: False
            
            # Parse XML with secure parser
            root = ET.fromstring(xml_content, parser)
            
            # Validate XML structure
            if not self._validate_xml_structure(root):
                return {"status": "error", "message": "Invalid XML structure"}
            
            # Extract data safely
            result = {}
            element_count = 0
            
            for element in root:
                if element_count >= 100:  # Limit number of elements
                    break
                
                if element.tag in self.allowed_tags:
                    # Sanitize text content
                    text = element.text or ''
                    if len(text) > 1000:  # Limit text length
                        text = text[:1000]
                    
                    # Remove potentially dangerous characters
                    text = self._sanitize_text_content(text)
                    result[element.tag] = text
                    element_count += 1
            
            return {"status": "success", "data": result}
            
        except ET.ParseError as e:
            return {"status": "error", "message": "XML parsing failed"}
        except Exception as e:
            return {"status": "error", "message": "XML processing failed"}
    
    def process_xml_file_secure(self, file_path: str) -> Dict[str, Any]:
        """Securely process XML file with validation"""
        try:
            # Validate file path
            if not self._is_safe_file_path(file_path):
                return {"status": "error", "message": "Invalid file path"}
            
            # Check file size
            path_obj = Path(file_path)
            if not path_obj.exists() or not path_obj.is_file():
                return {"status": "error", "message": "File not found"}
            
            if path_obj.stat().st_size > self.max_file_size:
                return {"status": "error", "message": "File too large"}
            
            # Read file content
            with open(file_path, 'r', encoding='utf-8') as f:
                xml_content = f.read()
            
            # Use secure parsing
            return self.parse_xml_document_secure(xml_content)
            
        except Exception as e:
            return {"status": "error", "message": "File processing failed"}
    
    def parse_xml_from_url_secure(self, url: str) -> Dict[str, Any]:
        """Securely parse XML from URL with SSRF protection"""
        try:
            # Validate URL
            if not self._is_safe_url(url):
                return {"status": "error", "message": "URL not allowed"}
            
            # Create secure HTTP request
            request = urllib.request.Request(url)
            request.add_header('User-Agent', 'SecureXMLProcessor/1.0')
            
            # Fetch with timeout and size limits
            with urllib.request.urlopen(request, timeout=10) as response:
                # Check content type
                content_type = response.headers.get('Content-Type', '')
                if 'xml' not in content_type.lower():
                    return {"status": "error", "message": "Invalid content type"}
                
                # Read with size limit
                xml_content = response.read(self.max_file_size)
                
                if len(xml_content) >= self.max_file_size:
                    return {"status": "error", "message": "Response too large"}
                
                # Decode and parse securely
                xml_text = xml_content.decode('utf-8')
                return self.parse_xml_document_secure(xml_text)
                
        except Exception as e:
            return {"status": "error", "message": "URL processing failed"}
    
    def create_xml_response_secure(self, user_data: str) -> Dict[str, Any]:
        """Create XML response with input sanitization"""
        try:
            # Input validation
            if not user_data or not isinstance(user_data, str):
                return {"error": "Invalid user data"}
            
            # Length limit
            if len(user_data) > 1000:
                user_data = user_data[:1000]
            
            # Sanitize user data to prevent entity injection
            sanitized_data = self._sanitize_xml_content(user_data)
            
            # Create XML safely using string formatting with sanitized data
            xml_template = f"""<?xml version="1.0" encoding="UTF-8"?>
<response>
    <message>{sanitized_data}</message>
    <timestamp>2024-01-01T00:00:00Z</timestamp>
</response>"""
            
            # Parse with secure parser
            return self.parse_xml_document_secure(xml_template)
            
        except Exception as e:
            return {"error": "Response creation failed"}
    
    def _contains_external_entities(self, xml_content: str) -> bool:
        """Check for external entity declarations"""
        # Check for various XXE patterns
        xxe_patterns = [
            r'<!ENTITY\s+\w+\s+SYSTEM',
            r'<!ENTITY\s+\w+\s+PUBLIC',
            r'&\w+;',
            r'file://',
            r'http://',
            r'https://',
            r'ftp://',
            r'<!DOCTYPE.*\[',
        ]
        
        xml_upper = xml_content.upper()
        for pattern in xxe_patterns:
            if re.search(pattern, xml_upper, re.IGNORECASE | re.MULTILINE):
                return True
        
        return False
    
    def _validate_xml_structure(self, root: ET.Element) -> bool:
        """Validate XML structure against whitelist"""
        # Check root element
        if root.tag not in self.allowed_tags:
            return False
        
        # Check all elements
        element_count = 0
        for elem in root.iter():
            element_count += 1
            
            # Limit total elements to prevent DoS
            if element_count > 200:
                return False
            
            # Check tag names against whitelist
            if elem.tag not in self.allowed_tags:
                return False
            
            # Check for suspicious attributes
            for attr_name in elem.attrib:
                if (attr_name.startswith('xmlns') or 
                    'entity' in attr_name.lower() or
                    'system' in attr_name.lower()):
                    return False
        
        return True
    
    def _sanitize_text_content(self, text: str) -> str:
        """Sanitize text content"""
        if not text:
            return ""
        
        # Remove control characters
        sanitized = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', text)
        
        # Remove XML entities and references
        sanitized = re.sub(r'&[a-zA-Z0-9#]+;', '', sanitized)
        
        # Remove potentially dangerous patterns
        sanitized = re.sub(r'<!.*?>', '', sanitized, flags=re.DOTALL)
        
        return sanitized
    
    def _sanitize_xml_content(self, content: str) -> str:
        """Sanitize content for safe XML inclusion"""
        if not content:
            return ""
        
        # XML escape special characters
        content = content.replace('&', '&')
        content = content.replace('<', '<')
        content = content.replace('>', '>')
        content = content.replace('"', '"')
        content = content.replace("'", '&apos;')
        
        # Remove control characters
        content = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', content)
        
        # Remove entity declarations
        content = re.sub(r'<!.*?>', '', content, flags=re.DOTALL)
        
        return content
    
    def _is_safe_file_path(self, file_path: str) -> bool:
        """Validate file path for safety"""
        if not file_path or len(file_path) > 255:
            return False
        
        # Prevent path traversal
        if '..' in file_path or file_path.startswith('/'):
            return False
        
        # Only allow XML files
        if not file_path.lower().endswith('.xml'):
            return False
        
        # Only allow safe characters
        if not re.match(r'^[a-zA-Z0-9._/-]+$', file_path):
            return False
        
        return True
    
    def _is_safe_url(self, url: str) -> bool:
        """Validate URL for SSRF protection"""
        try:
            parsed = urllib.parse.urlparse(url)
            
            # Only allow HTTP/HTTPS
            if parsed.scheme not in ['http', 'https']:
                return False
            
            # Check against trusted domains
            hostname = parsed.hostname
            if not hostname:
                return False
            
            # Only allow trusted domains
            hostname_lower = hostname.lower()
            if not any(hostname_lower == domain or hostname_lower.endswith('.' + domain) 
                      for domain in self.trusted_domains):
                return False
            
            # Block private IP ranges
            import socket
            try:
                ip = socket.gethostbyname(hostname)
                if self._is_private_ip(ip):
                    return False
            except:
                pass
            
            return True
            
        except Exception:
            return False
    
    def _is_private_ip(self, ip: str) -> bool:
        """Check if IP is in private range"""
        try:
            import ipaddress
            addr = ipaddress.ip_address(ip)
            return (addr.is_private or addr.is_loopback or 
                   addr.is_link_local or addr.is_multicast)
        except:
            return True  # Err on the side of caution

# Secure usage examples:
xml_processor = SecureXMLProcessor()

# Safe XML parsing with proper structure
safe_xml_content = '<?xml version="1.0" encoding="UTF-8"?><root><data>Safe content</data><message>Hello World</message></root>'
result = xml_processor.parse_xml_document_secure(safe_xml_content)

# Safe file processing
file_result = xml_processor.process_xml_file_secure("safe_data.xml")

# Safe URL processing (only trusted domains)
url_result = xml_processor.parse_xml_from_url_secure("https://api.example.com/data.xml")

# Safe response creation
response = xml_processor.create_xml_response_secure("User message content")

# Dangerous inputs are safely rejected by the secure implementation