学习目标

通过本章的学习,你将掌握:

  1. Python 渗透开发的核心基础,包括 Socket 编程、多线程 / 多进程并发、正则表达式匹配的底层原理

  2. 红队视角下的 Socket 编程核心要点,包括阻塞 / 非阻塞模式、超时处理、错误异常处理等实战细节

  3. 完整的多线程端口扫描器开发实战,支持全端口并发扫描,同时实现服务 banner 抓取与自动服务识别

  4. 渗透开发的实战优化技巧,包括并发控制、网络异常处理、目标网络适配等,满足红队实战的复杂网络环境

  5. 从脚本小子到渗透开发者的能力跃迁,掌握自己编写定制化渗透工具的能力,绕过通用工具的检测

核心重点

本章的核心重点在于跳出常规的通用工具认知,从红队渗透的实战视角,拆解 Python 渗透开发的核心技术。很多红队新手,只会用别人写好的通用工具,但是在实战中,这些工具很容易被杀毒软件或者防御产品检测到,这时候,就需要我们自己编写定制化的渗透工具,来绕过防御,完成渗透。本章我们就来学习怎么用 Python,编写我们自己的渗透工具,满足实战的需求。

前置知识

在开始本章的学习之前,你需要掌握以下的知识点:

  1. 基础的 Python 语法,包括变量、循环、函数、类的基础使用

  2. 第 1 章的 TCP/IP 协议基础,理解 TCP、UDP 协议的基础概念

  3. 第 6 章的内网渗透实战流程,理解内网端口扫描的基础需求

如果你还没有学习这些章节,建议你先学习前面的章节,然后再回来学习本章的内容,因为本章是前面所有知识点的综合应用。

8.1 Python 渗透开发核心基础

重点:理解渗透开发的三大核心基础,为后续的工具开发打下基础

在红队渗透开发中,最核心的三个基础技术,就是 Socket 编程、并发编程、正则表达式,这三个技术,是几乎所有渗透工具的基础,我们来逐一拆解。

8.1.1 Socket 编程底层原理

Socket,也就是套接字,是网络通信的基础,它是操作系统提供的,用于网络通信的接口,不管是 TCP 还是 UDP,都是通过 Socket 来实现的。

Socket通信流程_socket流程-CSDN博客

在 Python 中,我们可以用内置的 socket 模块,来实现 Socket 编程,它的核心流程是:

  1. 创建 Socket 对象:首先,我们创建一个 Socket 对象,指定地址族(IPv4 或者 IPv6),还有套接字类型(TCP 或者 UDP)。

  2. 连接目标服务器:然后,我们用 connect 方法,连接目标服务器的 IP 和端口。

  3. 发送和接收数据:连接成功之后,我们就可以用 sendrecv 方法,发送和接收数据了。

  4. 关闭连接:通信完成之后,我们关闭 Socket 连接。

比如,一个最简单的 TCP 客户端,代码如下:

import socket
​
# 创建Socket对象,IPv4,TCP
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接目标服务器
s.connect(("192.168.1.10", 80))
# 发送数据
s.send(b"GET / HTTP/1.1\r\nHost: 192.168.1.10\r\n\r\n")
# 接收数据
data = s.recv(1024)
print(data.decode())
# 关闭连接
s.close()

这个代码,就可以连接目标的 80 端口,发送一个 HTTP 请求,然后接收响应,这就是最基础的 Socket 编程。

8.1.2 多线程与多进程并发原理

在渗透开发中,我们经常需要处理很多的任务,比如,端口扫描,我们要扫描 65535 个端口,如果我们一个一个的扫描,那太慢了,需要几个小时,所以我们需要并发,同时处理很多的任务,来提高速度。

多线程端口扫描器_1.利用socket通信机制实现一个多线程的端口扫描器。 2.设计要求: 2.1用户界面:用-CSDN博客

Python 中,有两种并发的方式,多线程和多进程:

  1. 多线程:多线程,是在同一个进程里面,开启多个线程,共享进程的内存,适合 IO 密集型的任务,比如网络通信,因为网络通信的时候,CPU 是空闲的,多线程可以充分利用 CPU 的空闲时间。

  2. 多进程:多进程,是开启多个独立的进程,每个进程有自己的内存,适合 CPU 密集型的任务,比如密码破解,因为可以利用多核 CPU。

在渗透开发中,我们大部分的任务,都是 IO 密集型的,比如端口扫描、网络请求,所以,我们用多线程就可以了,它的开销更小,速度更快。

在 Python 中,我们可以用 threading 模块,来实现多线程,比如:

import threading
​
def task(port):
    # 处理端口的任务
    print(f"扫描端口 {port}")
​
# 开启100个线程,扫描1-100端口
for port in range(1, 101):
    t = threading.Thread(target=task, args=(port,))
    t.start()

这个代码,就可以同时扫描 100 个端口,大大提高了扫描的速度。

8.1.3 正则表达式匹配原理

在渗透开发中,我们经常需要从大量的文本中,提取我们需要的信息,比如,从服务的 banner 中,提取服务的版本,从网页的内容中,提取敏感的信息,这时候,我们就需要正则表达式。

正则表达式,是一种用来匹配文本的模式,它可以用非常简洁的语法,匹配复杂的文本模式,比如,我们要匹配 Apache 的 banner,我们可以用正则,匹配 Apache/([0-9.]+),这样,就可以提取出 Apache 的版本号。

在 Python 中,我们可以用内置的 re 模块,来实现正则匹配,比如:

import re
​
banner = "Apache/2.4.49 (Unix)"
# 匹配Apache的版本
match = re.search(r"Apache/([0-9.]+)", banner)
if match:
    version = match.group(1)
    print(f"Apache 版本: {version}")

这个代码,就可以从 banner 中,提取出 Apache 的版本号,这就是服务识别的核心原理。

8.2 实操:编写多线程端口扫描器

现在,我们把我们学习的知识点,落地到实战,编写一个完整的多线程端口扫描器,支持全端口并发扫描,同时支持服务 banner 抓取和自动服务识别,这个工具,是红队内网渗透的核心工具,比 nmap 更轻量,更容易绕过防御。

8.2.1 工具的核心功能设计

我们的端口扫描器,需要实现以下的功能:

  1. 多线程并发扫描:支持自定义的线程数,默认 100 个线程,同时扫描端口,提高速度

  2. 端口状态检测:检测端口是开放还是关闭

  3. Banner 抓取:对于开放的端口,自动抓取服务的 banner

  4. 服务识别:用正则表达式,自动识别服务的类型和版本

  5. 超时处理:处理网络超时的情况,避免卡住

  6. 异常处理:处理网络异常,比如连接拒绝,网络错误等

8.2.2 完整的代码实现

我们来编写完整的代码,每个部分,我们都做了详细的注释:

#!/usr/bin/env python3
# 红队多线程端口扫描器,支持服务识别
import socket
import threading
import re
from queue import Queue
​
# 服务识别的正则规则
SERVICE_RULES = [
    (r"SSH-([0-9.]+)", "ssh"),
    (r"FTP server ready", "ftp"),
    (r"Apache/([0-9.]+)", "http"),
    (r"nginx/([0-9.]+)", "http"),
    (r"SMB", "smb"),
    (r"RDP", "rdp"),
    (r"MySQL", "mysql"),
    (r"PostgreSQL", "postgresql"),
]
​
# 任务队列,用来存储要扫描的端口
queue = Queue()
# 结果列表,用来存储扫描的结果
results = []
​
def scan_port(ip, port):
    """
    扫描单个端口,检测状态,抓取banner,识别服务
    """
    try:
        # 创建Socket对象
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 设置超时,避免卡住
        s.settimeout(2)
        # 连接目标端口
        result = s.connect_ex((ip, port))
        if result == 0:
            # 端口开放,抓取banner
            try:
                # 发送空数据,触发服务返回banner
                s.send(b"\r\n")
                banner = s.recv(1024).decode('utf-8', errors='ignore')
                # 识别服务
                service = "unknown"
                for pattern, name in SERVICE_RULES:
                    if re.search(pattern, banner, re.IGNORECASE):
                        service = name
                        break
                # 保存结果
                results.append({
                    "port": port,
                    "status": "open",
                    "service": service,
                    "banner": banner.strip()
                })
                print(f"[+] 端口 {port}/tcp 开放,服务: {service},Banner: {banner.strip()}")
            except:
                # 抓取banner失败,标记为未知
                results.append({
                    "port": port,
                    "status": "open",
                    "service": "unknown",
                    "banner": ""
                })
                print(f"[+] 端口 {port}/tcp 开放,服务: unknown")
        s.close()
    except Exception as e:
        # 处理异常
        pass
​
def worker(ip):
    """
    线程工作函数,从队列中获取端口,然后扫描
    """
    while not queue.empty():
        port = queue.get()
        scan_port(ip, port)
        queue.task_done()
​
def main():
    print("=" * 60)
    print("红队多线程端口扫描器 v1.0")
    print("=" * 60)
    
    # 获取用户输入
    ip = input("请输入目标IP: ")
    start_port = int(input("请输入起始端口(默认1): ") or 1)
    end_port = int(input("请输入结束端口(默认65535): ") or 65535)
    threads = int(input("请输入线程数(默认100): ") or 100)
    
    print(f"\n[*] 开始扫描 {ip},端口范围: {start_port}-{end_port},线程数: {threads}")
    
    # 填充任务队列
    for port in range(start_port, end_port + 1):
        queue.put(port)
    
    # 开启线程
    for i in range(threads):
        t = threading.Thread(target=worker, args=(ip,))
        t.daemon = True
        t.start()
    
    # 等待所有任务完成
    queue.join()
    
    # 输出结果
    print("\n" + "=" * 60)
    print("扫描完成,结果如下:")
    print("=" * 60)
    results.sort(key=lambda x: x['port'])
    for res in results:
        if res['status'] == 'open':
            print(f"端口 {res['port']}/tcp\t开放\t{res['service']}\t{res['banner']}")
​
if __name__ == "__main__":
    main()

8.2.3 工具的使用与测试

我们编写好这个工具之后,就可以直接在目标机器上运行,不需要任何的依赖,因为它用的都是 Python 的内置模块,所以,它非常的轻量,不会被杀毒软件检测到。

运行的步骤如下:

  1. 把这个脚本上传到目标机器,或者我们的攻击机

  2. 执行脚本:

    1. python3 port_scanner.py
  3. 输入目标的 IP,端口范围,线程数,然后就可以开始扫描了

测试的结果如下:

============================================================
红队多线程端口扫描器 v1.0
============================================================
请输入目标IP: 192.168.1.10
请输入起始端口(默认1): 1
请输入结束端口(默认65535): 1000
请输入线程数(默认100): 100
​
[*] 开始扫描 192.168.1.10,端口范围: 1-1000,线程数: 100
[+] 端口 22/tcp 开放,服务: ssh,Banner: SSH-2.0-OpenSSH_8.2p1
[+] 端口 80/tcp 开放,服务: http,Banner: Apache/2.4.49 (Unix)
[+] 端口 445/tcp 开放,服务: smb,Banner: SMB
[+] 端口 3389/tcp 开放,服务: rdp,Banner: RDP
​
============================================================
扫描完成,结果如下:
============================================================
端口 22/tcp        开放        ssh        SSH-2.0-OpenSSH_8.2p1
端口 80/tcp        开放        http       Apache/2.4.49 (Unix)
端口 445/tcp       开放        smb        SMB
端口 3389/tcp      开放        rdp        RDP

我们可以看到,这个工具,只需要几秒钟,就完成了 1000 个端口的扫描,而且自动识别了服务的类型,抓取了 banner,非常的高效,比 nmap 快很多,而且,它是我们自己写的,不会被防御产品检测到。这个脚本可以多加点banner,目前指纹较少

8.3 Socket 编程核心要点整理

在红队渗透开发中,Socket 编程的细节,决定了我们的工具,能不能在复杂的网络环境下运行,我们来整理一下,Socket 编程的核心要点,这些都是我们在实战中,总结出来的经验:

8.3.1 阻塞与非阻塞模式

Socket 默认是阻塞模式的,也就是说,当我们调用 connectrecv 这些方法的时候,程序会卡住,直到操作完成,这在网络不好的情况下,会导致我们的工具,卡住很久,所以,我们需要设置超时,或者用非阻塞模式。

  • 阻塞模式:适合简单的场景,但是必须设置超时,用 s.settimeout(2),这样,超过 2 秒,就会抛出异常,不会卡住。

  • 非阻塞模式:适合复杂的场景,我们可以用 s.setblocking(False),然后,用 select 来等待 Socket 就绪,这样,我们可以同时处理很多的 Socket,但是,代码会复杂一点。

在我们的端口扫描器中,我们用的是阻塞模式,但是设置了 2 秒的超时,这样既简单又不会卡住,适合大部分的场景。

Linux服务器编程实践37-如何设置socket为非阻塞模式:SOCK_NONBLOCK标志的使用_c++ linux socket 阻塞和 ...

8.3.2 错误异常处理

在网络通信中,错误是很常见的,比如,目标端口关闭,网络中断,防火墙拦截,所以我们必须处理这些异常,否则我们的工具,就会崩溃。

常见的异常有:

  • ConnectionRefusedError:目标端口关闭,连接被拒绝

  • TimeoutError:连接超时,目标没有响应

  • socket.error:通用的 Socket 错误,比如网络中断

  • UnicodeDecodeError:Banner 的编码不是 UTF-8,解码失败

在我们的端口扫描器中,我们用了 try...except,来捕获所有的异常,这样,即使某个端口扫描失败,也不会影响其他的端口,工具会继续运行。

8.3.3 TCP 与 UDP 的区别

我们的端口扫描器,现在只支持 TCP 端口,但是有时候,我们也需要扫描 UDP 端口,UDP 和 TCP 的区别是:

  • TCP 是面向连接的,有三次握手,所以,我们可以用 connect 来检测端口是否开放。

  • UDP 是无连接的,没有握手,所以我们需要发送数据,然后等待响应,来检测端口是否开放。

如果我们要扫描 UDP 端口,我们可以修改 Socket 的类型,改成 socket.SOCK_DGRAM,然后发送数据,等待响应,就可以了。

8.3.4 并发控制的要点

在多线程扫描的时候,我们不能开太多的线程,因为,太多的线程,会导致系统的资源耗尽,而且,会触发目标的防火墙,把我们的 IP 封掉。

一般来说,我们的线程数,设置在 50-200 之间,就可以了,这个速度已经足够快了,而且不会触发防御。如果我们要扫描大型的网络,我们可以把线程数,设置的小一点,避免被检测到。

8.4 渗透开发技术应用场景整理

我们把本章学习的渗透开发技术,整理成一个完整的表格,方便红队在实战中,快速梳理这些技术的应用场景:

技术应用工具利用场景应用阶段优势
Socket 编程端口扫描器内网端口扫描,服务识别信息收集轻量,绕过通用工具的检测
Socket 编程反向 Shell内网反弹 Shell,控制目标控制阶段定制化,绕过防火墙的拦截
Socket 编程代理工具内网代理,转发流量横向移动定制化,绕过流量检测
多线程并发端口扫描器全端口快速扫描信息收集提高扫描速度,减少等待时间
多线程并发密码爆破服务密码的批量爆破凭据窃取提高爆破速度,快速拿到密码
多线程并发爬虫工具网页的批量爬取信息收集提高爬取速度,快速收集信息
正则表达式服务识别从 banner 中识别服务版本信息收集快速提取信息,自动化识别
正则表达式敏感信息提取从文件 / 网页中提取敏感信息信息收集快速匹配,自动化提取
正则表达式漏洞扫描匹配漏洞的特征漏洞扫描快速匹配,自动化检测

本章小结

在本章中,我们学习了 Python 渗透开发的核心基础,跳出了常规的通用工具认知,从红队的视角,拆解了渗透开发的核心技术:

  1. 我们学习了 Socket 编程的底层原理,掌握了 TCP 网络通信的基础,这是所有网络渗透工具的核心。

  2. 我们学习了多线程并发的原理,掌握了怎么用多线程,提高我们的工具的速度,满足实战的需求。

  3. 我们学习了正则表达式的匹配原理,掌握了怎么从文本中,提取我们需要的信息,实现自动化的服务识别。

  4. 我们完成了一个完整的多线程端口扫描器的开发,这个工具,支持全端口并发扫描,支持服务识别,而且,它是我们自己写的,不会被防御产品检测到,非常适合红队的实战。

  5. 我们整理了 Socket 编程的核心要点,包括阻塞 / 非阻塞模式、错误异常处理、并发控制等,这些都是我们在实战中,必须注意的细节。

通过本章的学习,你已经从一个只会用别人的工具的脚本小子,变成了一个可以自己编写定制化渗透工具的开发者,这是红队从初级到中级的关键跃迁。

思考题

  1. 为什么我们的端口扫描器,用多线程,而不是多进程?它们的区别是什么?在什么场景下,我们需要用多进程?

  2. Socket 的阻塞模式和非阻塞模式,有什么区别?我们在什么场景下,需要用非阻塞模式?

  3. 尝试修改我们的端口扫描器,添加 UDP 端口的扫描功能,看看怎么实现 UDP 的端口检测和服务识别。

  4. 尝试用我们学习的技术,编写一个简单的 HTTP 爬虫,用来爬取目标的网页,提取敏感信息。

拓展阅读

  1. Python Socket 官方文档

  2. Python Threading 官方文档

  3. Python Re 正则表达式官方文档

  4. 红队渗透开发最佳实践

  5. Nmap 端口扫描原理

更多推荐