记录黑客技术中优秀的内容, 传播黑客文化,分享黑客技术精华

如何搭建TCP代理(五)

2020-03-22 10:42

如何搭建TCP代理(一)

如何搭建TCP代理(二)

如何搭建TCP代理(三)

如何搭建TCP代理(四)

对TLS证书签名

但是,在验证证书的有效性时,客户更喜欢检查你的电话,而不是证书的常用名称。他们还会检查它的“密码签名”。密码签名是某个第三方(通常是“证书颁发机构”)附加在证书上的批准印章。CA是安全可靠的组织,其工作是颁发和签署TLS证书。毫不夸张地说,它们共同负责互联网上加密的完整性。在向某个客户颁发域名或主机名证书之前,CA会进行尽职调查,以验证该客户确实是该业务的真正所有者。

满足CA后,它将为主机名生成一个证书和私钥,并将加密签名附加到该证书上。 CA通过使用自己的私钥来加密证书的内容来创建签名。客户可以使用CA的公钥对签名进行解密,然后确认解密后的文本与证书中的文本匹配,从而验证签名是否有效。由于无法访问CA的私钥(无法对其进行高度保密和安全保护),就无法生成签名,因此可以安全地假定CA认可了证书的内容。他们已经确信证书的持有者(或更准确地说,是拥有相应私钥的组织)是证书所包含的域或主机名的真正所有者。此时,签名有效地表示一个语句。

像你的手机这样的客户端仅信任由其信任的CA签名的TLS证书,你的手机会使用经过手机制造商认可的根CA硬编码列表来决定信任哪些CA。例如,可以在线获取预加载到iOS11中的根CA列表。

可以为api.targetapp.com生成一个DIY证书,但是,由于我们不是api.targetapp.com的真正所有者,所以我们将永远无法说服真正可信的CA对其进行签名。如果签名都不保证我们的证书,那么你的手机将拒绝信任它。

我们可以通过启动自己的根CA“Robert 's trusted Certificate Corp”来解决这个问题,并将它放到智能手机的可信根CA列表中。一种非常困难的方法是建立一家真正的公司,雇用数百人来经营它,建立一些非常安全的证书生成基础架构,填写大量文书,以说服Google和Apple以及我们其他人应该被赋予数亿人的在线安全性,并最终在手机的下一个操作系统更新中包含我们的CA的公钥。之后,你可以继续进行此TCP代理项目的工作。

由于所有这些工作听起来都很繁琐,所以我们会做些简单的事情。我们仍将为Robert's Trusty Certificate Corp生成一个根CA证书。然后,我们将无需手动进行上述所有操作,而是将该证书手动添加到你手机的受信任根CA列表中。这告诉你的手机信任由Robert's Trusty Certificate Corp签名的其他证书。我们将使用RTCC的私钥对我们的api.targetapp.com证书进行签名,并将此签名的证书及其私钥传递到我们代理的listenSSL方法中。

当我们的代理人向你的手机提供最新签名的api.targetapp.com证书时,你的手机将看到该手机已由“ Robert’s Trusty Certificate Corp”签名。它将检查其受信任的CA的内部列表,并查看是否有这个出色的信誉机构。它将很乐意接受api.targetapp.com的证书为有效证书,并在开始发送TCP流量之前继续协商TLS会话。

请注意,掌握我们自制CA私钥的任何人都可以使用它为所需的任何主机名签署自己的证书。在你的手机(仅限你的手机)眼中,这些证书是合法且可信赖的。攻击者将能够使用它们对你执行自己的中间人攻击,从而允许他们读取你的加密流量。你应该小心确保CA的私钥非常安全。完成此项目后,你还应确保从手机的根CA列表中删除CA的公钥。

生成你的手机将信任的TLS证书后,我们可以通过listenSSL将其传递到我们的代理中。twisted将为我们处理其余的问题,然后我们的代理将启用TLS。

具体伪造过程

1.为我们的目标应用程序的主机名生成一个TLS证书和私钥;

2.生成根CA公钥/私钥对,并使用私钥对该TLS证书签名;

3.将我们的根CA的公钥安装为手机上受信任的授权机构;

4.将此证书和相应的私钥传递到twisted的listenSSL函数中,并使用listenSSL监控传入的TCP连接。

稍后,我们将针对英国广播公司的网站www.bbc.com测试我们的代理。

为我们的目标主机名生成证书

用于创建和签名TLS证书的库以大多数合理的编程语言提供,只需几行代码就可以生成bbc.com的第一个证书。

为我们的证书颁发机构生成证书

我们将为我们的原始根CA“ Robert's Trusty Certificate Corp”生成第二个TLS证书,该证书将是自签名的-使用证书自己的私钥签名。这是因为根CA是隐式受信任的,不需要其他任何人为其担保。

在目标证书上签名

现在,我们可以使用新的CA的私钥在www.bbc.com上签署我们的证书。此签名将对语句“ Robert's Trusty Certificate Corp断言该证书属于www.bbc.com的真实所有者”进行加密编码。

在手机上安装根CA的证书

最后,我们需要在你的手机上安装我们的根CA,以便你的手机信任由它签名的其他证书。最安全的方法是:

1. 运行下面的代理服务器脚本一次,它将生成一个根CA证书并打印出它的位置;

2. 打开证书文件,然后将后半部分(包括----- BEGIN CERTIFICATE -----之后的所有内容)复制到一个名为ca.pub的新文件中;

3. 通过电子邮件将ca.pub作为文件附件发送到你可以在手机上访问的电子邮件帐户;

4. 打开手机上的电子邮件,然后打开附件;

5. 其余步骤将取决于手机的品牌和型号,例如,在iOS上,你必须将密钥安装为根CA,然后在系统设置中找到它并启用它,并在此过程中勾选各种“是的,我知道这样做很冒险”选项。

在你的手机上安装根CA证书的方法更简单,更省力。

监控SSL

现在,我们拥有了启用TLS的中间人TCP代理所需的一切。我们将更新代理的启动脚本,以便它为目标主机名生成并签署自己的证书(在我们的测试中,它为www.bbc.com)。然后,我们将签名的证书传递到我们的代理中,并启动我们的代理。

此代码也在GitHub上。

import time
 
from twisted.internet import protocol, reactor
from twisted.internet import ssl as twisted_ssl
import dns.resolver
 
from OpenSSL.crypto import (X509Extension, X509,
        dump_privatekey, dump_certificate,
        load_certificate, load_privatekey,
        PKey, TYPE_RSA, X509Req)
from OpenSSL.SSL import FILETYPE_PEM
import tempfile
import os
import netifaces as ni
 
# Adapted from http://stackoverflow.com/a/15645169/221061
 
class TLSTCPProxyProtocol(protocol.Protocol):
    """
    TLSTCPProxyProtocol listens for TCP connections from a
    client (eg. a phone) and forwards them on to a specified
    destination (eg. an app's API server) over a second TCP
    connection, using a ProxyToServerProtocol.
 
    It assumes that both legs of this trip are encrypted
    using TLS.
    """
    def __init__(self):
        self.buffer = None
        self.proxy_to_server_protocol = None
 
    def connectionMade(self):
        """
        Called by twisted when a client connects to the
        proxy.  Makes an TLS connection from the proxy to
        the server to complete the chain.
        """
        print("Connection made from CLIENT => PROXY")
        proxy_to_server_factory = protocol.ClientFactory()
        proxy_to_server_factory.protocol = ProxyToServerProtocol
        proxy_to_server_factory.server = self
 
        reactor.connectSSL(DST_IP, DST_PORT,
                           proxy_to_server_factory,
                           twisted_ssl.CertificateOptions())
 
    def dataReceived(self, data):
        """
        Called by twisted when the proxy receives data from
        the client. Sends the data on to the server.
 
        CLIENT ===> PROXY ===> DST
        """
        print("")
        print("CLIENT => SERVER")
        print(FORMAT_FN(data))
        WRITE_TO_FILE(data)
        print("")
        if self.proxy_to_server_protocol:
            self.proxy_to_server_protocol.write(data)
        else:
            self.buffer = data
 
    def write(self, data):
        self.transport.write(data)
 
 
class ProxyToServerProtocol(protocol.Protocol):
    """
    ProxyToServerProtocol connects to a server over TCP.
    It sends the server data given to it by an
    TLSTCPProxyProtocol, and uses the TLSTCPProxyProtocol
    to send data that it receives back from the server on
    to a client.
    """
 
    def connectionMade(self):
        """
        Called by twisted when the proxy connects to the
        server.  Flushes any buffered data on the proxy
        to server.
        """
        print("Connection made from PROXY => SERVER")
        self.factory.server.proxy_to_server_protocol = self
        self.write(self.factory.server.buffer)
        self.factory.server.buffer = ''
 
    def dataReceived(self, data):
        """
        Called by twisted when the proxy receives data
        from the server. Sends the data on to to the
        client.
 
        DST ===> PROXY ===> CLIENT
        """
        print("")
        print("SERVER => CLIENT")
        print(FORMAT_FN(data))
        print("")
        self.factory.server.write(data)
 
    def write(self, data):
        if data:
            self.transport.write(data)
 
 
# A class that represents a CA. It wraps a root CA TLS
# certificate, and can generate and sign certificates using
# this root cert.
#
# Inpsiration from
# https://github.com/allfro/pymiproxy/blob/master/src/miproxy/proxy.py
class CertificateAuthority(object):
 
    CERT_PREFIX = 'fake-cert'
 
    def __init__(self, ca_file, cache_dir=tempfile.mkdtemp()):
        print("Initializing CertificateAuthority ca_file=%s cache_dir=%s" %
              (ca_file, cache_dir))
 
        self.ca_file = ca_file
        self.cache_dir = cache_dir
        if not os.path.exists(ca_file):
            raise Exception("No cert exists at %s" % ca_file)
        else:
            self._read_ca(ca_file)
 
    def get_cert_path(self, cn):
        cnp = os.path.sep.join([self.cache_dir, '%s-%s.pem' %
            (self.CERT_PREFIX, cn)])
        if os.path.exists(cnp):
            print("Cert already exists common_name=%s" % cn)
        else:
            print("Creating and signing cert common_name=%s" % cn)
            key = PKey()
            key.generate_key(TYPE_RSA, 2048)
 
            # Generate CSR
            req = X509Req()
            req.get_subject().CN = cn
            req.set_pubkey(key)
            req.sign(key, 'sha1')
 
            # Sign CSR
            cert = X509()
            cert.set_subject(req.get_subject())
            cert.set_serial_number(123)
            cert.gmtime_adj_notBefore(0)
            cert.gmtime_adj_notAfter(31536000)
            cert.set_issuer(self.cert.get_subject())
            cert.set_pubkey(req.get_pubkey())
            cert.sign(self.key, 'sha1')
 
            with open(cnp, 'wb+') as f:
                f.write(dump_privatekey(FILETYPE_PEM, key))
                f.write(dump_certificate(FILETYPE_PEM, cert))
 
            print("Created cert common_name=%s location=%s" % (cn, cnp))
 
        return cnp
 
    def _read_ca(self, file):
        self.cert = load_certificate(FILETYPE_PEM, open(file).read())
        self.key = load_privatekey(FILETYPE_PEM, open(file).read())
 
    @staticmethod
    def generate_ca_cert(path, common_name):
        if os.path.exists(path):
            print("Cert already exists at %s, not regenerating" % path)
            return
        # Generate key
        key = PKey()
        key.generate_key(TYPE_RSA, 2048)
 
        # Generate certificate
        cert = X509()
        cert.set_version(3)
        cert.set_serial_number(1)
        cert.get_subject().CN = common_name
        cert.gmtime_adj_notBefore(0)
        cert.gmtime_adj_notAfter(315360000)
        cert.set_issuer(cert.get_subject())
        cert.set_pubkey(key)
        cert.sign(key, "sha256")
 
        with open(path, 'wb+') as f:
            f.write(dump_privatekey(FILETYPE_PEM, key))
            f.write(dump_certificate(FILETYPE_PEM, cert))
 
 
def get_local_ip(iface):
    ni.ifaddresses(iface)
    return ni.ifaddresses(iface)[ni.AF_INET][0]['addr']
 
 
# Alternative functions for formating output data
def _side_by_side_hex(data):
    BLOCK_SIZE = 16
 
    output_lines = []
    for i in range(0, len(data), BLOCK_SIZE):
        block = data[i:i+BLOCK_SIZE]
        _hex = ["%.2x" % el for el in block]
        _str = [chr(el) if chr(el).isprintable() else "." for el in block]
        line = " ".join(_hex).ljust((3*BLOCK_SIZE)+4) + "".join(_str).replace("\n", ".")
        output_lines.append(line)
    return "\n".join(output_lines)
 
def _stacked_hex(data):
    BLOCK_SIZE = 32
 
    hex_lines = []
    plaintext_lines = []
    for i in range(0, len(data), BLOCK_SIZE):
        block = data[i:i+BLOCK_SIZE]
        _hex = ["%.2x" % el for el in block]
        _str = [chr(el) if chr(el).isprintable() else "." for el in block]
 
        hex_line = " ".join(_hex)
        hex_lines.append(hex_line)
 
        plaintext_line = "  ".join(_str).replace("\n", ".")
        plaintext_lines.append(plaintext_line)
 
    lines = hex_lines + ["\n"] + plaintext_lines
    return "\n".join(lines)
 
def _replayable(data):
    d = data[0:4000]
    _hex = "".join(["%.2x" % el for el in d])
    _str = "".join([chr(el) if chr(el).isprintable() else "." for el in d])
 
    return _hex + "\n" + _str
 
def _noop(data):
    return data
 
# Change this line to use an alternative formating function
FORMAT_FN = _noop
 
# Record data sent to the server to files
DIR_NAME = "replays/messages-%d/" % time.time()
os.mkdir(DIR_NAME)
f_n = 0
def _write_to_file(data):
    # Global variables are bad but they do the job
    global f_n
    with open(DIR_NAME + str(f_n), 'wb') as f:
        f.write(data)
        f_n += 1
WRITE_TO_FILE = _write_to_file
 
CA_CERT_PATH = "./ca-cert.pem"
 
LISTEN_PORT = 443
DST_PORT = 443
DST_HOST = "www.bbc.com"
local_ip = get_local_ip('en0')
 
print("Querying DNS records for %s..." % DST_HOST)
a_records = dns.resolver.query(DST_HOST, 'A')
print("Found %d A records:" % len(a_records))
for r in a_records:
    print("* %s" % r.address)
print("")
assert(len(a_records) > 0)
 
DST_IP = a_records[0].address
print("Choosing to proxy to %s" % DST_IP)
 
print("""
#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#
-#-#-#-#-#-RUNNING TLS TCP PROXY-#-#-#-#-#-
#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#-#
 
Root CA path:\t%s
 
Dst IP:\t%s
Dst port:\t%d
Dst hostname:\t%s
 
Listen port:\t%d
Local IP:\t%s
""" % (CA_CERT_PATH, DST_IP, DST_PORT, DST_HOST, LISTEN_PORT, local_ip))
 
CertificateAuthority.generate_ca_cert(CA_CERT_PATH, "Robert's Trusty Certificate Corp")
ca = CertificateAuthority(CA_CERT_PATH)
certfile = ca.get_cert_path(DST_HOST)
with open(certfile) as f:
    cert = twisted_ssl.PrivateCertificate.loadPEM(f.read())
 
print("""
Next steps:
 
1. Make sure you are spoofing DNS requests from the
device you are trying to proxy request from so that they
return your local IP (%s).
2. Make sure you have set the destination and listen
ports correctly (they should generally be the same).
3. Use the device you are proxying requests from to make
requests to %s and check that they are logged in this
terminal.
4. Look at the requests, write more code to replay them,
fiddle with them, etc.
 
Listening for requests on %s:%d...
""" % (local_ip, DST_HOST, local_ip, LISTEN_PORT))
 
factory = protocol.ServerFactory()
factory.protocol = TLSTCPProxyProtocol
reactor.listenSSL(LISTEN_PORT, factory, cert.options(),
                  interface=local_ip)
reactor.run()

为什么我们要使用BBC进行测试?

如前所述,我们将使用英国广播公司(BBC)的网站www.bbc.com测试我们的代理。这是因为www.bbc.com使用HTTPS,但未在其响应主体上使用压缩。

使用类似gzip的压缩算法可以大大减少服务器发送的数据大小,然而,这也增加了测试代理的难度。我们的代理只会看到被压缩的HTTP对象数据处于混乱的压缩状态,我们的代理不知道如何解压数据,因此无法将其以真实的、人类可读的形式显示给我们。但由于BBC不压缩其响应对象(截至2018年8月),我们在本文不讨论这个问题,只处理可读的非压缩数据。

在测试我们的启用了TLS的代理之前,请确保:

1. 你的DNS欺骗脚本被设置为欺骗www.bbc.com

2. 你的智能手机将其DNS服务器设置为你的笔记本电脑的IP地址;

3. 你的TCP代理脚本指向www.bbc.com(www很重要!);

4. 你的TCP代理脚本设置为监控并在端口443上发送;

5. 然后启动DNS欺骗脚本,启动TCP代理脚本,然后在手机上访问www.bbc.com/news(www.bbc.com主页不使用HTTPS,因此此处对我们没有用)。你应该看到HTTP请求和响应的未加密纯文本出现在代理的日志中!

完成此操作后,请通过www.google.com再次尝试。 HTTP响应标头仍然是可读的,但是由于Google确实使用gzip压缩,因此响应对象看起来像是难以理解的废话。

42.png

43.png

总结

现在,你可以代理和检查你的智能手机发送的任意TCP流。

这比构建仅处理HTTP请求的代理要困难得多,由于我们希望能够代理任何TCP流,而不仅仅是HTTP请求,因此我们无法利用你手机的任何内置HTTP代理功能。正如我们在开头中所看到的,如果你想通过笔记本电脑代理来自智能手机的HTTP请求,你要做的就是将设备连接到同一网络,告诉手机的代理使用笔记本电脑的IP地址作为代理。

在这几篇文章中,我们建立了一个伪造的DNS服务器,并用它来欺骗对你手机发送的对你手机的DNS请求的响应。这欺骗了你的手机,使其无法与笔记本电脑建立用于目标主机名的TCP连接。我们建立了一个代理来接收这些连接,并将数据从它们转发到目标服务器。它还将来自目标服务器的所有响应数据转发回你的智能手机,从而完成诱骗过程。这意味着,你的智能手机也不知道发生了什么事。

另外,我们还添加了TLS支持。创建了自己的虚假证书颁发机构,并将其公钥添加到你的智能手机的受信任CA列表中。这样就可以使用其私钥对已生成的TLS证书进行签名,并使用这些证书来协商手机和笔记本电脑之间的TLS会话。完成后,我们从你手机的受信任CA列表中删除了CA的公钥。

未来,代理过程的最高境界是“使TCP代理更像Burp Suite”,这样保存跨连接到文件或数据库的数据非常有用。稍后能够重播这些请求也很方便,如果我们可以先进行编辑,如果我们能先编辑它们那就太好了。

本文翻译自:https://robertheaton.com/2018/08/31/how-to-build-a-tcp-proxy-4/如若转载,请注明原文地址:
知识来源: https://www.4hou.com/posts/nNK7

阅读:22512 | 评论:0 | 标签:无

想收藏或者和大家分享这篇好文章→复制链接地址

“如何搭建TCP代理(五)”共有0条留言

发表评论

姓名:

邮箱:

网址:

验证码:

公告

九层之台,起于累土;黑客之术,始于阅读

推广

工具

标签云

本页关键词