python渗透测试入门之TCP代理

近期收到了电子工业出版社赠送的一本网络安全书籍《python黑帽子》，书中一共24个实验，今天复现第2个实验（开发一个TCP代理），我的测试环境是mbp电脑+kali虚拟机+centos虚拟机+conda开发环境。我测试了明文传输的FTP代理和加密传输的SSH代理，涉及到3方：1是mbp上跑的代理程序，2是kali上跑的ftp服务，3是centos上跑的ftp客户端，成功验证了mbp上的代理程序相当于kali ftp服务端和centos ftp客户端通信的中间人，不仅仅可以帮忙摆渡不通的网络，还可以抓取明文信息，所以使用代理一定要注意加密通信。这次实验总的来说，就是很卡～

ailx10

网络安全优秀回答者

网络安全硕士

去咨询

1、在kali上测试ftp，成功如下图所示

2、在kali上测试ftp代理，输入用户名和密码就挂掉了，因为ftp服务器是国外的延迟太高

3、在kali上测试ssh代理，也是输入用户名和密码之后，就挂掉了，也是因为ssh服务器是国外的，延迟太高

4、在mbp上可以看到代理的整个经过，ssh代理是加密的，但是ftp代理是明文传输的，代理作为中间人，可以嗅探到所有的明文信息，这一点尤其要注意，小心隐私泄漏～

5、由于上面ftp代理失败了，所以我在kali本地搭建了ftp服务器，然后在centos上测试，正常ftp连接如下

6、在centos通过代理连接kali上的ftp，结果和正常ftp一摸一样，说明这次成功了

代码如下：

# -*- coding: utf-8 -*-
# @Time    : 2022/6/2 7:29 PM
# @Author  : ailx10
# @File    : proxy.py

import sys
import socket
import threading

HEX_FILTER = ''.join([(len(repr(chr(i))) == 3) and chr(i) or '.' for i in range(256)])

# 类似 wireshark 以16进制和 可打印字符 形式打印内容
def hexdump(src,length=16,show=True):
    if isinstance(src,bytes):
        src = src.decode("utf8","ignore")

    results = list()
    for i in range(0,len(src),length):
        word = str(src[i:i+length])
        printable = word.translate(HEX_FILTER)
        hexa = ' '.join([f'{ord(c):02X}' for c in word])
        hexwidth = length * 3
        results.append(f'{i:04X} {hexa:<{hexwidth}} {printable}')
    if show:
        for line in results:
            print(line)
    else:
         return results

# 从代理两端 接收数据
def receive_from(connection):
    buffer = b""
    connection.settimeout(5)
    try:
        while True:
            data = connection.recv(4096)
            if not data:
                break
            buffer += data
    except Exception as e:
        pass
    return buffer

def request_handler(buffer):
    # 修改包
    return buffer

def response_handler(buffer):
    # 修改包
    return buffer

def proxy_handler(client_socket,remote_host,remote_port,receive_first):
    # 连接远程主机
    remote_socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    remote_socket.connect((remote_host,remote_port))
    # 是否先从服务端接收一段数据（比如欢迎信息）
    if receive_first:
        remote_buffer = receive_from(remote_socket)
        hexdump(remote_buffer)
        remote_buffer = response_handler(remote_buffer)
        if len(remote_buffer):
            print("[<==] Sending %d bytes to localhost." % len(remote_buffer))
            client_socket.send(remote_buffer)
    while True:
        local_buffer = receive_from(client_socket)
        if len(local_buffer):
            line = "[==>]Received %d bytes from localhost." % len(local_buffer)
            print(line)
            hexdump(local_buffer)

            local_buffer = request_handler(local_buffer)
            remote_socket.send(local_buffer)
            print("[==>] Send to remote.")

        remote_buffer = receive_from(remote_socket)
        if len(remote_buffer):
            print("[<==] Received %d bytes from remote." % len(remote_buffer))
            hexdump(remote_buffer)

            remote_buffer = response_handler(remote_buffer)
            client_socket.send(remote_buffer)
            print("[<==] Send to localhost.")

        # 异常处理
        if not len(local_buffer) or not len(remote_buffer):
            client_socket.close()
            remote_socket.close()
            print("[*] No more data. Closing connections.")
            break

def server_loop(local_host,local_port,remote_host,remote_port,receive_first):
    server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    try:
        server.bind((local_host,local_port))
    except Exception as e:
        print("problem on bind: %r" %e)

        print("[!!] Failed to listen on %s:%d" % (local_host,local_port))
        print("[!!] Check for other listening sockets or correct permissions.")
        sys.exit(0)

    print("[*] Listening on %s:%d" %(local_host,local_port))
    server.listen(5)
    while True:
        client_socket,addr = server.accept()
        line = "> Received incoming connection from %s:%d" % (addr[0],addr[1])
        print(line)
        proxy_thread = threading.Thread(target=proxy_handler,args=(client_socket,remote_host,remote_port,receive_first))
        proxy_thread.start()

def main():
    if len(sys.argv[1:]) != 5:
        print("Useage ./proxy.py [localhost] [localport]",end="")
        print("[remotehost] [remoteport] [receive_first]")
        print("Example: ./proxy.py 127.0.0.1 9000 10.12.132.1 9000 True")
        sys.exit(0)

    local_host = sys.argv[1]
    local_port = int(sys.argv[2])
    remote_host = sys.argv[3]
    remote_port = int(sys.argv[4])
    receive_first = sys.argv[5]

    if "True" in receive_first:
        receive_first = True
    else:
        receive_first = False

    server_loop(local_host,local_port,remote_host,remote_port,receive_first)

if __name__ == "__main__":
    main()

发布于 2022-06-02 21:11 知乎