opensource/wxdown-service
3
0
Fork 0
mirror of https://github.com/wechat-article/wxdown-service.git synced 2026-08-29 00:40:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
jock
2025-06-19 17:44:12 +08:00
parent 0b678382df
commit ea5587542b
7 changed files with 61 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cert.py
-6
View File
@@ -1,30 +1,24 @@
import platform
import subprocess
from logger import logger
def is_certificate_installed(cert_name = 'mitmproxy'):
if platform.system() == 'Windows':
import wincertstore
logger.debug(f"证书检测结果:")
stores = ["MY", "ROOT", "CA"]
for store_name in stores:
with wincertstore.CertSystemStore(store_name) as store:
for cert in store.itercerts():
name = cert.get_name()
logger.debug(f"{name}")
if name == cert_name:
return True
return False
elif platform.system() == 'Darwin':
try:
result = subprocess.run(['security', 'find-certificate', '-c', cert_name], capture_output=True, text=True)
logger.debug(f"证书检测结果: {result}")
return result.returncode == 0
except FileNotFoundError:
logger.error("此系统中未找到 security 命令")
raise NotImplementedError("此系统中未找到 security 命令")
else:
logger.error(f"暂不支持该系统: {platform.system()}")
raise NotImplementedError(f"暂不支持该系统: {platform.system()}")
main.py
+5 -3
View File
@@ -1,11 +1,12 @@
import argparse
import multiprocessing
import sys
import os
import mitm
import utils
import watcher
from ui.startup import startup_ui
from ui.startup import startup_ui_loop
from ui.console import console
@@ -20,7 +21,7 @@ def main():
# 启动 mitmproxy 进程
mitm_proxy_address = mitm.start(args.port)
if mitm_proxy_address is None:
console.print('[bold red]启动 mitmproxy 失败，请切换端口进行重试[/]')
console.print('[bold red]启动 mitmproxy 失败，请切换端口后重试[/]')
sys.exit(1)
# 启动文件监控及 ws 服务进程
@@ -30,11 +31,12 @@ def main():
sys.exit(1)
# 启动 UI
startup_ui(mitm_proxy_address, ws_address)
startup_ui_loop(mitm_proxy_address, ws_address)
if __name__ == '__main__':
multiprocessing.freeze_support()
try:
main()
except KeyboardInterrupt:
mitm.py
+24 -11
View File
@@ -1,10 +1,11 @@
import multiprocessing
import operator
import os
import queue
import re
import sys
import threading
import time
from multiprocessing import Process, Queue
from pathlib import Path
from mitmproxy.tools.main import mitmdump
@@ -17,26 +18,25 @@ PLUGIN_FILE = str(SRC_PATH / 'resources' / 'credential.py')
CREDENTIALS_FILE = str(SRC_PATH / 'resources' / 'data' / 'credentials.json')
def mitmproxy_process(args: list[str], q: Queue):
sys.stdout = sys.stderr = utils.Capture(q)
print(f'Run mitmdump process {args} ({os.getpid()})...')
def mitmproxy_process(args: list[str], output_queue: multiprocessing.Queue):
sys.stdout = sys.stderr = utils.Capture(output_queue)
print(f'Run mitmdump process {args} ({os.getpid()})...', flush=True)
mitmdump(args)
def start(port):
def start(port: str):
# 启动 mitmproxy 并加载 credentials 插件
args = ['-p', port, '-s', PLUGIN_FILE, '--set', 'credentials='+CREDENTIALS_FILE]
q = Queue()
p = Process(target=mitmproxy_process, args=(args, q))
p.start()
mitm_output_queue = multiprocessing.Queue()
mitm_process = multiprocessing.Process(target=mitmproxy_process, args=(args, mitm_output_queue))
mitm_process.start()
start_time = time.time()
proxy_address = None
while time.time() - start_time < 10:
try:
message = q.get_nowait()
message = mitm_output_queue.get(timeout=0.1)
logger.info(message)
if operator.contains(message, "HTTP(S) proxy listening at"):
match = re.search(r'\*:(\d+)', message)
@@ -46,6 +46,19 @@ def start(port):
elif operator.contains(message, "address already in use"):
break
except queue.Empty:
time.sleep(0.1)
pass
if proxy_address:
pass
# mitmproxy 启动成功，用一个守护线程处理后续日志
# threading.Thread(target=handle_mitm_output, args=(mitm_output_queue,), daemon=True).start()
else:
mitm_process.terminate()
return proxy_address
def handle_mitm_output(output_queue: multiprocessing.Queue):
while True:
message = output_queue.get()
logger.info(message)
ui/layout.py
+5 -2
View File
@@ -6,7 +6,10 @@ def make_layout() -> Layout:
layout.split(
Layout(name="header", size=10),
Layout(name="main", ratio=1, minimum_size=10),
Layout(name="footer", ratio=1, minimum_size=10),
Layout(name="main"),
)
layout['main'].split_row(
Layout(name="service"),
Layout(name="status"),
)
return layout
ui/startup.py
+6 -6
View File
@@ -9,14 +9,14 @@ import cert
import platform
def startup_ui(mitm_proxy_address = None, ws_address = None):
def startup_ui_loop(mitm_proxy_address = None, ws_address = None):
layout = make_layout()
layout['header'].update(Header())
layout['main'].update(make_message([
layout['service'].update(make_message([
{'name': 'mitmproxy', 'address': mitm_proxy_address},
{'name': 'websocket', 'address': ws_address},
]))
layout['footer'].update(StatusPanel())
layout['status'].update(StatusPanel())
with Live(layout, refresh_per_second=1, screen=False, transient=True):
while True:
@@ -29,17 +29,17 @@ def startup_ui(mitm_proxy_address = None, ws_address = None):
cmd = 'certutil -addstore root %userprofile%\\.mitmproxy\\mitmproxy-ca-cert.cer'
elif platform.system() == 'Darwin':
cmd = 'sudo security add-trusted-cert -d -p ssl -p basic -k /Library/Keychains/System.keychain ~/.mitmproxy/mitmproxy-ca-cert.pem'
layout['footer'].update(
layout['status'].update(
StatusPanel(is_success=False, reason="系统中未检测到 mitmproxy 的证书，请手动安装。",
details=f"执行以下命令安装证书:\n[bold green]{cmd}[/]"))
continue
except Exception as e:
layout['footer'].update(
layout['status'].update(
StatusPanel(is_success=False, reason="系统检测 mitmproxy 证书时异常。",
details="请将日志文件发送给开发者"))
continue
# 检查代理是否正确
success, reason, details = utils.check_system_proxy(mitm_proxy_address)
layout['footer'].update(StatusPanel(is_success=success, ws_address=ws_address, reason=reason, details=details))
layout['status'].update(StatusPanel(is_success=success, ws_address=ws_address, reason=reason, details=details))
continue
utils.py
+5 -15
View File
@@ -1,14 +1,13 @@
import io
import multiprocessing
import re
import urllib.request
from pathlib import Path
import io
import queue
import requests
from termcolor import colored
import version
from logger import logger
SRC_PATH = Path.absolute(Path(__file__)).parent
LOGO_FILE = str(SRC_PATH / 'resources' / 'logo.txt')
@@ -16,18 +15,12 @@ LOGO_FILE = str(SRC_PATH / 'resources' / 'logo.txt')
# 检查系统代理是否设置正确
def check_system_proxy(mitm_proxy_address):
proxy_obj = urllib.request.getproxies()
logger.debug(f"检测到系统代理设置为: {proxy_obj}")
logger.debug(f"mitmproxy 代理为: {mitm_proxy_address}")
details = f'将系统代理设置为 [bold green]{mitm_proxy_address.removeprefix('http://')}[/]\n当前系统代理为:\n{proxy_obj}'
try:
response = requests.get('http://mitm.it', proxies=proxy_obj, timeout=3).text
except requests.exceptions.ProxyError as e:
logger.error(f"检测 http://mitm.it 代理时出错: {e}")
return False, '代理配置有误，请检查设置', details
except requests.exceptions.ReadTimeout as e:
logger.error(f"检测 http://mitm.it 代理时超时: {e}")
except Exception as e:
return False, '代理配置有误，请检查设置', details
traffic_not_passing = re.search(r'If you can see this, traffic is not passing through mitmproxy', response)
@@ -55,7 +48,7 @@ def get_version():
return f"wxdown-service {version.version}"
class Capture(io.TextIOBase):
def __init__(self, q):
def __init__(self, q: multiprocessing.Queue):
self.queue = q
self.buffer = ""
@@ -66,7 +59,4 @@ class Capture(io.TextIOBase):
self.buffer += s
while '\n' in self.buffer:
line, _, self.buffer = self.buffer.partition('\n')
try:
self.queue.put_nowait(line)
except queue.Full:
pass
self.queue.put(line)
watcher.py
+16 -16
View File
@@ -1,10 +1,10 @@
import asyncio
import multiprocessing
import operator
import queue
import re
import sys
import time
from multiprocessing import Process, Queue
from pathlib import Path
import websockets
@@ -41,12 +41,12 @@ async def connect_handler(client: ServerConnection):
# 通知所有客户端
async def notify_clients(notification_queue: asyncio.Queue):
async def notify_clients(notification_queue: asyncio.Queue, output_queue: multiprocessing.Queue):
try:
while True:
data = await notification_queue.get()
if len(ws_clients) > 0:
utils.print_info_message('通知所有客户端最新 Credentials 数据')
logger.info('通知所有客户端最新 Credentials 数据')
for client in list(ws_clients):
try:
@@ -76,8 +76,8 @@ class CredentialsFileHandler(FileSystemEventHandler):
# 启动 websocket 服务
async def main(notification_queue):
asyncio.create_task(notify_clients(notification_queue))
async def main(notification_queue: asyncio.Queue, output_queue: multiprocessing.Queue):
asyncio.create_task(notify_clients(notification_queue, output_queue))
logger.info(f"开始启动 websocket 服务")
async with serve(connect_handler, "localhost") as server:
@@ -91,11 +91,8 @@ async def main(notification_queue):
await server.serve_forever()
def watcher_process(q: Queue):
sys.stdout = sys.stderr = utils.Capture(q)
def watcher_process(output_queue: multiprocessing.Queue):
sys.stdout = sys.stderr = utils.Capture(output_queue)
Path(CREDENTIALS_JSON_FILE).parent.mkdir(parents=True, exist_ok=True)
Path(CREDENTIALS_JSON_FILE).touch()
@@ -110,29 +107,32 @@ def watcher_process(q: Queue):
try:
observer.start()
loop.run_until_complete(main(notification_queue))
loop.run_until_complete(main(notification_queue, output_queue))
finally:
observer.stop()
observer.join()
def watch_credential_file():
pass
def start():
q = Queue()
p = Process(target=watcher_process, args=(q,))
p.start()
watcher_output_queue = multiprocessing.Queue()
process = multiprocessing.Process(target=watcher_process, args=(watcher_output_queue,))
process.start()
start_time = time.time()
ws_address = None
while time.time() - start_time < 10:
try:
message = q.get_nowait()
message = watcher_output_queue.get(timeout=0.1)
if operator.contains(message, "服务启动成功"):
match = re.search(r':(\d+)', message)
port = match.group(1)
ws_address = f"ws://127.0.0.1:{port}"
break
except queue.Empty:
time.sleep(0.1)
pass
return ws_address