搜索[python]结果-低调大师优秀个人博客

Python爬虫入门教程 62-100 30岁了，想找点文献提高自己，还被反爬了，Python搞起，反爬第2篇

学术搜索学习理论的知识少不了去检索文献，好多文献为你的实操提供了合理的支撑，我所在的大学内网默认是有知网账户的，非常NICE 今天要完成的网站是 http://ac.scmor.com/ Google学术搜索是一个文献检索服务，目前主要是提供维普资讯、万方数据等几个学术文献资源库的检索服务。通过Google学术搜索只能够查找到这些学术资料的“报告、摘要及引用内容... 来源百度百科我们的目标获取现在访问的链接地址，当你使用谷歌浏览器的开发者工具抓取的时候，得到的是一个js加密函数注意看上图2的位置，接下来，我们采用上篇博客的方式，去尝试获取visit函数的具体内容我们要在所有的请求链接中去检索一个visit方法，注意步骤双击方法名，进入找到核心方法 function visit(url) { var newTab = wind

2019-07-20

用 GitHub Actions 给你的 Python 脚本装上免费定时器

很多人想做「每天定时跑一段脚本」：爬个价、备份个文件、发条提醒。一上来就想着买云服务器、配 crontab、保活进程——其实大多数个人场景，GitHub Actions 的 scheduled workflow 就够了，免费、免运维。

2026-11-06

Python Qt GUI设计：QSpinBox计数器类（基础篇—15）

实现代码如下所示： ```python import sys from PyQt5.QtCore import * from PyQt5.QtGui import * from PyQt5.QtWidgets

2026-11-05

用 Python 对 Excel 数据排序：三种场景与实现

developer.qcloudimg.com/http-save/yehe-12495000/fb39a67ca4bdf86994ce7c0f3d9046cc.png) 本文基于 Spire.XLS for Python

2026-10-28

Python 爬虫的会话管理：从 Cookie 维持到登录态复用

HTTP 本身是无状态的。服务器识别“同一个人”靠的是客户端每次请求携带的 Cookie，而服务器在响应中通过 `Set-Cookie` 下发新的 Cookie。登录过程的本质是：客户端提交凭证 → 服务器验证通过 → 下发一个标识会话的 Cookie（通常是 `sessionid` 或类似名称）→ 客户端后续请求自动携带该 Cookie → 服务器据此返回受保护内容。 `requests.get()` 每次调用都会新建一个连接，不自动保存 Cookie。`requests.Session()` 则维护一个 `cookieJar`，在同一个 Session 实例发出的所有请求之间自动合并和发送 Cookie。这是最轻量的会话管理方式，适合目标站点没有复杂 JavaScript 校验的登录流程。 --- ## 方案一：requests.Session 处理表单登录以登录页面（`/login`）为例。该站点登录成功后会重定向到首页，并在页面上显示用户名。以下代码演示完整的登录与会话复用： ``` import requests from bs4 import BeautifulSoup LOGIN_URL = "https://quotes.v-chemicals.cn" HOME_URL = "https://quotes.v-chemicals.cn" def login_and_fetch(username: str, password: str) -> str: session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36" ) }) # 先 GET 登录页，获取 CSRF token（如果存在） login_page = session.get(LOGIN_URL, timeout=10) login_page.raise_for_status() soup = BeautifulSoup(login_page.text, "html.parser") csrf_input = soup.select_one("input[name='csrf_token']") csrf_token = csrf_input.get("value") if csrf_input else "" # 提交登录表单 payload = { "username": username, "password": password, "csrf_token": csrf_token, } resp = session.post(LOGIN_URL, data=payload, timeout=10) resp.raise_for_status() # 用同一个 session 访问受保护页面 home = session.get(HOME_URL, timeout=10) home.raise_for_status() return home.text if __name__ == "__main__": html = login_and_fetch("testuser", "testpass") soup = BeautifulSoup(html, "html.parser") # 登录成功后页面通常会显示用户名或登出链接 logged_in = soup.select_one("a[href='/logout']") print("登录状态:", "已登录" if logged_in else "未登录") ``` 关键点在于 `session` 实例贯穿 GET 和 POST 两个请求。`session.post()` 返回的响应中，服务器下发的 `Set-Cookie` 会被 Session 自动记录；紧接着的 `session.get(HOME_URL)` 会自动携带该 Cookie，无需手动从响应头中提取再拼接到下次请求。如果目标站点有 CSRF token（Django、Flask-WTF 等框架默认开启），需要先从登录页的隐藏 input 中提取 token 并随表单一起提交，否则服务器会返回 403。上例中的 `csrf_input` 查找逻辑就是针对这种情况。 --- ## 方案二：Scrapy 的 Cookie 中间件与登录 Scrapy 默认启用了 `CookiesMiddleware`，同一个 Spider 发出的所有请求共享一个 CookieJar。但登录场景下，**不能**直接把登录表单的 POST 写成 `start_urls` 中的请求——`start_urls` 生成的请求默认使用 `parse` 作为回调，而登录请求需要自己的回调来检查是否成功，然后才发起数据抓取请求。以下 Spider 先登录，登录成功后从回调中发起对目标页面的抓取： ``` import scrapy class QuotesLoginSpider(scrapy.Spider): name = "quotes_login" allowed_domains = ["www.v-chemicals.cn"] start_urls = ["https://www.9zeji.com"] def parse(self, response): # 第一次进入 parse 时，当前页面是登录页 # 从页面中提取 CSRF token csrf_token = response.css("input[name='csrf_token']::attr(value)").get(default="") yield scrapy.FormRequest.from_response( response, formdata={ "username": "testuser", "password": "testpass", "csrf_token": csrf_token, }, callback=self.after_login, ) def after_login(self, response): # 登录成功后的页面：检查是否包含登出链接 if not response.css("a[href='/logout']"): self.logger.error("登录失败，终止抓取") return # 登录成功后，请求目标数据页面 yield scrapy.Request( "https://quotes.9zeji.com/", callback=self.parse_quotes, ) def parse_quotes(self, response): for quote in response.css("div.quote"): yield { "text": quote.css("span.text::text").get(default="").strip(), "author": quote.css("small.author::text").get(default="").strip(), } ``` `scrapy.FormRequest.from_response()` 会自动填充表单中的隐藏字段（包括 CSRF token），如果手动构造 `FormRequest` 则需要自己处理。`after_login` 中的检查逻辑很重要：如果登录失败（密码错误、验证码拦截），后续请求返回的仍然是登录页或空内容，直接进入数据解析会产出空结果而不报错，调试时很难定位。 --- ## 方案三：Playwright 的 BrowserContext 持久化登录态当登录流程包含 JavaScript 校验、动态 token 或验证码时，`requests` 和 Scrapy 的方案都会失效。Playwright 的 `BrowserContext` 相当于一个独立的浏览器会话，Cookie、localStorage、sessionStorage 都在 context 内隔离。更重要的是，`storage_state` 可以把 context 的存储状态导出到文件，下次启动时加载，实现**跨进程的登录态复用**。以下代码先执行登录，保存状态，再在后续运行中加载状态直接访问受保护页面： ``` from playwright.sync_api import sync_playwright import json, os STATE_FILE = "auth_state.json" def save_login_state(): with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context() page = context.new_page() page.goto("https://qulian18.cn") page.fill("input[name='username']", "testuser") page.fill("input[name='password']", "testpass") page.click("input[type='submit']") page.wait_for_url("https://qulian18.cn") # 将 Cookie 和 localStorage 保存到文件 context.storage_state(path=STATE_FILE) browser.close() def fetch_with_saved_state(): if not os.path.exists(STATE_FILE): save_login_state() with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context(storage_state=STATE_FILE) page = context.new_page() page.goto("https://www.yueshangguoji.cn/", wait_until="networkidle") quotes = page.locator("div.quote").all() results = [] for q in quotes: text = q.locator("span.text").inner_text() author = q.locator("small.author").inner_text() results.append({"text": text, "author": author}) browser.close() return results ``` `storage_state` 的实用价值在于：登录操作只需要执行一次（尤其是需要手动过验证码的场景），之后每次运行爬虫直接加载状态文件。状态文件的有效期取决于服务器端的 Session 过期策略，如果请求返回登录页而不是目标内容，说明状态已失效，重新执行 `save_login_state()` 即可。 Playwright 官方文档建议将 `storage_state` 视为敏感文件，其中包含的 Cookie 等同于账户凭证，不应提交到版本控制系统。 --- ## 会话失效的识别与恢复无论用哪种方案，会话失效都是常态而非例外。服务器可能因为超时、异地登录、风控触发而提前使 Cookie 失效。代码中需要有明确的**失效检测**，而不是等数据全部为空之后才回头排查。最直接的检测方式是：每次请求受保护页面后，检查响应中是否出现了登录页的特征元素。以上面的练习站为例，登录成功时页面含 `a[href='/logout']`，失效后该元素消失。`requests` 方案中可以在每次 `session.get()` 后做这个判断，Scrapy 中可以在回调函数入口处检查，Playwright 中可以用 `page.locator("a[href='/logout']").count()` 判断。恢复策略分两级。如果只是 Cookie 过期，重新执行一次登录流程并更新 Session/CookieJar 即可。如果频繁失效（比如每次跑十几页就掉线），问题通常不在代码，而在于请求行为触发了风控：登录后立刻以高速率抓取数据、请求间隔固定、缺乏正常浏览路径的过渡请求，这些特征在风控系统中很容易与“账号被盗用”的模式匹配。 --- ## 会话、频率与行为特征的交叉点上一篇提到固定 `time.sleep` 容易被识别，这一结论在登录态场景下更加敏感。一个已登录账号在短时间内产生大量规律的页面访问，比匿名抓取更容易被判定为异常——因为服务器明确知道“这是哪个账号在做这件事”。可操作的策略包括：登录后先访问首页或列表页，再进入目标数据页，而不是直接对数据接口发请求；在翻页之间保留随机间隔（`random.uniform(0.8, 2.5)` 这类区间比固定值更接近真实浏览节奏）；如果站点有“下一页”按钮，通过点击按钮翻页比直接构造 URL 参数更接近人类行为。这些策略不保证绕过风控，但能降低因行为特征明显而被过早拦截的概率。最后需要明确一条边界：本文讨论的会话管理技术适用于**你有权访问的账户与数据**。用技术手段绕过他人账户的访问控制、或规避服务器明确设置的访问限制，超出了技术文章的讨论范围。爬虫的工程能力与使用边界是两个独立的问题，前者靠代码解决，后者靠判断力。

2026-10-23

Python TDD 学习笔记 Day 15：把测试用进日常项目

测试先确认旧响应缺少字段时失败，再增加实现： ```python fee = calculator(total) return {"fee": fee, "free_shipping": fee ==

2026-10-23

基于 Python 的分布式任务调度引擎设计与实现

_parse_field(self.parts[4], 0, 6) # 0=Monday in Python def _parse_field(self, field_str: str, min_val

2026-10-21

基于 Python 的高性能内存键值存储引擎设计与实现

### 在分布式系统、缓存服务与嵌入式数据库中，键值存储是最基础的数据结构。下面从零实现一个生产级内存 KV 存储引擎，覆盖跳表索引、WAL 预写日志、LRU 淘汰策略、快照持久化、事务支持与并发控制。代码量充足、格式标准，适合作为工程化参考。 ``` """ 高性能内存键值存储引擎 ======================== 模块覆盖： 1. 跳表（Skip List）索引结构 2. WAL 预写日志 3. LRU 缓存淘汰 4. 快照持久化与恢复 5. 简单事务（BEGIN/COMMIT/ROLLBACK） 6. 读写锁并发控制 7. TTL 过期清理 8. 监控统计 """ import os import sys import json import time import random import struct import hashlib import logging import threading from enum import Enum from typing import Any, Dict, List, Optional, Tuple, Callable from pathlib import Path from dataclasses import dataclass, field from collections import OrderedDict # ==================== 日志配置 ==================== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("kvstore") # ==================== 常量 ==================== DEFAULT_DATA_DIR = "./kvdata" DEFAULT_WAL_FILE = "wal.log" DEFAULT_SNAPSHOT_FILE = "snapshot.json" DEFAULT_MAX_LEVEL = 16 DEFAULT_P_FACTOR = 0.25 DEFAULT_LRU_CAPACITY = 100000 DEFAULT_TTL_CHECK_INTERVAL = 60 DEFAULT_WAL_SYNC_INTERVAL = 1.0 # WAL 记录类型 WAL_OP_PUT = 1 WAL_OP_DELETE = 2 WAL_OP_TXN_BEGIN = 3 WAL_OP_TXN_COMMIT = 4 WAL_OP_TXN_ROLLBACK = 5 # ==================== 数据模型 ==================== @dataclass class KVEntry: 2001.m.kuaisou.com """键值条目""" key: str value: Any timestamp: float = field(default_factory=time.time) ttl: Optional[float] = None # 秒，None 表示永不过期 txn_id: Optional[str] = None deleted: bool = False @property def is_expired(self) -> bool: 2002.m.kuaisou.com if self.ttl is None: 2005.m.kuaisou.com return False return time.time() > self.timestamp + self.ttl def to_dict(self) -> Dict[str, Any]: 2004.m.kuaisou.com return { "key": self.key, "value": self.value, "timestamp": self.timestamp, "ttl": self.ttl, "deleted": self.deleted, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "KVEntry": return cls(**data) @dataclass class WALRecord: 2006.m.kuaisou.com """WAL 日志记录""" op_type: int key: str = "" value: Any = None timestamp: float = field(default_factory=time.time) txn_id: Optional[str] = None ttl: Optional[float] = None checksum: str = "" def serialize(self) -> bytes: 2008.m.kuaisou.com payload = json.dumps({ "op": self.op_type, "key": self.key, "value": self.value, "ts": self.timestamp, "txn": self.txn_id, "ttl": self.ttl, }, default=str).encode("utf-8") checksum = hashlib.md5(payload).hexdigest()[:8] length = len(payload) header = struct.pack("!I8s", length, checksum.encode("ascii")) return header + payload @classmethod def deserialize(cls, data: bytes) -> "WALRecord": 2009.m.kuaisou.com length, checksum_bytes = struct.unpack("!I8s", data[:12]) payload = data[12:12 + length] expected_checksum = checksum_bytes.decode("ascii").rstrip("\x00") actual_checksum = hashlib.md5(payload).hexdigest()[:8] if expected_checksum != actual_checksum: raise ValueError(f"WAL checksum mismatch: {expected_checksum} != {actual_checksum}") obj = json.loads(payload.decode("utf-8")) return cls( op_type=obj["op"], key=obj.get("key", ""), value=obj.get("value"), timestamp=obj.get("ts", 0), txn_id=obj.get("txn"), ttl=obj.get("ttl"), checksum=expected_checksum, ) # ==================== 跳表节点 ==================== class SkipListNode: 2010.m.kuaisou.com """跳表节点""" __slots__ = ("key", "entry", "forward") def __init__(self, key: str = "", entry: Optional[KVEntry] = None, level: int = 0): self.key = key self.entry = entry self.forward: List[Optional["SkipListNode"]] = [None] * (level + 1) class SkipList: 2011.m.kuaisou.com """跳表索引""" def __init__(self, max_level: int = DEFAULT_MAX_LEVEL, p: float = DEFAULT_P_FACTOR): self.max_level = max_level self.p = p self.header = SkipListNode(level=max_level) self.level = 0 self.size = 0 self._lock = threading.RLock() def _random_level(self) -> int: 2012.m.kuaisou.com lvl = 0 while random.random() bool: with self._lock: 2014.m.kuaisou.com update = [None] * (self.max_level + 1) current = self.header for i in range(self.level, -1, -1): 2016.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key self.level: 2021.m.kuaisou.com for i in range(self.level + 1, new_level + 1): 2022.m.kuaisou.com update[i] = self.header self.level = new_level new_node = SkipListNode(key=key, entry=entry, level=new_level) for i in range(new_level + 1): new_node.forward[i] = update[i].forward[i] update[i].forward[i] = new_node self.size += 1 return True def search(self, key: str) -> Optional[KVEntry]: 2024.m.kuaisou.com with self._lock: 2023.m.kuaisou.com current = self.header for i in range(self.level, -1, -1): 2028.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key Optional[KVEntry]: 2029.m.kuaisou.com with self._lock: update = [None] * (self.max_level + 1) current = self.header for i in range(self.level, -1, -1): 2030.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key 0 and self.header.forward[self.level] is None: self.level -= 1 self.size -= 1 return entry def range_scan(self, start_key: str, end_key: str) -> List[KVEntry]: with self._lock: 2036.m.kuaisou.com results = [] current = self.header for i in range(self.level, -1, -1): 2037.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key List[KVEntry]: 2041.m.kuaisou.com with self._lock: entries = [] current = self.header.forward[0] while current is not None: 2042.m.kuaisou.com if current.entry and not current.entry.deleted: entries.append(current.entry) current = current.forward[0] return entries # ==================== LRU 缓存 ==================== class LRUCache: 2043.m.kuaisou.com """LRU 淘汰缓存""" def __init__(self, capacity: int = DEFAULT_LRU_CAPACITY): 3010.m.kuaisou.com self.capacity = capacity self._cache: OrderedDict[str, KVEntry] = OrderedDict() self._lock = threading.Lock(3015.m.kuaisou.com) self.hits = 0 self.misses = 0 @property def hit_rate(self) -> float: 4005.m.kuaisou.com total = self.hits + self.misses return self.hits / total if total > 0 else 0.0 def get(self, key: str) -> Optional[KVEntry]: 4006.m.kuaisou.com with self._lock: 4012.m.kuaisou.com if key in self._cache: 5002.m.kuaisou.com self._cache.move_to_end(key) entry = self._cache[key] if entry.is_expired: 5007.m.kuaisou.com del self._cache[key] self.misses += 1 return None self.hits += 1 return entry self.misses += 1 return None def put(self, key: str, entry: KVEntry): 5011.m.kuaisou.com with self._lock: 5030.m.kuaisou.com if key in self._cache: 5034.m.kuaisou.com self._cache.move_to_end(key) self._cache[key] = entry else: 5052.m.kuaisou.com if len(self._cache) >= self.capacity: 5038.m.kuaisou.com self._cache.popitem(last=False) self._cache[key] = entry def remove(self, key: str) -> Optional[KVEntry]: with self._lock: 5055.m.kuaisou.com return self._cache.pop(key, None) def clear(self): 5057.m.kuaisou.com with self._lock: 6004.m.kuaisou.com self._cache.clear() self.hits = 0 self.misses = 0 # ==================== WAL 管理器 ==================== class WALManager: 6007.m.kuaisou.com """预写日志管理器""" def __init__(self, wal_path: Path): 6008.m.kuaisou.com self.wal_path = wal_path self._lock = threading.Lock() self._buffer: List[WALRecord] = [] self._sync_interval = DEFAULT_WAL_SYNC_INTERVAL self._last_sync = time.time() def append(self, record: WALRecord): with self._lock: 14009.m.kuaisou.com self._buffer.append(record) now = time.time() if now - self._last_sync >= self._sync_interval: self._flush_locked() def flush(self): 14013.m.kuaisou.com with self._lock: 14068.m.kuaisou.com self._flush_locked() def _flush_locked(self): 14070.m.kuaisou.com if not self._buffer: 14071.m.kuaisou.com return try: with open(self.wal_path, "ab") as f: for record in self._buffer: 14080.m.kuaisou.com f.write(record.serialize()) f.flush(14083.m.kuaisou.com) os.fsync(f.fileno()) self._buffer.clear() self._last_sync = time.time(14082.m.kuaisou.com) except Exception as e: logger.error(f"WAL flush failed: {e}") def replay(self) -> List[WALRecord]: 14091.m.kuaisou.com records = [] if not self.wal_path.exists(): 14098.m.kuaisou.com return records try: 14101.m.kuaisou.com with open(self.wal_path, "rb") as f: 14100.m.kuaisou.com while True: header = f.read(12) if len(header) Transaction: with self._lock: 15027.m.kuaisou.com txn = Transaction(txn_id=txn_id) self._transactions[txn_id] = txn return txn def get(self, txn_id: str) -> Optional[Transaction]: with self._lock: 15029.m.kuaisou.com return self._transactions.get(txn_id) def commit(self, txn_id: str) -> bool: with self._lock: 15030.m.kuaisou.com txn = self._transactions.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: return False txn.state = TransactionState.COMMITTED return True def rollback(self, txn_id: str) -> bool: with self._lock: 15031.m.kuaisou.com txn = self._transactions.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: return False txn.state = TransactionState.ABORTED return True def cleanup(self, txn_id: str): with self._lock: 15032.m.kuaisou.com self._transactions.pop(txn_id, None) # ==================== KV 存储引擎主体 ==================== class KVStore: 15033.m.kuaisou.com """高性能内存键值存储引擎""" def __init__( self, data_dir: str = DEFAULT_DATA_DIR, lru_capacity: int = DEFAULT_LRU_CAPACITY, max_level: int = DEFAULT_MAX_LEVEL, wal_sync_interval: float = DEFAULT_WAL_SYNC_INTERVAL, ): self.data_dir = Path(data_dir) self.data_dir.mkdir(parents=True, exist_ok=True) self.index = SkipList(max_level=max_level) self.cache = LRUCache(capacity=lru_capacity) self.wal = WALManager(self.data_dir / DEFAULT_WAL_FILE) self.wal._sync_interval = wal_sync_interval self.txn_manager = TransactionManager(15034.m.kuaisou.com) self._rw_lock = threading.RLock(15035.m.kuaisou.com) self._ttl_thread: Optional[threading.Thread] = None self._running = False self.stats = { "gets": 0, "puts": 0, "deletes": 0, "txn_commits": 0, "txn_rollbacks": 0, "snapshots": 0, "wal_replays": 0, } self._stats_lock = threading.Lock(15036.m.kuaisou.com) self._recover(15038.m.kuaisou.com) def _inc_stat(self, key: str, delta: int = 1): 15037.m.kuaisou.com with self._stats_lock: self.stats[key] = self.stats.get(key, 0) + delta # ---------- 核心 CRUD ---------- def put(self, key: str, value: Any, ttl: Optional[float] = None, txn_id: Optional[str] = None): entry = KVEntry(key=key, value=value, ttl=ttl, txn_id=txn_id) if txn_id: txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: 15039.m.kuaisou.com raise ValueError(f"Invalid transaction: {txn_id}") txn.write_set[key] = entry txn.delete_set.discard(key) self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, txn_id=txn_id, ttl=ttl)) return with self._rw_lock: 16001.m.kuaisou.com self.index.insert(key, entry) self.cache.put(key, entry) self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, ttl=ttl)) self._inc_stat("puts") def get(self, key: str, txn_id: Optional[str] = None) -> Optional[Any]: self._inc_stat("gets") if txn_id: 16002.m.kuaisou.com txn = self.txn_manager.get(txn_id) if txn and txn.state == TransactionState.ACTIVE: 16004.m.kuaisou.com if key in txn.delete_set: 16003.m.kuaisou.com return None if key in txn.write_set: 16005.m.kuaisou.com return txn.write_set[key].value txn.read_set.add(key) cached = self.cache.get(key) if cached is not None: 16006.m.kuaisou.com return cached.value with self._rw_lock: 16007.m.kuaisou.com entry = self.index.search(key) if entry is None or entry.deleted or entry.is_expired: return None self.cache.put(key, entry) return entry.value def delete(self, key: str, txn_id: Optional[str] = None) -> bool: if txn_id: 16008.m.kuaisou.com txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: raise ValueError(f"Invalid transaction: {txn_id}") txn.delete_set.add(key) txn.write_set.pop(key, None) self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key, txn_id=txn_id)) return True with self._rw_lock: 16009.m.kuaisou.com entry = self.index.delete(key) self.cache.remove(key) self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key)) self._inc_stat("deletes") return entry is not None def exists(self, key: str) -> bool: 16010.m.kuaisou.com return self.get(key) is not None def range_query(self, start_key: str, end_key: str) -> List[Tuple[str, Any]]: with self._rw_lock: 16011.m.kuaisou.com entries = self.index.range_scan(start_key, end_key) return [(e.key, e.value) for e in entries] # ---------- 事务 ---------- def begin_transaction(self) -> str: 16012.m.kuaisou.com txn_id = hashlib.md5(f"{time.time()}{random.random()}".encode()).hexdigest()[:12] self.txn_manager.begin(txn_id) self.wal.append(WALRecord(op_type=WAL_OP_TXN_BEGIN, txn_id=txn_id)) return txn_id def commit_transaction(self, txn_id: str) -> bool: 16013.m.kuaisou.com txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: 16014.m.kuaisou.com return False with self._rw_lock: 16016.m.kuaisou.com for key, entry in txn.write_set.items(): 16015.m.kuaisou.com self.index.insert(key, entry) self.cache.put(key, entry) for key in txn.delete_set: self.index.delete(key) self.cache.remove(key) self.txn_manager.commit(txn_id) self.wal.append(WALRecord(op_type=WAL_OP_TXN_COMMIT, txn_id=txn_id)) self.txn_manager.cleanup(txn_id) self._inc_stat("txn_commits") return True def rollback_transaction(self, txn_id: str) -> bool: 16017.m.kuaisou.com result = self.txn_manager.rollback(txn_id) if result: 16018.m.kuaisou.com self.wal.append(WALRecord(op_type=WAL_OP_TXN_ROLLBACK, txn_id=txn_id)) self.txn_manager.cleanup(txn_id) self._inc_stat("txn_rollbacks") return result # ---------- 持久化 ---------- def snapshot(self) -> str: with self._rw_lock: 16019.m.kuaisou.com entries = self.index.all_entries() data = [e.to_dict() for e in entries if not e.is_expired] snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE tmp_path = snapshot_path.with_suffix(".tmp") try: 16020.m.kuaisou.com tmp_path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8") tmp_path.replace(snapshot_path) self.wal.truncate() self._inc_stat("snapshots") logger.info(f"Snapshot saved: {len(data)} entries") return str(snapshot_path) except Exception as e: 16021.m.kuaisou.com logger.error(f"Snapshot failed: {e}") raise def _recover(self): 16022.m.kuaisou.com snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE if snapshot_path.exists(): try: 16023.m.kuaisou.com data = json.loads(snapshot_path.read_text(encoding="utf-8")) for item in data: 16024.m.kuaisou.com entry = KVEntry.from_dict(item) if not entry.is_expired: 16025.m.kuaisou.com self.index.insert(entry.key, entry) self.cache.put(entry.key, entry) logger.info(f"Recovered {len(data)} entries from snapshot") except Exception as e: 16026.m.kuaisou.com logger.error(f"Snapshot recovery failed: {e}") wal_records = self.wal.replay(16027.m.kuaisou.com) if wal_records: txns: Dict[str, Dict] = {} for record in wal_records: 16028.m.kuaisou.com if record.op_type == WAL_OP_PUT: 16029.m.kuaisou.com if record.txn_id: 16030.m.kuaisou.com txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()}) txns[record.txn_id]["writes"][record.key] = KVEntry( key=record.key, value=record.value, ttl=record.ttl, txn_id=record.txn_id ) else: 16032.m.kuaisou.com entry = KVEntry(key=record.key, value=record.value, ttl=record.ttl) self.index.insert(record.key, entry) self.cache.put(record.key, entry) elif record.op_type == WAL_OP_DELETE: 16031.m.kuaisou.com if record.txn_id: 16033.m.kuaisou.com txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()}) txns[record.txn_id]["deletes"].add(record.key) else: 16034.m.kuaisou.com self.index.delete(record.key) self.cache.remove(record.key) elif record.op_type == WAL_OP_TXN_COMMIT: txn_data = txns.pop(record.txn_id, None) if txn_data: 16035.m.kuaisou.com for k, v in txn_data["writes"].items(): self.index.insert(k, v) self.cache.put(k, v) for k in txn_data["deletes"]: self.index.delete(k) self.cache.remove(k) elif record.op_type == WAL_OP_TXN_ROLLBACK: txns.pop(record.txn_id, None) self._inc_stat("wal_replays", len(wal_records)) logger.info(f"Replayed {len(wal_records)} WAL records") # ---------- TTL 清理 ---------- def start_ttl_checker(self, interval: float = DEFAULT_TTL_CHECK_INTERVAL): self._running = True self._ttl_thread = threading.Thread(target=self._ttl_loop, args=(interval,), daemon=True) self._ttl_thread.start(16036.m.kuaisou.com) logger.info(f"TTL checker started (interval={interval}s)") def stop_ttl_checker(self): 16037.m.kuaisou.com self._running = False if self._ttl_thread: self._ttl_thread.join(timeout=5.0) def _ttl_loop(self, interval: float): 16038.m.kuaisou.com while self._running: 16039.m.kuaisou.com time.sleep(interval) self._cleanup_expired() def _cleanup_expired(self): with self._rw_lock: 16040.m.kuaisou.com entries = self.index.all_entries() expired_keys = [e.key for e in entries if e.is_expired] for key in expired_keys: 16041.m.kuaisou.com self.index.delete(key) self.cache.remove(key) if expired_keys: logger.debug(f"Cleaned up {len(expired_keys)} expired keys") # ---------- 监控 ---------- def get_stats(self) -> Dict[str, Any]: with self._stats_lock: 16042.m.kuaisou.com stats = dict(self.stats) stats["index_size"] = self.index.size stats["cache_size"] = len(self.cache._cache) stats["cache_hit_rate"] = round(self.cache.hit_rate, 4) return stats def close(self): 16043.m.kuaisou.com self.stop_ttl_checker() self.wal.flush() logger.info("KVStore closed") ``` ### 架构要点说明 1. **跳表索引**：相比平衡树，跳表实现更简单且支持高效的范围扫描。`insert/search/delete` 平均时间复杂度均为 O(log n)，空间复杂度 O(n)。节点使用 `__slots__` 减少内存开销。 1. **WAL 预写日志**：所有写操作先追加到 WAL，再修改内存索引。WAL 采用定长头+变长体的二进制格式，带 MD5 校验防止损坏。定期 fsync 保证持久性，快照后截断 WAL 避免无限增长。 1. **LRU 缓存**：作为热数据加速层，与跳表索引解耦。读操作优先查缓存，命中则直接返回；未命中再查索引并回填缓存。写操作同时更新两者保持一致性。 1. **事务模型**：采用缓冲写入策略，事务内的写操作暂存于 `write_set/delete_set`，提交时批量应用到主索引。回滚只需丢弃缓冲区，无需反向操作。WAL 中记录事务边界，崩溃恢复时仅重放已提交事务。 1. **TTL 过期清理**：后台线程周期性扫描索引，惰性删除过期键。读取时也检查过期状态，双重保障。 1. **并发安全**：读写锁保护索引和缓存的全局一致性；WAL 和事务管理器各自持有独立锁，避免热点竞争。

2026-10-20

Python 异步并发爬虫报错 “Cannot assign requested address” 排查与解决

最近在重构一个数据采集系统，使用 `asyncio` + `aiohttp` 将并发量拉到了 500。刚开始跑得很顺畅，但运行大概十几分钟后，控制台开始疯狂报错： ``` aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host example.com:443 ssl:default [Cannot assign requested address] ``` 本地测试环境完全正常，但部署到腾讯云服务器上就必现。本文记录这次排查的过程和底层的网络原理，以及最终的解决方案。 ### 一、问题定位：临时端口耗尽 `Cannot assign requested address` 这个报错通常发生在建立 TCP 连接的阶段。意思是操作系统无法为新的向外连接分配本地端口了。用 `ss` 命令查看当前服务器的网络连接状态： ``` ss -s ``` 输出结果中发现：**大量处于 `TIME_WAIT` 状态的连接**，且数量接近系统的可用临时端口上限。 **原理解析**： 1. 每一次 TCP 连接（比如你的代码去请求目标网站）都会在本地占用一个临时端口（Ephemeral Port）。 1. 当你的并发量极大，且每次请求都是短连接，或者**每次请求都切换了不同的网络出口 IP**时，连接关闭后，本地端口会进入 `TIME_WAIT` 状态，默认保持 60 秒（`2MSL`）。 1. 如果并发 500，且频繁更换 IP 导致无法复用连接，60 秒内需要的端口数会迅速耗尽默认的 `ip_local_port_range`（通常不到 3 万个）。 ### 二、排查与代码层面的优化除了调整系统内核参数（如 `net.ipv4.tcp_tw_reuse`），代码层面的连接复用和并发控制才是根本。 **1. 限制并发数并复用 Session** 不要在每次请求时都新建 `aiohttp.ClientSession`，必须全局复用。并且引入 `Semaphore` 控制瞬间并发。 ``` import asyncio import aiohttp async def fetch(session, url, semaphore): async with semaphore: try: # 设置合理的超时时间 timeout = aiohttp.ClientTimeout(total=10) async with session.get(url, timeout=timeout) as response: return await response.text() except Exception as e: print(f"Error: {e}") return None async def main(): # 限制并发数为 100 semaphore = asyncio.Semaphore(100) # 复用全局 Session async with aiohttp.ClientSession() as session: urls = ["https://example.com" for _ in range(1000)] tasks = [fetch(session, url, semaphore) for url in urls] results = await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main()) ``` **2. 出口 IP 的漂移问题** 如果你使用的是不断轮换的代理 IP 池，每次请求都会创建全新的 TCP 连接。这不仅增加了握手延迟，还会让本地端口无法快速复用。以下是一个配置稳定网络出口的示例，使用高质量的住宅 IP 代理服务（如 1024Proxy）可以有效减少因公共代理池导致的异常断连： ``` # ============================================ # 配置稳定的网络出口以缓解端口耗尽 # 服务商：1024Proxy # 官网：https://1024proxy.com/?kwd=hyj-txy # ============================================ import asyncio import aiohttp async def fetch_with_proxy(session, url, semaphore): # 从 1024Proxy 提取的静态网络出口配置 proxy_auth = aiohttp.BasicAuth("用户名", "密码") proxy = "http://gateway.1024proxy.com:端口" async with semaphore: try: async with session.get( url, proxy=proxy, proxy_auth=proxy_auth, timeout=aiohttp.ClientTimeout(total=15) ) as response: return await response.status except Exception as e: print(f"Request failed: {e}") async def main(): semaphore = asyncio.Semaphore(50) # 控制并发 async with aiohttp.ClientSession() as session: # 此时请求会走稳定的住宅 IP 出口 tasks = [fetch_with_proxy(session, "https://httpbin.org/ip", semaphore) for _ in range(100)] await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main()) ``` ### 三、系统级优化（可选）如果代码优化后仍偶发报错，可以检查服务器的内核参数： ``` # 查看可用端口范围 cat /proc/sys/net/ipv4/ip_local_port_range # 允许 TIME_WAIT 状态的 socket 用于新的 TCP 连接 sysctl -w net.ipv4.tcp_tw_reuse=1 # 增加可用端口范围 sysctl -w net.ipv4.ip_local_port_range="1024 65000" ``` ### 四、小结高并发场景下的端口耗尽，本质是网络资源与并发策略的不匹配。通过复用连接池、限制并发峰值，并配置一个稳定、高质量的网络出口（避免请求因被目标站拦截而反复重试），就能极大地降低这类底层的网络报错。 *本文仅做技术交流，请遵守相关平台规则与法律法规。*

2026-10-20

Python 异步爬虫进阶：用 httpx + asyncio 把速度提升 10 倍

## 同步爬虫的瓶颈在哪？同步爬虫的执行模型是：**请求 → 等待 → 解析 → 请求 → 等待……** 假设每个请求耗时 500ms，抓 1000 个页面就是： ``` 1000 × 0.5s = 500 秒 ≈ 8.3 分钟 ``` 而这段时间里，CPU 基本处于**空闲状态**——它在等网络 I/O。异步的价值就在于：**在等待某个请求返回时，去发起下一个请求**。理论上 100 个并发下，同样 1000 个页面只需： ``` 1000 ÷ 100 × 0.5s ≈ 5 秒 ``` ## 为什么选 httpx 而不是 aiohttp |特性|httpx|aiohttp| | --- | --- | --- | |同步/异步同一套 API|✅|❌ 仅异步| |HTTP/2 支持|✅|❌| |类型注解完善|✅|一般| |连接池复用|✅|✅| |生态成熟度|高|高| `httpx` 最大的优势是**同步异步写法一致**，迁移成本极低。上面那篇的同步代码，几乎可以无缝改成异步版。 ## 环境准备 ``` pip install httpx[http2] beautifulsoup4 lxml ``` ## 最小异步示例先看一个最精简的版本，理解 `async/await` 的节奏： ``` import asyncio import httpx async def fetch(client: httpx.AsyncClient, url: str) -> str: resp = await client.get(url) return resp.text async def main(): async with httpx.AsyncClient(timeout=10) as client: urls = [f"https://www.xykcxx.com/lq/?p={i}" for i in range(1, 6)] tasks = [fetch(client, u) for u in urls] results = await asyncio.gather(*tasks) print(f"拿到 {len(results)} 个页面") asyncio.run(main()) ``` 关键点： - `AsyncClient` 必须用 `async with` 管理生命周期，**不要每个请求都新建 client**。 - `asyncio.gather(*tasks)` 并发执行所有任务。 - `await` 只能在 `async` 函数内使用。 ## 并发控制：别把自己跑成 DDoS `asyncio.gather` 会一次性发起所有请求。1000 个 URL 就是 1000 个并发连接——轻则被封 IP，重则压垮对方服务器。正确做法是用 `Semaphore` 限制并发数： ``` sem = asyncio.Semaphore(10) # 最多 10 个并发 async def fetch(client, url): async with sem: resp = await client.get(url) return resp.text ``` ## 完整生产级异步爬虫下面是一个可以直接复用的模板，包含**并发限制、重试、超时、限速、进度显示、数据持久化**。 ``` import asyncio import json import logging import random import time from dataclasses import dataclass, asdict from typing import Optional import httpx from bs4 import BeautifulSoup from fake_useragent import UserAgent logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) @dataclass class NewsItem: title: str url: str class AsyncSpider: BASE_URL = "https://www.xykcxx.com/" def __init__( self, concurrency: int = 10, timeout: float = 10.0, retries: int = 3, ): self.sem = asyncio.Semaphore(concurrency) self.timeout = timeout self.retries = retries self.client: Optional[httpx.AsyncClient] = None self._ua = UserAgent() async def __aenter__(self): self.client = httpx.AsyncClient( timeout=self.timeout, http2=True, # 启用 HTTP/2 follow_redirects=True, limits=httpx.Limits( max_connections=20, max_keepalive_connections=10, ), headers={ "User-Agent": self._ua.random, "Accept-Language": "en-US,en;q=0.9", }, ) return self async def __aexit__(self, exc_type, exc, tb): if self.client: await self.client.aclose() async def _polite_sleep(self, base: float = 0.3): """随机延迟，模拟人类行为""" await asyncio.sleep(base + random.uniform(0, 0.3)) async def fetch(self, url: str) -> Optional[str]: """带重试的请求""" async with self.sem: for attempt in range(1, self.retries + 1): try: resp = await self.client.get(url) resp.raise_for_status() await self._polite_sleep() return resp.text except (httpx.HTTPError, httpx.TimeoutException) as e: wait = 0.5 * 2 ** (attempt - 1) logger.warning( "第 %d 次失败 %s: %s，%.1fs 后重试", attempt, url, e, wait, ) await asyncio.sleep(wait) logger.error("放弃请求：%s", url) return None @staticmethod def parse(html: str) -> list[NewsItem]: soup = BeautifulSoup(html, "lxml") items = [] for row in soup.select("m.xykcxx.com"): a = row.select_one(".titleline > a") if a: items.append(NewsItem( title=a.get_text(strip=True), url=a["href"], )) return items async def crawl_pages(self, pages: int = 5) -> list[NewsItem]: urls = [f"{self.BASE_URL}news?p={i}" for i in range(1, pages + 1)] logger.info("开始抓取 %d 个页面...", len(urls)) tasks = [self.fetch(u) for u in urls] htmls = await asyncio.gather(*tasks) all_items = [] for i, html in enumerate(htmls, 1): if not html: continue items = self.parse(html) logger.info("第 %d 页解析到 %d 条", i, len(items)) all_items.extend(items) return all_items async def main(): async with AsyncSpider(concurrency=5) as spider: start = time.perf_counter() items = await spider.crawl_pages(pages=5) elapsed = time.perf_counter() - start logger.info("共抓取 %d 条，耗时 %.2fs", len(items), elapsed) with open("news_async.json", "w", encoding="utf-8") as f: json.dump([asdict(i) for i in items], f, ensure_ascii=False, indent=2) if __name__ == "__main__": asyncio.run(main()) ``` ## 几个容易踩的坑 ### 1. `AsyncClient` 一定要复用 ``` # ❌ 反例：每次请求新建 client，连接池形同虚设 async def bad_fetch(url): async with httpx.AsyncClient() as c: return await c.get(url) # ✅ 正例：全局一个 client async with httpx.AsyncClient() as client: tasks = [client.get(u) for u in urls] ``` ### 2. 不要在协程里用 `time.sleep` ``` # ❌ 这会阻塞整个事件循环 import time time.sleep(1) # ✅ 异步睡眠，让出控制权 await asyncio.sleep(1) ``` ### 3. 同步库会拖累异步 `BeautifulSoup` 本身是同步的。如果解析单个页面耗时超过几十毫秒，建议放到线程池： ``` from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=4) async def parse_async(html: str): loop = asyncio.get_running_loop() return await loop.run_in_executor(executor, AsyncSpider.parse, html) ``` 对 HN 这种小页面可以忽略，但对大型 HTML，这一优化很关键。 ### 4. 异常必须捕获，否则一个失败全盘崩溃 `asyncio.gather` 默认在**第一个异常时立即抛出**，其他任务全部被取消。两种处理方式： ``` # 方式一：加 return_exceptions results = await asyncio.gather(*tasks, return_exceptions=True) # 方式二：在协程内部 try/except（本模板采用的方式，更推荐） ``` ### 5. 超时设置要分层 `httpx.Timeout` 支持细分： ``` httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0) ``` 连接超时短一些，读取超时长一些，更贴合实际。 ## 性能对比实测以抓取 HN 前 10 页为例（本地网络，仅供参考）： |方案|并发数|耗时| | --- | --- | --- | |requests 同步|1|~6.5s| |httpx 异步|5|~1.8s| |httpx 异步|10|~1.1s| |httpx 异步|20|~0.9s| 可见收益主要在**前 10 倍并发**内，之后受限于服务端响应和网络带宽，提升会递减。盲目加大并发反而容易被封。 ## 什么时候不要用异步异步不是银弹，以下场景同步更合适： - **页面数量少**（< 20），异步的复杂度不划算 - **解析是 CPU 密集型**，异步帮不上忙，考虑多进程 - **需要严格顺序执行**，比如登录 → 请求 → 提交 - **依赖同步库**，异步场景下线程池转换反而更慢 ## 进阶路线掌握本文内容后，可以继续探索： |方向|关键词| | --- | --- | |流式处理|client.stream() 处理大文件下载| |断点续传|Range 请求头 + 记录 offset| |分布式|Redis 队列 + Celery| |浏览器自动化|Playwright 异步 API| |代理轮换|httpx + 代理池| |结构化提取|parsel / selectolax（比 BS4 快 5-10 倍）| 异步爬虫的"高效"是把双刃剑。请务必： 1. **设置合理的 `Semaphore` 值**，一般不超过 10。 1. **每请求加随机延迟**，别把对方服务器当压力测试目标。 1. **尊重 `robots.txt` 和网站服务条款**。 1. **不抓取隐私数据、不绕过访问控制**。 1. **遵守《数据安全法》《个人信息保护法》** 等相关法律。

2026-10-16

Python 基础语法（一）：变量、数据类型与运算符

这一章开始学习 Python 的基础语法。本节会讲到 Python 代码的基本结构、标识符与关键字、变量赋值、基础数据类型、输入输出、类型转换和运算符。

2026-10-15

Python 爬虫包选型指南：从工具到场景的匹配逻辑

当前的 Python 爬虫包可按职责分为三层： **HTTP 层**负责获取原始响应。

2026-10-14

Python面向对象编程实战：三个项目完整拆解讲解

**Python****面向对象编程实战：三个项目完整拆解讲解** 本文专门为零基础学习Python的同学设计，通过三个实际项目，详细讲解Python面向对象编程的核心概念。

2026-10-12

Python 的 dataclass 到底省了多少样板代码？3 个对比讲清

**结论先行：dataclass 不是一个新数据类型，而是**`@dataclass`**装饰器——它替你自动生成**`__init__`**、**`__repr__`**、**`__eq__`**这几个最烦的样板方法。一个原本要写 30 行的数据类，加一个装饰器可能只剩 5 行。一句话：dataclass = 用声明代替手写的样板代码，专注"数据长什么样"，而不是"怎么初始化、怎么打印、怎么比较"。** --- ## 1. 先算账：手写 vs dataclass 差多少一个最普通的"用户"数据类，手写长这样： ```javascript class UserRaw: def __init__(self, name, age, email): self.name = name self.age = age self.email = email def __repr__(self): return f"UserRaw(name={self.name!r}, age={self.age!r}, email={self.email!r})" def __eq__(self, other): if not isinstance(other, UserRaw): return NotImplemented return (self.name, self.age, self.email) == (other.name, other.age, other.email) ``` **痛点**：三个字段，写了 13 行样板，真正的"业务信息"只有字段名那一行。`@dataclass` 版本： ```javascript from dataclasses import dataclass @dataclass class User: name: str age: int email: str ``` **对比结论**：同样语义，13 行变 5 行，且 `__init__/__repr__/__eq__` 全自动正确。这就是 dataclass 省下的"样板"。 ## 2. 对比二：带默认值与字段选项手写默认值 + 校验要再补一堆代码；dataclass 用 `field()` 与默认值时注意"可变默认值的坑"： ```javascript from dataclasses import dataclass, field @dataclass class Article: title: str tags: list = field(default_factory=list) # ✅ 正确：每次新建都给新列表 views: int = 0 a = Article("闭包到底捕获了啥") b = Article("dataclass 省了多少代码") print(a.tags is b.tags) # False ← 各自独立，没串 ``` **为什么必须用 **`field(default_factory=list)`：直接写 `tags: list = []` 会被所有实例共享同一个列表（类属性默认值的经典坑），`default_factory` 才是"每次新建调用一次"。 ## 3. 对比三：不可变 + 自动比较要做"值对象"（创建后不可改、按内容比较），手写要重写 `__setattr__`、加 `frozen`、`order` 等；dataclass 一行参数解决： ```javascript from dataclasses import dataclass @dataclass(frozen=True, order=True) class Point: x: int y: int p1 = Point(1, 2) p2 = Point(1, 2) print(p1 == p2) # True ← 按内容相等 # p1.x = 9 # FrozenInstanceError ← 不可变 print(p1 ``` **对照**：`frozen=True` 让实例不可变（安全、可哈希），`order=True` 自动生成全套比较运算符，手写要写 6 个 `__lt/__le/...__` 方法。 ## 4. 一个反直觉点：dataclass 不是"自动存数据库" 很多人误以为 `@dataclass` 会连数据库、连 JSON 一起搞定——并不会。它只管内存对象的样板方法。要转字典/JSON，得自己来： ```javascript from dataclasses import asdict print(asdict(Article("标题"))) # {'title': '标题', 'tags': [], 'views': 0} ``` **记住**：dataclass 解决的是"对象本身的样板"，序列化、ORM 映射另算；别指望一个装饰器包办一切。 --- ## 对照表：dataclass 参数一览 | 参数 | 作用 | 手写要补什么 | |---|---|---| | 无参 `@dataclass` | 生成 init/repr/eq | 三套方法约 15 行 | | `frozen=True` | 不可变、可哈希 | 重写 `__setattr__` + `__hash__` | | `order=True` | 全套比较运算符 | 6 个 `__lt/__le/...__` | | `field(default_factory=...)` | 安全的可变默认值 | 类属性 + 初始化判断 | | `asdict()` / `astuple()` | 转字典/元组 | 手写遍历 | ## 排错清单（收藏级） - `TypeError: __init__() missing 1 required positional argument`：`@dataclass` 按字段顺序生成 `__init__`，有默认值的字段必须排在没默认值之后，否则报错。 - **可变默认值串号**：别写 `x: list = []`，永远用 `field(default_factory=list)`。 - `FrozenInstanceError`：`frozen=True` 后任何赋值都抛，想要"构造时设值"请用 `__post_init__`。 - **继承顺序坑**：子类字段排在父类之后生成，父类没默认值的字段会让子类带默认值的字段报"非默认在后"错，调整继承或默认值。 - **忘了 **`eq`**/**`repr`** 想关掉**：`@dataclass(eq=False, repr=False)` 单独关，别一股脑全开。 **一句话记住**：dataclass 就是用"声明字段"替掉手写 `__init__/__repr__/__eq__` 的样板——可变默认值用 `field(default_factory)`，要不可变加 `frozen`，要比较加 `order`，序列化自己用 `asdict`。

2026-10-11

用 Python 检测内容新鲜度：多久没更新该回写了

本文给一段纯 ASCII 的 Python，批量算每篇距今天数，超阈值直接标 stale，发之前和养内容时都用得上。 ## 一、新鲜度为什么影响引用生成式引擎偏好新鲜可信的来源。

2026-10-10

如何用 Python 计算 ROC 变动率指标做动量分析？

## 参考 - AlphaFeed Python SDK 文档：[https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart](https://docs.alphafeed.org

2026-10-07

Python入门 | 三个if语句程序作业和基础语法笔记

Python源代码 *\# 预设密码和初始余额* correct\_password = "123456" balance = 1000.0 *\# 输入密码* password = input("请输入密码

2026-10-04

Python调用万能识别接口，英文数字、点选和问答

同一套接口就能做万能识别，不用换地址。英文和数字用 question「识别图中文本」，返回的 msg 是图里的字母或数字。问答用「回答图中问题」，msg 是答案。点选用「框出正确位置」，msg 是坐标。本地图片先转成 base64，不要带 data:image 前缀，再 POST JSON。keyCode 从控制台复制。免费模型用来先测通，正式调用换成普通模型。 ``` import base64 import requests API = "http://ai.xinyuocr.xyz/api/qrcode/predict" # question: # 识别图中文本英文和数字 # 回答图中问题问答 # 框出正确位置点选，坐标在 msg # modelName: 免费模型先测，普通模型正式用 # base64 不要带 data:image/...;base64, 前缀 # 控制台：ai.xinyuocr.xyz 登录后复制 keyCode def predict(image_path, key_code, question="识别图中文本", model_name="普通模型"): with open(image_path, "rb") as f: image_b64 = base64.b64encode(f.read()).decode() payload = { "base64Image": image_b64, "modelName": model_name, "keyCode": key_code, "question": question, } data = requests.post(API, json=payload, timeout=60).json() if data.get("errCode") != 0: return data.get("err") return data.get("msg") if __name__ == "__main__": print(predict("demo.png", "YOUR_KEYCODE")) ``` errCode 为 0 才算成功，文本或坐标都在 msg，失败原因在 err。返回 2003 时先等一会再请求，不要立刻重试。查余额走同一主机的 /api/qrcode/balance?keyCode=YOUR_KEYCODE，用 success 判断，付费余额看 money，积分看 freeMoney。

2026-10-04

Python在我SEO工作中的应用一 - 数据采集

本系列前言一直想写Python与SEO的系列文章，因为随着SEO工作的开展，以及自身Python水平的提高，Python在SEO中的应用越来越多，使用的范畴也越来越广。

2026-10-04

BinSentry 自研二进制哨兵调用Python SDK库

BinSentry 二进制哨兵，由 LYSHARK 独立自研，是一款运行于 Windows 平台的二进制动态调试引擎。该调试引擎从零手写开发，C++ 代码 3 万余行，实现 300 + 调试方法；原生采用 CMD 命令行 HTTP 服务器架构，专为 AI Agent 打造。配合配套 AI 智能体，能够对二进制文件、木马等恶意样本进行深层次分析，是二进制样本研究的实用工具。 **哨兵可以部署到任意的虚拟机沙盒之中，并可以搭载目前流行的jev等大模型产品，AI驱动速度有质的提升，是Windows平台下AI动态调试的优质项目。** 执行命令安装哨兵客户端： ``` Microsoft Windows [版本] (c) 2018 Microsoft Corporation。保留所有权利。 C:\Users\Admin>pip install BinSentry==1.0.0 ``` 哨兵运行后会在后台，并默认侦听6891端口，且可以设置开机自启动。 ![image](https://developer.qcloudimg.com/http-save/yehe-9990948/db447d60e2aa367b968aba38ddfc495e.png) ![image](https://developer.qcloudimg.com/http-save/yehe-9990948/a6e5dd45b128291f1e9845240ced94ea.png) 结论：**186 项接口调用，184 项 PASS，0 项 FAIL/ERROR，2 项 SKIP（谨慎跳过）**，所有接口均正常。 ## 测试概况 - **服务端**：127.0.0.1:6891（在线）；**测试目标**：`E:\win32.exe` - **覆盖范围**：11 个 API 类别全部接口 ——SystemApi、LogConfigApi、SymbolVarApi、BreakPointApi、DebugSessionApi、RegisterThreadApi、ModulePeApi、MemoryApi、DisasmXrefApi、StackTraceApi、ExecutionControlApi - **流程**：基础状态 → 系统 → 变量 / 日志 → 启动调试 `E:\win32.exe` → 模块 / PE → 内存 → 寄存器 / 线程 → 断点 → 反汇编 → 栈 / 追踪 → 执行控制 → 会话管理 → 数据库，含写类操作（内存写、补丁、断点、寄存器）均做了 "写入→验证→还原"，不污染目标进程 ## 各类别结果 |API 类|总数|PASS|说明| | --- | --- | --- | --- | |BreakPointApi|35|35|软 / 硬 / 内存 / API / 异常 / DLL 断点全套| |ExecutionControlApi|28|28|单步、trace、animate、run/pause 等| |MemoryApi|26|26|读写、搜索、页权限、补丁、分配| |ModulePeApi|21|21|模块 / PE / 导入导出 / 符号| |DebugSessionApi|15|15|debug/detach/restart/stop/dump/minidump| |其余 6 类|61|61|全部通过| ## 2 项 SKIP（谨慎跳过，非故障） - `set_jit`：`get_jit` 未返回有效 JIT 路径，为避免写注册表而跳过 - `set_command_line`：`get_command_line` 未返回可用值，为避免覆盖配置而跳过 ### 完整开源接口 ``` import json import socket from urllib.parse import urlparse from typing import List, Dict, Union, Optional, Callable from functools import wraps def check_server_available(func: Callable) -> Callable: """装饰器：检测服务是否可用，不可用直接返回error json""" @wraps(func) def wrapper(self, *args, **kwargs): if not self.config.is_server_available(): return json.dumps({ "status": "error", "message": "Server unavailable" }, ensure_ascii=False) return func(self, *args, **kwargs) return wrapper def validate_hex_address(address: Union[int, str]) -> Optional[str]: """校验并格式化十六进制地址，返回 0xxxxx 格式字符串，非法返回None""" if isinstance(address, int): return hex(address) addr_str = str(address).strip() if not addr_str: return None if addr_str.startswith(('0x', '0X')): try: int(addr_str, 16) return addr_str except ValueError: return None try: int(addr_str, 10) return addr_str except ValueError: return None class Config: def __init__(self, address: str = "127.0.0.1", port: int = 6891): self.address = address self.port = port self.server_addr = f"http://{address}:{port}" self.timeout = 5 def is_server_available(self, timeout: Optional[int] = None) -> bool: """检测端口是否开放""" timeout = timeout or self.timeout try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(timeout) result = sock.connect_ex((self.address, self.port)) return result == 0 except socket.error as e: print(f"WARNING: Server check failed: {str(e)}") return False class BaseHttpClient: def __init__(self, config: Optional[Config] = None): self.config = config or Config() parsed_url = urlparse(self.config.server_addr) self.address = parsed_url.hostname self.port = parsed_url.port self.scheme = parsed_url.scheme self.path = parsed_url.path or '/' self.verify_ssl = True def custom_post(self, payload: Optional[Dict] = None, timeout: Optional[int] = None) -> str: """发送POST请求，payload对应BinSentry接口 {interface, params}""" import http.client headers = {"Content-Type": "application/json"} body = json.dumps(payload).encode("utf-8") if payload else None timeout = timeout or self.config.timeout try: if self.scheme == "https": import ssl context = ssl._create_unverified_context() if not self.verify_ssl else None conn = http.client.HTTPSConnection(self.address, self.port, timeout=timeout, context=context) else: conn = http.client.HTTPConnection(self.address, self.port, timeout=timeout) conn.request("POST", self.path, body=body, headers=headers) response = conn.getresponse() response_text = response.read().decode("utf-8", errors="ignore") conn.close() return response_text except socket.timeout: return json.dumps({"status": "error", "message": "Request timed out"}, ensure_ascii=False) except ConnectionRefusedError: return json.dumps({"status": "error", "message": "Connection refused by server"}, ensure_ascii=False) except Exception as e: return json.dumps({"status": "error", "message": f"Request failed: {str(e)}"}, ensure_ascii=False) class SystemApi(BaseHttpClient): """系统与进程相关接口""" @check_server_available def system_info(self) -> str: return self.custom_post({"interface": "SystemInfo", "params": {}}) @check_server_available def process_list(self) -> str: return self.custom_post({"interface": "ProcessList", "params": {}}) @check_server_available def enum_windows(self, pid: int) -> str: return self.custom_post({"interface": "EnumWindows", "params": {"pid": pid}}) @check_server_available def is_process_elevated(self) -> str: return self.custom_post({"interface": "IsProcessElevated", "params": {}}) @check_server_available def enum_error_codes(self) -> str: return self.custom_post({"interface": "EnumErrorCodes", "params": {}}) @check_server_available def enum_exceptions(self) -> str: return self.custom_post({"interface": "EnumExceptions", "params": {}}) @check_server_available def help(self) -> str: return self.custom_post({"interface": "Help", "params": {}}) @check_server_available def get_jit(self) -> str: return self.custom_post({"interface": "GetJIT", "params": {}}) @check_server_available def set_jit(self, path: str) -> str: return self.custom_post({"interface": "SetJIT", "params": {"path": path}}) @check_server_available def get_command_line(self) -> str: return self.custom_post({"interface": "GetCommandLine", "params": {}}) @check_server_available def set_command_line(self, args: str) -> str: return self.custom_post({"interface": "SetCommandLine", "params": {"args": args}}) class LogConfigApi(BaseHttpClient): """日志与配置数据库""" @check_server_available def clear_log(self) -> str: return self.custom_post({"interface": "ClearLog", "params": {}}) @check_server_available def logs(self) -> str: return self.custom_post({"interface": "Logs", "params": {}}) @check_server_available def save_log(self, path: str) -> str: return self.custom_post({"interface": "SaveLog", "params": {"path": path}}) @check_server_available def load_database(self, path: str) -> str: return self.custom_post({"interface": "LoadDatabase", "params": {"path": path}}) @check_server_available def save_database(self, path: str) -> str: return self.custom_post({"interface": "SaveDatabase", "params": {"path": path}}) @check_server_available def load_config(self, module: str) -> str: return self.custom_post({"interface": "LoadConfig", "params": {"module": module}}) class SymbolVarApi(BaseHttpClient): """变量、参数、注释、标签、函数""" @check_server_available def set_var(self, name: str, value: str) -> str: return self.custom_post({"interface": "SetVar", "params": {"name": name, "value": value}}) @check_server_available def del_var(self, name: str) -> str: return self.custom_post({"interface": "DelVar", "params": {"name": name}}) @check_server_available def get_vars(self) -> str: return self.custom_post({"interface": "GetVars", "params": {}}) @check_server_available def add_argument(self, start: Union[int, str], end: Union[int, str] = "", name: str = "") -> str: start_addr = validate_hex_address(start) payload: Dict = {"start": start_addr} if end: payload["end"] = validate_hex_address(end) if name: payload["name"] = name return self.custom_post({"interface": "AddArgument", "params": payload}) @check_server_available def del_argument(self, start: Union[int, str]) -> str: s = validate_hex_address(start) return self.custom_post({"interface": "DelArgument", "params": {"start": s}}) @check_server_available def get_arguments(self) -> str: return self.custom_post({"interface": "GetArguments", "params": {}}) @check_server_available def get_comments(self) -> str: return self.custom_post({"interface": "GetComments", "params": {}}) @check_server_available def get_labels(self) -> str: return self.custom_post({"interface": "GetLabels", "params": {}}) @check_server_available def get_bookmarks(self) -> str: return self.custom_post({"interface": "GetBookMarks", "params": {}}) @check_server_available def functions(self) -> str: return self.custom_post({"interface": "Functions", "params": {}}) class BreakPointApi(BaseHttpClient): """断点：软件/硬件/内存/api/异常/dll断点""" @check_server_available def show_breakpoint(self) -> str: return self.custom_post({"interface": "ShowBreakPoint", "params": {}}) @check_server_available def show_hbreakpoint(self) -> str: return self.custom_post({"interface": "ShowHbreakPoint", "params": {}}) @check_server_available def show_mem_breakpoint(self) -> str: return self.custom_post({"interface": "ShowMemBreakPoint", "params": {}}) @check_server_available def show_api_breakpoint(self) -> str: return self.custom_post({"interface": "ShowApiBreakPoint", "params": {}}) @check_server_available def set_bpx_options(self, option: str, enable: int) -> str: return self.custom_post({"interface": "SetBPXOptions", "params": {"option": option, "enable": enable}}) @check_server_available def set_exception_bpx(self, code: str) -> str: return self.custom_post({"interface": "SetExceptionBPX", "params": {"code": code}}) @check_server_available def del_exception_bpx(self, code: str) -> str: return self.custom_post({"interface": "DelExceptionBPX", "params": {"code": code}}) @check_server_available def get_exception_bpx_list(self) -> str: return self.custom_post({"interface": "GetExceptionBPXList", "params": {}}) @check_server_available def set_dll_breakpoint(self, dll: str) -> str: return self.custom_post({"interface": "SetDllBreakPoint", "params": {"dll": dll}}) @check_server_available def del_dll_breakpoint(self, dll: str) -> str: return self.custom_post({"interface": "DelDllBreakPoint", "params": {"dll": dll}}) @check_server_available def get_dll_breakpoints(self) -> str: return self.custom_post({"interface": "GetDllBreakPoints", "params": {}}) @check_server_available def set_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakPoint", "params": {"address": addr}}) @check_server_available def del_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelBreakPoint", "params": {"address": addr}}) @check_server_available def get_breakpoint_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointInfo", "params": {"address": addr}}) @check_server_available def get_breakpoint_type(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointType", "params": {"address": addr}}) @check_server_available def set_breakpoint_name(self, address: Union[int, str], name: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointName", "params": {"address": addr, "name": name}}) @check_server_available def set_breakpoint_singleshoot(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointSingleshoot", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_fast_resume(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointFastResume", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_silent(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointSilent", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_log(self, address: Union[int, str], text: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointLog", "params": {"address": addr, "text": text}}) @check_server_available def set_breakpoint_log_file(self, address: Union[int, str], file: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointLogFile", "params": {"address": addr, "file": file}}) @check_server_available def set_breakpoint_hit_count(self, address: Union[int, str], count: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointHitCount", "params": {"address": addr, "count": count}}) @check_server_available def get_breakpoint_hit_count(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointHitCount", "params": {"address": addr}}) @check_server_available def reset_breakpoint_hit_count(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "ResetBreakpointHitCount", "params": {"address": addr}}) @check_server_available def set_cond_breakpoint(self, address: Union[int, str], cond: str, thread: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetCondBreakPoint", "params": {"address": addr, "cond": cond, "thread": thread}}) @check_server_available def del_cond_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelCondBreakPoint", "params": {"address": addr}}) @check_server_available def set_hbreakpoint(self, address: Union[int, str], len_: str = "1", flag: str = "e") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetHbreakPoint", "params": {"address": addr, "len": len_, "flag": flag}}) @check_server_available def del_hbreakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelHbreakPoint", "params": {"address": addr}}) @check_server_available def set_mem_breakpoint(self, address: Union[int, str], flag: str = "e") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetMemBreakPoint", "params": {"address": addr, "flag": flag}}) @check_server_available def del_mem_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelMemBreakPoint", "params": {"address": addr}}) @check_server_available def set_api_breakpoint(self, dll: str, api: str) -> str: return self.custom_post({"interface": "SetApiBreakPoint", "params": {"dll": dll, "api": api}}) @check_server_available def del_api_breakpoint(self, dll: str, api: str) -> str: return self.custom_post({"interface": "DelApiBreakPoint", "params": {"dll": dll, "api": api}}) @check_server_available def disable_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DisableBreakpoint", "params": {"address": addr}}) @check_server_available def enable_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "EnableBreakpoint", "params": {"address": addr}}) @check_server_available def breakpoint_command(self, address: Union[int, str], command: str = "") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "BreakpointCommand", "params": {"address": addr, "command": command}}) class DebugSessionApi(BaseHttpClient): """会话管理：启动调试、停止、分离、dump""" @check_server_available def debug(self, path: str, args: str = "", cwd: str = "") -> str: return self.custom_post({"interface": "Debug", "params": {"path": path, "args": args, "cwd": cwd}}) @check_server_available def set_debug(self, command: str = "Logs") -> str: return self.custom_post({"interface": "SetDebug", "params": command}) @check_server_available def stop(self) -> str: return self.custom_post({"interface": "Stop", "params": {}}) @check_server_available def detach(self) -> str: return self.custom_post({"interface": "Detach", "params": {}}) @check_server_available def restart(self) -> str: return self.custom_post({"interface": "Restart", "params": {}}) @check_server_available def dump_process(self, path: str, base: Union[int, str], size: int) -> str: base_addr = validate_hex_address(base) return self.custom_post({"interface": "DumpProcess", "params": {"path": path, "base": base_addr, "size": size}}) @check_server_available def minidump(self, path: str) -> str: return self.custom_post({"interface": "minidump", "params": {"path": path}}) @check_server_available def status(self) -> str: return self.custom_post({"interface": "Status", "params": {}}) class RegisterThreadApi(BaseHttpClient): """寄存器、线程操作""" @check_server_available def register(self) -> str: return self.custom_post({"interface": "Register", "params": {}}) @check_server_available def set_register(self, reg: str, value: str) -> str: return self.custom_post({"interface": "SetRegister", "params": {"reg": reg, "value": value}}) @check_server_available def threads(self) -> str: return self.custom_post({"interface": "Threads", "params": {}}) @check_server_available def thread_info(self, tid: int) -> str: return self.custom_post({"interface": "ThreadInfo", "params": {"tid": tid}}) @check_server_available def get_active_thread(self) -> str: return self.custom_post({"interface": "GetActiveThread", "params": {}}) @check_server_available def set_active_thread(self, tid: int) -> str: return self.custom_post({"interface": "SetActiveThread", "params": {"tid": tid}}) @check_server_available def get_thread_last_error(self, tid: int) -> str: return self.custom_post({"interface": "GetThreadLastError", "params": {"tid": tid}}) @check_server_available def get_thread_priority(self, tid: int) -> str: return self.custom_post({"interface": "GetThreadPriority", "params": {"tid": tid}}) @check_server_available def set_thread_priority(self, tid: int, priority: str = "NORMAL") -> str: return self.custom_post({"interface": "SetThreadPriority", "params": {"tid": tid, "priority": priority}}) @check_server_available def set_thread_name(self, tid: int, name: str) -> str: return self.custom_post({"interface": "SetThreadName", "params": {"tid": tid, "name": name}}) @check_server_available def set_flag(self, flag: str, value: int) -> str: return self.custom_post({"interface": "SetFlag", "params": {"flag": flag, "value": value}}) class ModulePeApi(BaseHttpClient): """模块、PE、导入导出、符号""" @check_server_available def modules(self) -> str: return self.custom_post({"interface": "Modules", "params": {}}) @check_server_available def module_info(self, module: str) -> str: return self.custom_post({"interface": "ModuleInfo", "params": {"module": module}}) @check_server_available def sections(self, module: str) -> str: return self.custom_post({"interface": "Sections", "params": {"module": module}}) @check_server_available def pe_info(self, path: str) -> str: return self.custom_post({"interface": "PEInfo", "params": {"path": path}}) @check_server_available def rich_header(self, module: str) -> str: return self.custom_post({"interface": "RichHeader", "params": {"module": module}}) @check_server_available def tls_callbacks(self, module: str) -> str: return self.custom_post({"interface": "TLSCallbacks", "params": {"module": module}}) @check_server_available def relocation_list(self, module: str) -> str: return self.custom_post({"interface": "RelocationList", "params": {"module": module}}) @check_server_available def debug_directory(self, module: str) -> str: return self.custom_post({"interface": "DebugDirectory", "params": {"module": module}}) @check_server_available def import_list(self, module: str) -> str: return self.custom_post({"interface": "ImportList", "params": {"module": module}}) @check_server_available def export_list(self, module: str) -> str: return self.custom_post({"interface": "ExportList", "params": {"module": module}}) @check_server_available def get_import_address(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetImportAddress", "params": {"module": module, "name": name}}) @check_server_available def get_export_address(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetExportAddress", "params": {"module": module, "name": name}}) @check_server_available def gpa(self, dll: str, api: str) -> str: return self.custom_post({"interface": "gpa", "params": {"dll": dll, "api": api}}) @check_server_available def addr_to_module(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "AddrToModule", "params": {"address": addr}}) @check_server_available def get_section_data(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetSectionData", "params": {"module": module, "name": name}}) @check_server_available def get_section_info(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetSectionInfo", "params": {"module": module, "name": name}}) @check_server_available def symbol(self, expr: str) -> str: return self.custom_post({"interface": "Symbol", "params": {"expr": expr}}) @check_server_available def get_symbol_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetSymbolInfo", "params": {"address": addr}}) @check_server_available def get_function_size(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetFunctionSize", "params": {"address": addr}}) @check_server_available def heaps(self) -> str: return self.custom_post({"interface": "Heaps", "params": {}}) @check_server_available def handles(self, max_num: int = 20) -> str: return self.custom_post({"interface": "Handles", "params": {"max": max_num}}) @check_server_available def seh_list(self) -> str: return self.custom_post({"interface": "SEHList", "params": {}}) class MemoryApi(BaseHttpClient): """内存读写、搜索、补丁、内存分配""" @check_server_available def memory_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "MemoryInfo", "params": {"address": addr}}) @check_server_available def regions(self) -> str: return self.custom_post({"interface": "Regions", "params": {}}) @check_server_available def memory(self, address: Union[int, str], size: int = 16) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "Memory", "params": {"address": addr, "size": size}}) @check_server_available def read_memory_value(self, address: Union[int, str], size: int = 4) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "ReadMemoryValue", "params": {"address": addr, "size": size}}) @check_server_available def write_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "WriteMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def get_page_rights(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetPageRights", "params": {"address": addr}}) @check_server_available def set_page_rights(self, address: Union[int, str], protect: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetPageRights", "params": {"address": addr, "protect": protect}}) @check_server_available def set_page_memory(self, address: Union[int, str], protect: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetPageMemory", "params": {"address": addr, "protect": protect}}) @check_server_available def allocate_memory(self, size: int) -> str: return self.custom_post({"interface": "AllocateMemory", "params": {"size": size}}) @check_server_available def free_memory(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FreeMemory", "params": {"address": addr}}) @check_server_available def set_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def fill_memory(self, address: Union[int, str], size: int, value: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FillMemory", "params": {"address": addr, "size": size, "value": value}}) @check_server_available def memcpy(self, src: Union[int, str], dst: Union[int, str], size: int) -> str: src_a = validate_hex_address(src) dst_a = validate_hex_address(dst) return self.custom_post({"interface": "Memcpy", "params": {"src": src_a, "dst": dst_a, "size": size}}) @check_server_available def va_to_file_offset(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "VaToFileOffset", "params": {"address": addr}}) @check_server_available def file_offset_to_va(self, offset: str) -> str: return self.custom_post({"interface": "FileOffsetToVa", "params": {"offset": offset}}) @check_server_available def get_string(self, address: Union[int, str], max_num: int = 64, type_: str = "ascii") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetString", "params": {"address": addr, "max": max_num, "type": type_}}) @check_server_available def search_memory(self, pattern: str, start: Union[int, str], size: int) -> str: s = validate_hex_address(start) return self.custom_post({"interface": "SearchMemory", "params": {"pattern": pattern, "start": s, "size": size}}) @check_server_available def search_all_memory(self, pattern: str, max_num: int = 5) -> str: return self.custom_post({"interface": "SearchAllMemory", "params": {"pattern": pattern, "max": max_num}}) @check_server_available def search_strings(self, address: Union[int, str], size: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SearchStrings", "params": {"address": addr, "size": size}}) @check_server_available def match_pattern(self, address: Union[int, str], pattern: str, size: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "MatchPattern", "params": {"address": addr, "pattern": pattern, "size": size}}) @check_server_available def hash_memory(self, address: Union[int, str], size: int, algo: str = "md5") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "HashMemory", "params": {"address": addr, "size": size, "algo": algo}}) @check_server_available def patch_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "PatchMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def patches(self) -> str: return self.custom_post({"interface": "Patches", "params": {}}) @check_server_available def revert_patch(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "RevertPatch", "params": {"address": addr}}) @check_server_available def delete_patch(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DeletePatch", "params": {"address": addr}}) class DisasmXrefApi(BaseHttpClient): """反汇编、汇编、交叉引用""" @check_server_available def disassemble_at(self, address: Union[int, str], count: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DisassembleAt", "params": {"address": addr, "count": count}}) @check_server_available def dissasembler(self, address: Union[int, str], count: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "Dissasembler", "params": {"address": addr, "count": count}}) @check_server_available def get_opcode_size(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetOpcodeSize", "params": {"address": addr}}) @check_server_available def mnemonicbrief(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "mnemonicbrief", "params": {"address": addr}}) @check_server_available def get_branch_target(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBranchTarget", "params": {"address": addr}}) @check_server_available def assemble(self, instr: str, cip: Union[int, str]) -> str: c = validate_hex_address(cip) return self.custom_post({"interface": "Assemble", "params": {"instr": instr, "cip": c}}) @check_server_available def assemble_at(self, address: Union[int, str], instr: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "AssembleAt", "params": {"address": addr, "instr": instr}}) @check_server_available def xrefs(self, address: Union[int, str], max_num: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "xrefs", "params": {"address": addr, "max": max_num}}) @check_server_available def find_ref(self, address: Union[int, str], max_num: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FindRef", "params": {"address": addr, "max": max_num}}) class StackTraceApi(BaseHttpClient): """栈、调用栈、追踪记录""" @check_server_available def call_stack(self, max_num: int = 16) -> str: return self.custom_post({"interface": "CallStack", "params": {"max": max_num}}) @check_server_available def stack(self, count: int = 8) -> str: return self.custom_post({"interface": "Stack", "params": {"count": count}}) @check_server_available def stack_push(self, value: str) -> str: return self.custom_post({"interface": "StackPush", "params": {"value": value}}) @check_server_available def stack_pop(self) -> str: return self.custom_post({"interface": "StackPop", "params": {}}) @check_server_available def stack_peek(self, offset: int = 0) -> str: return self.custom_post({"interface": "StackPeek", "params": {"offset": offset}}) @check_server_available def start_trace_record(self) -> str: return self.custom_post({"interface": "StartTraceRecord", "params": {}}) @check_server_available def get_trace_record(self, count: int = 5) -> str: return self.custom_post({"interface": "GetTraceRecord", "params": {"count": count}}) @check_server_available def stop_trace_record(self) -> str: return self.custom_post({"interface": "StopTraceRecord", "params": {}}) @check_server_available def jmp_history(self) -> str: return self.custom_post({"interface": "JmpHistory", "params": {}}) @check_server_available def last_exception(self) -> str: return self.custom_post({"interface": "LastException", "params": {}}) @check_server_available def eval_expr(self, expr: str) -> str: return self.custom_post({"interface": "Eval", "params": {"expr": expr}}) @check_server_available def show_debugger(self) -> str: return self.custom_post({"interface": "ShowDebugger", "params": {}}) @check_server_available def hide_debugger(self) -> str: return self.custom_post({"interface": "HideDebugger", "params": {}}) class ExecutionControlApi(BaseHttpClient): """执行控制：运行、单步、暂停、run_to""" @check_server_available def run(self) -> str: return self.custom_post({"interface": "Run", "params": {}}) @check_server_available def pause(self) -> str: return self.custom_post({"interface": "Pause", "params": {}}) @check_server_available def e_run(self) -> str: return self.custom_post({"interface": "ERun", "params": {}}) @check_server_available def se_run(self) -> str: return self.custom_post({"interface": "SERun", "params": {}}) @check_server_available def step_in(self) -> str: return self.custom_post({"interface": "StepIn", "params": {}}) @check_server_available def step_over(self) -> str: return self.custom_post({"interface": "StepOver", "params": {}}) @check_server_available def step_out(self) -> str: return self.custom_post({"interface": "StepOut", "params": {}}) @check_server_available def e_step_into(self) -> str: return self.custom_post({"interface": "EStepInto", "params": {}}) @check_server_available def e_step_over(self) -> str: return self.custom_post({"interface": "EStepOver", "params": {}}) @check_server_available def e_step_out(self) -> str: return self.custom_post({"interface": "EStepOut", "params": {}}) @check_server_available def step_user(self) -> str: return self.custom_post({"interface": "StepUser", "params": {}}) @check_server_available def step_system(self) -> str: return self.custom_post({"interface": "StepSystem", "params": {}}) @check_server_available def skip(self, count: int = 1) -> str: return self.custom_post({"interface": "Skip", "params": {"count": count}}) @check_server_available def instr_undo(self) -> str: return self.custom_post({"interface": "InstrUndo", "params": {}}) @check_server_available def execute_command(self, command: str) -> str: return self.custom_post({"interface": "ExecuteCommand", "params": command}) @check_server_available def trace_into(self, count: int = 3) -> str: return self.custom_post({"interface": "TraceInto", "params": {"count": count}}) @check_server_available def trace_over(self, count: int = 3) -> str: return self.custom_post({"interface": "TraceOver", "params": {"count": count}}) @check_server_available def trace_line(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "TraceLine", "params": {"address": addr}}) @check_server_available def run_to(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "RunTo", "params": {"address": addr}}) @check_server_available def run_to_user_code(self) -> str: return self.custom_post({"interface": "RunToUserCode", "params": {}}) @check_server_available def debug_continue(self, status: int = 0) -> str: return self.custom_post({"interface": "DebugContinue", "params": {"status": status}}) @check_server_available def pause_all_threads(self) -> str: return self.custom_post({"interface": "PauseAllThreads", "params": {}}) @check_server_available def resume_all_threads(self) -> str: return self.custom_post({"interface": "ResumeAllThreads", "params": {}}) @check_server_available def thread_pause(self, tid: int) -> str: return self.custom_post({"interface": "ThreadPause", "params": {"tid": tid}}) @check_server_available def thread_resume(self, tid: int) -> str: return self.custom_post({"interface": "ThreadResume", "params": {"tid": tid}}) @check_server_available def animate_into(self, count: int = 3) -> str: return self.custom_post({"interface": "AnimateInto", "params": {"count": count}}) @check_server_available def animate_over(self, count: int = 3) -> str: return self.custom_post({"interface": "AnimateOver", "params": {"count": count}}) @check_server_available def animate_stop(self) -> str: return self.custom_post({"interface": "AnimateStop", "params": {}}) # 统一入口类把所有API聚合在一起 class BinSentryClient( SystemApi, LogConfigApi, SymbolVarApi, BreakPointApi, DebugSessionApi, RegisterThreadApi, ModulePeApi, MemoryApi, DisasmXrefApi, StackTraceApi, ExecutionControlApi ): def __init__(self, config: Optional[Config] = None): super().__init__(config) ``` 调用接口测试 ``` from binsentry import BinSentryClient, Config if __name__ == "__main__": cfg = Config(address="127.0.0.1", port=6891) client = BinSentryClient(cfg) # 获取状态 resp = client.status() print(resp) # 设置断点，支持数字int或者字符串地址 resp = client.set_breakpoint(0x401000) print(resp) ``` 输出内容： ``` {"status":"success","result":{"sessionActive":false,"mode":"None","pid":0,"tid":0,"stopped":false,"firstRun":true,"softBreakpoints":0,"memoryBreakpoints":0," hardwareBreakpoints":0,"apiBreakpoints":0,"labels":0,"comments":0,"logCount":0,"watches":0,"bookmarks":0,"variables":0,"traceActive":false,"traceIntoActive": false,"traceResult":"","interfaces":[{"interface":"Debug","usage":"启动调试会话: params.path = 进程路径"},{"interface":"Attach","usage":"附加进程: params.pid = 进程ID"},{"interface":"Detach","usage":"分离调试会话"},{"interface":"Stop","usage":"终止被调试进程"},{"interface":"Status","usage":"查询调试器状态(含接口列表)"},{"interface":"WaitStop","usage":"等待调试器停止: params.timeout = 超时秒数(默认60)"},{"interface":"Run","usage":"继续运行程序"},{"interface":"StepIn" ,"usage":"单步步入"},{"interface":"StepOver","usage":"单步步过"},{"interface":"StepOut","usage":"单步步出"},{"interface":"SetBreakPoint","usage":"设置软件断点: address, thread(可选), temp(可选)"},{"interface":"DelBreakPoint","usage":"删除软件断点: address"},{"interface":"ShowBreakPoint","usage":"显示软件断点列表 "},{"interface":"SetMemBreakPoint","usage":"设置内存断点: address, flag(r/w/e)"},{"interface":"DelMemBreakPoint","usage":"删除内存断点: address"},{"interface ":"ShowMemBreakPoint","usage":"显示内存断点列表"},{"interface":"SetHbreakPoint","usage":"设置硬件断点: address, len(1/2/4), flag(r/w/e)"},{"interface":"DelHb reakPoint","usage":"删除硬件断点: dr(可选, 默认全部)"},{"interface":"ShowHbreakPoint","usage":"显示硬件断点列表"},{"interface":"SetApiBreakPoint","usage":"设置API断点: dll, api(可选--end为函数返回处)"},{"interface":"DelApiBreakPoint","usage":"删除API断点: dll, api(可选--end)"},{"interface":"ShowApiBreakPoint","us age":"显示API断点列表"},{"interface":"SearchMemory","usage":"搜索内存: pattern(hex, ?=通配), start(可选), size(可选)"},{"interface":"PatchMemory","usage":"内存补丁: address, hex(字节串), nop(可选)"},{"interface":"DumpProcess","usage":"转储进程(脱壳): path, entry(可选)"},{"interface":"HideDebugger","usage":"隐藏调试器(CheckRemoteDebuggerPresent等)"},{"interface":"ShowDebugger","usage":"恢复调试器可见"},{"interface":"ThreadPause","usage":"暂停线程: tid"},{"interface":" ThreadResume","usage":"恢复线程: tid"},{"interface":"PauseAllThreads","usage":"暂停所有线程"},{"interface":"ResumeAllThreads","usage":"恢复所有线程"},{"inter face":"PEInfo","usage":"PE文件信息: path"},{"interface":"Register","usage":"显示寄存器"},{"interface":"SetRegister","usage":"修改寄存器: reg(eax/ebx/...), va lue"},{"interface":"Stack","usage":"显示栈: count(可选)"},{"interface":"Memory","usage":"显示内存: address, size(可选)"},{"interface":"SetMemory","usage":"修改内存: address, value"},{"interface":"Dissasembler","usage":"反汇编: address(可选, 默认EIP), size(可选)"},{"interface":"Modules","usage":"显示模块列表"},{"i nterface":"Threads","usage":"显示线程列表"},{"interface":"SetDebug","usage":"执行调试命令字符串: params = \"命令行\" 或 {command: \"命令行\"}"},{"interface": "ProcessList","usage":"进程列表: filter(可选, 名称模糊过滤)"},{"interface":"Regions","usage":"内存区域列表(VirtualQueryEx)"},{"interface":"CallStack","usage" :"调用栈: count(可选, 默认32, StackWalk64+符号)"},{"interface":"SearchStrings","usage":"搜索进程字符串: max(可选, 默认100, ASCII+Unicode)"},{"interface":"Fil lMemory","usage":"内存填充: address, size, value (FillEx)"},{"interface":"SetCondBreakPoint","usage":"设置条件断点: address, cond(表达式, 如 \"eax == 0x5\"), thread(可选)"},{"interface":"DelCondBreakPoint","usage":"清除断点条件: address (转为普通断点)"},{"interface":"SetLabel","usage":"设置标签: address, name"},{ "interface":"DelLabel","usage":"删除标签: address"},{"interface":"GetLabels","usage":"显示标签列表"},{"interface":"SetComment","usage":"设置注释: address, te xt"},{"interface":"DelComment","usage":"删除注释: address"},{"interface":"GetComments","usage":"显示注释列表"},{"interface":"Logs","usage":"查询调试日志: cou nt(可选, 默认50)"},{"interface":"Eval","usage":"表达式求值: expr(寄存器/0x十六进制/十进制/运算, 需停止态)"},{"interface":"SetFlag","usage":"设置标志位: flag( CF PF AF ZF SF TF IF DF OF), value(0/1)"},{"interface":"GetBranchTarget","usage":"分支指令目标地址: address (GetJumpDestinationEx)"},{"interface":"GetArgumen ts","usage":"参数标记列表"},{"interface":"AddArgument","usage":"添加参数标记: start, end(可选), name(可选)"},{"interface":"DelArgument","usage":"删除参数标记 : start"},{"interface":"RevertPatch","usage":"还原补丁: address(恢复原字节并移除记录)"},{"interface":"DeletePatch","usage":"删除补丁记录: address(不写内存)"} ,{"interface":"ReadMemoryValue","usage":"读数值: address, size(1/2/4/8)"},{"interface":"GetThreadPriority","usage":"读线程优先级: tid"},{"interface":"SystemI nfo","usage":"系统信息(架构/页大小/CPU/内存/OS版本)"},{"interface":"GetSectionInfo","usage":"指定节详情: module, name"},{"interface":"GetBreakpointInfo","usa ge":"指定地址断点详情: address (软/内存/硬件)"},{"interface":"InjectDll","usage":"远程注入DLL: path (LoadLibraryA+CreateRemoteThread)"},{"interface":"HashMem ory","usage":"内存哈希: address, size, algo(可选 md5/sha1/crc32)"},{"interface":"AddrToModule","usage":"地址归属模块: address (模块名+基址+RVA)"},{"interface ":"RunTo","usage":"运行到目标地址: address (临时断点, 命中自动移除)"},{"interface":"SetExceptionIgnore","usage":"设置异常忽略: code, ignore(0/1; 断点/单步不可忽略)"},{"interface":"GetExceptionSettings","usage":"异常忽略列表"},{"interface":"StartTraceRecord","usage":"开始执行足迹记录(每次停止记录EIP, 上限4096)"}, {"interface":"StopTraceRecord","usage":"停止执行足迹记录"},{"interface":"GetTraceRecord","usage":"执行足迹列表"},{"interface":"GetImportAddress","usage":"导入函数IAT地址: module, name (或 #序号)"},{"interface":"GetSectionData","usage":"节内容转储: module, name (hex, 限4KB)"},{"interface":"TraceOver","usage":"步过跟踪: count(步数,可选), condition(表达式,可选); call 自动跳过"},{"interface":"LoadLibrary","usage":"目标进程远程加载DLL: path (返回 moduleHandle)"},{"inter face":"FreeLibrary","usage":"目标进程远程卸载DLL: handle (LoadLibrary 返回值)"},{"interface":"gpa","usage":"远程API地址解析: dll, api 或 expr(模块:函数)"},{" interface":"VaToFileOffset","usage":"虚拟地址->文件偏移: address"},{"interface":"FileOffsetToVa","usage":"文件偏移->虚拟地址: offset, module(可选,默认主模块) "},{"interface":"GetPageRights","usage":"页权限查询: address (返回 rights 如 RX/RWX)"},{"interface":"SetPageRights","usage":"页权限修改: address, protect(hex 或RWX串)"},{"interface":"minidump","usage":"生成目标进程转储: path (.dmp, 含全内存)"},{"interface":"xrefs","usage":"代码引用查找: address, max(可选)"},{"inte rface":"mnemonicbrief","usage":"指令助记符简介: address(可选,默认EIP)"},{"interface":"EnumWindows","usage":"枚举目标进程窗口: pid(可选,默认会话进程)"},{"inte rface":"IsProcessElevated","usage":"进程提权检查: pid(可选,默认本进程)"},{"interface":"ClearLog","usage":"清空调试日志缓冲区"},{"interface":"SaveLog","usage" :"保存日志到文件: path"},{"interface":"EnumErrorCodes","usage":"常见Win32错误码表: code(可选过滤)"},{"interface":"EnumExceptions","usage":"常见异常码表: code (可选过滤)"},{"interface":"GetFunctionSize","usage":"函数边界: address (函数表优先, 反汇编扫描兜底)"},{"interface":"BreakpointCommand","usage":"断点命令: add ress, command (命中时自动入队执行; 空清除)"},{"interface":"JmpHistory","usage":"跳转历史(恢复运行前记录分支指令)"},{"interface":"GetSymbolInfo","usage":"符号详情: address (dbghelp: 名称/模块/偏移/行号)"},{"interface":"TCPConnections","usage":"目标进程 TCP 连接列表"},{"interface":"StackPush","usage":"栈压入: value (ESP-4 写入)"},{"interface":"StackPop","usage":"栈弹出(读栈顶+ESP+4)"},{"interface":"StackPeek","usage":"栈窥视: offset (ESP+offset*4)"},{"interface":"SetTh readName","usage":"线程命名: tid, name (SetThreadDescription)"},{"interface":"Restart","usage":"重启目标进程(分离旧会话并重新创建)"},{"interface":"SetJIT","u sage":"设置系统JIT调试器: path (HKLM AeDebug, 需管理员)"},{"interface":"GetJIT","usage":"查询当前JIT调试器"},{"interface":"TraceLine","usage":"单步跟踪到目标地址: address, timeout(可选, 秒)"},{"interface":"Handles","usage":"句柄列表: max(可选, NtQuerySystemInformation+对象名)"},{"interface":"FindRef","usage":"查找引用: address(目标地址), max(可选, 反汇编扫描代码区)"},{"interface":"SearchAllMemory","usage":"全内存搜索: pattern(hex, ?=通配), max(可选)"},{"interface":" ExportList","usage":"模块导出表: module(模块名)"},{"interface":"SetWatch","usage":"添加/更新监视: name, expr(表达式, 含 $变量)"},{"interface":"DelWatch","usa ge":"删除监视: id 或 name"},{"interface":"GetWatches","usage":"查询监视(实时求值, 需停止态)"},{"interface":"SetVar","usage":"设置用户变量: name, value (表达式 $name 引用)"},{"interface":"DelVar","usage":"删除用户变量: name"},{"interface":"GetVars","usage":"查询用户变量列表"},{"interface":"SetBookMark","usage":" 添加书签: address"},{"interface":"DelBookMark","usage":"删除书签: address"},{"interface":"GetBookMarks","usage":"查询书签列表"},{"interface":"Assemble","usag e":"汇编指令为字节码(XEDParse): instr, cip(可选)"},{"interface":"AssembleAt","usage":"汇编并写入目标进程: address, instr, fillnop(可选)"},{"interface":"Enabl eBreakpoint","usage":"启用软断点: address"},{"interface":"DisableBreakpoint","usage":"禁用软断点: address"},{"interface":"SetThreadPriority","usage":"设置线程优先级: tid, priority(数值或名称)"},{"interface":"Memcpy","usage":"进程内内存复制: src, dst, size"},{"interface":"AllocateMemory","usage":"目标进程分配内存 : size, protect(可选)"},{"interface":"FreeMemory","usage":"释放目标进程内存: address"},{"interface":"Symbol","usage":"符号解析(模块名+偏移或表达式): expr"} ,{"interface":"MemoryInfo","usage":"地址内存信息(区域/保护/模块归属): address"},{"interface":"IsValidPointer","usage":"地址是否可读: address"},{"interface":" LastException","usage":"最近一次异常信息"},{"interface":"Pause","usage":"暂停运行中的程序(DebugBreakProcess)"},{"interface":"CreateRemoteThread","usage":"目标进程创建远程线程: start 或 hex(shellcode), param(可选)"},{"interface":"TerminateThread","usage":"结束线程: tid"},{"interface":"GetThreadLastError","usage": "读取线程 TEB.LastErrorValue: tid"},{"interface":"WriteMemory","usage":"按 hex 字节串写内存: address, hex"},{"interface":"GetString","usage":"读取内存字符串: address, max(可选), type(ascii/unicode)"},{"interface":"GetOpcodeSize","usage":"指令长度: address (LengthDisassembleEx)"},{"interface":"Functions","usage":" 函数标记列表"},{"interface":"AddFunction","usage":"添加函数标记: start, end(可选), name(可选)"},{"interface":"DelFunction","usage":"删除函数标记: start"},{"i nterface":"ProcessInfo","usage":"进程详细信息: pid(可选, 默认会话进程)"},{"interface":"GetExportAddress","usage":"模块导出函数地址: module, name 或 ordinal"} ,{"interface":"Help","usage":"显示帮助"}]},"timestamp":4107265} ``` 改进 ``` import json import re import socket import time from urllib.parse import urlparse from typing import List, Dict, Union, Optional, Callable from functools import wraps def check_server_available(func: Callable) -> Callable: """装饰器：检测服务是否可用，不可用直接返回error json""" @wraps(func) def wrapper(self, *args, **kwargs): if not self.config.is_server_available(): return json.dumps({ "status": "error", "message": "Server unavailable" }, ensure_ascii=False) return func(self, *args, **kwargs) return wrapper def validate_hex_address(address: Union[int, str]) -> Optional[str]: """校验并格式化十六进制地址，返回 0xxxxx 格式字符串，非法返回None""" if isinstance(address, int): return hex(address) addr_str = str(address).strip() if not addr_str: return None if addr_str.startswith(('0x', '0X')): try: int(addr_str, 16) return addr_str except ValueError: return None try: int(addr_str, 10) return addr_str except ValueError: return None class Config: def __init__(self, address: str = "127.0.0.1", port: int = 6891): self.address = address self.port = port self.server_addr = f"http://{address}:{port}" self.timeout = 5 def is_server_available(self, timeout: Optional[int] = None) -> bool: """检测端口是否开放""" timeout = timeout or self.timeout try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(timeout) result = sock.connect_ex((self.address, self.port)) return result == 0 except socket.error as e: print(f"WARNING: Server check failed: {str(e)}") return False class BaseHttpClient: def __init__(self, config: Optional[Config] = None): self.config = config or Config() parsed_url = urlparse(self.config.server_addr) self.address = parsed_url.hostname self.port = parsed_url.port self.scheme = parsed_url.scheme self.path = parsed_url.path or '/' self.verify_ssl = True def custom_post(self, payload: Optional[Dict] = None, timeout: Optional[int] = None) -> str: """发送POST请求，payload对应BinSentry接口 {interface, params}""" import http.client headers = {"Content-Type": "application/json"} body = json.dumps(payload).encode("utf-8") if payload else None timeout = timeout or self.config.timeout try: if self.scheme == "https": import ssl context = ssl._create_unverified_context() if not self.verify_ssl else None conn = http.client.HTTPSConnection(self.address, self.port, timeout=timeout, context=context) else: conn = http.client.HTTPConnection(self.address, self.port, timeout=timeout) conn.request("POST", self.path, body=body, headers=headers) response = conn.getresponse() response_text = response.read().decode("utf-8", errors="ignore") conn.close() return response_text except socket.timeout: return json.dumps({"status": "error", "message": "Request timed out"}, ensure_ascii=False) except ConnectionRefusedError: return json.dumps({"status": "error", "message": "Connection refused by server"}, ensure_ascii=False) except Exception as e: return json.dumps({"status": "error", "message": f"Request failed: {str(e)}"}, ensure_ascii=False) class SystemApi(BaseHttpClient): """系统与进程相关接口""" @check_server_available def system_info(self) -> str: return self.custom_post({"interface": "SystemInfo", "params": {}}) @check_server_available def process_list(self) -> str: return self.custom_post({"interface": "ProcessList", "params": {}}) @check_server_available def enum_windows(self, pid: int) -> str: return self.custom_post({"interface": "EnumWindows", "params": {"pid": pid}}) @check_server_available def is_process_elevated(self) -> str: return self.custom_post({"interface": "IsProcessElevated", "params": {}}) @check_server_available def enum_error_codes(self) -> str: return self.custom_post({"interface": "EnumErrorCodes", "params": {}}) @check_server_available def enum_exceptions(self) -> str: return self.custom_post({"interface": "EnumExceptions", "params": {}}) @check_server_available def help(self) -> str: return self.custom_post({"interface": "Help", "params": {}}) @check_server_available def get_jit(self) -> str: return self.custom_post({"interface": "GetJIT", "params": {}}) @check_server_available def set_jit(self, path: str) -> str: return self.custom_post({"interface": "SetJIT", "params": {"path": path}}) @check_server_available def get_command_line(self) -> str: return self.custom_post({"interface": "GetCommandLine", "params": {}}) @check_server_available def set_command_line(self, args: str) -> str: return self.custom_post({"interface": "SetCommandLine", "params": {"args": args}}) class LogConfigApi(BaseHttpClient): """日志与配置数据库""" @check_server_available def clear_log(self) -> str: return self.custom_post({"interface": "ClearLog", "params": {}}) @check_server_available def logs(self) -> str: return self.custom_post({"interface": "Logs", "params": {}}) @check_server_available def save_log(self, path: str) -> str: return self.custom_post({"interface": "SaveLog", "params": {"path": path}}) @check_server_available def load_database(self, path: str) -> str: return self.custom_post({"interface": "LoadDatabase", "params": {"path": path}}) @check_server_available def save_database(self, path: str) -> str: return self.custom_post({"interface": "SaveDatabase", "params": {"path": path}}) @check_server_available def load_config(self, module: str) -> str: return self.custom_post({"interface": "LoadConfig", "params": {"module": module}}) class SymbolVarApi(BaseHttpClient): """变量、参数、注释、标签、函数""" @check_server_available def set_var(self, name: str, value: str) -> str: return self.custom_post({"interface": "SetVar", "params": {"name": name, "value": value}}) @check_server_available def del_var(self, name: str) -> str: return self.custom_post({"interface": "DelVar", "params": {"name": name}}) @check_server_available def get_vars(self) -> str: return self.custom_post({"interface": "GetVars", "params": {}}) @check_server_available def add_argument(self, start: Union[int, str], end: Union[int, str] = "", name: str = "") -> str: start_addr = validate_hex_address(start) payload: Dict = {"start": start_addr} if end: payload["end"] = validate_hex_address(end) if name: payload["name"] = name return self.custom_post({"interface": "AddArgument", "params": payload}) @check_server_available def del_argument(self, start: Union[int, str]) -> str: s = validate_hex_address(start) return self.custom_post({"interface": "DelArgument", "params": {"start": s}}) @check_server_available def get_arguments(self) -> str: return self.custom_post({"interface": "GetArguments", "params": {}}) @check_server_available def get_comments(self) -> str: return self.custom_post({"interface": "GetComments", "params": {}}) @check_server_available def get_labels(self) -> str: return self.custom_post({"interface": "GetLabels", "params": {}}) @check_server_available def get_bookmarks(self) -> str: return self.custom_post({"interface": "GetBookMarks", "params": {}}) @check_server_available def functions(self) -> str: return self.custom_post({"interface": "Functions", "params": {}}) class BreakPointApi(BaseHttpClient): """断点：软件/硬件/内存/api/异常/dll断点""" @check_server_available def show_breakpoint(self) -> str: return self.custom_post({"interface": "ShowBreakPoint", "params": {}}) @check_server_available def show_hbreakpoint(self) -> str: return self.custom_post({"interface": "ShowHbreakPoint", "params": {}}) @check_server_available def show_mem_breakpoint(self) -> str: return self.custom_post({"interface": "ShowMemBreakPoint", "params": {}}) @check_server_available def show_api_breakpoint(self) -> str: return self.custom_post({"interface": "ShowApiBreakPoint", "params": {}}) @check_server_available def set_bpx_options(self, option: str, enable: int) -> str: return self.custom_post({"interface": "SetBPXOptions", "params": {"option": option, "enable": enable}}) @check_server_available def set_exception_bpx(self, code: str) -> str: return self.custom_post({"interface": "SetExceptionBPX", "params": {"code": code}}) @check_server_available def del_exception_bpx(self, code: str) -> str: return self.custom_post({"interface": "DelExceptionBPX", "params": {"code": code}}) @check_server_available def get_exception_bpx_list(self) -> str: return self.custom_post({"interface": "GetExceptionBPXList", "params": {}}) @check_server_available def set_dll_breakpoint(self, dll: str) -> str: return self.custom_post({"interface": "SetDllBreakPoint", "params": {"dll": dll}}) @check_server_available def del_dll_breakpoint(self, dll: str) -> str: return self.custom_post({"interface": "DelDllBreakPoint", "params": {"dll": dll}}) @check_server_available def get_dll_breakpoints(self) -> str: return self.custom_post({"interface": "GetDllBreakPoints", "params": {}}) @check_server_available def set_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakPoint", "params": {"address": addr}}) @check_server_available def del_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelBreakPoint", "params": {"address": addr}}) @check_server_available def get_breakpoint_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointInfo", "params": {"address": addr}}) @check_server_available def get_breakpoint_type(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointType", "params": {"address": addr}}) @check_server_available def set_breakpoint_name(self, address: Union[int, str], name: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointName", "params": {"address": addr, "name": name}}) @check_server_available def set_breakpoint_singleshoot(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointSingleshoot", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_fast_resume(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointFastResume", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_silent(self, address: Union[int, str], enable: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointSilent", "params": {"address": addr, "enable": enable}}) @check_server_available def set_breakpoint_log(self, address: Union[int, str], text: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointLog", "params": {"address": addr, "text": text}}) @check_server_available def set_breakpoint_log_file(self, address: Union[int, str], file: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointLogFile", "params": {"address": addr, "file": file}}) @check_server_available def set_breakpoint_hit_count(self, address: Union[int, str], count: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetBreakpointHitCount", "params": {"address": addr, "count": count}}) @check_server_available def get_breakpoint_hit_count(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBreakpointHitCount", "params": {"address": addr}}) @check_server_available def reset_breakpoint_hit_count(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "ResetBreakpointHitCount", "params": {"address": addr}}) @check_server_available def set_cond_breakpoint(self, address: Union[int, str], cond: str, thread: int = 0) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetCondBreakPoint", "params": {"address": addr, "cond": cond, "thread": thread}}) @check_server_available def del_cond_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelCondBreakPoint", "params": {"address": addr}}) @check_server_available def set_hbreakpoint(self, address: Union[int, str], len_: str = "1", flag: str = "e") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetHbreakPoint", "params": {"address": addr, "len": len_, "flag": flag}}) @check_server_available def del_hbreakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelHbreakPoint", "params": {"address": addr}}) @check_server_available def set_mem_breakpoint(self, address: Union[int, str], flag: str = "e") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetMemBreakPoint", "params": {"address": addr, "flag": flag}}) @check_server_available def del_mem_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DelMemBreakPoint", "params": {"address": addr}}) @check_server_available def set_api_breakpoint(self, dll: str, api: str) -> str: return self.custom_post({"interface": "SetApiBreakPoint", "params": {"dll": dll, "api": api}}) @check_server_available def del_api_breakpoint(self, dll: str, api: str) -> str: return self.custom_post({"interface": "DelApiBreakPoint", "params": {"dll": dll, "api": api}}) @check_server_available def disable_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DisableBreakpoint", "params": {"address": addr}}) @check_server_available def enable_breakpoint(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "EnableBreakpoint", "params": {"address": addr}}) @check_server_available def breakpoint_command(self, address: Union[int, str], command: str = "") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "BreakpointCommand", "params": {"address": addr, "command": command}}) class DebugSessionApi(BaseHttpClient): """会话管理：启动调试、停止、分离、dump""" @check_server_available def debug(self, path: str, args: str = "", cwd: str = "") -> str: return self.custom_post({"interface": "Debug", "params": {"path": path, "args": args, "cwd": cwd}}) @check_server_available def set_debug(self, command: str = "Logs") -> str: return self.custom_post({"interface": "SetDebug", "params": command}) @check_server_available def stop(self) -> str: return self.custom_post({"interface": "Stop", "params": {}}) @check_server_available def detach(self) -> str: return self.custom_post({"interface": "Detach", "params": {}}) @check_server_available def restart(self) -> str: return self.custom_post({"interface": "Restart", "params": {}}) @check_server_available def dump_process(self, path: str, base: Union[int, str], size: int) -> str: base_addr = validate_hex_address(base) return self.custom_post({"interface": "DumpProcess", "params": {"path": path, "base": base_addr, "size": size}}) @check_server_available def minidump(self, path: str) -> str: return self.custom_post({"interface": "minidump", "params": {"path": path}}) @check_server_available def status(self) -> str: return self.custom_post({"interface": "Status", "params": {}}) class RegisterThreadApi(BaseHttpClient): """寄存器、线程操作""" @check_server_available def register(self) -> str: return self.custom_post({"interface": "Register", "params": {}}) @check_server_available def set_register(self, reg: str, value: str) -> str: return self.custom_post({"interface": "SetRegister", "params": {"reg": reg, "value": value}}) @check_server_available def threads(self) -> str: return self.custom_post({"interface": "Threads", "params": {}}) @check_server_available def thread_info(self, tid: int) -> str: return self.custom_post({"interface": "ThreadInfo", "params": {"tid": tid}}) @check_server_available def get_active_thread(self) -> str: return self.custom_post({"interface": "GetActiveThread", "params": {}}) @check_server_available def set_active_thread(self, tid: int) -> str: return self.custom_post({"interface": "SetActiveThread", "params": {"tid": tid}}) @check_server_available def get_thread_last_error(self, tid: int) -> str: return self.custom_post({"interface": "GetThreadLastError", "params": {"tid": tid}}) @check_server_available def get_thread_priority(self, tid: int) -> str: return self.custom_post({"interface": "GetThreadPriority", "params": {"tid": tid}}) @check_server_available def set_thread_priority(self, tid: int, priority: str = "NORMAL") -> str: return self.custom_post({"interface": "SetThreadPriority", "params": {"tid": tid, "priority": priority}}) @check_server_available def set_thread_name(self, tid: int, name: str) -> str: return self.custom_post({"interface": "SetThreadName", "params": {"tid": tid, "name": name}}) @check_server_available def set_flag(self, flag: str, value: int) -> str: return self.custom_post({"interface": "SetFlag", "params": {"flag": flag, "value": value}}) class ModulePeApi(BaseHttpClient): """模块、PE、导入导出、符号""" @check_server_available def modules(self) -> str: return self.custom_post({"interface": "Modules", "params": {}}) @check_server_available def module_info(self, module: str) -> str: return self.custom_post({"interface": "ModuleInfo", "params": {"module": module}}) @check_server_available def sections(self, module: str) -> str: return self.custom_post({"interface": "Sections", "params": {"module": module}}) @check_server_available def pe_info(self, path: str) -> str: return self.custom_post({"interface": "PEInfo", "params": {"path": path}}) @check_server_available def rich_header(self, module: str) -> str: return self.custom_post({"interface": "RichHeader", "params": {"module": module}}) @check_server_available def tls_callbacks(self, module: str) -> str: return self.custom_post({"interface": "TLSCallbacks", "params": {"module": module}}) @check_server_available def relocation_list(self, module: str) -> str: return self.custom_post({"interface": "RelocationList", "params": {"module": module}}) @check_server_available def debug_directory(self, module: str) -> str: return self.custom_post({"interface": "DebugDirectory", "params": {"module": module}}) @check_server_available def import_list(self, module: str) -> str: return self.custom_post({"interface": "ImportList", "params": {"module": module}}) @check_server_available def export_list(self, module: str) -> str: return self.custom_post({"interface": "ExportList", "params": {"module": module}}) @check_server_available def get_import_address(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetImportAddress", "params": {"module": module, "name": name}}) @check_server_available def get_export_address(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetExportAddress", "params": {"module": module, "name": name}}) @check_server_available def gpa(self, dll: str, api: str) -> str: return self.custom_post({"interface": "gpa", "params": {"dll": dll, "api": api}}) @check_server_available def addr_to_module(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "AddrToModule", "params": {"address": addr}}) @check_server_available def get_section_data(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetSectionData", "params": {"module": module, "name": name}}) @check_server_available def get_section_info(self, module: str, name: str) -> str: return self.custom_post({"interface": "GetSectionInfo", "params": {"module": module, "name": name}}) @check_server_available def symbol(self, expr: str) -> str: return self.custom_post({"interface": "Symbol", "params": {"expr": expr}}) @check_server_available def get_symbol_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetSymbolInfo", "params": {"address": addr}}) @check_server_available def get_function_size(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetFunctionSize", "params": {"address": addr}}) @check_server_available def heaps(self) -> str: return self.custom_post({"interface": "Heaps", "params": {}}) @check_server_available def handles(self, max_num: int = 20) -> str: return self.custom_post({"interface": "Handles", "params": {"max": max_num}}) @check_server_available def seh_list(self) -> str: return self.custom_post({"interface": "SEHList", "params": {}}) class MemoryApi(BaseHttpClient): """内存读写、搜索、补丁、内存分配""" @check_server_available def memory_info(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "MemoryInfo", "params": {"address": addr}}) @check_server_available def regions(self) -> str: return self.custom_post({"interface": "Regions", "params": {}}) @check_server_available def memory(self, address: Union[int, str], size: int = 16) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "Memory", "params": {"address": addr, "size": size}}) @check_server_available def read_memory_value(self, address: Union[int, str], size: int = 4) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "ReadMemoryValue", "params": {"address": addr, "size": size}}) @check_server_available def write_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "WriteMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def get_page_rights(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetPageRights", "params": {"address": addr}}) @check_server_available def set_page_rights(self, address: Union[int, str], protect: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetPageRights", "params": {"address": addr, "protect": protect}}) @check_server_available def set_page_memory(self, address: Union[int, str], protect: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetPageMemory", "params": {"address": addr, "protect": protect}}) @check_server_available def allocate_memory(self, size: int) -> str: return self.custom_post({"interface": "AllocateMemory", "params": {"size": size}}) @check_server_available def free_memory(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FreeMemory", "params": {"address": addr}}) @check_server_available def set_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SetMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def fill_memory(self, address: Union[int, str], size: int, value: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FillMemory", "params": {"address": addr, "size": size, "value": value}}) @check_server_available def memcpy(self, src: Union[int, str], dst: Union[int, str], size: int) -> str: src_a = validate_hex_address(src) dst_a = validate_hex_address(dst) return self.custom_post({"interface": "Memcpy", "params": {"src": src_a, "dst": dst_a, "size": size}}) @check_server_available def va_to_file_offset(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "VaToFileOffset", "params": {"address": addr}}) @check_server_available def file_offset_to_va(self, offset: str) -> str: return self.custom_post({"interface": "FileOffsetToVa", "params": {"offset": offset}}) @check_server_available def get_string(self, address: Union[int, str], max_num: int = 64, type_: str = "ascii") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetString", "params": {"address": addr, "max": max_num, "type": type_}}) @check_server_available def search_memory(self, pattern: str, start: Union[int, str], size: int) -> str: s = validate_hex_address(start) return self.custom_post({"interface": "SearchMemory", "params": {"pattern": pattern, "start": s, "size": size}}) @check_server_available def search_all_memory(self, pattern: str, max_num: int = 5) -> str: return self.custom_post({"interface": "SearchAllMemory", "params": {"pattern": pattern, "max": max_num}}) @check_server_available def search_strings(self, address: Union[int, str], size: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "SearchStrings", "params": {"address": addr, "size": size}}) @check_server_available def match_pattern(self, address: Union[int, str], pattern: str, size: int) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "MatchPattern", "params": {"address": addr, "pattern": pattern, "size": size}}) @check_server_available def hash_memory(self, address: Union[int, str], size: int, algo: str = "md5") -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "HashMemory", "params": {"address": addr, "size": size, "algo": algo}}) @check_server_available def patch_memory(self, address: Union[int, str], hex_: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "PatchMemory", "params": {"address": addr, "hex": hex_}}) @check_server_available def patches(self) -> str: return self.custom_post({"interface": "Patches", "params": {}}) @check_server_available def revert_patch(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "RevertPatch", "params": {"address": addr}}) @check_server_available def delete_patch(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DeletePatch", "params": {"address": addr}}) class DisasmXrefApi(BaseHttpClient): """反汇编、汇编、交叉引用""" @check_server_available def disassemble_at(self, address: Union[int, str], count: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "DisassembleAt", "params": {"address": addr, "count": count}}) @check_server_available def dissasembler(self, address: Union[int, str], count: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "Dissasembler", "params": {"address": addr, "count": count}}) @check_server_available def get_opcode_size(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetOpcodeSize", "params": {"address": addr}}) @check_server_available def mnemonicbrief(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "mnemonicbrief", "params": {"address": addr}}) @check_server_available def get_branch_target(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "GetBranchTarget", "params": {"address": addr}}) @check_server_available def assemble(self, instr: str, cip: Union[int, str]) -> str: c = validate_hex_address(cip) return self.custom_post({"interface": "Assemble", "params": {"instr": instr, "cip": c}}) @check_server_available def assemble_at(self, address: Union[int, str], instr: str) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "AssembleAt", "params": {"address": addr, "instr": instr}}) @check_server_available def xrefs(self, address: Union[int, str], max_num: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "xrefs", "params": {"address": addr, "max": max_num}}) @check_server_available def find_ref(self, address: Union[int, str], max_num: int = 5) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "FindRef", "params": {"address": addr, "max": max_num}}) class StackTraceApi(BaseHttpClient): """栈、调用栈、追踪记录""" @check_server_available def call_stack(self, max_num: int = 16) -> str: return self.custom_post({"interface": "CallStack", "params": {"max": max_num}}) @check_server_available def stack(self, count: int = 8) -> str: return self.custom_post({"interface": "Stack", "params": {"count": count}}) @check_server_available def stack_push(self, value: str) -> str: return self.custom_post({"interface": "StackPush", "params": {"value": value}}) @check_server_available def stack_pop(self) -> str: return self.custom_post({"interface": "StackPop", "params": {}}) @check_server_available def stack_peek(self, offset: int = 0) -> str: return self.custom_post({"interface": "StackPeek", "params": {"offset": offset}}) @check_server_available def start_trace_record(self) -> str: return self.custom_post({"interface": "StartTraceRecord", "params": {}}) @check_server_available def get_trace_record(self, count: int = 5) -> str: return self.custom_post({"interface": "GetTraceRecord", "params": {"count": count}}) @check_server_available def stop_trace_record(self) -> str: return self.custom_post({"interface": "StopTraceRecord", "params": {}}) @check_server_available def jmp_history(self) -> str: return self.custom_post({"interface": "JmpHistory", "params": {}}) @check_server_available def last_exception(self) -> str: return self.custom_post({"interface": "LastException", "params": {}}) @check_server_available def eval_expr(self, expr: str) -> str: return self.custom_post({"interface": "Eval", "params": {"expr": expr}}) @check_server_available def show_debugger(self) -> str: return self.custom_post({"interface": "ShowDebugger", "params": {}}) @check_server_available def hide_debugger(self) -> str: return self.custom_post({"interface": "HideDebugger", "params": {}}) class ExecutionControlApi(BaseHttpClient): """执行控制：运行、单步、暂停、run_to""" @check_server_available def run(self) -> str: return self.custom_post({"interface": "Run", "params": {}}) @check_server_available def pause(self) -> str: return self.custom_post({"interface": "Pause", "params": {}}) @check_server_available def e_run(self) -> str: return self.custom_post({"interface": "ERun", "params": {}}) @check_server_available def se_run(self) -> str: return self.custom_post({"interface": "SERun", "params": {}}) @check_server_available def step_in(self) -> str: return self.custom_post({"interface": "StepIn", "params": {}}) @check_server_available def step_over(self) -> str: return self.custom_post({"interface": "StepOver", "params": {}}) @check_server_available def step_out(self) -> str: return self.custom_post({"interface": "StepOut", "params": {}}) @check_server_available def e_step_into(self) -> str: return self.custom_post({"interface": "EStepInto", "params": {}}) @check_server_available def e_step_over(self) -> str: return self.custom_post({"interface": "EStepOver", "params": {}}) @check_server_available def e_step_out(self) -> str: return self.custom_post({"interface": "EStepOut", "params": {}}) @check_server_available def step_user(self) -> str: return self.custom_post({"interface": "StepUser", "params": {}}) @check_server_available def step_system(self) -> str: return self.custom_post({"interface": "StepSystem", "params": {}}) @check_server_available def skip(self, count: int = 1) -> str: return self.custom_post({"interface": "Skip", "params": {"count": count}}) @check_server_available def instr_undo(self) -> str: return self.custom_post({"interface": "InstrUndo", "params": {}}) @check_server_available def execute_command(self, command: str) -> str: return self.custom_post({"interface": "ExecuteCommand", "params": command}) @check_server_available def trace_into(self, count: int = 3) -> str: return self.custom_post({"interface": "TraceInto", "params": {"count": count}}) @check_server_available def trace_over(self, count: int = 3) -> str: return self.custom_post({"interface": "TraceOver", "params": {"count": count}}) @check_server_available def trace_line(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "TraceLine", "params": {"address": addr}}) @check_server_available def run_to(self, address: Union[int, str]) -> str: addr = validate_hex_address(address) return self.custom_post({"interface": "RunTo", "params": {"address": addr}}) @check_server_available def run_to_user_code(self) -> str: return self.custom_post({"interface": "RunToUserCode", "params": {}}) @check_server_available def debug_continue(self, status: int = 0) -> str: return self.custom_post({"interface": "DebugContinue", "params": {"status": status}}) @check_server_available def pause_all_threads(self) -> str: return self.custom_post({"interface": "PauseAllThreads", "params": {}}) @check_server_available def resume_all_threads(self) -> str: return self.custom_post({"interface": "ResumeAllThreads", "params": {}}) @check_server_available def thread_pause(self, tid: int) -> str: return self.custom_post({"interface": "ThreadPause", "params": {"tid": tid}}) @check_server_available def thread_resume(self, tid: int) -> str: return self.custom_post({"interface": "ThreadResume", "params": {"tid": tid}}) @check_server_available def animate_into(self, count: int = 3) -> str: return self.custom_post({"interface": "AnimateInto", "params": {"count": count}}) @check_server_available def animate_over(self, count: int = 3) -> str: return self.custom_post({"interface": "AnimateOver", "params": {"count": count}}) @check_server_available def animate_stop(self) -> str: return self.custom_post({"interface": "AnimateStop", "params": {}}) # 统一入口类把所有API聚合在一起 class BinSentryClient( SystemApi, LogConfigApi, SymbolVarApi, BreakPointApi, DebugSessionApi, RegisterThreadApi, ModulePeApi, MemoryApi, DisasmXrefApi, StackTraceApi, ExecutionControlApi ): def __init__(self, config: Optional[Config] = None): super().__init__(config) # ====================================================================== # 扩展功能区：以下方法全部由上方 11 个 API 类的现有接口组合而成， # 不改变、不遮蔽任何原始接口，仅新增面向逆向分析/漏洞挖掘的实用封装。 # 说明：扩展方法一律直发 HTTP（不套 check_server_available 装饰器）， # 服务端离线时快速抛 RuntimeError（Connection refused），而非每次 5s 探测。 # ====================================================================== # ---------------- 解析辅助 ---------------- def _parse_ok(self, raw: str, what: str) -> dict: """解析 JSON 并校验 status=success，返回 result dict；失败抛 RuntimeError""" try: obj = json.loads(raw) except Exception as e: raise RuntimeError(f"{what} 响应不是合法 JSON: {str(raw)[:160]}") from e if obj.get("status") != "success": res = obj.get("result") msg = res.get("message") if isinstance(res, dict) else (res if isinstance(res, str) else None) msg = msg or obj.get("message") or "未知错误" raise RuntimeError(f"{what} 失败: {msg}") return obj.get("result") or {} def _find(self, obj, *keys): """递归查找第一个命中的键值，容忍服务端字段名差异""" if isinstance(obj, dict): for k in keys: if k in obj and obj[k] is not None: return obj[k] for v in obj.values(): r = self._find(v, *keys) if r is not None: return r elif isinstance(obj, list): for v in obj: r = self._find(v, *keys) if r is not None: return r return None def _int(self, v): """宽容转 int：支持 int / '0x..' / 十进制字符串""" if v is None: return None if isinstance(v, int): return v s = str(v).strip() if not s: return None try: return int(s, 16) if s.lower().startswith("0x") else int(s) except ValueError: return None def _hex(self, v): iv = self._int(v) return hex(iv) if iv is not None else None def _collect_addresses(self, res) -> list: """从响应中收集所有地址（搜索/引用/匹配结果通用）""" out = [] def walk(o): if isinstance(o, dict): for k, v in o.items(): if k.lower() in ("address", "addr", "base", "target", "found") and self._int(v) is not None: out.append(self._int(v)) else: walk(v) elif isinstance(o, list): for v in o: walk(v) walk(res) return sorted(set(out)) @staticmethod def _region_is_exec(r) -> bool: rt = str(r.get("rights") or "").upper() if any(x in rt for x in ("X", "E")): return True p = r.get("protect") return isinstance(p, int) and bool(p & 0x10) # PAGE_EXECUTE 系列 @staticmethod def _region_is_writable(r) -> bool: rt = str(r.get("rights") or "").upper() if "W" in rt: return True p = r.get("protect") return isinstance(p, int) and bool(p & 0x04) # PAGE_READWRITE/WRITECOPY 系列 # ---------------- 会话管理 ---------------- def session_status(self) -> dict: """结构化 Status：会话激活、模式、pid/tid、停止态、断点计数""" res = self._parse_ok(self.custom_post({"interface": "Status", "params": {}}), "Status") return { "sessionActive": bool(res.get("sessionActive")), "mode": res.get("mode"), "pid": self._int(res.get("pid")), "tid": self._int(res.get("tid")), "stopped": bool(res.get("stopped")), "firstRun": bool(res.get("firstRun")), "traceActive": bool(res.get("traceActive")), "breakpoints": { "soft": res.get("softBreakpoints"), "mem": res.get("memoryBreakpoints"), "hw": res.get("hardwareBreakpoints"), "api": res.get("apiBreakpoints"), }, } def wait_stopped(self, timeout: int = 60) -> bool: """轮询等待调试器进入停止态；超时返回 False""" t0 = time.time() while time.time() - t0 bool: """运行中则 Pause，等待停止态""" st = self.session_status() if not st["sessionActive"]: raise RuntimeError("没有活动的调试会话") if st["stopped"]: return True self.custom_post({"interface": "Pause", "params": {}}) return self.wait_stopped(timeout) def restart_and_wait(self, timeout: int = 10) -> dict: """Restart 后轮询等待新会话出现（restart 是异步的，约 0.5s 后新会话才可见）""" self._parse_ok(self.custom_post({"interface": "Restart", "params": {}}), "Restart") t0 = time.time() while time.time() - t0 dict: """从异常运行态安全停下：animate_stop → Pause → Detach → 重新 Debug 重建会话""" self.custom_post({"interface": "AnimateStop", "params": {}}) self.custom_post({"interface": "Pause", "params": {}}) try: self.wait_stopped(timeout) except RuntimeError: pass self.custom_post({"interface": "Detach", "params": {}}) time.sleep(0.5) return self.session_status() # ---------------- 进程 ---------------- def processes(self) -> list: """结构化进程列表：[{pid, name, path}]""" res = self._parse_ok(self.process_list(), "ProcessList") procs = self._find(res, "processes", "list", "items") out = [] if isinstance(procs, list): for p in procs: if isinstance(p, dict) and p.get("pid") is not None: out.append({"pid": self._int(p.get("pid")), "name": p.get("name"), "path": p.get("path")}) return out def attach(self, pid: int) -> int: """附加指定 PID（服务端 Attach 接口，原文件未封装，此处补上）""" self._parse_ok(self.custom_post({"interface": "Attach", "params": {"pid": pid}}), f"Attach({pid})") return pid def attach_by_name(self, name: str) -> dict: """按进程名子串匹配附加；返回 {attached, candidates}""" cands = [p for p in self.processes() if name.lower() in str(p.get("name", "")).lower()] if not cands: raise RuntimeError(f"进程列表中没有匹配 '{name}' 的进程") self.attach(cands[0]["pid"]) return {"attached": cands[0]["pid"], "candidates": cands} def windows_of(self, pid: int): """进程窗口列表（枚举窗口接口）""" res = self._parse_ok(self.enum_windows(pid), f"EnumWindows({pid})") return self._find(res, "windows", "list", "items") or res def is_elevated(self) -> bool: """当前调试进程是否以管理员权限运行""" res = self._parse_ok(self.is_process_elevated(), "IsProcessElevated") return bool(self._find(res, "elevated", "isElevated", "result", "ok")) # ---------------- 模块 / PE ---------------- def get_modules(self) -> list: """模块列表：[{name, base, size, path}]""" res = self._parse_ok(self.modules(), "Modules") mods = self._find(res, "modules", "list", "items") out = [] if isinstance(mods, list): for m in mods: if isinstance(m, dict) and m.get("name"): out.append({ "name": m.get("name"), "base": self._int(m.get("baseAddressHex") or m.get("baseAddress") or m.get("base")), "size": self._int(m.get("size") or m.get("imageSize")), "path": m.get("path"), }) return out def find_module(self, name: str = None, addr=None) -> dict: """按名称（子串/精确）或地址定位模块；未找到返回 None""" for m in self.get_modules(): if name and (m["name"].lower() == name.lower() or name.lower() in m["name"].lower()): return m if addr is not None and m["base"] is not None and m["size"] is not None: if m["base"] str: """全模块摘要文本（基址/大小/名称/路径，按基址排序）""" lines = [] for m in sorted(self.get_modules(), key=lambda x: x["base"] or 0): base = f"0x{m['base']:08X}" if m["base"] else "?" size = f"0x{m['size']:X}" if m["size"] else "?" lines.append(f"{base} {size: list: """节表：[{name, va, vsize, raw, rsize, flags}]""" res = self._parse_ok(self.sections(module), f"Sections({module})") secs = self._find(res, "sections", "list", "items") out = [] if isinstance(secs, list): for s in secs: if isinstance(s, dict) and s.get("name"): out.append({ "name": s.get("name"), "va": self._int(s.get("virtualAddress") or s.get("va") or s.get("address")), "vsize": self._int(s.get("virtualSize") or s.get("vsize") or s.get("size")), "raw": self._int(s.get("rawOffset") or s.get("pointerToRawData")), "rsize": self._int(s.get("rawSize") or s.get("sizeOfRawData")), "flags": s.get("flags"), }) return out def module_headers(self, module: str) -> dict: """PE 头摘要（入口点/机器/节数量等），path 由模块信息自动解析； entry 为 RVA，entry_va 为加基址后的虚拟地址（可直接用于反汇编/设断点）""" m = self.find_module(module) path = (m or {}).get("path") or module base = (m or {}).get("base") res = self._parse_ok(self.pe_info(path), f"PEInfo({path})") entry = self._int(self._find(res, "entry", "entryPoint", "addressOfEntryPoint")) return { "entry": entry, "entry_va": (base + entry) if (base is not None and entry is not None) else None, "machine": self._find(res, "machine", "machineType"), "characteristics": self._find(res, "characteristics", "imageCharacteristics"), "sections": self._find(res, "sectionCount", "numberOfSections"), "magic": self._find(res, "magic", "optionalMagic"), } def module_detail(self, module: str) -> dict: """ModuleInfo 原样结构化（入口/基址/大小等）""" res = self._parse_ok(self.module_info(module), f"ModuleInfo({module})") return res def imports_of(self, module: str) -> list: """导入表（展开 DLL 下的函数级条目）：[{name, dll, hint, iat}]""" res = self._parse_ok(self.import_list(module), f"ImportList({module})") lst = self._find(res, "imports", "list", "items", "functions") out = [] if isinstance(lst, list): for d in lst: if not isinstance(d, dict): continue dll = d.get("dll") or d.get("module") funcs = d.get("funcs") or d.get("functions") or [] if isinstance(funcs, list) and funcs: for f in funcs: if isinstance(f, dict): out.append({ "name": f.get("name"), "dll": dll, "hint": f.get("hint") or f.get("ordinal"), "iat": self._int(f.get("iat") or f.get("address") or f.get("thunk")), }) else: out.append({ "name": d.get("name"), "dll": dll, "hint": d.get("hint") or d.get("ordinal"), "iat": self._int(d.get("iat") or d.get("address") or d.get("thunk")), }) return [x for x in out if x["name"] or x["dll"]] def exports_of(self, module: str) -> list: """导出表：[{name, ordinal, address}]""" res = self._parse_ok(self.export_list(module), f"ExportList({module})") lst = self._find(res, "exports", "list", "items", "functions") out = [] if isinstance(lst, list): for i in lst: if isinstance(i, dict): out.append({ "name": i.get("name"), "ordinal": self._int(i.get("ordinal")), "address": self._int(i.get("address") or i.get("va")), }) return out def imports_by_name(self, module: str, keyword: str) -> list: """按关键字过滤导入函数（含 DLL 名匹配）""" kw = keyword.lower() return [i for i in self.imports_of(module) if kw in (i["name"] or "").lower() or kw in (i["dll"] or "").lower()] def exports_by_name(self, module: str, keyword: str) -> list: """按关键字过滤导出函数（返回含地址，便于下 API 断点）""" kw = keyword.lower() return [e for e in self.exports_of(module) if kw in (e["name"] or "").lower()] def import_address(self, module: str, name: str) -> int: """导入函数 IAT 地址""" res = self._parse_ok(self.get_import_address(module, name), f"GetImportAddress({module}!{name})") return self._int(self._find(res, "address", "va", "iat", "value")) def export_address(self, module: str, name: str) -> int: """导出函数 VA""" res = self._parse_ok(self.get_export_address(module, name), f"GetExportAddress({module}!{name})") return self._int(self._find(res, "address", "va", "value")) def gpa_resolve(self, dll: str, api: str) -> int: """GetProcAddress 式解析：dll!api → 地址""" res = self._parse_ok(self.gpa(dll, api), f"gpa({dll}!{api})") return self._int(self._find(res, "address", "va", "value", "result")) def symbol_at(self, addr) -> dict: """地址 → 符号信息（模块/名称/偏移）""" res = self._parse_ok(self.get_symbol_info(self._hex(addr)), f"GetSymbolInfo(0x{self._int(addr):X})") return { "module": self._find(res, "module", "mod"), "name": self._find(res, "name", "symbol", "label"), "offset": self._int(self._find(res, "offset", "delta")), } def symbol_lookup(self, expr: str): """表达式 → 符号/地址信息""" res = self._parse_ok(self.symbol(expr), f"Symbol({expr})") return res def addr_to_module_name(self, addr) -> str: """地址所属模块名""" res = self._parse_ok(self.addr_to_module(self._hex(addr)), f"AddrToModule(0x{self._int(addr):X})") return self._find(res, "module", "name", "mod") or "" def function_size(self, addr) -> int: """函数大小（服务端按签名/下个函数推断）""" res = self._parse_ok(self.get_function_size(self._hex(addr)), f"GetFunctionSize(0x{self._int(addr):X})") return self._int(self._find(res, "size", "length", "count")) def section_data(self, module: str, name: str) -> bytes: """节原始数据 → bytes""" res = self._parse_ok(self.get_section_data(module, name), f"GetSectionData({module}!{name})") hexdata = self._find(res, "hex", "data", "bytes", "content") if isinstance(hexdata, list): return bytes(int(b) for b in hexdata) return bytes.fromhex(re.sub(r"[^0-9a-fA-F]", "", str(hexdata or ""))) def section_info(self, module: str, name: str) -> dict: """节信息详情""" res = self._parse_ok(self.get_section_info(module, name), f"GetSectionInfo({module}!{name})") return res def rich_header(self, module: str) -> dict: res = self._parse_ok(self.rich_header(module), f"RichHeader({module})") return res def tls_callbacks_list(self, module: str) -> list: """TLS 回调地址列表（反调试/初始化代码分析）""" res = self._parse_ok(self.tls_callbacks(module), f"TLSCallbacks({module})") return self._collect_addresses(res) def relocations(self, module: str) -> list: """重定位表（基址随机化分析）""" res = self._parse_ok(self.relocation_list(module), f"RelocationList({module})") return self._collect_addresses(res) def debug_directory(self, module: str) -> dict: """调试目录（PDB/源码信息）""" res = self._parse_ok(self.debug_directory(module), f"DebugDirectory({module})") return res # ---------------- 内存读写 ---------------- _CHUNK = 0x400 # 单次读写分块大小（保守值） def read_bytes(self, addr, size: int, chunk: int = None) -> bytes: """分块读取原始字节 → bytes""" chunk = chunk or self._CHUNK base = self._int(addr) if base is None: raise RuntimeError(f"非法地址: {addr}") out = bytearray() off = 0 while off str: """读取内存 → hex 字符串""" return self.read_bytes(addr, size).hex() def read_mem_value(self, addr, size: int = 4) -> int: """ReadMemoryValue：任意宽度整数读取""" res = self._parse_ok(self.read_memory_value(self._hex(addr), size), f"ReadMemoryValue(0x{self._int(addr):X},{size})") v = self._find(res, "value", "val", "result", "number") if v is None: raise RuntimeError(f"ReadMemoryValue 响应缺少值: {str(res)[:200]}") return self._int(v) def read_u8(self, addr) -> int: return self.read_mem_value(addr, 1) def read_u16(self, addr) -> int: return self.read_mem_value(addr, 2) def read_u32(self, addr) -> int: return self.read_mem_value(addr, 4) def read_u64(self, addr) -> int: return self.read_mem_value(addr, 8) def write_bytes(self, addr, data) -> int: """写内存（bytes 或 hex 字符串），分块；返回写入字节数""" if isinstance(data, str): data = bytes.fromhex(re.sub(r"[^0-9a-fA-F]", "", data)) base = self._int(addr) off = 0 while off str: return self.read_string(addr, "ascii", max_len) def read_wstring(self, addr, max_len: int = 256) -> str: return self.read_string(addr, "unicode", max_len) def read_string(self, addr, type_: str = "ascii", max_len: int = 256) -> str: """GetString：按类型（ascii/unicode）读取字符串""" res = self._parse_ok(self.get_string(self._hex(addr), max_len, type_), f"GetString(0x{self._int(addr):X},{type_})") s = self._find(res, "string", "value", "text", "content") if s is None: return "" if isinstance(s, (list, dict)): s = self._find(s, "string", "value", "text") return str(s) def memory_region(self, addr) -> dict: """地址所在内存区域（基址/大小/权限/类型）""" res = self._parse_ok(self.memory_info(self._hex(addr)), f"MemoryInfo(0x{self._int(addr):X})") return { "base": self._int(self._find(res, "base", "baseAddress", "regionBase")), "size": self._int(self._find(res, "size", "regionSize")), "rights": self._find(res, "rights", "protect", "protection"), "type": self._find(res, "type", "state"), "module": self._find(res, "module", "moduleName"), } def memory_map(self) -> list: """内存区域表：[{base, size, rights, protect, type}]""" res = self._parse_ok(self.regions(), "Regions") regs = self._find(res, "regions", "list", "items") out = [] if isinstance(regs, list): for r in regs: if isinstance(r, dict) and (r.get("base") is not None or r.get("address") is not None): rights = self._find(r, "rights", "protect", "protection", "flags") out.append({ "base": self._int(r.get("base") or r.get("address") or r.get("baseAddress")), "size": self._int(r.get("size") or r.get("regionSize")), "rights": rights, "protect": self._int(r.get("protect") or r.get("protection")), "type": r.get("type") or r.get("state"), }) return out def executable_regions(self) -> list: """可执行区域（代码扫描/设断点）""" return [r for r in self.memory_map() if self._region_is_exec(r)] def writable_regions(self) -> list: """可写区域（数据区/堆分析）""" return [r for r in self.memory_map() if self._region_is_writable(r)] def page_rights(self, addr) -> dict: """页面权限详情""" res = self._parse_ok(self.get_page_rights(self._hex(addr)), f"GetPageRights(0x{self._int(addr):X})") return res def set_protect(self, addr, protect: str) -> dict: """修改页面保护属性（如 'rwx'），用于可写可执行改造实验""" res = self._parse_ok(self.set_page_rights(self._hex(addr), protect), f"SetPageRights(0x{self._int(addr):X},{protect})") return res def alloc(self, size: int = 0x1000) -> int: """分配内存，返回地址""" res = self._parse_ok(self.allocate_memory(size), f"AllocateMemory({size})") addr = self._int(res.get("addressHex") or res.get("address")) if addr is None: raise RuntimeError(f"AllocateMemory 响应缺少地址: {str(res)[:200]}") return addr def free(self, addr) -> bool: """释放分配的内存""" self._parse_ok(self.free_memory(self._hex(addr)), f"FreeMemory(0x{self._int(addr):X})") return True def fill(self, addr, size: int, value: str) -> bool: """填充内存（value 为字节值）""" self._parse_ok(self.fill_memory(self._hex(addr), size, value), f"FillMemory(0x{self._int(addr):X})") return True def mem_copy(self, src, dst, size: int) -> bool: """进程内 memcpy""" self._parse_ok(self.memcpy(self._hex(src), self._hex(dst), size), f"Memcpy(0x{self._int(src):X}->0x{self._int(dst):X})") return True def va_to_offset(self, addr) -> int: """虚拟地址 → 文件偏移（PE 静态分析）""" res = self._parse_ok(self.va_to_file_offset(self._hex(addr)), f"VaToFileOffset(0x{self._int(addr):X})") return self._int(self._find(res, "offset", "fileOffset", "value")) def offset_to_va(self, offset) -> int: """文件偏移 → 虚拟地址（注意：offset 传 0 会被服务端判非法，需传非零值）""" res = self._parse_ok(self.file_offset_to_va(str(offset)), f"FileOffsetToVa({offset})") return self._int(self._find(res, "va", "address", "value")) def snapshot(self, addr, size: int) -> str: """内存快照 → hex（配合 restore 做 fuzz 回滚/补丁备份）""" return self.read_hex(addr, size) def restore(self, addr, hexdata: str) -> int: """按快照还原内存""" return self.write_bytes(addr, hexdata) def hash_region(self, addr, size: int, algo: str = "md5") -> str: """区域哈希（校验完整性/样本去重）""" res = self._parse_ok(self.hash_memory(self._hex(addr), size, algo), f"HashMemory({algo})") return self._find(res, "hash", "digest", "value") or "" # ---------------- 搜索 ---------------- def search_in_module(self, module: str, pattern: str, max_num: int = 5) -> list: """模块映射范围内搜索字节模式（pattern 支持 '?' 通配，如 'E8 ? ? ? ?'）""" m = self.find_module(module) if not m or m["base"] is None or m["size"] is None: raise RuntimeError(f"无法定位模块 {module}") res = self._parse_ok(self.search_memory(pattern, hex(m["base"]), m["size"]), f"SearchMemory({pattern} in {module})") return self._collect_addresses(res)[:max_num] def search_all(self, pattern: str, max_num: int = 10) -> list: """全内存搜索""" res = self._parse_ok(self.search_all_memory(pattern, max_num), f"SearchAllMemory({pattern})") return self._collect_addresses(res)[:max_num] def pattern_match(self, addr, pattern: str, size: int) -> bool: """检查 addr 处是否匹配模式（'?' 通配）""" res = self._parse_ok(self.match_pattern(self._hex(addr), pattern, size), f"MatchPattern(0x{self._int(addr):X})") return bool(self._find(res, "match", "matched", "found", "ok")) def strings_in_module(self, module: str, max_num: int = 200) -> list: """模块内字符串提取：[{address, string}]""" m = self.find_module(module) if not m or m["base"] is None: raise RuntimeError(f"无法定位模块 {module}") size = m["size"] or 0x100000 res = self._parse_ok(self.search_strings(hex(m["base"]), size), f"SearchStrings({module})") lst = self._find(res, "strings", "list", "items", "results") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): s = it.get("string") or it.get("text") or it.get("value") if s: out.append({"address": self._int(it.get("address") or it.get("addr")), "string": str(s)}) return out[:max_num] def read_pointer_chain(self, addr, depth: int = 8, mask: int = None) -> list: """解引用链：addr → *addr → **addr ...（对象/链表遍历）""" chain = [] cur = self._int(addr) seen = set() width = 8 # 默认按 x64；win32 等 x86 目标由调用方按需传参或用 system_info 判断 try: si = self._parse_ok(self.system_info(), "SystemInfo") arch = str(si.get("arch") or si.get("processorArchitecture") or "").lower() if "86" in arch or "x86" in arch or arch == "x32": width = 4 except RuntimeError: pass for _ in range(depth): if cur is None or cur in seen: break seen.add(cur) chain.append(cur) try: val = self.read_u64(cur) if width == 8 else self.read_u32(cur) except RuntimeError: break if val is None or val == 0: break if mask: val &= mask cur = val return chain # ---------------- 反汇编 / 静态分析 ---------------- def disasm(self, addr, count: int = 10) -> list: """反汇编 count 条：[{addr, mnemonic, text, size}]；text 优先取 titan 完整指令""" res = self._parse_ok(self.disassemble_at(self._hex(addr), count), f"DisassembleAt(0x{self._int(addr):X})") insns = self._find(res, "instructions", "lines", "list", "disasm") out = [] if isinstance(insns, list): for i in insns: if isinstance(i, dict): mn = (i.get("mnemonic") or "").strip() op = (i.get("opStr") or "").strip() text = (i.get("titan") or i.get("text") or i.get("disasm") or i.get("instruction") or "").strip() if not text and op: text = f"{mn} {op}".strip() size = self._int(i.get("size") or i.get("length")) if size is None and i.get("bytes"): size = len(str(i["bytes"]).replace(" ", "")) // 2 out.append({"addr": self._int(i.get("address") or i.get("addr")), "mnemonic": mn, "text": text, "size": size}) return out @staticmethod def _is_ret(insn) -> bool: return (insn.get("mnemonic") or "").lower().startswith("ret") @staticmethod def _is_branch(insn) -> bool: mn = (insn.get("mnemonic") or "").lower() return mn.startswith(("j", "call", "loop")) or BinSentryClient._is_ret(insn) def basic_block(self, addr, max_insns: int = 64) -> list: """顺序反汇编到分支/返回（不含 call 也保留），返回指令列表""" out = [] cur = self._int(addr) while len(out) = max_insns: return out cur = (out[-1]["addr"] or cur) + (out[-1]["size"] or 1) return out def function_prologue(self, addr) -> tuple: """识别函数序言：返回 (模式名, 匹配指令数, 前若干条指令)""" insns = self.disasm(addr, 12) if not insns: return None, 0, [] mns = [(i.get("mnemonic") or "").lower() for i in insns] txts = [(i.get("text") or "").lower() for i in insns] if mns[:2] == ["push", "mov"] and "ebp" in txts[1]: return "push ebp; mov ebp,esp", 2, insns[:6] if mns and mns[0] == "push" and "esp" in txts[0]: return "push regs prologue", 1, insns[:6] if mns[:2] == ["mov", "push"] and "esp" in txts[1]: return "mov; push regs prologue", 2, insns[:6] if mns[:2] == ["push", "mov"] and "rsp" in txts[1]: return "push rbp; mov rbp,rsp (x64)", 2, insns[:6] if mns and mns[0] == "endbr64": return "endbr64 (CET)", 1, insns[:6] if mns and mns[0] == "sub" and "rsp" in txts[0]: return "sub rsp, imm (x64 leaf)", 1, insns[:6] return "未知序言", 0, insns[:6] def analyze_function(self, addr) -> dict: """函数分析：序言 + 函数大小 + 前 6 条指令""" prologue, n, insns = self.function_prologue(addr) size = None try: size = self.function_size(addr) except RuntimeError: pass return {"entry": self._int(addr), "prologue": prologue, "prologue_insns": n, "size": size, "instructions": insns} _GADGET_PATTERNS = { "ret": "C3", "jmp esp": "FF E4", "jmp eax": "FF E0", "jmp ecx": "FF E1", "call esp": "FF D4", "pop eax; ret": "58 C3", "pop ebx; ret": "5B C3", "pop ecx; ret": "59 C3", "pop edx; ret": "5A C3", "pop esi; ret": "5E C3", "pop edi; ret": "5F C3", "pop ebp; ret": "5D C3", "leave; ret": "C9 C3", "jmp [esp]": "FF 24 24", "int3": "CC", } def find_gadgets(self, module: str, patterns: dict = None, max_per_gadget: int = 20) -> dict: """ROP gadget 扫描：在模块可执行范围搜常见字节模式（启发式）""" patterns = patterns or self._GADGET_PATTERNS m = self.find_module(module) if not m or m["base"] is None or m["size"] is None: raise RuntimeError(f"无法定位模块 {module}") found = {} for name, pat in patterns.items(): try: res = self._parse_ok(self.search_memory(pat, hex(m["base"]), m["size"]), f"gadget {name}") addrs = self._collect_addresses(res)[:max_per_gadget] if addrs: found[name] = addrs except RuntimeError: continue return found def xrefs_to(self, addr, max_num: int = 10) -> list: """交叉引用（谁引用了该地址）""" res = self._parse_ok(self.xrefs(self._hex(addr), max_num), f"xrefs(0x{self._int(addr):X})") return self._collect_addresses(res)[:max_num] def find_refs(self, addr, max_num: int = 10) -> list: """反汇编扫描引用""" res = self._parse_ok(self.find_ref(self._hex(addr), max_num), f"FindRef(0x{self._int(addr):X})") return self._collect_addresses(res)[:max_num] def opcode_size_at(self, addr) -> int: """指令长度""" res = self._parse_ok(self.get_opcode_size(self._hex(addr)), f"GetOpcodeSize(0x{self._int(addr):X})") return self._int(self._find(res, "size", "length", "count")) def branch_target(self, addr) -> int: """分支指令的目标地址""" res = self._parse_ok(self.get_branch_target(self._hex(addr)), f"GetBranchTarget(0x{self._int(addr):X})") return self._int(self._find(res, "target", "address", "dest")) def mnemonic_at(self, addr) -> str: """指令助记符简述""" res = self._parse_ok(self.mnemonicbrief(self._hex(addr)), f"mnemonicbrief(0x{self._int(addr):X})") return self._find(res, "mnemonic", "brief", "description", "text") or "" def asm(self, instr: str, cip=None) -> str: """汇编一条指令为字节 hex；cip 缺省取当前 eip""" if cip is None: try: cip = self.reg("eip") or self.reg("rip") except RuntimeError: cip = 0 res = self._parse_ok(self.assemble(instr, self._hex(cip)), f"Assemble({instr})") return self._find(res, "hex", "bytes", "opcode", "code") or "" def asm_at(self, addr, instr: str) -> int: """汇编并写入目标进程（就地补丁/改指令）""" self._parse_ok(self.assemble_at(self._hex(addr), instr), f"AssembleAt({instr})") return self._int(addr) # ---------------- 断点 / 运行监控 ---------------- def set_bp(self, addr, name: str = None, log: str = None, cond: str = None, singleshoot: bool = False) -> int: """组合式设软断点：地址 + 可选名称/日志/条件/单次命中""" self._parse_ok(self.set_breakpoint(self._hex(addr)), f"SetBreakPoint(0x{self._int(addr):X})") if name: self._parse_ok(self.set_breakpoint_name(self._hex(addr), name), "SetBreakpointName") if log: self._parse_ok(self.set_breakpoint_log(self._hex(addr), log), "SetBreakpointLog") if cond: self._parse_ok(self.set_cond_breakpoint(self._hex(addr), cond, 0), "SetCondBreakPoint") if singleshoot: self._parse_ok(self.set_breakpoint_singleshoot(self._hex(addr), 1), "SetBreakpointSingleshoot") return self._int(addr) def del_bp(self, addr) -> int: self._parse_ok(self.del_breakpoint(self._hex(addr)), f"DelBreakPoint(0x{self._int(addr):X})") return self._int(addr) def bp_list(self) -> list: """软件断点列表：[{id, address, temporary, threadId, condition}]""" res = self._parse_ok(self.show_breakpoint(), "ShowBreakPoint") lst = self._find(res, "breakpoints", "list", "items", "bpx") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "id": it.get("id"), "address": self._int(it.get("addressHex") or it.get("address") or it.get("addr")), "temporary": it.get("temporary"), "threadId": self._int(it.get("threadId")), "condition": it.get("condition"), }) return out def bp_info(self, addr) -> dict: """断点详情（类型/状态/命中数）""" res = self._parse_ok(self.get_breakpoint_info(self._hex(addr)), f"GetBreakpointInfo(0x{self._int(addr):X})") return {"address": self._int(self._find(res, "address", "addr")), "type": self._find(res, "type", "kind"), "enabled": self._find(res, "enabled", "active"), "hits": self._int(self._find(res, "hitCount", "count", "hits"))} def bp_type(self, addr) -> str: """断点类型（软/硬/内存/api）""" res = self._parse_ok(self.get_breakpoint_type(self._hex(addr)), f"GetBreakpointType(0x{self._int(addr):X})") return self._find(res, "type", "kind") or "" def bp_hit_count(self, addr) -> int: """断点命中数""" res = self._parse_ok(self.get_breakpoint_hit_count(self._hex(addr)), f"GetBreakpointHitCount(0x{self._int(addr):X})") return self._int(self._find(res, "count", "hitCount", "hits")) def set_bp_hits(self, addr, count: int) -> bool: """设置断点命中阈值""" self._parse_ok(self.set_breakpoint_hit_count(self._hex(addr), count), "SetBreakpointHitCount") return True def reset_bp_hits(self, addr) -> bool: self._parse_ok(self.reset_breakpoint_hit_count(self._hex(addr)), "ResetBreakpointHitCount") return True def enable_bp(self, addr) -> bool: self._parse_ok(self.enable_breakpoint(self._hex(addr)), f"EnableBreakpoint(0x{self._int(addr):X})") return True def disable_bp(self, addr) -> bool: self._parse_ok(self.disable_breakpoint(self._hex(addr)), f"DisableBreakpoint(0x{self._int(addr):X})") return True def bp_command(self, addr, command: str = "") -> bool: """断点命中时执行命令（脚本化调试）""" self._parse_ok(self.breakpoint_command(self._hex(addr), command), "BreakpointCommand") return True def bp_fast_resume(self, addr, enable: int = 1) -> bool: self._parse_ok(self.set_breakpoint_fast_resume(self._hex(addr), enable), "SetBreakpointFastResume") return True def bp_silent(self, addr, enable: int = 1) -> bool: self._parse_ok(self.set_breakpoint_silent(self._hex(addr), enable), "SetBreakpointSilent") return True def bp_log_file(self, addr, file: str) -> bool: self._parse_ok(self.set_breakpoint_log_file(self._hex(addr), file), "SetBreakpointLogFile") return True def del_cond_bp(self, addr) -> bool: self._parse_ok(self.del_cond_breakpoint(self._hex(addr)), "DelCondBreakPoint") return True def set_hw_bp(self, addr, len_: str = "1", flag: str = "e") -> int: """硬件断点（可能回退软件断点）""" self._parse_ok(self.set_hbreakpoint(self._hex(addr), len_, flag), f"SetHbreakPoint(0x{self._int(addr):X})") return self._int(addr) def del_hw_bp(self, addr) -> bool: self._parse_ok(self.del_hbreakpoint(self._hex(addr)), f"DelHbreakPoint(0x{self._int(addr):X})") return True def hw_bp_list(self) -> list: res = self._parse_ok(self.show_hbreakpoint(), "ShowHbreakPoint") lst = self._find(res, "hbreakpoints", "list", "items", "bpx") return lst if isinstance(lst, list) else [] def set_mem_bp(self, addr, flag: str = "r") -> int: """内存访问断点（r/w/x 访问监控）""" self._parse_ok(self.set_mem_breakpoint(self._hex(addr), flag), f"SetMemBreakPoint(0x{self._int(addr):X},{flag})") return self._int(addr) def del_mem_bp(self, addr) -> bool: self._parse_ok(self.del_mem_breakpoint(self._hex(addr)), f"DelMemBreakPoint(0x{self._int(addr):X})") return True def mem_bp_list(self) -> list: res = self._parse_ok(self.show_mem_breakpoint(), "ShowMemBreakPoint") lst = self._find(res, "membreakpoints", "list", "items", "bpx") return lst if isinstance(lst, list) else [] def set_api_bp(self, dll: str, api: str) -> str: """API 断点：dll!api""" self._parse_ok(self.set_api_breakpoint(dll, api), f"SetApiBreakPoint({dll}!{api})") return f"{dll}!{api}" def del_api_bp(self, dll: str, api: str) -> str: self._parse_ok(self.del_api_breakpoint(dll, api), f"DelApiBreakPoint({dll}!{api})") return f"{dll}!{api}" def api_bp_list(self) -> list: res = self._parse_ok(self.show_api_breakpoint(), "ShowApiBreakPoint") lst = self._find(res, "apibreakpoints", "list", "items", "bpx") return lst if isinstance(lst, list) else [] def set_exception_bp(self, code: str) -> str: """异常断点（如 'C0000005' 访问违例）""" self._parse_ok(self.set_exception_bpx(code), f"SetExceptionBPX({code})") return code def del_exception_bp(self, code: str) -> str: self._parse_ok(self.del_exception_bpx(code), f"DelExceptionBPX({code})") return code def exception_bp_list(self) -> list: res = self._parse_ok(self.get_exception_bpx_list(), "GetExceptionBPXList") lst = self._find(res, "exceptions", "list", "items", "bpx") return lst if isinstance(lst, list) else [] def set_dll_bp(self, dll: str) -> str: """DLL 加载断点""" self._parse_ok(self.set_dll_breakpoint(dll), f"SetDllBreakPoint({dll})") return dll def del_dll_bp(self, dll: str) -> str: self._parse_ok(self.del_dll_breakpoint(dll), f"DelDllBreakPoint({dll})") return dll def dll_bp_list(self) -> list: res = self._parse_ok(self.get_dll_breakpoints(), "GetDllBreakPoints") lst = self._find(res, "dlls", "list", "items", "breakpoints") return lst if isinstance(lst, list) else [] def bpx_options(self, option: str, enable: int) -> bool: """断点全局选项（如 single_shot 等）""" self._parse_ok(self.set_bpx_options(option, enable), f"SetBPXOptions({option},{enable})") return True def run_to_bp(self, addr, timeout: int = 60) -> dict: """设临时断点 → Run → 等待命中；返回 {hit, info, registers}""" self.set_bp(addr) self.custom_post({"interface": "Run", "params": {}}) hit = self.wait_stopped(timeout) info = self.bp_info(addr) if hit else None regs = self.registers() if hit else None return {"hit": hit, "info": info, "registers": regs} def trace_to(self, addr, timeout: int = 30) -> dict: """TraceLine 运行到目标地址（TF 单步跟踪）""" self._parse_ok(self.trace_line(self._hex(addr)), f"TraceLine(0x{self._int(addr):X})") hit = self.wait_stopped(timeout) return {"hit": hit, "registers": self.registers() if hit else None} # ---------------- 寄存器 / 线程 / 栈 ---------------- def registers(self) -> dict: """全部寄存器 dict""" res = self._parse_ok(self.register(), "Register") regs = self._find(res, "registers", "regs") if not isinstance(regs, dict): raise RuntimeError(f"Register 响应缺少寄存器表: {str(res)[:200]}") return dict(regs) def reg(self, name: str) -> int: """读取单个寄存器（eax/eip/eflags...）""" v = self.registers().get(name) return self._int(v) def set_reg(self, name: str, value) -> bool: self._parse_ok(self.set_register(name, str(value)), f"SetRegister({name})") return True def expr_eval(self, expr: str) -> int: """表达式求值（需停止态）：'eax' / '0x401000+0x20' / '[esp]' 等""" res = self._parse_ok(self.eval_expr(expr), f"Eval({expr})") return self._int(self._find(res, "value", "result", "val")) def context(self) -> dict: """调试上下文：寄存器 + EIP 指令 + 栈顶""" regs = self.registers() eip = self._int(regs.get("eip") or regs.get("rip")) ctx = {"registers": regs} if eip: try: ctx["eip"] = eip ctx["mnemonic"] = self.mnemonic_at(eip) ctx["instructions"] = self.disasm(eip, 3) except RuntimeError: pass try: ctx["stack_top"] = self.stack_dump(8) except RuntimeError: pass return ctx def stack_dump(self, count: int = 16) -> list: """栈转储：[{address, value, ascii}]""" res = self._parse_ok(self.stack(count), f"Stack({count})") lst = self._find(res, "stack", "items", "list", "entries") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "address": self._int(it.get("address") or it.get("addr")), "value": self._int(it.get("value") or it.get("data")), "ascii": it.get("ascii") or it.get("text"), }) return out def stack_frames(self, max_num: int = 16) -> list: """调用栈帧：[{frame, address, sp, module, name}]（name 含符号+偏移）""" res = self._parse_ok(self.call_stack(max_num), f"CallStack({max_num})") lst = self._find(res, "frames", "stack", "callstack", "list", "items") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "frame": self._int(it.get("frame") or it.get("index")), "address": self._int(it.get("pcHex") or it.get("pc") or it.get("address") or it.get("addr")), "sp": self._int(it.get("spHex") or it.get("sp")), "frame_addr": self._int(it.get("frameHex") or it.get("frameAddr")), "module": it.get("module") or it.get("mod"), "name": it.get("symbol") or it.get("name"), }) return out def threads_info(self) -> list: """线程列表：[{tid, teb, status, name}]""" res = self._parse_ok(self.threads(), "Threads") lst = self._find(res, "threads", "list", "items") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "tid": self._int(it.get("threadId") or it.get("tid")), "teb": it.get("tebBaseHex") or it.get("teb"), "status": it.get("status"), "name": it.get("name"), }) return out def thread_detail(self, tid: int) -> dict: """单线程详情""" res = self._parse_ok(self.thread_info(tid), f"ThreadInfo({tid})") return res def active_thread_id(self) -> int: """活动线程 TID""" res = self._parse_ok(self.get_active_thread(), "GetActiveThread") return self._int(self._find(res, "activeThread", "tid", "thread")) def set_active_thread_id(self, tid: int) -> bool: self._parse_ok(self.set_active_thread(tid), f"SetActiveThread({tid})") return True def thread_last_error(self, tid: int) -> int: """线程最近错误码""" res = self._parse_ok(self.get_thread_last_error(tid), f"GetThreadLastError({tid})") return self._int(self._find(res, "error", "code", "lastError", "value")) def thread_priority(self, tid: int) -> str: """线程优先级""" res = self._parse_ok(self.get_thread_priority(tid), f"GetThreadPriority({tid})") return self._find(res, "priority", "value", "level") or "" def thread_set_priority(self, tid: int, priority: str = "NORMAL") -> bool: """设置线程优先级（HIGHEST/ABOVE_NORMAL/NORMAL/BELOW_NORMAL/LOWEST）""" self._parse_ok(self.set_thread_priority(tid, priority), f"SetThreadPriority({tid},{priority})") return True def thread_set_name(self, tid: int, name: str) -> bool: self._parse_ok(self.set_thread_name(tid, name), f"SetThreadName({tid})") return True def flag_get(self, name: str) -> int: """读取标志位（CF/PF/AF/ZF/SF/TF/IF/DF/OF）""" regs = self.registers() ef = self._int(regs.get("eflags") or regs.get("flags")) if ef is None: raise RuntimeError("寄存器中无 eflags") bits = {"CF": 0, "PF": 2, "AF": 4, "ZF": 6, "SF": 7, "TF": 8, "IF": 9, "DF": 10, "OF": 11} name = name.upper() if name not in bits: raise RuntimeError(f"未知标志位: {name}") return (ef >> bits[name]) & 1 def flag_set(self, name: str, value: int) -> bool: self._parse_ok(self.set_flag(name.upper(), int(value)), f"SetFlag({name})") return True def trace_record_start(self) -> bool: self._parse_ok(self.start_trace_record(), "StartTraceRecord") return True def trace_record_get(self, count: int = 5) -> list: """最近执行轨迹记录""" res = self._parse_ok(self.get_trace_record(count), f"GetTraceRecord({count})") lst = self._find(res, "records", "trace", "list", "items") return lst if isinstance(lst, list) else [] def trace_record_stop(self) -> bool: self._parse_ok(self.stop_trace_record(), "StopTraceRecord") return True def jmp_history_list(self) -> list: """跳转历史""" res = self._parse_ok(self.jmp_history(), "JmpHistory") lst = self._find(res, "history", "jumps", "list", "items") return lst if isinstance(lst, list) else [] def last_exception_info(self) -> dict: """最近一次异常信息""" res = self._parse_ok(self.last_exception(), "LastException") return {"code": self._find(res, "code", "exceptionCode"), "address": self._int(self._find(res, "address", "exceptionAddress")), "flags": self._find(res, "flags", "exceptionFlags")} # ---------------- 执行控制包装（自动等待停止态） ---------------- def run_and_wait(self, timeout: int = 60) -> bool: self.custom_post({"interface": "Run", "params": {}}) return self.wait_stopped(timeout) def step_into_wait(self, timeout: int = 60) -> bool: self.custom_post({"interface": "StepIn", "params": {}}) return self.wait_stopped(timeout) def step_over_wait(self, timeout: int = 60) -> bool: self.custom_post({"interface": "StepOver", "params": {}}) return self.wait_stopped(timeout) def step_out_wait(self, timeout: int = 60) -> bool: self.custom_post({"interface": "StepOut", "params": {}}) return self.wait_stopped(timeout) def skip_wait(self, count: int = 1, timeout: int = 60) -> bool: self.custom_post({"interface": "Skip", "params": {"count": count}}) return self.wait_stopped(timeout) def run_to_and_wait(self, addr, timeout: int = 60) -> bool: self._parse_ok(self.run_to(self._hex(addr)), f"RunTo(0x{self._int(addr):X})") return self.wait_stopped(timeout) def run_to_user_wait(self, timeout: int = 60) -> bool: self.custom_post({"interface": "RunToUserCode", "params": {}}) return self.wait_stopped(timeout) def trace_into_wait(self, count: int = 3, timeout: int = 60) -> bool: self.custom_post({"interface": "TraceInto", "params": {"count": count}}) return self.wait_stopped(timeout) def trace_over_wait(self, count: int = 3, timeout: int = 60) -> bool: self.custom_post({"interface": "TraceOver", "params": {"count": count}}) return self.wait_stopped(timeout) def animate_into_wait(self, count: int = 3, timeout: int = 60) -> bool: self.custom_post({"interface": "AnimateInto", "params": {"count": count}}) return self.wait_stopped(timeout) def animate_over_wait(self, count: int = 3, timeout: int = 60) -> bool: self.custom_post({"interface": "AnimateOver", "params": {"count": count}}) return self.wait_stopped(timeout) def animate_stop_now(self) -> bool: self._parse_ok(self.custom_post({"interface": "AnimateStop", "params": {}}), "AnimateStop") return True def pause_all(self) -> bool: self.custom_post({"interface": "PauseAllThreads", "params": {}}) return True def resume_all(self) -> bool: self.custom_post({"interface": "ResumeAllThreads", "params": {}}) return True def thread_pause_by_id(self, tid: int) -> bool: self.custom_post({"interface": "ThreadPause", "params": {"tid": tid}}) return True def thread_resume_by_id(self, tid: int) -> bool: self.custom_post({"interface": "ThreadResume", "params": {"tid": tid}}) return True def debug_continue_wait(self, status: int = 0, timeout: int = 60) -> bool: """处理完异常/事件后继续执行""" self.custom_post({"interface": "DebugContinue", "params": {"status": status}}) return self.wait_stopped(timeout) def exec_cmd(self, command: str) -> str: """执行调试器命令（如 'Help'、'SetDebug' 等命令串）""" res = self._parse_ok(self.execute_command(command), f"ExecuteCommand({command})") return json.dumps(res, ensure_ascii=False) # ---------------- 漏洞挖掘 / 深度取证 ---------------- def exception_chain(self) -> list: """SEH 异常处理链：[{address, handler, next}]（栈溢出利用分析）""" res = self._parse_ok(self.seh_list(), "SEHList") lst = self._find(res, "seh", "chain", "list", "items", "entries") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "address": self._int(it.get("address") or it.get("addr")), "handler": self._int(it.get("handler") or it.get("handlerAddress")), "next": self._int(it.get("next") or it.get("nextAddress")), }) return out def heap_layout(self) -> list: """堆布局：[{base, size, flags}]（堆喷/堆溢出分析）""" res = self._parse_ok(self.heaps(), "Heaps") lst = self._find(res, "heaps", "list", "items") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "base": self._int(it.get("base") or it.get("address") or it.get("baseAddress")), "size": self._int(it.get("size") or it.get("regionSize")), "flags": it.get("flags") or it.get("type"), }) return out def handles_of(self, max_num: int = 50) -> list: """句柄列表：[{handle, type, name}]""" res = self._parse_ok(self.handles(max_num), f"Handles({max_num})") lst = self._find(res, "handles", "list", "items") out = [] if isinstance(lst, list): for it in lst: if isinstance(it, dict): out.append({ "handle": self._int(it.get("handle") or it.get("value")), "type": it.get("type") or it.get("objectType"), "name": it.get("name"), }) return out def patch(self, addr, hexstr: str) -> int: """打补丁（写字节）并返回字节数""" self._parse_ok(self.patch_memory(self._hex(addr), hexstr), f"PatchMemory(0x{self._int(addr):X})") return len(re.sub(r"[^0-9a-fA-F]", "", hexstr)) // 2 def patch_list(self) -> list: """已打补丁列表""" res = self._parse_ok(self.patches(), "Patches") lst = self._find(res, "patches", "list", "items") return lst if isinstance(lst, list) else [] def del_patch(self, addr) -> bool: self._parse_ok(self.delete_patch(self._hex(addr)), f"DeletePatch(0x{self._int(addr):X})") return True def verify_patch(self, addr, hexstr: str) -> tuple: """校验补丁生效（读回对比），返回 (是否一致, 实际字节 hex)""" want = bytes.fromhex(re.sub(r"[^0-9a-fA-F]", "", hexstr)) got = self.read_bytes(addr, len(want)) return got == want, got.hex() def minidump_to(self, path: str) -> str: """生成 minidump（崩溃现场取证）""" self._parse_ok(self.minidump(path), f"minidump({path})") return path def dump_to(self, path: str, addr, size: int) -> str: """按区域 dump 进程内存到文件""" self._parse_ok(self.dump_process(path, self._hex(addr), size), f"DumpProcess({path})") return path _DANGEROUS_PATTERNS = [ (r"%[0-9]*[n]", "格式串 %n（任意写原语）"), (r"%[0-9]*\$[a-zA-Z]", "位置参数格式串"), (r"system\s*\(", "system() 调用"), (r"(/bin/)?sh\b", "shell 命令"), (r"cmd\.exe", "cmd.exe"), (r"http[s]?://", "URL"), (r"password|passwd", "口令字段"), (r"secret|token|apikey|api[_-]?key", "密钥字段"), (r"SELECT\s|INSERT\s|DELETE\s|DROP\s|UPDATE\s", "SQL 语句"), (r"\.\./|\.\.\\", "路径穿越"), (r"GetProcAddress|LoadLibrary|VirtualAlloc|WriteProcessMemory|CreateRemoteThread", "危险 API 名"), (r"flag\{|ctf\{", "CTF flag"), ] def string_scan_report(self, module: str, max_strings: int = 2000) -> dict: """危险字符串扫描：提取模块字符串并标记漏洞相关模式（格式串/SQL/口令/危险API等）""" strs = self.strings_in_module(module, max_strings) hits = [] for item in strs: s = item["string"] for pat, desc in self._DANGEROUS_PATTERNS: if re.search(pat, s, re.IGNORECASE): hits.append({"address": item["address"], "string": s[:120], "pattern": desc}) return {"total_strings": len(strs), "dangerous_hits": hits} ```

2026-09-26

精选列表

Python爬虫入门教程 62-100 30岁了，想找点文献提高自己，还被反爬了，Python搞起，反爬第2篇

用 GitHub Actions 给你的 Python 脚本装上免费定时器

Python Qt GUI设计：QSpinBox计数器类（基础篇—15）

用 Python 对 Excel 数据排序：三种场景与实现

Python 爬虫的会话管理：从 Cookie 维持到登录态复用

Python TDD 学习笔记 Day 15：把测试用进日常项目

基于 Python 的分布式任务调度引擎设计与实现

基于 Python 的高性能内存键值存储引擎设计与实现

Python 异步并发爬虫报错 “Cannot assign requested address” 排查与解决

Python 异步爬虫进阶：用 httpx + asyncio 把速度提升 10 倍

Python 基础语法（一）：变量、数据类型与运算符

Python 爬虫包选型指南：从工具到场景的匹配逻辑

Python面向对象编程实战：三个项目完整拆解讲解

Python 的 dataclass 到底省了多少样板代码？3 个对比讲清

用 Python 检测内容新鲜度：多久没更新该回写了

如何用 Python 计算 ROC 变动率指标做动量分析？

Python入门 | 三个if语句程序作业和基础语法笔记

Python调用万能识别接口，英文数字、点选和问答

Python在我SEO工作中的应用一 - 数据采集

BinSentry 自研二进制哨兵调用Python SDK库

资源下载

Spring

Rocky Linux

优质分享App

Sublime Text

欢迎您来访！