-
Notifications
You must be signed in to change notification settings - Fork 39
/
util.py
84 lines (65 loc) · 2.07 KB
/
util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster
import asyncio
import logging
import json
import re
import os
import pickle
import requests
import random
from jsonpath import jsonpath
from time import sleep, time
from urllib.request import urlretrieve
from datetime import datetime, timedelta
def except_decorative(func):
def decorator(*args, **keyargs):
try:
return func(*args, **keyargs)
except Exception as e:
logging.error(f'handle {func.__name__} error: {e}')
return decorator
def append_txt_file(save_item, file_path='1.txt', end='\n'):
with open(file_path, 'a', encoding='utf8') as txt_file:
txt_file.write(save_item + end)
@except_decorative
def get_json_file(file_path):
with open(file_path, 'r', encoding='utf-8') as json_file:
return json.load(json_file)
def save_json_file(file_path, save_item):
with open(file_path, 'w', encoding='utf-8') as json_file:
json.dump(save_item, json_file, ensure_ascii = False)
def create_folder(path):
if os.path.exists(path):
return
os.mkdir(path)
def get_file_suffix(path):
return os.path.splitext(path)[-1]
def get_pickle_file(file_path):
with open(file_path, 'rb') as pickle_file:
return pickle.load(pickle_file)
def save_pickle(file_path, data):
with open(file_path, 'wb') as code_file:
pickle.dump(data, code_file)
def get_json_val(item, path, get_first=False):
res = jsonpath(item, path)
if res and get_first:
return res[0]
return res
# mitmweb -p 8887 --listen-host 10.2.147.130 --set validate_inbound_headers=false
async def start_proxy(add_on, ip="10.2.147.164", port=8887):
# ip = '192.168.1.6'
# ip = '127.0.0.1'
opts = Options(listen_host=ip, listen_port=port)
opts.add_option("body_size_limit", int, 0, "")
# opts.add_option("validate_inbound_headers", bool, False, "")
# opts.add_option("allow_hosts", list, ["api.weibo.cn"], "")
# opts.add_option("ssl_insecure", bool, True, "")
m = DumpMaster(opts, with_termlog=False, with_dumper=False)
m.addons.add(add_on)
try:
print('\nproxy:', ip, port)
await m.run()
except KeyboardInterrupt:
m.shutdown()
return m