惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知
Stack Overflow Blog
Stack Overflow Blog
云风的 BLOG
云风的 BLOG
G
Google Developers Blog
J
Java Code Geeks
C
Check Point Blog
Last Week in AI
Last Week in AI
Microsoft Azure Blog
Microsoft Azure Blog
Blog — PlanetScale
Blog — PlanetScale
月光博客
月光博客
Vercel News
Vercel News
The GitHub Blog
The GitHub Blog
L
LangChain Blog
有赞技术团队
有赞技术团队
Cyber Security Advisories - MS-ISAC
Cyber Security Advisories - MS-ISAC
博客园 - 司徒正美
IT之家
IT之家
Martin Fowler
Martin Fowler
让小产品的独立变现更简单 - ezindie.com
让小产品的独立变现更简单 - ezindie.com
T
Tailwind CSS Blog
U
Unit 42
Jina AI
Jina AI
Microsoft Security Blog
Microsoft Security Blog
I
InfoQ

OhYee 博客

小鹏辅助驾驶测评|OhYee 博客 小鹏非支持手机开启自动解锁|OhYee 博客 使用函数计算实现 301 重定向|OhYee 博客 针对 HTML 内容使用 Ant Design 图片弹框|OhYee 博客 博客进程泄露及僵尸进程解决|OhYee 博客 蓝易云服务器体验|OhYee 博客 SSH 调起本地 VSCode|OhYee 博客 【2022 秋招内推】阿里云后端研发工程师|OhYee 博客 使用函数计算获取 IP 地址信息|OhYee 博客 正确获取客户端 IP/HTTP Header 也可能重复|OhYee 博客 评测 Oculus Quest2 及 BigScreen|OhYee 博客 NextJS 热重载保留状态|OhYee 博客 如何优雅地贴 gist 代码|OhYee 博客 Linux 精细化文件权限|OhYee 博客 VSCode 容器开发环境|OhYee 博客 Clash 的不兼容更新排查|OhYee 博客 Zeek 导出 PCAP|OhYee 博客 记一次 ssh 配置问题|OhYee 博客 Git Commit 规范化工具|OhYee 博客 谈谈《星之卡比-探索发现》|OhYee 博客 VSCode 快捷键绑定 Shell 命令|OhYee 博客 ASN.1 语法及 X.509 证书格式解析解析|OhYee 博客 腾讯企业邮箱忽略 MX 记录发信|OhYee 博客 Chrome/Edge 标签组插件|OhYee 博客 【应届内推】阿里云后端研发工程师|OhYee 博客 损坏的 Typecho 备份处理为 JSON|OhYee 博客 VS Code VIM 插件高效使用|OhYee 博客 SSH 正反向代理|OhYee 博客 Let's Encrypt 根证书过期引发的问题|OhYee 博客 OpenWRT 忽略内核依赖|OhYee 博客
使用 Python Stem 操作 Tor|OhYee 博客
2020-07-30 · via OhYee 博客

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

使用 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 的测试结果,六个小时内,只有 120∼130120 \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)

参考资料