分享个几年前实现的grin代理充值的奇思妙想

分享 / 0 / 1 / 创建于 2年前 / 更新于 2年前

当时我们团队要做一个grin币充值服务，因该服务无法修改介入，那么用户对此服务发起的充值，我们平台无法感知到，而且该服务没有钱包地址概念，也就实现不了充值功能，该问题一直困扰了团队挺久。
后经过我的不懈努力，以及大量的对该服务的通讯抓包分析，我产生了一个奇妙的主意，做一个代理充值服务，代码使用laravel结合workerman实现，部分贴出来和大家分享一下，欢迎大家尽情批判该代码，一晃3年过去了，想起3年前初出茅庐的自己能有这样的想法还是挺难得的。

服务启动命令：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Workerman\Worker;
use App\Workerman\Events;

/**
 * Grin钱包监听
 *
 * @author wen <756307849@qq.com>
 * @date 2019-03-20
 */
class Grin extends Command
{

    protected $signature = 'grin {action} {--d}';

    protected $description = 'Start a grin wallet listen server';

    public function handle()
    {
        global $argv;
        $action = $this->argument('action');

        $argv[0] = 'wk';
        $argv[1] = $action;
        $argv[2] = $this->option('d') ? '-d' : '';

        $this->start();
    }

    private function start()
    {
        $this->startWorker();
        Worker::runAll();
    }

    private function startWorker()
    {
        $socket_name = 'Grin://' . env('GRIN_PROXY_HOST', '127.0.0.1') . ':' . env('GRIN_PROXY_PORT', '9501');
        // 是否启用https
        if ('on' == env('HTTPS', 'off')) {
            // wss 证书最好是申请的证书
            $context = array(
                // 更多ssl选项请参考手册 http://php.net/manual/zh/context.ssl.php
                'ssl' => array(
                    // 请使用绝对路径
                    'local_cert'        => env('SSL_LOCAL_CERT'), // 也可以是crt文件
                    'local_pk'          => env('SSL_LOCAL_PK'),
                    'verify_peer'       => false
                )
            );
            $worker = new Worker($socket_name, $context);
            // 设置transport开启ssl，websocket+ssl即wss
            $worker->transport = 'ssl';
        } else {
            $worker = new Worker($socket_name);
        }

        $worker->name  = 'GrinWalletListen';
        $worker->count = 1;

        $worker->onWorkerStart = function ($worker) {
            Events::onWorkerStart($worker);
        };

        $worker->onConnect = function ($connection) {
            Events::onConnect($connection);
        };

        $worker->onMessage = function ($connection, $buffer) {
            Events::onMessage($connection, $buffer);
        };

        $worker->onClose = function ($connection) {
            Events::onClose($connection);
        };
    }

}

关键业务代码：

<?php

namespace App\Workerman;

use Workerman\Connection\AsyncTcpConnection;
use App\JsonRpc\Blockchain as BC;

/**
 * Grin钱包监听回调
 *
 * @author wen <756307849@qq.com>
 * @date 2019-04-29
 */
class Events
{
    /**
     * 充值状态
     */
    const STATUS = [
        'unconfirmed' => 1, // 未确认的充值
    ];

    public static function onWorkerStart($worker)
    {
    }

    public static function onConnect($connection)
    {
        echo "--onConnect\n";
    }

    /**
     * @param $connection
     * @param $buffer
     * @throws \Exception
     */
    public static function onMessage($connection, $buffer)
    {
        $data = decode($buffer);

        // 验证充值地址是否存在
        try {
            $result = self::verifyAddress($data['ADDRESS']);
        } catch (\Exception $e) {
            $connection->send('Account Address Error');
            custom_log('info', 'grin/info.log', '--充值地址验证失败：' . $e->getMessage());
            echo '--充值地址验证失败：' . $e->getMessage() . "\n";
            return;
        }
        echo "--充值地址验证通过\n";

        $body = json_decode($data['BODY'], true);

        // 写入充值记录
        $recharge_data = [
            'symbol' => 'GRIN',
            'amount' => bcdiv($body['amount'], 1000000000, 8),
            'address' => $data['ADDRESS'],
            'tx_hash' => $body['id'],
            'status' => self::STATUS['unconfirmed'],
            'confirms' => 0,
        ];

        try {
            $result = self::save($recharge_data);
        } catch (\Exception $e) {
            $connection->send('Create failure');
            custom_log('info', 'grin/info.log', '--写入充值记录失败：' . $e->getMessage());
            echo '--写入充值记录失败：' . $e->getMessage() . "\n";
            return;
        }
        echo "--写入充值记录成功\n";

        $remote_address = 'GrinServer://' . env('GRIN_HOST') . ':' . env('GRIN_PORT');
        $remote_connection = new AsyncTcpConnection($remote_address);

        // 将原始报文发送给服务端
        $remote_connection->send($data['BUFFER']);

        // Pipe.
        $remote_connection->pipe($connection);
        $connection->pipe($remote_connection);

        $remote_connection->connect();
    }

    public static function onClose($connection)
    {
        echo "--onClose\n";
    }

    /**
     * 保存交易数据
     *
     * @param $data
     * @return mixed
     * @throws \Exception
     */
    protected static function save($data)
    {
        try {
            $result = BC::post('/api/recharge/add', [$data]);
            if (200 != $result->code)
                throw new \Exception('blockchain error: ' . $result->msg, $result->code);

            return $result->data;
        } catch (\Exception $e) {
            throw new \Exception('blockchain error: ' . $e->getMessage());
        }
    }

    /**
     * 验证地址是否有效
     *
     * @param string $address
     * @return mixed
     * @throws \Exception
     */
    protected static function verifyAddress(string $address)
    {
        try {
            $result = BC::post('/api/address/check', [['symbol' => 'grin', 'address' => $address]]);
            if (200 != $result->code)
                throw new \Exception('blockchain error: ' . $result->msg, $result->code);

            return $result;
        } catch (\Exception $e) {
            throw new \Exception('blockchain error: ' . $e->getMessage());
        }
    }


}

助手函数：

<?php
/**
 * 自定义助手类
 *
 * 函数定义规范: 下划线法
 * 注释要写 author
 */

use GuzzleHttp\Client;
use GuzzleHttp\Exception\GuzzleException;
use Monolog\Handler\StreamHandler;
use Monolog\Logger;

if (!function_exists('decode')) {
    /**
     * grin client to server data decode
     *
     * @author wen <756307849@qq.com>
     * @access public
     * @return array
     */
    function decode($buffer)
    {
        $server = [
            'REQUEST_METHOD'  => '',
            'REQUEST_URI'     => '',
            'SERVER_PROTOCOL' => '',
            'REQUEST_TIME'    => time(),
            'BUFFER'          => '',
            'ADDRESS'         => '',
            'BODY'            => '',
        ];

        // Parse headers.
        list($http_header, $http_body) = explode("\r\n\r\n", $buffer, 2);
        $header_data = explode("\r\n", $http_header);
        $server['BODY'] = $http_body;

        list($server['REQUEST_METHOD'], $server['REQUEST_URI'], $server['SERVER_PROTOCOL']) = explode(' ',
            $header_data[0]);

        $address = str_replace('/v1/wallet/foreign/receive_tx', '', $server['REQUEST_URI']);
        $server['ADDRESS'] = ltrim($address, '/');
        $server['BUFFER'] = str_replace($server['REQUEST_URI'], '/v1/wallet/foreign/receive_tx', $buffer);

        return $server;
    }
}

if (!function_exists('grin_date_format')) {
    /**
     * grin client to server data format
     *
     * @author wen <756307849@qq.com>
     * @access public
     * @return array
     */
    function grin_date_format($data)
    {
        // 用户充值地址
        list($method, $url) = explode(' ', $data);
        $addr = str_replace('/v1/wallet/foreign/receive_tx', '', $url);
        $addr = ltrim($addr, '/');

        // 合约内容
        $data = explode("\n", $data);
        $tx = end($data);
        $tx = json_decode($tx, true);

        $res['method'] = $method;
        $res['addr'] = $addr;
        $res['tx'] = $tx;
        $res['url'] = $url;
        return $res;
    }
}

if (!function_exists('graphql_post')) {
    /**
     * @param $query
     * @return mixed
     */
    function graphql_post($query)
    {
        $result = ['code' => 400, 'msg' => ''];

        $client = new Client(['base_uri' => env('BC_HOST')]);

        $options = [
            'form_params' => ['query' => $query],
            'stream' => true,
            'read_timeout' => 3
        ];
        try {
            $response = $client->request('POST', '', $options)->getBody();
        } catch (GuzzleException $e) {
            $result['msg'] = $e->getMessage();
            return $result;
        }
        $contents = json_decode($response->getContents(), true);
        if (!isset($contents['code']) || 200 != $contents['code']) {
            $result['code'] = $contents['code'];
            $result['msg'] = $contents['msg'];
            return $result;
        }

        $result['code'] = 200;
        $result['data'] = $contents['data'];
        return $result;
    }
}

if (!function_exists('custom_log')) {
    /**
     * 自定义日志
     *
     * @param string $level
     * @param string $path
     * @param $msg
     */
    function custom_log(string $level, string $path, $msg)
    {
        $log = new Logger('');
        try {
            $log->pushHandler(new StreamHandler(storage_path('logs/' . $path)));
        } catch (Exception $e) {

        }

        $msg = is_array($msg) ? json_encode($msg) : $msg;
        $log->log($level, $msg);
    }
}

两个自定义基于TCP的GRIN通讯协议：

<?php

namespace Protocols;

use Workerman\Connection\TcpConnection;

/**
 * 自定义Grin通讯协议
 *
 * Class Grin
 * @package Protocols
 * @author wen <756307849@qq.com>
 * @date 2019-03-30
 */
class Grin
{
    public static $methods = array('POST');

    /**
     * 检查包的完整性
     * 如果能够得到包长，则返回包的在buffer中的长度，否则返回0继续等待数据
     * 如果协议有问题，则可以返回false，当前客户端连接会因此断开
     * @param string $recv_buffer
     * @return int
     */
    public static function input($recv_buffer, TcpConnection $connection)
    {
        echo "原始报文：--------------------\n$recv_buffer\n--------------------------------\n";
        if (!strpos($recv_buffer, "\r\n\r\n")) {
            // Judge whether the package length exceeds the limit.
            if (strlen($recv_buffer) >= $connection->maxPackageSize) {
                $connection->close();
                return 0;
            }
            return 0;
        }

        list($header,) = explode("\r\n\r\n", $recv_buffer, 2);
        $method = substr($header, 0, strpos($header, ' '));

        if(in_array($method, static::$methods)) {
            return static::getRequestSize($header);
        }else{
            $connection->send("HTTP/1.1 400 Bad Request\r\n\r\n", true);
            return 0;
        }
    }

    /**
     * Get whole size of the request
     * includes the request headers and request body.
     * @param string $header The request headers
     * @return integer
     */
    protected static function getRequestSize($header)
    {
        $match = array();
        if (preg_match("/\r\nContent-Length: ?(\d+)/i", $header, $match)) {
            $content_length = isset($match[1]) ? $match[1] : 0;
            return $content_length + strlen($header) + 4;
        }
        return 0;
    }

    /**
     * 打包，当向客户端发送数据的时候会自动调用
     * @param string $buffer
     * @return string
     */
    public static function encode($buffer)
    {
        return $buffer;
    }

    /**
     * 解包，当接收到的数据字节数等于input返回的值（大于0的值）自动调用
     * 并传递给onMessage回调函数的$data参数
     * @param string $buffer
     * @return string
     */
    public static function decode($buffer)
    {
        return $buffer;
    }
}

<?php

namespace Protocols;

use Workerman\Connection\TcpConnection;

/**
 * 与服务端的通讯协议
 *
 * Class GrinServer
 * @package Protocols
 * @author wen <756307849@qq.com>
 * @date 2019-05-24
 */
class GrinServer
{
    /**
     * 检查包的完整性
     * 如果能够得到包长，则返回包的在buffer中的长度，否则返回0继续等待数据
     * 如果协议有问题，则可以返回false，当前客户端连接会因此断开
     * @param string $recv_buffer
     * @return int
     */
    public static function input($recv_buffer, TcpConnection $connection)
    {
        echo "服务端原始报文：--------------------\n$recv_buffer\n--------------------------------\n";
        if (!strpos($recv_buffer, "\r\n\r\n")) {
            // Judge whether the package length exceeds the limit.
            if (strlen($recv_buffer) >= $connection->maxPackageSize) {
                $connection->close();
                return 0;
            }
            return 0;
        }

        list($header,) = explode("\r\n\r\n", $recv_buffer, 2);

        return static::getRequestSize($header);
    }

    /**
     * Get whole size of the request
     * includes the request headers and request body.
     * @param string $header The request headers
     * @return integer
     */
    protected static function getRequestSize($header)
    {
        $match = array();
        if (preg_match("/\r\nContent-Length: ?(\d+)/i", $header, $match)) {
            $content_length = isset($match[1]) ? $match[1] : 0;
            return $content_length + strlen($header) + 4;
        }
        return 0;
    }

    /**
     * 打包，当向客户端发送数据的时候会自动调用
     * @param string $buffer
     * @return string
     */
    public static function encode($buffer)
    {
        return $buffer;
    }

    /**
     * 解包，当接收到的数据字节数等于input返回的值（大于0的值）自动调用
     * 并传递给onMessage回调函数的$data参数
     * @param string $recv_buffer
     * @return string
     */
    public static function decode($recv_buffer)
    {
        // Grin版本升级有bug，暂时处理自行处理
        list($http_header, $http_body) = explode("\r\n\r\n", $recv_buffer, 2);

        // 处理服务端报文可能出现反引号的问题
        $http_body = trim($http_body, '"');
        $http_body = str_replace("\\", '', $http_body);

        preg_match("/\r\n(Content-Length: ?(\d+))/i", $http_header, $match);

        $http_header = str_replace($match[1], 'content-length: ' . strlen($http_body), $http_header);

        $buffer = $http_header . "\r\n\r\n" . $http_body;
        echo "服务端处理后报文：--------------------\n$buffer\n--------------------------------\n";
        return $buffer;
    }
}