简介这是一套面向网络安全学习者的Python恶意软件检测系统源码基于威胁情报思路构建适合具备一定Python基础、希望深入理解恶意流量识别与威胁数据应用的开发者参考。压缩包内含19个文件以py源码和pyc编译文件为主辅以json配置、README说明及txt依赖清单整体仅13KB结构紧凑可直接查看核心实现。系统围绕E-Search-master项目组织包括应用初始化、接口路由、状态码校验、验证逻辑与主程序入口等模块便于从入口到出口完整梳理检测流程。代码中涉及Pandas数据处理、Scapy协议解析、Yara规则匹配等常见安全库的典型用法并借助IP黑名单、域名黑名单及签名库完成恶意文件的比对与预警读者可从中获得从数据清洗到模式匹配的完整链路。资源已有799人学习对想构建小型安全分析工具或开展课设二次开发的读者是一份轻量而完整的入门样例。1. 威胁情报驱动的恶意软件检测这套系统到底在防什么某天凌晨内网一台服务器突然向外发起陌生域名的 DNS 请求流量审计设备没告警因为域名注册不到 24 小时本地设备情报库里根本没有这条记录。但外部威胁情报平台在域名注册当天就打上了“恶意 C2”标签。如果你有一套基于威胁情报的恶意软件检测系统这一刻就能比对手快一步。这套 Python 系统做的事是把威胁情报平台上的失陷指标IOC包括恶意哈希、IP、域名、URL周期性地拉回本地再配合 PE 静态特征提取、哈希匹配和置信度打分对文件和流量做离线检测。适合安全工程师、蓝队成员和恶意代码分析入门者一台能跑 Python 3 的机器就是全部前置条件。2. 系统架构与数据流情报源选型、IOC 归一化和存储设计先看数据在系统里怎么流再谈选型。常见做法是把任务拆成三条流水线情报采集、检测扫描、告警输出。采集模块从外部情报源拿到 IOC经过归一化和去重后写入本地库检测模块用这些 IOC 去匹配待检文件和网络会话命中后按置信度打分超过阈值就产生一条告警。下面三个小节分别回答情报从哪来、来了之后长什么样、存到哪里。2.1 情报源的三种接入方式API 拉取、订阅推送与离线导入威胁情报源有很多但接入方式基本就三种API 拉取、订阅推送和离线导入。我一般把选型做成一张表方便对照。情报源常见接入方式主要 IOC 类型适合场景MISPREST API / 订阅推送事件内全部 Attribute有自建平台或社区源OpenCTIAPI / 导出STIX 对象需要做知识图谱联动AbuseIPDBHTTP APIIP临时查单个 IP 信誉MalwareBazaarAPI / CSV 下载样本哈希补充恶意文件指纹API 拉取用得最多因为它支持定时批量拉取还能用时间戳做增量同步。订阅推送适合情报更新非常频繁的源但要额外维护一个常驻接收服务复杂度高。离线导入则留给内网隔离环境运维定期把情报包拷进机器程序读文件入库。三种方式不是互斥的生产环境里往往是“API 拉取为主、离线导入兜底”。选型有一条原则宁可要一个每周更新的高质量私有情报也不要十个三个月不更新的公开情报。威胁情报的第一要素是新鲜度一次有效的 C2 域名标记价值远大于一百条过期的“可疑 IP”。2.2 IOC 归一化把 IP、域名、哈希、URL 统一成一种格式同一个恶意域名在情报源 A 里被写成evil.example.com带空格在情报源 B 里被写成EVIL.EXAMPLE.COM.带尾部句点同一个恶意文件一个源给 MD5另一个源给 SHA256。归一化不做好漏报是必然的。IOC 类型归一化规则失败案例IP去掉空格与无意义前导零 127.0.0.1 -127.0.0.1域名转小写、去尾部句点EVIL.COM.-evil.com哈希转小写、去空格A*64-a*64URL仅保留 schemehostpath去掉 fragment 和默认端口https://evil.com:443/a#frag-https://evil.com/a归一化函数不要写得太复杂够用就好def normalize_ioc(ioc_type: str, value: str) - str: value value.strip().lower() if ioc_type domain: value value.rstrip(.) elif ioc_type ip: value value.lstrip(0) or 0 elif ioc_type url: from urllib.parse import urlsplit, urlunsplit parts urlsplit(value) parts parts._replace(fragment, portNone) value urlunsplit(parts) return value这段代码的逻辑是先做strip().lower()把所有可见的格式差异压平再针对类型做特殊处理。域名去尾部句点是为了消除根域写法差异URL 处理注意parts._replace返回的是新对象不能直接改原 tuple。四个类型统一成小写字符串后后续建索引和匹配都会省事很多。2.3 存储设计为什么第一版我选 SQLite 而不是 MySQL单机场景下IOC 量级在万级到百万级SQLite 的读写性能完全够用而且零运维。MySQL、PostgreSQL 要等系统有多节点需求再迁表结构先按兼容 SQL 的方式来写。SQLite 唯一的短板是并发写弱但检测系统只需要一个采集进程写库读取走 WAL 模式不会成为瓶颈。CREATE TABLE sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT UNIQUE NOT NULL, base_url TEXT, enabled INTEGER DEFAULT 1 ); CREATE TABLE iocs ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER NOT NULL, ioc_type TEXT NOT NULL CHECK(ioc_type IN (ip,domain,url,sha256,md5)), ioc_value TEXT NOT NULL, confidence INTEGER NOT NULL DEFAULT 50, tlp_level TEXT NOT NULL DEFAULT amber, first_seen DATETIME NOT NULL, last_seen DATETIME NOT NULL, expire_at DATETIME, UNIQUE(source_id, ioc_type, ioc_value) );confidence字段范围 0 到 100默认 50来自不同情报源的同一指标会在检测环节叠加expire_at是 IOC 的生命周期终点为空表示永不过期适合处理勒索软件钱包地址这类长期有效的指标。UNIQUE约束直接兜住重复入库配合INSERT OR IGNORE使用采集任务跑多少次都不会把库撑爆。注意建表时ioc_type加CHECK约束是个好习惯。接口字段拼错一次数据库层面就能拦下来比在 Python 里做一堆 if 判断可靠得多。3. 用 Python 实现最小可用检测系统情报拉取、静态检测与调度这套源码包的核心我一般会按三块来组织情报拉取、检测匹配和任务调度。三块独立成文件彼此只通过数据库交互拿到任何一份同类源码先找这三个文件总是没错的。3.1 情报拉取从 MISP 拉事件并入库的完整脚本第一步先写采集脚本从 MISP 拉取最近七天的全部事件解析出 IOC 后入库。API Key 不要写死在代码里用环境变量传入。import os import sqlite3 import requests from urllib.parse import urljoin MISP_URL os.getenv(MISP_URL, ) MISP_API_KEY os.getenv(MISP_API_KEY, ) DB_PATH os.getenv(DB_PATH, ioc.db) def pull_events(within_days: int 7): headers { Authorization: MISP_API_KEY, Accept: application/json, } params {returnFormat: json, timestamp: f{within_days}d, limit: 50} url urljoin(MISP_URL, /events/index) resp requests.get(url, headersheaders, paramsparams, timeout20) resp.raise_for_status() for event in resp.json(): for attr in event.get(Attribute, []): yield attr[type], attr[value] def save_iocs(rows): conn sqlite3.connect(DB_PATH) conn.execute(PRAGMA journal_modewal) for ioc_type, ioc_value in rows: ioc_type ioc_type.lower() if ioc_type not in (ip, domain, url, sha256, md5): continue normalized normalize_ioc(ioc_type, ioc_value) conn.execute( INSERT OR IGNORE INTO iocs (source_id, ioc_type, ioc_value, confidence, first_seen, last_seen) VALUES (1, ?, ?, 60, datetime(now), datetime(now)), (ioc_type, normalized) ) conn.commit() conn.close()这段代码的核心是两个边界控制timestamp参数用7d把拉取量框在最近一周避免每次全量同步拖垮情报源和本地库INSERT OR IGNORE配合UNIQUE约束重复事件直接忽略不会污染数据。PRAGMA journal_modewal打开 WAL 模式读写不再互相阻塞实际跑采集和扫描并行时能明显感觉到差别。timeout20是必须的不写超时网络一抖整个任务就挂在那。3.2 检测模块PE 静态特征提取与 IOC 匹配的代码骨架有了 IOC 库接下来是检测侧。第一版先做两种检测哈希命中以及 PE 导入表特征提取。哈希命中是基础款先把不能做的漏报兜住。import hashlib import sqlite3 from pathlib import Path import pefile def sha256_of_file(path: Path) - str: h hashlib.sha256() with open(path, rb) as fp: while chunk : fp.read(1 20): h.update(chunk) return h.hexdigest() def scan_file(path: Path, conn: sqlite3.Connection) - int: sha256 sha256_of_file(path) row conn.execute( SELECT confidence FROM iocs WHERE ioc_typesha256 AND ioc_value?, (sha256,) ).fetchone() if row: return row[0] return 0哈希匹配的逻辑很简单但sha256_of_file里的分块读是防止大文件吃满内存的关键。1MB 分块是经验值兼顾速度和内存占用while chunk : fp.read(1 20)是 Python 3.8 的海象运算符写法低版本环境要提前确认。scan_file返回情报源里存的置信度0 表示没命中方便上层按分数累加。PE 静态特征提取用pefile库先不做复杂检测只把导入表拉出来备用。def extract_pe_features(path: Path) - dict: pe pefile.PE(str(path), fast_loadTrue) features { entry_point: hex(pe.OPTIONAL_HEADER.AddressOfEntryPoint), imports: [], } for entry in pe.DIRECTORY_ENTRY_IMPORT: features[imports].append(entry.dll.decode(errorsignore).lower()) return featuresfast_loadTrue只加载必要头不解析全部分区速度能快一倍以上。导入表的基本逻辑是恶意样本经常加载ws2_32.dll做 socket 通信、加载ntdll.dll做进程注入正常业务 exe 很少同时出现这两个库。提取的 DLL 名单可以留给后续 YARA 规则或机器学习模型用第一版先不做判定避免误报。注意pefile只能解析 PE 文件碰到 ELF 或 Mach-O 会直接抛异常。实际部署时先用文件头判断格式不是 PE 的走另一套检测逻辑别硬解析。3.3 调度与结果输出APScheduler 定时跑批与 JSON 告警采集和扫描要自动化跑起来用 APScheduler 做定时任务是最省事的方式。先把依赖装齐pip install requests pefile apscheduler。这一步经常有人卡住Python 环境没配好就先跑pip list确认三个轮子都在再往下走。from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.add_job( pull_iocs_and_save, interval, hours1, idpull_iocs, misfire_grace_time300, ) scheduler.add_job( scan_watch_dir, interval, minutes10, idscan_watch_dir, misfire_grace_time120, ) if __name__ __main__: scheduler.start()采集任务一小时一次是基准值MISP 类平台的事件更新不会比这更频繁扫描任务十分钟一次则要配合待检测目录的文件量来调。misfire_grace_time是任务线程卡住后允许补跑的时间窗口不设置的话任务错过执行窗口会被直接丢弃这在检测场景里等于漏报。每个任务显式设置id后面要停用或替换任务时能直接用 id 操作。扫描命中后输出一条 JSON 告警至少包含时间、文件路径、sha256、命中的 IOC 类型和总分方便下游 SIEM 或企业微信机器人消费。4. 三个必调参数可信度阈值、IOC 过期时间与白名单优先级这套系统跑起来容易跑好难难点全在参数上。我自己踩过一轮之后认为最值得花时间调的就三个可信度阈值、IOC 过期时间和白名单优先级。4.1 可信度阈值source_confidence 和 score_threshold 怎么定新系统最容易犯的错是把情报源里每一条 IOC 都当确凿证据结果内网误报一片。开源情报源质量参差不齐同一 IP 可能在列表里躺了半年都没人复查。正确做法是给每个命中项加权累计超过阈值才告警。def risk_score(hits: list[dict]) - int: weights {hash: 80, domain: 40, url: 25, ip: 15} return sum(weights.get(h.get(ioc_type), 10) * h.get(confidence, 50) for h in hits) def should_alert(score: int, threshold: int 6000) - bool: return score threshold这个打分公式里权重不是百分比而是对“证据可靠性”的估计哈希权重 80因为样本哈希几乎不会误报域名权重 40因为恶意域名可能换手后被正常使用IP 权重只有 15因为动态 IP 被重复分配的案例太多了。最终分数还要乘上情报源的confidence一个默认 50 的源命中域名得到40 * 50 2000分离阈值很远就不会打扰你。阈值初始值建议设成 6000跑一周看误报率再往下压。我的调参节奏是第一周误报多就提到 8000确认安全运营团队能接受告警量之后再一点点降到 5000。阈值调参不是玄学是拿真实误报数据做依据千万别凭感觉乱改。参数初始值调参方向source_confidence50可靠商业源提到 80开源源降到 30score_threshold6000误报多调高漏报多调低hash 权重80基本不用动误报率极低ip 权重15内网动态 IP 多的场景降到 54.2 IOC 过期时间为什么 30 天前的恶意域名必须降权恶意基础设施的平均存活时间只有几十天C2 域名会被频繁更换。把三个月前的恶意域名当今天的告警依据就是把历史当新闻。真正该做的是两段式处理未过期正常计分过期后降权但不删除再过一段宽限期才清理。UPDATE iocs SET confidence CAST(confidence * 0.1 AS INTEGER) WHERE expire_at IS NOT NULL AND expire_at datetime(now);降权而不是直接删除是为了保留关联分析的线索。比如一个 IP 曾经和某勒索软件家族绑定它现在虽然不活跃了但未来某个新样本如果也连接这个 IP分析师查询历史时还能关联到旧情报。expire_at设为空则表示永不过期用于钱包地址、特定样本哈希这类长期有效指标。过期时间不是一刀切IP 的过期要短域名略长哈希基本可以放宽到一年。我的经验值是 IP 45 天、域名 70 天、URL 35 天、哈希 365 天按你内网实际观察到的误报情况再微调。4.3 白名单优先级内网域名永远比情报源“大”最容易翻车的是内网域名。情报源标记了bad.example.com而你的内网恰好有一堆oa.example.com、git.example.com业务子域域名同根时误报率能冲到难以接受。白名单匹配必须放在威胁情报匹配之前执行。def match_with_whitelist(ioc_value: str, hits: list[dict]) - list[dict]: if ioc_value.endswith((.corp.example.com, .internal.example.net)): return [] return hits这里用endswith而不是in是因为in会把notbad.example.com.evil.com这种边界情况也匹配上。白名单的匹配优先级是绝对的白名单 本地历史情报 外部实时情报。曾经有个同事把阈值调了三天误报率纹丝不动最后发现是白名单放在情报匹配之后执行等于没放。顺序错了调参就全靠玄学。5. 部署避坑zip 源码跑起来之前先看这 5 个常见问题源码打包成 zip 分发最常见的坑不在代码逻辑而在环境、编码和资源消耗。下面五条都是我实际跑这类系统时遇到过的每条按“现象 → 原因 → 解决”列清楚。5.1 坑一情报源请求超时导致整个采集任务退出现象定时任务跑着跑着不执行了查看日志发现某个情报源请求超时异常一路抛到最外层任务进程直接退出。原因requests.get没有配置timeout网络一抖动就永远挂起异常没有被捕获后续任务全部中断。很多人只在本地环境测试网络通畅根本触发不了这个分支。解决所有请求统一走带超时和重试的封装函数并对超时异常做指数退避重试。from time import sleep def fetch_with_retry(url, headers, params, timeout20, retries3): for attempt in range(retries): try: return requests.get(url, headersheaders, paramsparams, timeouttimeout) except requests.exceptions.Timeout: if attempt retries - 1: raise sleep(2 ** attempt)timeout20表示连接和读取各 20 秒上限不是总时间。重试间隔按 2 的幂指数退避第一次失败等 2 秒第二次等 4 秒既给了网络恢复时间又不至于把情报源打挂。如果业务允许还可以在第三次重试失败后切换到离线导入流程保证当天有数据可用。5.2 坑二IOC 变多以后哈希匹配越来越慢现象IOC 表从几千条涨到十万级之后扫描一个文件耗时暴涨全量扫描一轮要几个小时。原因检测代码里用 SQLIN查几十万条哈希或者对每条 IOC 做字符串in判断导致全表扫描。解决建索引只是基础更有效的是把常用哈希在启动时加载进内存集合。HASH_SET set() def load_hash_set(): cur.execute(SELECT ioc_value FROM iocs WHERE typesha256 AND expire_at datetime(now)) for row in cur: HASH_SET.add(row[0]) def match_hash_file(sha256: str) - bool: return sha256 in HASH_SET一百万个 SHA256 哈希大约是 100MB 内存大部分服务器都扛得住。用内存集合换掉 SQL 查询单个文件的哈希匹配从毫秒变微秒。这个优化做完扫描 10 万文件的耗时能从小时降到分钟级值得在部署前就做。5.3 坑三zip 解压后中文文件名和代码注释在 Linux 上报错现象源码 zip 在 Windows 下压缩拷贝到 Linux 服务器解压后中文文件名变成乱码代码里的中文注释导致 Python 解释器报SyntaxError。原因zip 格式对非 ASCII 文件名的处理有历史包袱。Windows 压缩时中文名默认 GBK 编码而 Linux 解压工具按 UTF-8 解两边就对不上。另外老项目源码里经常有coding声明缺失的.py文件。解决解压前先识别文件名编码再决定解码方式。import zipfile with zipfile.ZipFile(source.zip) as zf: for info in zf.infolist(): if info.flag_bits 0x800: name info.filename.encode(cp437).decode(gbk) else: name info.filenamezipflag_bits的第 11 位置位表示文件名使用了非 UTF-8 编码此时需要用cp437先还原原始字节流再用gbk解码成中文。对应的.py文件统一在首行加# -*- coding: utf-8 -*-运行时就不会再报编码错误。我一般还会顺手把解压后的文本文件全部转一遍 UTF-8省得后面日志和数据库里出现乱码字符。5.4 坑四样本还没检测完就被本机杀毒软件隔离现象检测程序刚扫描到样本文件文件就被本机杀毒软件强制隔离程序报FileNotFoundError整批扫描中断。原因很多人习惯直接在当前工作机上跑检测而样本目录正好在杀毒软件实时防护的监控范围内。杀软识别出恶意样本后抢先处理你连读取它的机会都没有。解决本地测试时把样本目录加入杀软白名单生产环境则放到隔离虚拟机上跑主机上不保留任何样本文件。同时在代码里对文件读取异常做跳过处理不能因为单个文件访问失败就中断整轮扫描。try: scan_file(path) except FileNotFoundError: log.warning(fsample disappeared during scan: {path})这段血泪经验总结成一句话样本的处置页面永远轮不到你的检测程序来判你只是管道的一环保证管道不堵就行。5.5 坑五同一批情报重复入库数据库快速膨胀现象采集任务每次拉全量情报IOC 表行数暴涨检测结果重复出现告警翻倍。原因入库时没有按“源 类型 值”做唯一约束同一个 IOC 被多次插入。这条最隐蔽因为短时间内不会报错只是库越来越大。解决建表时加UNIQUE(source_id, ioc_type, ioc_value)插入用INSERT OR REPLACE更新最新时间INSERT OR REPLACE INTO iocs (source_id, ioc_type, ioc_value, confidence, first_seen, last_seen) VALUES (1, domain, evil.com, 60, datetime(now), datetime(now));INSERT OR REPLACE与普通INSERT的差别在于碰到唯一键冲突时它会删掉旧行写新行last_seen始终是最新时间库里的数据不会无限膨胀。配合定期清理expire_at过期的记录数据库体积能长期稳定在一个合理水平。6. 进阶用法用基准样本集量化检测效果再接入沙箱闭环6.1 基准样本集怎么搭恶意哈希打点良性 exe 打底系统跑通只是开始更要回答“它到底能抓多少真样本”。我的做法是搭一个基准样本集从 MalwareBazaar 拉最近 30 天的恶意样本哈希作为阳性集再从系统目录抽 200 个常见 exe 作为阴性集。每天跑一次现网检测算出两个数。def run_benchmark(conn: sqlite3.Connection) - dict: positives load_malicious_hashes(malware_hashes.txt) negatives list(Path(C:/Windows/system32).glob(*.exe))[:200] tp sum(1 for h in positives if check_hash(conn, h)) fp sum(1 for f in negatives if scan_file(f, conn) 0) return { detection_rate: tp / len(positives), false_positive_rate: fp / len(negatives), }检测率要盯下限误报率要盯上限。我个人的验收线是检测率不低于 70%、误报率不超过 2%。低于这条线说明情报源的质量或数量有问题优先补源而不是调高阈值自欺欺人。6.2 从静态命中升级到沙箱联动把结果回吐给情报库静态命中只能覆盖已知样本未知样本靠哈希和字符串根本发现不了。我的进阶路线是静态检测未命中但“可疑”的文件转发到沙箱做动态行为分析跑完把结果回写进 IOC 表形成私有情报闭环。常见做法是起一个本地队列目录沙箱消费完一个文件就把新增的域名或 IP 写回iocs表下次检测时全内网都能用上这条情报。这样做三个月本地情报库会越来越贴合自己网络里真实出现的攻击工具比任何公开源都准。我现在还保持一个习惯每次更新情报源或调整阈值先跑一遍基准集确认检测率没掉、误报率没涨再切到生产。数据比手感靠谱。希望帮到你。本文还有配套的精品资源点击获取
