如何搭建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代理项目的工作。

由于所有这些工作听起来都很繁琐,所以我们会做些简单的事情。我们仍将为Roberts Trusty Certificate Corp生成一个根CA证书。然后,我们将无需手动进行上述所有操作,而是将该证书手动添加到你手机的受信任根CA列表中。这告诉你的手机信任由Roberts 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“ Roberts Trusty Certificate Corp”生成第二个TLS证书,该证书将是自签名的-使用证书自己的私钥签名。这是因为根CA是隐式受信任的,不需要其他任何人为其担保。

在目标证书上签名

现在,我们可以使用新的CA的私钥在www.bbc.com上签署我们的证书。此签名将对语句“ Roberts 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 apps 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()

            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(0)

        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, "Roberts 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压缩,因此响应对象看起来像是难以理解的废话。

总结

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

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

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

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

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

参考及来源: