proxy test

synapse_invite_checker/invite_checker.py

@@ -1,7 +1,12 @@
 import treq
 import json
 import time
+import os
 from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet import reactor
+from twisted.web.client import ProxyAgent, Agent
+from twisted.web.http_headers import Headers
+from twisted.internet.endpoints import TCP4ClientEndpoint
 from synapse.module_api import ModuleApi, errors
 from synapse.types import UserID
 import logging
@@ -47,17 +52,49 @@ class InviteChecker:
     @inlineCallbacks
     def fetch_json(self, url):
         logger.info(f"Fetching JSON data from: {url}")
+
+        # Fetch proxy from the environment variable
+        https_proxy = os.getenv('https_proxy')
+
+        if https_proxy:
+            logger.info(f"Using HTTPS proxy: {https_proxy}")
+            
+            # Parse the proxy URL with optional username and password
+            if '@' in https_proxy:  # Check if authentication info is present
+                credentials, proxy_hostport = https_proxy.split('@')
+                username, password = credentials.split(':')
+                proxy_host, proxy_port = proxy_hostport.split(':')
+                
+                # Create basic authentication header
+                auth_string = f'{username}:{password}'
+                auth_bytes = base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')
+                headers = Headers({
+                    'Proxy-Authorization': [f'Basic {auth_bytes}']
+                })
+            else:
+                headers = None
+                proxy_host, proxy_port = https_proxy.split(':')
+
+            proxy_endpoint = TCP4ClientEndpoint(reactor, proxy_host, int(proxy_port))
+
+            # Create a ProxyAgent with authentication if needed
+            agent = ProxyAgent(proxy_endpoint)
+        else:
+            # If no proxy is set, use the default agent
+            agent = Agent(reactor)
+
         try:
-            response = yield treq.get(url)
+            # Pass headers to the request if authentication is required
+            if headers:
+                response = yield treq.get(url, agent=agent, headers=headers)
+            else:
+                response = yield treq.get(url, agent=agent)
+
             if response.code == 200:
-                try:
-                    content = yield response.content()
-                    data = json.loads(content.decode('utf-8'))
-                    logger.debug(f"Received JSON data: {data}")
-                    returnValue(data)
-                except Exception as json_error:
-                    logger.error(f"Failed to decode JSON data: {json_error}")
-                    returnValue(None)
+                content = yield response.content()
+                data = json.loads(content.decode('utf-8'))
+                logger.debug(f"Received JSON data: {data}")
+                returnValue(data)
             else:
                 logger.error(f"Failed to fetch JSON data. Status code: {response.code}")
                 returnValue(None)
@@ -65,6 +102,8 @@ class InviteChecker:
             logger.error(f"Error while fetching JSON: {str(e)}")
             returnValue(None)
 
+
+
     @inlineCallbacks
     def fetch_policy_room_banlist(self):
         """Fetches the ban lists from multiple policy rooms using Synapse API."""