Enofirm-Temp/odoo/_monkeypatches/requests_blocker.py

# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
"""
Monkeypatch to block all HTTP requests to odoo.com domains.
This prevents any telemetry or data transmission to Odoo servers.
"""
import logging
from urllib.parse import urlparse

_logger = logging.getLogger(__name__)

# List of blocked domains
BLOCKED_DOMAINS = [
    'odoo.com',
    'www.odoo.com',
    'services.odoo.com',
    'iot-proxy.odoo.com',
    'accounts.odoo.com',
    'partner-autocomplete.odoo.com',
    'iap.odoo.com',
    'iap-services.odoo.com',
    'sms.api.odoo.com',
    'media-api.odoo.com',
    'olg.api.odoo.com',
    'iap-snailmail.odoo.com',
    'l10n-in-edi.api.odoo.com',
    'l10n-in-edi-demo.api.odoo.com',
    'nightly.odoo.com',
    'apps.odoo.com',
    # 'website.api.odoo.com',
]

# Also block any subdomain of odoo.com
def is_blocked_domain(url):
    """Check if a URL belongs to a blocked domain."""
    try:
        parsed = urlparse(url)
        hostname = parsed.hostname or ''
        hostname_lower = hostname.lower()
        
        # Check exact matches
        if hostname_lower in BLOCKED_DOMAINS:
            return True
        
        # Check if it's a subdomain of odoo.com
        # if hostname_lower.endswith('.odoo.com'):
        #     return True
            
        return False
    except Exception:
        return False


def patch_requests():
    """Monkeypatch requests library to block requests to odoo.com domains."""
    try:
        import requests
        
        # Store original methods
        _original_post = requests.post
        _original_get = requests.get
        _original_put = requests.put
        _original_patch = requests.patch
        _original_delete = requests.delete
        _original_request = requests.request
        _original_session_request = requests.Session.request if hasattr(requests.Session, 'request') else None
        
        def blocked_request(method, url, *args, **kwargs):
            """Block requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked request to Odoo server: %s %s", method, url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_request(method, url, *args, **kwargs)
        
        def blocked_post(url, *args, **kwargs):
            """Block POST requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked POST request to Odoo server: %s", url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_post(url, *args, **kwargs)
        
        def blocked_get(url, *args, **kwargs):
            """Block GET requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked GET request to Odoo server: %s", url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_get(url, *args, **kwargs)
        
        def blocked_put(url, *args, **kwargs):
            """Block PUT requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked PUT request to Odoo server: %s", url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_put(url, *args, **kwargs)
        
        def blocked_patch(url, *args, **kwargs):
            """Block PATCH requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked PATCH request to Odoo server: %s", url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_patch(url, *args, **kwargs)
        
        def blocked_delete(url, *args, **kwargs):
            """Block DELETE requests to odoo.com domains."""
            if is_blocked_domain(url):
                _logger.warning("Blocked DELETE request to Odoo server: %s", url)
                raise ConnectionError(f"Request to Odoo server blocked: {url}")
            return _original_delete(url, *args, **kwargs)
        
        # Patch requests module
        requests.post = blocked_post
        requests.get = blocked_get
        requests.put = blocked_put
        requests.patch = blocked_patch
        requests.delete = blocked_delete
        requests.request = blocked_request
        
        # Patch Session class
        if _original_session_request:
            def blocked_session_request(self, method, url, *args, **kwargs):
                """Block Session requests to odoo.com domains."""
                if is_blocked_domain(url):
                    _logger.warning("Blocked Session request to Odoo server: %s %s", method, url)
                    raise ConnectionError(f"Request to Odoo server blocked: {url}")
                return _original_session_request(self, method, url, *args, **kwargs)
            
            requests.Session.request = blocked_session_request
        
        _logger.info("Requests library patched to block Odoo server connections")
        
    except ImportError:
        _logger.warning("requests library not available, skipping patch")
    except Exception as e:
        _logger.error("Failed to patch requests library: %s", e)


def patch_urllib3_requests():
    """Monkeypatch urllib3 to block requests to odoo.com domains."""
    try:
        import urllib3
        
        _original_urlopen = urllib3.poolmanager.PoolManager.urlopen if hasattr(urllib3.poolmanager.PoolManager, 'urlopen') else None
        
        if _original_urlopen:
            def blocked_urlopen(self, method, url, *args, **kwargs):
                """Block urllib3 requests to odoo.com domains."""
                url_str = str(url) if not isinstance(url, str) else url
                if is_blocked_domain(url_str):
                    _logger.warning("Blocked urllib3 request to Odoo server: %s %s", method, url_str)
                    raise ConnectionError(f"Request to Odoo server blocked: {url_str}")
                return _original_urlopen(self, method, url, *args, **kwargs)
            
            urllib3.poolmanager.PoolManager.urlopen = blocked_urlopen
            _logger.info("urllib3 PoolManager patched to block Odoo server connections")
        
    except ImportError:
        _logger.warning("urllib3 library not available, skipping patch")
    except Exception as e:
        _logger.error("Failed to patch urllib3 library: %s", e)