feat: override feature requests for dream11

examples/hybrid/utils/proxy-server.py

@@ -0,0 +1,98 @@
+from mitmproxy import http, ctx
+import json
+import os
+from datetime import datetime
+
+# Configuration
+TARGET_HOST = "instabug.com"
+ORIGINAL_DOMAIN = "api.instabug.com"
+NEW_DOMAIN = "st001012dream11.instabug.com"
+TARGET_TOKEN = "dfed1c768730afbd56efafd571b4cbaa"
+ALL_REQUESTS_FILE = "InterceptedRequests.json"
+def save_to_json(data, filename):
+    try:
+        if os.path.exists(filename):
+            with open(filename, 'r') as f:
+                existing_data = json.load(f)
+        else:
+            existing_data = []
+
+        existing_data.append(data)
+
+        with open(filename, 'w') as f:
+            json.dump(existing_data, f, indent=2)
+    except Exception as e:
+        ctx.log.error(f"Error saving to {filename}: {str(e)}")
+
+def should_intercept(url: str) -> bool:
+    return TARGET_HOST in url
+
+def request(flow: http.HTTPFlow) -> None:
+    original_url = flow.request.pretty_url
+
+    if not should_intercept(original_url):
+        return
+
+    ctx.log.info(f"Intercepted request: {original_url}")
+
+    if ORIGINAL_DOMAIN in original_url and "/api/sdk/v3/features" in original_url:
+        flow.metadata['original_url'] = original_url
+        flow.metadata['original_host'] = flow.request.host
+        flow.metadata['original_headers'] = dict(flow.request.headers)
+
+        ctx.log.info(f"Modifying request to go to new domain first")
+        new_url = original_url.replace(ORIGINAL_DOMAIN, NEW_DOMAIN)
+        new_url = new_url.split('application_token=')[0] + f'application_token={TARGET_TOKEN}'
+        flow.request.url = new_url
+        flow.request.host = NEW_DOMAIN
+        flow.request.headers["host"] = NEW_DOMAIN
+
+    timestamp = datetime.now().isoformat()
+    request_data = {
+        "timestamp": timestamp,
+        "method": flow.request.method,
+        "url": original_url,
+        "headers": dict(flow.request.headers),
+        "body": flow.request.get_text() if flow.request.get_text() else None
+    }
+    save_to_json(request_data, ALL_REQUESTS_FILE)
+
+def response(flow: http.HTTPFlow) -> None:
+    if not should_intercept(flow.request.pretty_url):
+        return
+
+    if NEW_DOMAIN in flow.request.pretty_url and hasattr(flow, 'metadata'):
+        try:
+            timestamp = datetime.now().isoformat()
+            captured_response = {
+                'status_code': flow.response.status_code,
+                'headers': dict(flow.response.headers),
+                'content': flow.response.get_text()
+            }
+            flow.metadata['captured_response'] = captured_response
+
+            ctx.log.info("Restoring original request details")
+            flow.request.url = flow.metadata['original_url']
+            flow.request.host = flow.metadata['original_host']
+            flow.request.headers.update(flow.metadata['original_headers'])
+        except json.JSONDecodeError:
+            ctx.log.error(f"Error: Response is not valid JSON for URL: {flow.request.pretty_url}")
+        except Exception as e:
+            ctx.log.error(f"Error processing response: {str(e)}")
+
+    timestamp = datetime.now().isoformat()
+    request_response_data = {
+        "timestamp": timestamp,
+        "request": {
+            "method": flow.request.method,
+            "url": flow.request.pretty_url,
+            "headers": dict(flow.request.headers),
+            "body": flow.request.get_text() if flow.request.get_text() else None
+        },
+        "response": {
+            "status_code": flow.response.status_code,
+            "headers": dict(flow.response.headers),
+            "body": flow.response.get_text() if flow.response.get_text() else None
+        }
+    }
+    save_to_json(request_response_data, ALL_REQUESTS_FILE)