关键代码基本重构完成，但目前尚未调通

2026-10-04 19:42:00 +08:00 · 2021-10-21 17:33:16 +08:00
1 parent 3309b8e12b
commit 5eed1d942b
@@ -17,8 +17,6 @@ import android.os.Build;
 import android.os.Handler;
 import android.os.Message;
 import android.os.ParcelFileDescriptor;
-import android.util.Log;
-import android.util.Pair;
 import android.widget.Toast;

 import androidx.annotation.NonNull;
@@ -31,19 +29,28 @@ import com.pkusz.min_vpn_client.model.MINVpnSettingAPI;
 import com.pkusz.min_vpn_client.model.Response.VpnServerInfo;
 import com.pkusz.min_vpn_client.model.event.ConnectEvent;
 import com.pkusz.min_vpn_client.model.event.TimeoutEvent;
+import com.pkusz.min_vpn_client.services.Runnables.Ip2MinTask;
+import com.pkusz.min_vpn_client.services.Runnables.Min2IpTask;
+import com.pkusz.min_vpn_client.services.Runnables.ReadTunTask;
+import com.pkusz.min_vpn_client.services.Runnables.ReceiveTask;
+import com.pkusz.min_vpn_client.services.Runnables.SendTask;
+import com.pkusz.min_vpn_client.services.Runnables.WriteTunTask;
 import com.pkusz.min_vpn_client.utils.SM4HelperForConnection;

 import org.greenrobot.eventbus.EventBus;

+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;

-import common.LoggerHelper;
 import logicface.LogicFace;
 import packet.CPacket;

@@ -53,6 +60,7 @@ public class MINVpnService extends VpnService implements Handler.Callback{
    public static final String ACTION_DISCONNECT = "com.pkusz.min_vpn_client.STOP";

    private Handler mHandler;
+    AtomicReference<VpnPreparation> vpnPreparation = new AtomicReference<>(); // 握手工具类

    // 标记是否在运行
    private static boolean isRunning = false;
@@ -175,47 +183,90 @@ public class MINVpnService extends VpnService implements Handler.Callback{
 //            vpnServerInfo.IP="43.242.142.226"; // gdcni13
            vpnServerInfo.Port=14922;
            // 2. 连接建立阶段
-            Pair<LogicFace,ParcelFileDescriptor> pair;
+            System.out.println("new 1");
+            VpnConnectionInfo connectionInfo;
            if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.Q) {
-                pair=handshakeAndBuildTun(
+                System.out.println("new 2");
+                connectionInfo=handshakeAndBuildTun(
                        vpnServerInfo,secret,proxyHost,proxyPort,allow,packages,
                        GlobalSetting.isSM4InExternal,allocateIP
                );
-                if(pair==null){
-                    System.out.println("连接建立阶段出现问题!!!");
+                System.out.println("new 3");
+                if(connectionInfo==null){
+                    System.out.println("连接建立阶段出现问题1!!!");
                    return;
                }
+            }else {
+                System.out.println("连接建立阶段出现问题2!!!");
+                return;
            }
+            System.out.println("new 4");
            // 3. 开启线程池，进入数据中转阶段
-
+            setupVPN(connectionInfo.face, connectionInfo.mParcelFileDescriptor,
+                    connectionInfo.allocatedAddress,connectionInfo.localFacePrefix);
        }
    }

-    // 握手，并创建tun
+    // 握手，并创建tun。注意，安卓要求只能在线程中进行网络请求
    @RequiresApi(api = Build.VERSION_CODES.Q)
-    private Pair<LogicFace,ParcelFileDescriptor>
+    private VpnConnectionInfo
    handshakeAndBuildTun(final VpnServerInfo vpnServerInfo, final byte[] sharedSecret,
                         final String proxyHostName, final int proxyHostPort, boolean allow,
                         final Set<String> packages,
                         final boolean isSM4,
                         final String allocatedAddressFromInner){
-        VpnPreparation vpnPreparation=new VpnPreparation(this,vpnServerInfo,sharedSecret,
-                proxyHostName,proxyHostPort,allow,packages,isSM4,allocatedAddressFromInner);
-        // 对vpn connection进行各种监听设置
-        startConnectionListener(vpnPreparation);
-        return vpnPreparation.startPreparation();
+        Thread netRequestThread = new Thread(() -> {
+            vpnPreparation.set(new VpnPreparation(this, vpnServerInfo, sharedSecret,
+                    proxyHostName, proxyHostPort, allow, packages, isSM4, allocatedAddressFromInner));
+            // 对vpn connection进行各种监听设置
+            startConnectionListener(vpnPreparation.get());
+            vpnPreparation.get().startPreparation();
+        });
+        netRequestThread.start();
+        // 等待五秒，五秒内依然线程没有执行到对象实例化并获取握手结果，就放弃
+        long startTime=System.currentTimeMillis();
+        while ((System.currentTimeMillis()-startTime)/1000<5){
+            if(vpnPreparation.get()!=null){
+                if(vpnPreparation.get().getInfo()!=null) {
+                    return vpnPreparation.get().getInfo();
+                }
+            }
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        return null;
+    }
+
+    static class VpnConnectionInfo{
+        LogicFace face; //建立的face连接
+        ParcelFileDescriptor mParcelFileDescriptor; // tun句柄
+        String allocatedAddress; // 分配的tun地址
+        String localFacePrefix; // 本机本次连接使用的客户端前缀
+        byte[] realSecret; // 协商密钥，暂时没用
    }

    // 开启数据中转阶段
-    private void setupVPN(LogicFace face,ParcelFileDescriptor mParcelFileDescriptor){
+    private void setupVPN(LogicFace face,ParcelFileDescriptor mParcelFileDescriptor,
+                          String allocatedAddress,String localFacePrefix){
        // 1. 初始化缓冲队列
        readTunDataQueue = new ConcurrentLinkedQueue<>();
        readTunCPackets = new ConcurrentLinkedQueue<>();
        writeTunCPackets = new ConcurrentLinkedQueue<>();
        writeTunDataQueue = new ConcurrentLinkedQueue<>();
-        // 2. 开启线程池
+        // 2. 初始化收发channel
+        FileChannel tunInput = new FileInputStream(mParcelFileDescriptor.getFileDescriptor()).getChannel();
+        FileChannel tunOutput = new FileOutputStream(mParcelFileDescriptor.getFileDescriptor()).getChannel();
+        // 3. 开启线程池
        executorService = Executors.newFixedThreadPool(6);
-        executorService.submit();
+        executorService.submit(new ReadTunTask(readTunDataQueue,tunInput));
+        executorService.submit(new Ip2MinTask(readTunDataQueue,readTunCPackets,allocatedAddress,localFacePrefix));
+        executorService.submit(new SendTask(readTunCPackets,face));
+        executorService.submit(new ReceiveTask(writeTunCPackets,face));
+        executorService.submit(new Min2IpTask(writeTunCPackets,writeTunDataQueue));
+        executorService.submit(new WriteTunTask(writeTunDataQueue,tunOutput));
    }

    /**
@@ -1,82 +1,139 @@
 package com.pkusz.min_vpn_client.services.Runnables;
 /*
 * @Author: Wang Feng
- * @Description:
+ * @Description: 将tun读取的Ip包转为min包
 * @Version: 1.0.0
 * @Date: 11:35 2021/10/21
 * @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
 */

+import com.pkusz.min_vpn_client.model.MINVpnSettingAPI;
 import com.pkusz.min_vpn_client.services.ByteBufferPool;
 import com.pkusz.min_vpn_client.utils.IPPackageUtil;
-import com.pkusz.min_vpn_client.utils.SM4HelperForConnection;

-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;

+import component.ComponentException;
+import component.Identifier;
+import component.IdentifierComponent;
 import packet.CPacket;

 public class Ip2MinTask implements Runnable{
    private  ConcurrentLinkedQueue<ByteBuffer> readTunDataQueue;
    private ConcurrentLinkedQueue<CPacket> readTunCPackets;
-    private String allocatedAddress;
-    private boolean isSM4=false;
+    private String allocatedAddress; // tun地址
+    private boolean isSM4=false; // 加密方式
+    private final AtomicInteger interestSeq=new AtomicInteger(10); // 前缀序号
+    private String localFacePrefix; // 本地前缀
+    private String serverPrefix=MINVpnSettingAPI.INSTANCE.getServicePrefix(); // 服务器前缀

    public Ip2MinTask(ConcurrentLinkedQueue<ByteBuffer> readTunDataQueue,
                      ConcurrentLinkedQueue<CPacket> readTunCPackets,
-                      String allocatedAddress){
+                      String allocatedAddress,String localFacePrefix){
        this.readTunDataQueue=readTunDataQueue;
        this.readTunCPackets=readTunCPackets;
        this.allocatedAddress=allocatedAddress;
+        this.localFacePrefix=localFacePrefix;
    }

    @Override
    public void run() {
        try {
+            // Tun地址
            byte[] allocatedAddressBytes= IPPackageUtil.StringAddress2byteAddress(allocatedAddress);
-            // 循环读数据
-            while (!Thread.interrupted()) {
-                int readLen; // tun读包原始长促
-                int encryptedLen; // 加密后的包长度+tunIP地址长度
-                // 取tun读取到的包
-                ByteBuffer buff1 = readTunDataQueue.poll();
-                if(buff1==null){
-                    continue;
+            assert allocatedAddressBytes != null;
+            int ipLen=allocatedAddressBytes.length;
+            // 当前线程
+            Thread currentThread = Thread.currentThread();
+            // 最终去做CPacket负载的buff
+            ByteBuffer ipAndData=ByteBufferPool.acquire();
+            // 循环处理数据：ip to min
+            while (true) {
+                ipAndData.clear();
+                // 1.阻塞，直到读取出待处理的IP包
+                ByteBuffer pendingProcessIpPacket;
+                do {
+                    pendingProcessIpPacket=readTunDataQueue.poll();
+                    if(pendingProcessIpPacket!=null){
+                        break;
+                    }
+                    Thread.sleep(10);
+                }while(!currentThread.isInterrupted());
+
+                // 如果线程被中断，则退出循环
+                if(currentThread.isInterrupted()){
+                    break;
                }
-                readLen = buff1.position();
-                if (readLen > 0) {
-                    System.out.println("read from tun queue, readLen = " + readLen);

-                    // 2. 对数据做加密，然后将服务器分配的地址[allocatedAddress]和加密数据一起放入buff2
-                    assert allocatedAddressBytes != null;
-                    System.arraycopy(allocatedAddressBytes, 0, buff2, 0, allocatedAddressBytes.length);
+                assert pendingProcessIpPacket != null;
+                pendingProcessIpPacket.flip();

-                    byte[] ipPackageData = new byte[readLen];
-                    System.arraycopy(buff1.array(), 0, ipPackageData, 0, readLen);
-                    byte[] encrytedIPPackage;
-                    if (isSM4) {
-                        encrytedIPPackage = SM4HelperForConnection.encryptWithSM4(realSecret, ipPackageData);
-                    } else { // 不加密
-                        encrytedIPPackage = ipPackageData;
-                    }
-                    if (encrytedIPPackage == null) {
-                        System.out.println("对称加密错误!!! 对称加密是否为sm4: " + isSM4);
-                        return;
-                    }
-                    System.arraycopy(encrytedIPPackage, 0, buff2, allocatedAddressBytes.length, encrytedIPPackage.length);
-                    encryptedLen = allocatedAddressBytes.length + encrytedIPPackage.length;
-
-                    // 3. 将cpacket放入发包缓冲队列
-                    CPacket cPacket = genInterestPkt(buff2, encryptedLen);
-                    // 查看这个时候的包长度
-                    if (cPacket != null) {
-                        readTunCPackets.offer(cPacket);
-                    }
+                // 2.先将ip头放入buff
+                ipAndData.put(allocatedAddressBytes);
+                // 再将ip包放入buff（视情况决定是否加密）
+                int readLen=pendingProcessIpPacket.position();
+                byte[] ipPackageData = new byte[readLen];
+                pendingProcessIpPacket.get(ipPackageData);
+                int encryptedLen=readLen;
+                if(isSM4){
+                    System.out.println("暂不支持sm4!!!");
+                }else {
+                    ipAndData.put(ipPackageData);
                }
+                ipAndData.flip();
+
+                // 制造CPacket，入队
+                int totalLen=ipLen+encryptedLen;
+                byte[] allData=new byte[totalLen];
+                ipAndData.get(allData);
+                CPacket cPacket=genInterestPkt(allData,totalLen);
+                if(cPacket!=null){
+                    readTunCPackets.offer(cPacket);
+                }else{
+                    System.out.println("error: 生成CPacket出现问题!!!请修复程序!!!");
+                }
+
+                // 释放读出的bytebuffer到pool中
+                ByteBufferPool.release(pendingProcessIpPacket);
            }
-        } catch (IOException | InterruptedException e) {
+        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
+
+    // 将按照规则生成的IP包数据包裹在CPacket中
+    private CPacket genInterestPkt(byte[] buf,int bufLen){
+        // 构造目的标识：/{service prefix}/{cPacket seq}/{defraudString}
+        Identifier serverIdentifier;
+        try {
+            serverIdentifier = new Identifier(serverPrefix);
+        } catch (ComponentException e) {
+            e.printStackTrace();
+            System.out.println("genInterestPkt error 1: serverIdentifier init error: " + e.getMessage());
+            return null;
+        }
+        serverIdentifier.appendFragmentNumber(interestSeq.getAndAdd(1));
+        serverIdentifier.append(new IdentifierComponent(MINVpnSettingAPI.defraudString));
+
+        // 构造本地标识：/{local_prefix}
+        Identifier clientIdentifier;
+        try {
+            clientIdentifier = new Identifier(localFacePrefix);
+        } catch (ComponentException e) {
+            e.printStackTrace();
+            System.out.println("genInterestPkt error 1: clientIdentifier init error: " + e.getMessage());
+            return null;
+        }
+
+        // 构造CPacket
+        CPacket cPacket = new CPacket();
+        cPacket.setSrcIdentifier(clientIdentifier);
+        cPacket.setDstIdentifier(serverIdentifier);
+        cPacket.ttl.setTtl(64);
+        cPacket.payload.setValue(buf);
+
+        return cPacket;
+    }
 }
@@ -7,5 +7,63 @@ package com.pkusz.min_vpn_client.services.Runnables;
 * @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
 */

-public class Min2IpTask {
+import com.pkusz.min_vpn_client.services.ByteBufferPool;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import packet.CPacket;
+
+public class Min2IpTask implements Runnable{
+    private ConcurrentLinkedQueue<CPacket> writeTunCPackets;
+    private ConcurrentLinkedQueue<ByteBuffer> writeTunDataQueue;
+    private boolean isSM4=false; // 加密方式
+
+    public Min2IpTask(ConcurrentLinkedQueue<CPacket> writeTunCPackets,
+                      ConcurrentLinkedQueue<ByteBuffer> writeTunDataQueue){
+        this.writeTunCPackets=writeTunCPackets;
+        this.writeTunDataQueue=writeTunDataQueue;
+    }
+
+    @Override
+    public void run() {
+        try{
+            // 当前线程
+            Thread currentThread = Thread.currentThread();
+            // 循环处理数据: min to ip
+            while (true){
+                // 1.阻塞，直到读取出待处理的CPacket包
+                CPacket pendingProcessCPacket;
+                do {
+                    pendingProcessCPacket=writeTunCPackets.poll();
+                    if(pendingProcessCPacket!=null){
+                        break;
+                    }
+                    Thread.sleep(10);
+                }while(!currentThread.isInterrupted());
+
+                // 如果线程被中断，则退出循环
+                if(currentThread.isInterrupted()){
+                    break;
+                }
+
+                // 2.读取负载字段
+                assert pendingProcessCPacket != null;
+                byte[] rawIpDataBytes = pendingProcessCPacket.payload.getValue();
+                // 解密
+                ByteBuffer ipData= ByteBufferPool.acquire();
+                if(isSM4){ //解密
+                    System.out.println("暂不支持sm4!!!");
+                }else{
+                    ipData.put(rawIpDataBytes);
+                }
+                ipData.flip();
+
+                // 3.入队
+                writeTunDataQueue.offer(ipData);
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
 }
@@ -53,6 +53,7 @@ public class ReadTunTask implements Runnable{
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
+            System.out.println("error: tun读包出现问题，请及时修复!!!");
        }
    }
 }
@@ -7,5 +7,52 @@ package com.pkusz.min_vpn_client.services.Runnables;
 * @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
 */

-public class ReceiveTask {
+import java.io.IOException;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import logicface.LogicFace;
+import logicface.LogicFaceException;
+import packet.CPacket;
+
+public class ReceiveTask implements Runnable{
+    private ConcurrentLinkedQueue<CPacket> writeTunCPackets;
+    private LogicFace face;
+
+    public ReceiveTask(ConcurrentLinkedQueue<CPacket> writeTunCPackets,
+                       LogicFace face){
+        this.writeTunCPackets=writeTunCPackets;
+        this.face=face;
+    }
+
+    @Override
+    public void run() {
+        try {
+            boolean dataReceive;
+            // 循环收数据
+            while (!Thread.interrupted()) {
+                // 读包入队 todo: 这里收包的时间要根据face收发是否互相影响来定!!!
+                CPacket cPacket;
+                try {
+                    cPacket = face.receiveCPacket(10000);
+                }catch (LogicFaceException e){
+                    e.printStackTrace();
+                    System.out.println("error: face收包出现问题，请及时修复(inner)!!!");
+                    continue;
+                }
+                if(cPacket!=null){
+                    writeTunCPackets.offer(cPacket);
+                    dataReceive=true;
+                }else{
+                    dataReceive=false;
+                }
+                // 收不到就暂停线程
+                if(!dataReceive){
+                    Thread.sleep(10);
+                }
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            System.out.println("error: face收包出现问题，请及时修复!!!");
+        }
+    }
 }
@@ -7,5 +7,54 @@ package com.pkusz.min_vpn_client.services.Runnables;
 * @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
 */

-public class SendTask {
+import com.pkusz.min_vpn_client.services.ByteBufferPool;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import logicface.LogicFace;
+import logicface.LogicFaceException;
+import packet.CPacket;
+
+public class SendTask implements Runnable{
+    private ConcurrentLinkedQueue<CPacket> readTunCPackets;
+    private LogicFace face;
+
+    public SendTask(ConcurrentLinkedQueue<CPacket> readTunCPackets, LogicFace face){
+        this.readTunCPackets=readTunCPackets;
+        this.face=face;
+    }
+
+    @Override
+    public void run() {
+        try {
+            boolean dataSend;
+            // 循环发数据
+            while (!Thread.interrupted()) {
+                // 出队
+                CPacket cPacket = readTunCPackets.poll();
+                // 发包
+                if(cPacket!=null){
+                    try {
+                        face.sendCPacket(cPacket);
+                    }catch (LogicFaceException e){
+                        e.printStackTrace();
+                        System.out.println("error: face发包出现问题，请及时修复(inner)!!!");
+                        continue;
+                    }
+                    dataSend = true;
+                }else{
+                    dataSend = false;
+                }
+                // 无包可发，就暂停线程
+                if(!dataSend){
+                    Thread.sleep(10);
+                }
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            System.out.println("error: face发包出现问题，请及时修复!!!");
+        }
+    }
 }
@@ -14,7 +14,6 @@ import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.concurrent.ConcurrentLinkedQueue;

-import jnr.ffi.Struct;

 public class WriteTunTask implements Runnable{
    private ConcurrentLinkedQueue<ByteBuffer> writeTunQueue;
@@ -52,6 +51,7 @@ public class WriteTunTask implements Runnable{
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
+            System.out.println("error: tun写包出现问题，请及时修复!!!");
        }
    }
 }
@@ -75,6 +75,9 @@ public class VpnPreparation implements VpnPreparationInterface{
    // 服务器给客户端的tun网卡分配的IP地址
    private String allocatedAddress;

+    // 握手结果
+    private MINVpnService.VpnConnectionInfo vpnConnectionInfo=null;
+
    // 待决Intent、回调函数（建立连接成功；建立连接失败；长时间不操作的超时回调）
    private PendingIntent mConfigureIntent;
    private OnEstablishListener mOnEstablishListener;
@@ -143,23 +146,45 @@ public class VpnPreparation implements VpnPreparationInterface{
    }

    @RequiresApi(api = Build.VERSION_CODES.Q)
-    public Pair<LogicFace, ParcelFileDescriptor> startPreparation(){
+    public void startPreparation(){
        // 1. 建立一条连接服务器的TCP-LogicFace
        LogicFace face=initTCPLogicFace(this.mServerName,this.mServerPort);
        if(face==null){
            LoggerHelper.severe("vpn connection run error: tcp logic face init error");
-            return null;
+            return;
        }

        // 2. VPN握手，交换信息
        ParcelFileDescriptor mParcelFileDescriptor=handShake(face);
        if(mParcelFileDescriptor==null){
            LoggerHelper.severe("vpn connection run error: handle shake error");
-            return null;
+            return;
        }

        // 返回建立的face及tun句柄
-        return new Pair<>(face,mParcelFileDescriptor);
+        MINVpnService.VpnConnectionInfo vpnConnectionInfo=new MINVpnService.VpnConnectionInfo();
+        vpnConnectionInfo.face=face;
+        vpnConnectionInfo.mParcelFileDescriptor=mParcelFileDescriptor;
+        vpnConnectionInfo.allocatedAddress=this.allocatedAddress;
+        vpnConnectionInfo.localFacePrefix=this.localFacePrefix;
+        vpnConnectionInfo.realSecret=realSecret;
+        this.vpnConnectionInfo=vpnConnectionInfo;
+
+        // 开启一个线程，保证本对象一直存在
+//        new Thread(()->{
+//            while(true){
+//                try {
+//                    Thread.sleep(10000);
+//                } catch (InterruptedException e) {
+//                    e.printStackTrace();
+//                }
+//                System.out.println("心跳线程还健在!!!");
+//            }
+//        }).start();
+    }
+
+    public MINVpnService.VpnConnectionInfo getInfo(){
+        return this.vpnConnectionInfo;
    }

    /**