1
+ from mitmproxy import http , ctx
2
+ import json
3
+ import os
4
+ from datetime import datetime
5
+
6
+ # Configuration
7
+ TARGET_HOST = "instabug.com"
8
+ ORIGINAL_DOMAIN = "api.instabug.com"
9
+ NEW_DOMAIN = "st001012dream11.instabug.com"
10
+ TARGET_TOKEN = "dfed1c768730afbd56efafd571b4cbaa"
11
+ ALL_REQUESTS_FILE = "InterceptedRequests.json"
12
+ def save_to_json (data , filename ):
13
+ try :
14
+ if os .path .exists (filename ):
15
+ with open (filename , 'r' ) as f :
16
+ existing_data = json .load (f )
17
+ else :
18
+ existing_data = []
19
+
20
+ existing_data .append (data )
21
+
22
+ with open (filename , 'w' ) as f :
23
+ json .dump (existing_data , f , indent = 2 )
24
+ except Exception as e :
25
+ ctx .log .error (f"Error saving to { filename } : { str (e )} " )
26
+
27
+ def should_intercept (url : str ) -> bool :
28
+ return TARGET_HOST in url
29
+
30
+ def request (flow : http .HTTPFlow ) -> None :
31
+ original_url = flow .request .pretty_url
32
+
33
+ if not should_intercept (original_url ):
34
+ return
35
+
36
+ ctx .log .info (f"Intercepted request: { original_url } " )
37
+
38
+ if ORIGINAL_DOMAIN in original_url and "/api/sdk/v3/features" in original_url :
39
+ flow .metadata ['original_url' ] = original_url
40
+ flow .metadata ['original_host' ] = flow .request .host
41
+ flow .metadata ['original_headers' ] = dict (flow .request .headers )
42
+
43
+ ctx .log .info (f"Modifying request to go to new domain first" )
44
+ new_url = original_url .replace (ORIGINAL_DOMAIN , NEW_DOMAIN )
45
+ new_url = new_url .split ('application_token=' )[0 ] + f'application_token={ TARGET_TOKEN } '
46
+ flow .request .url = new_url
47
+ flow .request .host = NEW_DOMAIN
48
+ flow .request .headers ["host" ] = NEW_DOMAIN
49
+
50
+ timestamp = datetime .now ().isoformat ()
51
+ request_data = {
52
+ "timestamp" : timestamp ,
53
+ "method" : flow .request .method ,
54
+ "url" : original_url ,
55
+ "headers" : dict (flow .request .headers ),
56
+ "body" : flow .request .get_text () if flow .request .get_text () else None
57
+ }
58
+ save_to_json (request_data , ALL_REQUESTS_FILE )
59
+
60
+ def response (flow : http .HTTPFlow ) -> None :
61
+ if not should_intercept (flow .request .pretty_url ):
62
+ return
63
+
64
+ if NEW_DOMAIN in flow .request .pretty_url and hasattr (flow , 'metadata' ):
65
+ try :
66
+ timestamp = datetime .now ().isoformat ()
67
+ captured_response = {
68
+ 'status_code' : flow .response .status_code ,
69
+ 'headers' : dict (flow .response .headers ),
70
+ 'content' : flow .response .get_text ()
71
+ }
72
+ flow .metadata ['captured_response' ] = captured_response
73
+
74
+ ctx .log .info ("Restoring original request details" )
75
+ flow .request .url = flow .metadata ['original_url' ]
76
+ flow .request .host = flow .metadata ['original_host' ]
77
+ flow .request .headers .update (flow .metadata ['original_headers' ])
78
+ except json .JSONDecodeError :
79
+ ctx .log .error (f"Error: Response is not valid JSON for URL: { flow .request .pretty_url } " )
80
+ except Exception as e :
81
+ ctx .log .error (f"Error processing response: { str (e )} " )
82
+
83
+ timestamp = datetime .now ().isoformat ()
84
+ request_response_data = {
85
+ "timestamp" : timestamp ,
86
+ "request" : {
87
+ "method" : flow .request .method ,
88
+ "url" : flow .request .pretty_url ,
89
+ "headers" : dict (flow .request .headers ),
90
+ "body" : flow .request .get_text () if flow .request .get_text () else None
91
+ },
92
+ "response" : {
93
+ "status_code" : flow .response .status_code ,
94
+ "headers" : dict (flow .response .headers ),
95
+ "body" : flow .response .get_text () if flow .response .get_text () else None
96
+ }
97
+ }
98
+ save_to_json (request_response_data , ALL_REQUESTS_FILE )
0 commit comments