本文章为纯技术向文章,不涉及非法内容。同时抵制使用匿名网络进行非法行为
使用 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
- 直接请求,返回的是本机 IP
- 使用 Tor 请求,返回的是某个出口节点的 IP
- 使用 Tor 请求,返回的 IP 与 2 相同
- 要求重建电路,使用 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 的测试结果,六个小时内,只有 个可用 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 |
中继节点摘要, 位大写十六进制字符串 | v3 |
bandwidth |
带宽,单位为 | 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 或更高版本的目录协议 |
在测试中,共有 个节点,其中带有各种标志的节点数目如下:
标志 | 个数 |
---|---|
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()
电路构建
通过上一步获得所有节点的数据后,即可从中选出任意个节点(至少需要 个才能确保匿名性),并建立电路。
使用 controller.new_circuit
可以根据自己的需要建立任意跳数的电路。
但是需要注意的以下几点:
- 并非所有获取到的节点都是有效的
- 并非所有的节点都可以作为出口节点(出口节点是实际访问网站节点,如果被用于访问非法内容可能会承担相应的责任,因此很多节点设置为不充当出口节点,可以根据
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)