Ratchet PHP WAMP - React/ZeroMQ - 特定用户广播 [英] Ratchet PHP WAMP - React / ZeroMQ - Specific user broadcast

问题描述

注意:这与这个问题 使用 MessageComponentInterface.我正在使用 WampServerInterface 代替，所以这个问题特别与那部分有关.我需要一个包含代码示例和解释的答案，因为我认为这在未来对其他人有帮助.

Note: This is not the same as this question which utilises MessageComponentInterface. I am using WampServerInterface instead, so this question pertains to that part specifically. I need an answer with code examples and an explanation, as I can see this being helpful to others in the future.

尝试为个人用户循环推送

我正在使用 Ratchet 和 ZeroMQ 的 WAMP 部分，并且我目前有一个 push 集成的工作版本教程.

我正在尝试执行以下操作:

I'm attempting to perform the following:

• zeromq 服务器已启动并运行，准备好记录订阅者和取消订阅者
• 用户在浏览器中通过 websocket 协议进行连接
• 启动一个循环，将数据发送给请求它的特定用户
• 当用户断开连接时，该用户数据的循环停止

• The zeromq server is up and running, ready to log subscribers and unsubscribers
• A user connects in their browser over the websocket protocol
• A loop is started which sends data to the specific user who requested it
• When the user disconnects, the loop for that user's data is stopped

我有点 (1) 和 (2) 工作，但是我遇到的问题是第三点:

I have points (1) and (2) working, however the issue I have is with the third one:

首先:如何只将数据发送给每个特定用户?广播将数据发送给所有人，除非主题"最终可能是个人用户 ID?

Firstly: How can I send data to each specific user only? Broadcast sends it to everyone, unless maybe the 'topics' end up being individual user IDs maybe?

第二:我有一个很大的安全问题.如果我发送哪个用户 ID 想要从客户端订阅，这似乎是我需要的，那么用户可以更改将变量设置为另一个用户的 ID，而是返回他们的数据.

Secondly: I have a big security issue. If I'm sending which user ID wants to subscribe from the client-side, which it seems like I need to, then the user could just change the variable to another user's ID and their data is returned instead.

第三:我必须运行一个单独的 php 脚本，其中包含用于启动实际循环的 zeromq 代码.我不确定这是最好的方法，我宁愿让这个完全在代码库中工作，而不是一个单独的 php 文件.这是我需要整理的一个主要领域.

Thirdly: I'm having to run a separate php script containing the code for zeromq to start the actual looping. I'm not sure this is the best way to do this and I would rather having this working completely within the codebase as opposed to a separate php file. This is a major area I need sorted.

以下代码显示了我目前拥有的内容.

The following code shows what I currently have.

我直接输入 php bin/push-server.php 来运行它.订阅和取消订阅将输出到此终端以进行调试.

I literally type php bin/push-server.php to run this. Subscriptions and un-subscriptions are output to this terminal for debugging purposes.

$loop   = React\EventLoop\Factory::create();
$pusher = Pusher;

$context = new React\ZMQ\Context($loop);
$pull = $context->getSocket(ZMQ::SOCKET_PULL);
$pull->bind('tcp://127.0.0.1:5555');
$pull->on('message', array($pusher, 'onMessage'));

$webSock = new React\Socket\Server($loop);
$webSock->listen(8080, '0.0.0.0'); // Binding to 0.0.0.0 means remotes can connect
$webServer = new Ratchet\Server\IoServer(
    new Ratchet\WebSocket\WsServer(
        new Ratchet\Wamp\WampServer(
            $pusher
        )
    ),
    $webSock
);

$loop->run();

通过 websockets 发送数据的 Pusher

我省略了无用的东西，专注于 onMessage() 和 onSubscribe() 方法.

public function onSubscribe(ConnectionInterface $conn, $topic) 
{
    $subject = $topic->getId();
    $ip = $conn->remoteAddress;

    if (!array_key_exists($subject, $this->subscribedTopics)) 
    {
        $this->subscribedTopics[$subject] = $topic;
    }

    $this->clients[] = $conn->resourceId;

    echo sprintf("New Connection: %s" . PHP_EOL, $conn->remoteAddress);
}

public function onMessage($entry) {
    $entryData = json_decode($entry, true);

    var_dump($entryData);

    if (!array_key_exists($entryData['topic'], $this->subscribedTopics)) {
        return;
    }

    $topic = $this->subscribedTopics[$entryData['topic']];

    // This sends out everything to multiple users, not what I want!!
    // I can't send() to individual connections from here I don't think :S
    $topic->broadcast($entryData);
}

开始循环使用上述Pusher代码的脚本

这是我的问题 - 这是一个单独的 php 文件，希望将来可以集成到其他代码中，但目前我不确定如何正确使用它.我是否从会话中获取用户 ID?我仍然需要从客户端发送它...

The script to start using the above Pusher code in a loop

This is my issue - this is a separate php file that hopefully may be integrated into other code in the future, but currently I'm not sure how to use this properly. Do I grab the user's ID from the session? I still need to send it from client-side...

// Thought sessions might work here but they don't work for subscription
session_start();
$userId = $_SESSION['userId'];

$loop   = React\EventLoop\Factory::create();

$context = new ZMQContext();
$socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher');
$socket->connect("tcp://localhost:5555");

$i = 0;
$loop->addPeriodicTimer(4, function() use ($socket, $loop, $userId, &$i) {

   $entryData = array(
       'topic'     => 'subscriptionTopicHere',
       'userId'    => $userId
    );
    $i++;

    // So it doesn't go on infinitely if run from browser
    if ($i >= 3)
    {
        $loop->stop();
    }

    // Send stuff to the queue
    $socket->send(json_encode($entryData));
});

最后是要订阅的客户端js

$(document).ready(function() { 

    var conn = new ab.Session(
        'ws://localhost:8080' 
      , function() {            
            conn.subscribe('topicHere', function(topic, data) {
                console.log(topic);
                console.log(data);
            });
        }
      , function() {          
            console.warn('WebSocket connection closed');
        }
      , {                       
            'skipSubprotocolCheck': true
        }
    );
});

结论

以上是有效的，但我真的需要弄清楚以下几点:

Conclusion

The above is working, but I really need to figure out the following:

• 如何向个人用户发送个人消息?当他们访问在 JS 中启动 websocket 连接的页面时，我是否也应该启动将内容推入 PHP 队列的脚本(zeromq)?这就是我目前手动执行的操作，只是感觉不对.

从 JS 订阅用户时，从会话中获取用户 ID 并从客户端发送是不安全的.这可能是伪造的.请告诉我有更简单的方法，如果有，怎么做?

When subscribing a user from JS, it can't be safe to grab the users id from the session and send that from client-side. This could be faked. Please tell me there is an easier way, and if so, how?

推荐答案

注意:我在这里的回答不包括对 ZeroMQ 的引用，因为我不再使用它.但是，我相信如果需要，您将能够弄清楚如何将 ZeroMQ 与此答案一起使用.

Note: My answer here does not include references to ZeroMQ, as I am not using it any more. However, I'm sure you will be able to figure out how to use ZeroMQ with this answer if you need to.

使用 JSON

首先，Websocket RFC 和 WAMP Spec 声明要订阅的主题必须是字符串.我在这里有点作弊，但我仍然遵守规范:我正在传递 JSON.

Use JSON

First and foremost, the Websocket RFC and WAMP Spec state that the topic to subscribe to must be a string. I'm cheating a little here, but I'm still adhering to the spec: I'm passing JSON through instead.

{
    "topic": "subject here",
    "userId": "1",
    "token": "dsah9273bui3f92h3r83f82h3"
}

JSON 仍然是一个字符串，但它允许我传递更多数据来代替主题"，而且 PHP 在另一端执行 json_decode() 很简单.当然，您应该验证您确实收到了 JSON，但这取决于您的实现.

JSON is still a string, but it allows me to pass through more data in place of the "topic", and it's simple for PHP to do a json_decode() on the other end. Of course, you should validate that you actually receive JSON, but that's up to your implementation.

• 主题

主题是用户订阅的主题.您可以使用它来决定将哪些数据传回给用户.

The topic is the subject the user is subscribing to. You use this to decide what data you pass back to the user.

• 用户 ID

显然是用户的 ID.您必须使用下一部分验证此用户是否存在并允许订阅:

Obviously the ID of the user. You must verify that this user exists and is allowed to subscribe, using the next part:

• 令牌

这应该是一个一次性随机生成的令牌，在您的 PHP 中生成，并传递给一个 JavaScript 变量.当我说一次使用"时，我的意思是每次您重新加载页面时(以及，通过扩展，在每个 HTTP 请求上)，您的 JavaScript 变量中都应该有一个新令牌.此令牌应根据用户 ID 存储在数据库中.

This should be a one use randomly generated token, generated in your PHP, and passed to a JavaScript variable. When I say "one use", I mean every time you reload the page (and, by extension, on every HTTP request), your JavaScript variable should have a new token in there. This token should be stored in the database against the User's ID.

然后，一旦发出 websocket 请求，您将令牌和用户 ID 与数据库中的那些进行匹配，以确保用户确实是他们所说的那样，并且他们没有弄乱 JS 变量.

Then, once a websocket request is made, you match the token and user id to those in the database to make sure the user is indeed who they say they are, and they haven't been messing around with the JS variables.

注意:在您的事件处理程序中，您可以使用 $conn->remoteAddress 来获取连接的 IP，因此如果有人试图恶意连接，您可以阻止他们(记录他们什么的).

Note: In your event handler, you can use $conn->remoteAddress to get the IP of the connection, so if someone is trying to connect maliciously, you can block them (log them or something).

为什么会这样?

之所以有效，是因为每次有新连接通过时，唯一令牌可确保任何用户都无法访问其他人的订阅数据.

Why does this work?

It works because every time a new connection comes through, the unique token ensures that no user will have access to anyone else's subscription data.

这是我用于运行循环和事件处理程序的内容.我正在创建循环，完成所有装饰器样式对象的创建，并将我的 EventHandler(我将很快介绍)与其中的循环一起传递.

Here's what I am using for running the loop and event handler. I am creating the loop, doing all the decorator style object creation, and passing in my EventHandler (which I'll come to soon) with the loop in there too.

$loop = Factory::create();

new IoServer(
    new WsServer(
        new WampServer(
            new EventHandler($loop) // This is my class. Pass in the loop!
        )
    ),
    $webSock
);

$loop->run();

事件处理程序

class EventHandler implements WampServerInterface, MessageComponentInterface
{
    /**
     * @var \React\EventLoop\LoopInterface
     */
    private $loop;

    /**
     * @var array List of connected clients
     */
    private $clients;

    /**
     * Pass in the react event loop here
     */
    public function __construct(LoopInterface $loop)
    {
        $this->loop = $loop;
    }

    /**
     * A user connects, we store the connection by the unique resource id
     */
    public function onOpen(ConnectionInterface $conn)
    {
        $this->clients[$conn->resourceId]['conn'] = $conn;
    }

    /**
     * A user subscribes. The JSON is in $subscription->getId()
     */
    public function onSubscribe(ConnectionInterface $conn, $subscription)
    {
        // This is the JSON passed in from your JavaScript
        // Obviously you need to validate it's JSON and expected data etc...
        $data = json_decode(subscription->getId());
        
        // Validate the users id and token together against the db values
        
        // Now, let's subscribe this user only
        // 5 = the interval, in seconds
        $timer = $this->loop->addPeriodicTimer(5, function() use ($subscription) {
            $data = "whatever data you want to broadcast";
            return $subscription->broadcast(json_encode($data));
        });

        // Store the timer against that user's connection resource Id
        $this->clients[$conn->resourceId]['timer'] = $timer;
    }

    public function onClose(ConnectionInterface $conn)
    {
        // There might be a connection without a timer
        // So make sure there is one before trying to cancel it!
        if (isset($this->clients[$conn->resourceId]['timer']))
        {
            if ($this->clients[$conn->resourceId]['timer'] instanceof TimerInterface)
            {
                $this->loop->cancelTimer($this->clients[$conn->resourceId]['timer']);
            }
        }
    
        unset($this->clients[$conn->resourceId]);
    }

    /** Implement all the extra methods the interfaces say that you must use **/
}

基本上就是这样.这里的要点是:

That's basically it. The main points here are:

• 唯一令牌、用户 ID 和连接 ID 提供所需的唯一组合，以确保一个用户无法看到另一个用户的数据.
• 唯一令牌意味着如果同一用户打开另一个页面并请求订阅，他们将拥有自己的连接 ID + 令牌组合，因此同一用户在同一页面上的订阅不会翻倍(基本上，每个连接拥有自己的个人数据).

在对数据进行任何操作之前，您应该确保所有数据都经过验证，而不是进行黑客攻击.使用诸如 Monolog 之类的内容记录所有连接尝试，并在发生任何关键事件时设置电子邮件转发(如服务器停止工作，因为有人是个混蛋并试图入侵您的服务器).

You should be ensuring all data is validated and not a hack attempt before you do anything with it. Log all connection attempts using something like Monolog, and set up e-mail forwarding if any critical's occur (like the server stops working because someone is being a bastard and attempting to hack your server).

• 验证一切.我怎么强调都不过分.每次请求都会更改的唯一令牌很重要.
• 请记住，如果您在每个 HTTP 请求上重新生成令牌，并且在尝试通过 websockets 连接之前发出 POST 请求，则您必须在尝试连接之前将重新生成的令牌传回给您的 JavaScript (否则您的令牌将无效).
• 记录一切.记录每个连接、询问主题和断开连接的人.Monolog 对此非常有用.

• Validate Everything. I can't stress this enough. Your unique token that changes on every request is important.
• Remember, if you re-generate the token on every HTTP request, and you make a POST request before attempting to connect via websockets, you'll have to pass back the re-generated token to your JavaScript before trying to connect (otherwise your token will be invalid).
• Log everything. Keep a record of everyone that connects, asks for what topic, and disconnects. Monolog is great for this.

这篇关于Ratchet PHP WAMP - React/ZeroMQ - 特定用户广播的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！