Python黑帽子阅读笔记 发表于 2025-03-18 | 更新于 2025-03-18
| 浏览量:
Python黑帽子阅读笔记 前置准备 安装好python3和kali之后，下载venv虚拟环境
1 2 3 4 5 6 7 8 sudo apt-get install python3-venv #安装python3-venv软件包 mkdir bhp #创建bhp目录 cd bhp python3 -m venv venv3 #向python3传递-m选项来调用venv包，命名环境名为venv3 source venv3/bin/activate deactivate #退出虚拟环境
在venv虚拟环境下用pip来搜索安装软件包
1 2 例如 (venv3) ➜ bhp$ pip install lxml
进入python的shell当中查看是否安装成功
1 2 3 4 Python 3.11.8 (main, Feb 7 2024, 21:52:08) [GCC 13.2.0] on linux Type "help", "copyright", "credits" or "license" for more information. >>> from lxml import etree #从lxml当中导入etree >>> exit()
看到报错或者python版本不对的话，就再检查检查是kali问题还是python的问题
一些python代码的小叮嘱 1 2 3 4 5 6 7 8 9 10 11 程序开头导包，按照字母顺序排列，方便检查 函数和类定义没有硬性规定，但当发现自己正在尝试使用全局变量记录某些状态的时候，或者是将同样数据结构传给多个函数的时候，就要考虑写类来重构程序了 有关main的注意事项： main代码块部分都是放在结尾的，这能够让我们用两种不同的方式来调用我们的程序 一是能从命令行当中启动程序，此时模块内部名是_main_，所以main代码块就会被执行。 譬如：python scan.py 这条命令会加载sacn.py当中所有的函数和类，然后执行main代码块 二是能被当作包来被其他程序导入 当我们并不希望执行main函数里的程序，而只希望使用这个包里的其他函数时，我们就可以在我们写的另一个程序当中这样操作： import scan 这样我们就可以把scan.py当成一个包来导入了，因为此时模块的内部名是python包名scan，而不是_main_，因此我们可以调用里面定义的所有函数和类，同时不会执行main代码块
基础的网络编程工具 socket模块提供了所有必须的接口，是我们可以快速开发出TCP/UDP客户端和服务端，或者直接调用原始socket等等。
TCP客户端 渗透测试当中，经常需要创建一个TCP客户端用以测试服务，发送垃圾数据和fuzz等等。如果在某大型企业的内网环境当中，不太可能直接获取网络工具或者编译器，甚至复制粘贴和连接外网都难，因此需要创建TCP客户端
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 import socket target_host = "www.baidu.com" target_port = 80 #创建socket对象 client = socket.socket(socket.AF_INET,socket.SOCK_STREAM) #连接客户端 client.connect((target_host,target_port)) #发送数据 client.send(b"GET / HTTP/1.1\r\nHost:baidu.com\r\n\r\n") #接收数据 response = client.recv(4096) print(response.decode()) client.close()
这里注意一下，socket这个库在3.11的python当中已经是内置库了，我当初还以为需要手动导一个，导半天版本错误也没找出是什么问题，自己查才发现
另外，这个AF_INET参数表示我们将使用标准的IPV4地址或者主机名，而SOCK_STREAM则表示这是一个TCP客户端。我们将该客户端连接到服务器，再发送一些bytes类型的数据，接收到返回的数据并打印到屏幕上，再关闭scoket
UDP客户端 Python UDP客户端和TCP客户端相去不远，只需要两处小改动就能发送UDP形式的数据了
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 import socket target_host = "127.0.0.1" target_port = 9997 #创建一个socket对象 client = socket.socket(socket.AF_INET,socket.SOCK_DGRAM) #发送数据 client.sendto(b"hello",(target_host,target_port)) #接收数据 data,addr = client.recvfrom(4096) print(data.decode()) client.close()
创建socket对象时，我们把socket类型改成了SOCK_DGRAM，之后简单调用sendto()函数，填好要发送的数据和接收数据的服务器地址就可以了。
因为UDP是一个无连接协议，所以开始通信前不需要用connect()函数建立连接。最后用recvfrom()函数接收数据，这个函数不仅会返回接收到的数据，还会返回详细的数据来源信息(主机名和端口号)
TCP服务端 编写一个标准的多线程TCP服务器
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 import socket import threading IP = '0.0.0.0' PORT = 9998 def main(): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.bind((IP, PORT)) server.listen(5) # 修正了拼写错误 print(f'[*] Listening on {IP}:{PORT}') while True: client_socket, address = server.accept() print(f'[*] Accepted connection from {address[0]}:{address[1]}') client_handler = threading.Thread(target=handle_client, args=(client_socket,)) client_handler.start() def handle_client(client_socket): # 修正了函数名的拼写 with client_socket: # 这里的 with 语句可能不正确，因为 socket 对象没有 __enter__ 和 __exit__ 方法 request = client_socket.recv(1024) # 修正了变量名的拼写 print(f'[*] Received: {request.decode("utf-8")}') client_socket.send(b'ACK') # 修正了变量名的拼写 if __name__ == '__main__': main()
在这段代码当中，我们先指定了服务器监听的IP地址和端口。然后我们让服务器开始监听，并且把最大连接数设置为5
下一步就是让服务器进入主循环中，并在该循环中等待外来连接。当一个客户端成功建立连接的时候，就会将收到的客户端socket对象保存到client变量当中，将远程连接的详细信息保存到address变量当中，然后再创建一个新的线程，指向handle_client函数，并且传入client变量
创建好了，我们启动线程来处理刚刚收到的连接，同时服务端主循环也已经准备好处理下一个外来的连接了。handle_client函数会调用recv()来接收数据，并且给客户端发送一段简单回复
这里可以用前面写过的TCP客户端对服务端进行请求，能看到有回应
接下来就是在这个基础上拓展了
netcat平替 有的时候系统里面妹有netcat，但是有Python，我们就可以创建一个简单的网络客户端和服务端用以传送文件，rce
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 import argparse import socket import shlex import subprocess import sys import textwrap import threading def execute(cmd): # 使用 subprocess 执行命令，并捕获输出 cmd = cmd.strip() if not cmd: return output = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT) return output.decode() def send(self): # 连接到 target 和 port self.socket.connect((self.args.target, self.args.port)) if self.buffer: # 若缓冲区有数据，则发送缓冲区数据 self.socket.send(self.buffer) try: while True: # 循环接收 target 返回的数据 response = self.socket.recv(4096) # 接收数据 if response: # 检查是否读取到内容 print(response.decode()) # 打印接收到的数据 buffer = input('>').encode() # 读取用户输入 self.socket.send(buffer) # 发送数据到服务端 except KeyboardInterrupt: # Ctrl+C 终止循环 print('用户操作终止') self.socket.close() # 关闭 socket 对象 def listen(self): # 将 socket 对象绑定到 target 和 port 上 self.socket.bind((self.args.target, self.args.port)) self.socket.listen(5) print(f'[*] Listening on {self.args.target}:{self.args.port}') while True: # 循环监听新连接 client_socket, _ = self.socket.accept() print(f'[*] Accepted connection from {client_socket.getpeername()}') client_thread = threading.Thread( target=self.handle, args=(self, client_socket) # 传递 client_socket 和 self 给 handle 函数 ) client_thread.start() def handle(self, client_socket): # 处理接收到的数据或执行命令的函数 if self.args.execute: # 如果需要执行命令 output = execute(self.args.execute) client_socket.send(output.encode()) # 发送命令执行结果 elif self.args.upload: # 如果需要上传文件 with open(self.args.upload, 'wb') as f: while True: data = client_socket.recv(4096) if not data: break f.write(data) client_socket.send(f'File {self.args.upload} saved.'.encode()) elif self.args.command: # 如果需要创建一个shell，则创建一个循环 prompt = b'BHP:#>' # 定义 shell 提示符 while True: try: client_socket.send(prompt) cmd_buffer = b'' while b'\n' not in cmd_buffer: cmd_buffer += client_socket.recv(4096) response = execute(cmd_buffer.decode()) if response: client_socket.send(response.encode()) except Exception as e: print(f'Error: {e}') break class NetCat: def __init__(self, args, buffer=None): # 初始化一个 NetCat 对象，然后创建一个 socket 对象 self.args = args self.buffer = buffer self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) def run(self): # run 函数作为 NetCat 对象的执行入口 if self.args.listen: # 如果是接收方，执行 listen 函数 self.listen() else: # 如果是发送方，执行 send 函数 self.send() if __name__ == '__main__': parser = argparse.ArgumentParser( description='BHP Net Tool', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent(''' Example: netcat.py -t 192.168.0.1 -p 5555 -l -c # 交互式命令行 shell netcat.py -t 192.168.0.1 -p 5555 -l -u=mytest.txt # 指定上传的文件 netcat.py -t 192.168.0.1 -p 5555 -l -e="ls -la" # 命令执行 echo 'ABC' | netcat.py -t 192.168.0.1 -p 5555 # 发送数据到端口 5555 netcat.py -t 192.168.0.1 -p 5555 # 连接服务器 ''').strip() ) parser.add_argument('-l', '--listen', action='store_true', help='Listen for incoming connections') parser.add_argument('-p', '--port', type=int, default=5555, help='Target port') parser.add_argument('-t', '--target', help='Target IP') parser.add_argument('-u', '--upload', help='Specify file to upload') parser.add_argument('-e', '--execute', help='Execute command') parser.add_argument('-c', '--command', action='store_true', help='Create interactive command line shell') args = parser.parse_args() # 根据是否监听来决定 buffer 的内容 buffer = sys.stdin.read() if not args.listen else b'' nc = NetCat(args, buffer.encode() if buffer else None) nc.run()
