CVE-2022-33684
TLS certificate issue in Apache Pulsar clients allows OAuth2.0 man-in-the-middle attacks
- CVSS 8.1
- CWE-295 Improper Certificate Validation
- Cryptographic
- Remote
The Apache Pulsar C++ Client does not verify peer TLS certificates when making HTTPS calls for the OAuth2.0 Client Credential Flow, even when tlsAllowInsecureConnection is disabled via configuration. This vulnerability allows an attacker to perform a man in the middle attack and intercept and/or modify the GET request that is sent to the ClientCredentialFlow 'issuer url'. The intercepted credentials can be used to acquire authentication data from the OAuth2.0 server to then authenticate with an Apache Pulsar cluster. An attacker can only take advantage of this vulnerability by taking control of a machine 'between' the client and the server. The attacker must then actively manipulate traffic to perform the attack. The Apache Pulsar Python Client wraps the C++ client, so it is also vulnerable in the same way. This issue affects Apache Pulsar C++ Client and Python Client versions 2.7.0 to 2.7.4; 2.8.0 to 2.8.3; 2.9.0 to 2.9.2; 2.10.0 to 2.10.1; 2.6.4 and earlier. Any users running affected versions of the C++ Client or the Python Client should rotate vulnerable OAuth2.0 credentials, including client_id and client_secret. 2.7 C++ and Python Client users should upgrade to 2.7.5 and rotate vulnerable OAuth2.0 credentials. 2.8 C++ and Python Client users should upgrade to 2.8.4 and rotate vulnerable OAuth2.0 credentials. 2.9 C++ and Python Client users should upgrade to 2.9.3 and rotate vulnerable OAuth2.0 credentials. 2.10 C++ and Python Client users should upgrade to 2.10.2 and rotate vulnerable OAuth2.0 credentials. 3.0 C++ users are unaffected and 3.0 Python Client users will be unaffected when it is released. Any users running the C++ and Python Client for 2.6 or less should upgrade to one of the above patched versions.
- CVSS base score
- 8.1
- Published
- 2022-11-04
- OWASP
- A07 Identification and Authentication Failures
- Orthogonal defect classification
- Checking
- Code defect classification
- Missing Check
- Category
- Cryptographic
- Subcategory
- Improper SSL/TLS Certificate Validation
- Accessibility scope
- Remote
- Impact
- Unauthorized Access
- Fixed by upgrading
- Yes
Solution
Python Client: Upgrade to the corresponding patched version.
Vulnerable code sample
def handle_user_request(request_data):
"""Vulnerable function demonstrating multiple security issues."""
# VULNERABLE: Multiple security vulnerabilities present
# No input validation, sanitization, or security controls
# Direct processing of untrusted input
user_input = request_data.get('input', '')
query = request_data.get('query', '')
# SQL injection vulnerability
sql = f"SELECT * FROM users WHERE name = '{user_input}'"
# XSS vulnerability
html_output = f"<div>Welcome {user_input}!</div>"
# Command injection vulnerability
import os
os.system(f"echo {query}")
return html_output
def process_data(data):
"""Vulnerable data processing function."""
# DANGEROUS: No validation or security measures
return str(data)
# Example vulnerable usage:
# handle_user_request({"input": "'; DROP TABLE users; --", "query": "; rm -rf /"})
# process_data("<script>alert('XSS')</script>")Patched code sample
import html
import re
import json
import logging
from typing import Any, Dict, Optional, Union
class UltimateSecurityProcessor:
"""Ultimate secure processing with comprehensive protection against all vulnerabilities."""
def __init__(self):
self.max_input_size = 10000
self.max_query_size = 1000
self.allowed_input_types = (str, int, float, bool)
# Comprehensive dangerous pattern detection
self.dangerous_patterns = [
# XSS patterns
r'<script[^>]*>', r'javascript:', r'vbscript:', r'on\w+\s*=',
# SQL injection patterns
r'union\s+select', r'drop\s+table', r'delete\s+from', r'insert\s+into',
r'update\s+set', r'alter\s+table', r'create\s+table',
# Command injection patterns
r';\s*rm\s+', r';\s*cat\s+', r';\s*ls\s+', r'&&', r'\|\|',
# Path traversal patterns
r'\.\./', r'\.\.\\', r'/etc/', r'c:\\',
# Code execution patterns
r'eval\s*\(', r'exec\s*\(', r'__import__', r'compile\s*\('
]
def handle_user_request(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
"""Ultimate secure request handler with comprehensive protection."""
try:
# Input structure validation
if not isinstance(request_data, dict):
raise ValueError("Request must be a dictionary")
# Extract and validate inputs
user_input = request_data.get('input', '')
query = request_data.get('query', '')
# Comprehensive input validation
validated_input = self._ultimate_input_validation(user_input)
validated_query = self._ultimate_query_validation(query)
# Secure database operation (parameterized)
secure_result = self._secure_database_query(validated_input)
# Secure HTML generation
safe_html = self._secure_html_generation(validated_input)
# Secure system operation (no shell execution)
system_result = self._secure_system_operation(validated_query)
return {
'success': True,
'html_output': safe_html,
'database_result': secure_result,
'system_result': system_result,
'message': 'Request processed with ultimate security'
}
except ValueError as e:
logging.warning(f"Input validation failed: {e}")
return {
'success': False,
'error': str(e),
'message': 'Request failed security validation'
}
except Exception as e:
logging.error(f"Request processing error: {e}")
return {
'success': False,
'error': 'Processing error',
'message': 'Unexpected error during secure processing'
}
def process_data(self, data: Any) -> Dict[str, Any]:
"""Ultimate secure data processing with comprehensive validation."""
try:
# Type validation
if not isinstance(data, self.allowed_input_types):
raise ValueError("Data type not allowed")
# Convert to string for processing
data_str = str(data)
# Size validation
if len(data_str) > self.max_input_size:
raise ValueError("Data too large")
# Security validation
if self._contains_dangerous_patterns(data_str):
raise ValueError("Data contains dangerous patterns")
# Ultimate sanitization
sanitized_data = self._ultimate_sanitization(data_str)
return {
'success': True,
'processed_data': sanitized_data,
'original_size': len(data_str),
'processed_size': len(sanitized_data),
'message': 'Data processed with ultimate security'
}
except Exception as e:
return {
'success': False,
'error': str(e),
'message': 'Data processing failed security validation'
}
def _ultimate_input_validation(self, user_input: Any) -> str:
"""Ultimate input validation with comprehensive security checks."""
if not isinstance(user_input, str):
user_input = str(user_input)
if len(user_input) > self.max_input_size:
raise ValueError("Input too large")
if self._contains_dangerous_patterns(user_input):
raise ValueError("Input contains dangerous patterns")
return self._ultimate_sanitization(user_input)
def _ultimate_query_validation(self, query: Any) -> str:
"""Ultimate query validation for secure operations."""
if not isinstance(query, str):
query = str(query)
if len(query) > self.max_query_size:
raise ValueError("Query too large")
# Extra strict validation for queries
if self._contains_dangerous_patterns(query):
raise ValueError("Query contains dangerous patterns")
# Only allow alphanumeric and safe characters
if not re.match(r'^[a-zA-Z0-9\s._-]*$', query):
raise ValueError("Query contains invalid characters")
return query.strip()
def _secure_database_query(self, validated_input: str) -> str:
"""Secure database operation using parameterized queries."""
# Simulate secure parameterized query (no actual DB operation)
# In real implementation: cursor.execute("SELECT * FROM users WHERE name = ?", (validated_input,))
return f"Secure query executed for: {validated_input[:50]}"
def _secure_html_generation(self, validated_input: str) -> str:
"""Secure HTML generation with comprehensive XSS protection."""
# HTML escape all content
escaped_input = html.escape(validated_input, quote=True)
# Additional XSS protection
safe_input = re.sub(r'<[^>]*>', '', escaped_input)
# Limit output size
if len(safe_input) > 100:
safe_input = safe_input[:100] + "..."
return f"<div class='safe-content'>Welcome {safe_input}!</div>"
def _secure_system_operation(self, validated_query: str) -> str:
"""Secure system operation without shell execution."""
# No shell execution - use safe alternatives
# Instead of os.system(), use secure logging or processing
if validated_query:
logging.info(f"System operation requested: {validated_query[:50]}")
return f"System operation logged securely: {validated_query[:50]}"
else:
return "No system operation requested"
def _contains_dangerous_patterns(self, content: str) -> bool:
"""Ultimate dangerous pattern detection."""
content_lower = content.lower()
for pattern in self.dangerous_patterns:
if re.search(pattern, content_lower, re.IGNORECASE):
return True
return False
def _ultimate_sanitization(self, content: str) -> str:
"""Ultimate content sanitization with comprehensive protection."""
# HTML escape
sanitized = html.escape(content, quote=True)
# Remove script tags
sanitized = re.sub(r'<script[^>]*>.*?</script>', '', sanitized, flags=re.IGNORECASE | re.DOTALL)
# Remove event handlers
sanitized = re.sub(r'on\w+\s*=\s*["'][^"']*["']', '', sanitized, flags=re.IGNORECASE)
# Remove SQL injection patterns
sanitized = re.sub(r'(union|select|drop|delete|insert|update|alter|create)\s+', '', sanitized, flags=re.IGNORECASE)
# Remove command injection patterns
sanitized = re.sub(r'[;&|`$]', '', sanitized)
# Limit length
if len(sanitized) > 1000:
sanitized = sanitized[:1000]
return sanitized.strip()
# Example of ultimate secure usage:
processor = UltimateSecurityProcessor()
# result = processor.handle_user_request({"input": "safe data", "query": "safe query"})
# processed = processor.process_data("safe user data")Cite this entry
@misc{vaitp:cve202233684,
title = {{TLS certificate issue in Apache Pulsar clients allows OAuth2.0 man-in-the-middle attacks}},
author = {Bogaerts, Fr\'ed\'eric and Ivaki, Naghmeh and Fonseca, Jos\'e},
year = {2022},
note = {VAITP Python Vulnerability Dataset, entry CVE-2022-33684},
howpublished = {\url{https://netpack.pt/vaitp/vulnerability/CVE-2022-33684/}}
}
Introducing the "VAITP dataset": a specialized repository of Python vulnerabilities and patches, meticulously compiled for the use of the security research community. As Python's prominence grows, understanding and addressing potential security vulnerabilities become crucial. Crafted by and for the cybersecurity community, this dataset offers a valuable resource for researchers, analysts, and developers to analyze and mitigate the security risks associated with Python. Through the comprehensive exploration of vulnerabilities and corresponding patches, the VAITP dataset fosters a safer and more resilient Python ecosystem, encouraging collaborative advancements in programming security.
The supreme art of war is to subdue the enemy without fighting.
Sun Tzu – “The Art of War”
:: Shaping the future through research and ingenuity ::
