完善 cookie 监控系统:

1. 修复 browser_login.py 中的 cookie 处理错误
2. 重构 cookie_cloud.py 实现浏览器实例级 cookie 共享
3. 创建 cookie_extractor.py GUI 应用程序,支持打开、保存和恢复功能
4. 优化 cookie 导入方式,支持批量导入和回退机制
5. 修复 localStorage 注入问题,确保特殊字符正确处理
This commit is contained in:
2026-03-03 17:10:50 +08:00
parent 894acc3191
commit c3d07f7325
8 changed files with 1324 additions and 82 deletions
+117 -14
View File
@@ -1,4 +1,4 @@
"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取 cookie 数据"""
"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取并处理 cookie 数据"""
import json
import hashlib
import base64
@@ -6,11 +6,21 @@ from typing import Dict, List, Optional, Tuple
import urllib.request
import urllib.error
from Crypto.Cipher import AES
from dataclasses import dataclass
from applogger import get_logger
logger = get_logger(__name__)
@dataclass
class CookieCloudConfig:
"""CookieCloud 配置"""
api_url: str
uuid: str
password: str
timeout: int = 30
class CookieCloud:
"""Cookie Cloud 客户端"""
@@ -23,13 +33,16 @@ class CookieCloud:
uuid: 用户 UUID
password: 用户密码
"""
self.api_url = api_url.rstrip('/')
self.uuid = uuid
self.password = password
self.config = CookieCloudConfig(
api_url=api_url.rstrip('/'),
uuid=uuid,
password=password,
timeout=30
)
def _get_crypt_key(self) -> bytes:
"""生成加密密钥"""
combined_string = f"{self.uuid}-{self.password}"
combined_string = f"{self.config.uuid}-{self.config.password}"
return hashlib.md5(combined_string.encode('utf-8')).hexdigest()[:16].encode("utf-8")
def _bytes_to_key(self, data: bytes, salt: bytes, output: int = 48) -> bytes:
@@ -58,10 +71,10 @@ class CookieCloud:
padding_length = ord(padding_length)
return decrypted_padded[:-padding_length]
def _download_data(self) -> Tuple[Optional[Dict], str]:
"""下载并解密所有数据"""
def _download_all(self) -> Tuple[Optional[Dict], str]:
"""下载所有cookie和local storage数据"""
try:
url = f"{self.api_url}/get/{self.uuid}"
url = f"{self.config.api_url}/get/{self.config.uuid}"
request = urllib.request.Request(
url,
headers={
@@ -71,7 +84,7 @@ class CookieCloud:
method='GET'
)
response = urllib.request.urlopen(request, timeout=30)
response = urllib.request.urlopen(request, timeout=self.config.timeout)
if response.status != 200:
return None, f"服务器返回错误状态码: {response.status}"
@@ -98,7 +111,10 @@ class CookieCloud:
return result, ""
except urllib.error.HTTPError as e:
return None, f"HTTP错误: {e.code} {e.reason}"
if e.code == 401:
return None, "认证失败,请检查用户名和密码"
else:
return None, f"HTTP错误: {e.code} {e.reason}"
except urllib.error.URLError as e:
return None, f"网络连接失败: {e.reason}"
except Exception as e:
@@ -112,7 +128,7 @@ class CookieCloud:
按域名分组的 cookie 数据字典
格式: {"domain.com": [{"name": "cookie1", "value": "val1", ...}, ...], ...}
"""
data, error = self._download_data()
data, error = self._download_all()
if error:
logger.error(error)
@@ -139,8 +155,32 @@ class CookieCloud:
cookie['sameSite'] = 'Lax'
processed_cookies[domain].append(cookie)
logger.info(f"获取到 {len(processed_cookies)} 个域名的 cookies")
return processed_cookies
def get_local_storage(self) -> Dict[str, Dict]:
"""
从 Cookie Cloud 服务器获取 local storage 数据
Returns:
按域名分组的 local storage 数据字典
格式: {"domain.com": {"key1": "value1", ...}, ...}
"""
data, error = self._download_all()
if error:
logger.error(error)
return {}
# 处理数据结构
local_storage_data = {}
if isinstance(data, dict) and data.get('local_storage_data'):
local_storage_data = data.get('local_storage_data', {})
logger.info(f"获取到 {len(local_storage_data)} 个域名的 local storage")
return local_storage_data
def get_cookies_for_domain(self, domain: str) -> List[Dict]:
"""
获取指定域名的 cookie
@@ -159,8 +199,61 @@ class CookieCloud:
if domain in clean_domain or clean_domain in domain:
matched_cookies.extend(cookies)
logger.info(f"获取到 {len(matched_cookies)}{domain} 的 cookies")
return matched_cookies
def get_local_storage_for_domain(self, domain: str) -> Dict:
"""
获取指定域名的 local storage
Args:
domain: 域名
Returns:
local storage 字典
"""
all_local_storage = self.get_local_storage()
for ls_domain, data in all_local_storage.items():
clean_domain = ls_domain.lstrip('.')
if domain in clean_domain or clean_domain in domain:
logger.info(f"获取到 {domain} 的 local storage")
return data
return {}
def get_all_data(self) -> Dict:
"""
获取所有数据(包括 cookies 和 local storage
Returns:
包含所有数据的字典
"""
data, error = self._download_all()
if error:
logger.error(error)
return {}
result = {
'cookies': {},
'local_storage': {}
}
# 处理 cookies
if isinstance(data, dict):
if not data.get('cookie_data'):
result['cookies'] = data
elif data.get('cookie_data'):
result['cookies'] = data.get('cookie_data', {})
# 处理 local storage
if data.get('local_storage_data'):
result['local_storage'] = data.get('local_storage_data', {})
logger.info(f"获取到所有数据: {len(result['cookies'])} 个域名的 cookies, {len(result['local_storage'])} 个域名的 local storage")
return result
if __name__ == "__main__":
cc = CookieCloud(
@@ -169,8 +262,18 @@ if __name__ == "__main__":
password="jtDM5dV9AyqXkZdQVeA9f6"
)
cookies = cc.get_cookies()
logger.info(f"获取到 {len(cookies)} 个域名的 cookies")
# 测试获取所有数据
all_data = cc.get_all_data()
logger.info(f"所有数据: {json.dumps(all_data, indent=2, ensure_ascii=False)[:500]}...")
# 测试获取 cookies
cookies = cc.get_cookies()
# 测试获取指定域名的 cookies
lmkbi_cookies = cc.get_cookies_for_domain('lmkbi.95155.com')
logger.info(f"获取到 {len(lmkbi_cookies)} 个 lmkbi.95155.com 的 cookies")
# 测试获取 local storage
local_storage = cc.get_local_storage()
# 测试获取指定域名的 local storage
lmkbi_local_storage = cc.get_local_storage_for_domain('lmkbi.95155.com')