本文章为纯技术向文章,不涉及非法内容。同时抵制使用匿名网络进行非法行为

使用 Python Stem 操作 Tor

Stem 是一个 Python 的模块,从而可以实现在 Python 中操作 Tor。

安装依赖

首先要在 Python 中安装 Stem,同时为了可以让 requests 使用 Socks5 连接,还需要下载相应的 Socks5 模块。

python -m pip install stem "request[socks]"

使用 Stem 切换电路

__init__.py 中,可以看到NEWNYM,其功能是重建一条新的电路。新的请求会使用重新建立的电路发送。
在 Tor 中,为了安全性,每条电路建立后是会复用一段时间的。(每次建立电路都需要花费时间,并且频繁建立的电路可以被诸如前驱攻击的方式进行攻击)。而使用NEWNYM命令即可要求洋葱代理(Onion Proxy, OP)重新建立电路。

这里使用http://ip-api.com/json的接口来获取实际用于访问的 IP。

首先,配置torrc,在其中配置如下两行,开启控制端口并设置其为9051

ControlPort 9051      
CookieAuthentication 1

在代码中,使用Controller.from_port(port=9051)来与控制端口建立连接,并调用controller.signal(Singal.NEWNYM)向 OP 发送NEWNYM命令,要求重新建立电路。

在下面的代码中,会请求四次 IP

  1. 直接请求,返回的是本机 IP
  2. 使用 Tor 请求,返回的是某个出口节点的 IP
  3. 使用 Tor 请求,返回的 IP 与 2 相同
  4. 要求重建电路,使用 Tor 请求,返回新的出口节点 IP

并且,如果在此执行程序,新的输出中,2、3 的 IP 与上一次执行的 4 相同(短时间内不会重建电路)

from stem import Signal
from stem.control import Controller
import requests

def switchIP():
    with Controller.from_port(port=9051) as controller:
        controller.authenticate()
        controller.signal(Signal.NEWNYM)

def getIP(proxy=None):
    proxies = {}
    if proxy != None:
        proxies["http"] = proxy
        proxies["https"] = proxy
    data = requests.get("http://ip-api.com/json", proxies=proxies).json()
    if data["status"] == "success":
        print(data["query"], data["city"])

def main():
    getIP()
    getIP("socks5://127.0.0.1:9050")
    getIP("socks5://127.0.0.1:9050")
    switchIP()
    getIP("socks5://127.0.0.1:9050")


if __name__ == '__main__':
    main()

如果单纯作为匿名通信工具而言,重建电路实际上并没有什么意义(会降低性能和安全性)。但是如果要写一个爬虫,但是有没有足够的代理池,不失为一种作为代理 IP 的提供方案。(当然,由于出口节点质量良莠不齐,且基本上都在海外,用于爬取国内信息会非常之慢,而且可能已经被很多站点封禁)。
根据 东东's Blog 的测试结果,六个小时内,只有 120130120 \sim 130 个可用 IP,但毕竟不需要额外的费用,已经算是很好的解决方案了。(但如果结合下面提到的自主电路构建,还可以获取更多的可用 IP)

自主电路构建

Stem 不仅仅是一个单纯的 IP 切换工具,他是一个可以深层次操作 Tor 的模块。用户可以根据 Stem 自行进行路由选择。那么第一步应该是获取中继信息

获取中继信息

使用controller.get_network_statuses()可以获取中继的信息,返回的是本地 OP 已知的所有 OR 的信息(与前文描述的一致,这里 OR 只是整个 Tor 网络中的一部分,并非所有 OR)

返回的是一个对象,但是思路 VS Code 无法很好地对内容进行自动补全,不过根据 文档 依旧可以获知所有可用的属性(返回的类为stem.descriptor.router_status_entry.RouterStatusEntryV3,其继承自stem.descriptor.router_status_entry.RouterStatusEntry

属性 解释 版本
document 文档信息(本地测试全部是 None v1
nickname 节点昵称 v1
fingerprint 节点指纹信息 v1
published 节点发布时间 v1
address 节点 IP 地址 v1
or_port 节点端口号 v1
dir_port 文档信息(本地测试全部是 None v1
flags 节点的标志 v1
version 节点版本 v1
version_line 节点声明的版本 v1
or_address 中继的地址,返回的是一个列表,共三项,分别是地址、端口、是否为IPv6(有可能是空列表) v3
identifier_type 身份摘要类型(本地测试全部是 None v3
identifier 身份摘要(本地测试全部是 None v3
digest 中继节点摘要,4040 位大写十六进制字符串 v3
bandwidth 带宽,单位为 kb/skb/s v3
measured 节点声明的带宽是否经过验证(本地测试全部是 None v3
is_unmeasured 带宽验证方案不是根据三次以上验证(本地测试全部是 False v3
unrecognized_bandwidth_entries 无法识别的带宽权重信息(本地测试全部是空列表) v3
exit_policy 节点退出策略(本地测试全部是 None v3
protocols 支持的协议(本地测试全部是空对象) v3
microdescriptor_hashes 摘要及生成方法(本地测试全部是空列表) v3

可以看出,这里实际上大部分信息尽管提供了,但是实际上并没有什么用处。

节点标志

原则上,每一个节点都应该有一个标志,用于对节点特点进行简单的描述

标志 解释
Authority 表明节点是一个目录服务器
BadExit 表明节点不是一个很差的出口节点(可能因为其处于一个进行内容审查的 ISP,也可能因为其通过限制性代理服务联网)
Exit 节点适合用于出口节点
Fast 节点适合用于高速电路
Guard 节点适合用于入口守卫
HSDir 节点可作为 V2 版本的隐藏服务目录服务器
NoEdConsensus 描述中无 Ed25519 密钥标识
Stable 节点适合用于长时间电路连接(很稳定)
Running 节点发布的端口都是可用的(原则上所有可获取的节点都应该是 Running)
Valid 节点是经过验证的(老版本无此标志不会被使用,新版本原则上都存在该标志)
V2Dir 节点实现了 V2 或更高版本的目录协议

在测试中,共有 62896289 个节点,其中带有各种标志的节点数目如下:

标志 个数
Fast 5728
Running 6289
Stable 5574
Valid 6289
Guard 3180
HSDir 3833
V2Dir 5462
Exit 1076
StaleDesc 7
Authority 9
BadExit 8

存储节点信息

在这里,我们可以获取所有可用的节点,可以对这些信息进行存储,以便后续进行节点分析。

可以使用下述代码将其存储到本地

from stem.control import Controller

def main():
    with Controller.from_port(port=9051) as controller:
        controller.authenticate()
        count = 0
        fields = [
            "fingerprint", "or_addresses", "identifier_type",
            "identifier", "digest", "bandwidth",
            "measured", "is_unmeasured", "unrecognized_bandwidth_entries",
            "exit_policy", "protocols", "microdescriptor_hashes",
            "document", "nickname", "published",
            "address", "or_port", "dir_port",
            "flags", "version", "version_line",
        ]
        print("No.\t"+'\t'.join(fields))

        for relay in controller.get_network_statuses():
            print("{}\t{}".format(
                count,
                "\t".join(map(lambda f: str(getattr(relay, f)), fields))
            ))
            count += 1

if __name__ == '__main__':
    main()

电路构建

通过上一步获得所有节点的数据后,即可从中选出任意个节点(至少需要 22 个才能确保匿名性),并建立电路。
使用 controller.new_circuit 可以根据自己的需要建立任意跳数的电路。

但是需要注意的以下几点:

  1. 并非所有获取到的节点都是有效的
  2. 并非所有的节点都可以作为出口节点(出口节点是实际访问网站节点,如果被用于访问非法内容可能会承担相应的责任,因此很多节点设置为不充当出口节点,可以根据 Exit 标志来选择出口节点,也可以在 出口节点列表 中选择)
with Controller.from_port(port=9051) as controller:
        controller.authenticate()
        circuit_id = controller.new_circuit(
            ["0006DE2E77E3C3EC5E1825B076E5257FD200ED0A",
                "0011BD2485AD45D984EC4159C88FC066E5E3300E"],
            await_build=True
        )

        def attach_stream(stream):
            if stream.status == 'NEW':
                controller.attach_stream(stream.id, circuit_id)
                
        controller.add_event_listener(attach_stream, EventType.STREAM)
        try:
            controller.set_conf('__LeaveStreamsUnattached', '1')
            # 在这里通过 Socks5://127.0.0.1:9050 即可使用上面选择的节点(2 跳)进行通信
        finally:
            controller.remove_event_listener(attach_stream)
            controller.reset_conf('__LeaveStreamsUnattached')

获取已构建的电路

在 OP 中,往往是同时维护着多条电路以供备用,使用controller.get_circuits() 即可获得每条电路的信息。
在这里获取的中继节点只有指纹和昵称,但可以采用前面的节点信息反查 IP 等信息,也可以使用 官方的接口 查询信息

with Controller.from_port(port=9051) as controller:
        controller.authenticate()

        for circuit in controller.get_circuits():
            print(circuit.id)
            for relay in circuit.path:
                print(" ",relay)

参考资料