socket - Ratchet PHP WAMP-React/ZeroMQ-Difusión de usuario específica

sockets web php

Question

Nota : Esta no es la misma pregunta que utiliza MessageComponentInterface . Estoy utilizando WampServerInterface en WampServerInterface lugar, por lo que esta pregunta pertenece específicamente a esa parte. Necesito una respuesta con ejemplos de código y una explicación, ya que puedo ver que esto es útil para otros en el futuro.

Intentando pulsaciones en bucle para usuarios individuales.

Estoy usando la parte WAMP de Ratchet y ZeroMQ, y actualmente tengo una versión funcional del tutorial de integración push .

Estoy intentando realizar lo siguiente:

• El servidor zeromq está en funcionamiento, listo para registrar a los suscriptores y anular la suscripción
• Un usuario se conecta en su navegador a través del protocolo websocket.
• Se inicia un bucle que envía datos al usuario específico que lo solicitó.
• Cuando el usuario se desconecta, el bucle para los datos de ese usuario se detiene

Tengo puntos (1) y (2) trabajando, sin embargo, el problema que tengo es con el tercero:

En primer lugar: ¿Cómo puedo enviar datos a cada usuario específico solamente? La difusión se envía a todos, a menos que los "temas" terminen siendo ID de usuarios individuales, tal vez.

En segundo lugar: tengo un gran problema de seguridad. Si estoy enviando qué ID de usuario desea suscribirse desde el lado del cliente, lo que parece que necesito, entonces el usuario podría simplemente cambiar la variable a la ID de otro usuario y se devolverán sus datos.

En tercer lugar: tengo que ejecutar un script php separado que contenga el código para que zeromq inicie el bucle real. No estoy seguro de que esta sea la mejor manera de hacer esto y preferiría que esto funcione completamente dentro de la base de código en lugar de un archivo php separado. Esta es un área importante que necesito ordenada.

El siguiente código muestra lo que tengo actualmente.

El servidor que solo se ejecuta desde la consola.

Literalmente php bin/push-server.php para ejecutar esto. Las suscripciones y las no suscripciones se envían a este terminal para fines de depuración.

$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();

El empujador que envía datos a través de websockets

He omitido las cosas inútiles y me he concentrado en los onMessage() y 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);
}

La secuencia de comandos para comenzar a utilizar el código de empuje anterior en un bucle

Este es mi problema: este es un archivo php separado que esperamos se pueda integrar en otro código en el futuro, pero actualmente no estoy seguro de cómo usarlo correctamente. ¿Agarro la identificación del usuario de la sesión? Todavía tengo que enviarlo desde el lado del cliente ...

// 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));
});

Finalmente, el lado del cliente js para suscribirse con

$(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
        }
    );
});

Conclusión

Lo anterior está funcionando, pero realmente necesito averiguar lo siguiente:

• ¿Cómo puedo enviar mensajes individuales a usuarios individuales? Cuando visitan la página que inicia la conexión de websocket en JS, ¿también debería estar iniciando el script que mete cosas en la cola en PHP (el zeromq)? Eso es lo que actualmente estoy haciendo manualmente, y simplemente se siente mal .

• Al suscribir a un usuario de JS, no puede ser seguro tomar el ID de los usuarios de la sesión y enviarlo desde el lado del cliente. Esto podría ser falso. Por favor, dime que hay una manera más fácil, y si es así, ¿cómo?

Answer 1

Nota: Mi respuesta aquí no incluye referencias a ZeroMQ, ya que no lo estoy usando más. Sin embargo, estoy seguro de que podrá averiguar cómo usar ZeroMQ con esta respuesta si lo necesita.

Usar json

En primer lugar, las especificaciones de RFC y WAMP de Websocket establecen que el tema al que se debe suscribir debe ser una cadena . Estoy haciendo trampa un poco aquí, pero aún estoy cumpliendo con la especificación: en su lugar, estoy pasando a JSON.

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

JSON sigue siendo una cadena, pero me permite pasar más datos en lugar del "tema", y es simple para PHP hacer un json_decode() en el otro extremo. Por supuesto, debe validar que en realidad recibe JSON, pero eso depende de su implementación.

Entonces, ¿qué estoy pasando por aquí, y por qué?

• Tema

El tema es el tema al que se suscribe el usuario. Usted usa esto para decidir qué datos le devuelve al usuario.

• ID de usuario

Obviamente el ID del usuario. Debe verificar que este usuario existe y se le permite suscribirse, usando la siguiente parte:

• Simbólico

Este debe ser un token generado de forma aleatoria, generado en su PHP y pasado a una variable de JavaScript. Cuando digo "un uso", me refiero a que cada vez que vuelva a cargar la página (y, por extensión, en cada solicitud HTTP), su variable de JavaScript debería tener un nuevo token allí. Este token se debe almacenar en la base de datos con el ID del usuario.

Luego, una vez que se realiza una solicitud de websocket, hace coincidir el token y la identificación del usuario con los de la base de datos para asegurarse de que el usuario sea quien dice ser, y no ha estado jugando con las variables JS.

Nota: En su controlador de eventos, puede usar $conn->remoteAddress para obtener la IP de la conexión, por lo que si alguien está tratando de conectarse maliciosamente, puede bloquearlos (registrarlos o algo).

¿Por qué funciona esto?

Funciona porque cada vez que se realiza una nueva conexión, el token único garantiza que ningún usuario tendrá acceso a los datos de suscripción de nadie más.

El servidor

Esto es lo que estoy usando para ejecutar el bucle y el controlador de eventos. Estoy creando el bucle, haciendo toda la creación de objetos de estilo decorador y pasando mi EventHandler (al que llegaré pronto) con el bucle allí también.

$loop = Factory::create();

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

$loop->run();

El controlador de eventos

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 **/
}

Eso es básicamente eso. Los puntos principales aquí son:

• El token, el ID de usuario y la identificación de conexión únicos proporcionan la combinación única necesaria para garantizar que un usuario no pueda ver los datos de otro usuario.
• El token único significa que si el mismo usuario abre otra página y solicita suscribirse, tendrá su propio id de conexión + token combo para que el mismo usuario no tenga el doble de suscripciones en la misma página (básicamente, cada conexión tiene su propia cuenta). datos individuales).

Extensión

Debe asegurarse de que todos los datos estén validados y no un intento de pirateo antes de hacer algo con ellos. Registre todos los intentos de conexión usando algo como Monolog , y configure el reenvío de correo electrónico si ocurre alguna crítica (como si el servidor dejara de funcionar porque alguien está siendo un bastardo y está intentando hackear su servidor).

Puntos de cierre

• Validar todo . No puedo enfatizar esto lo suficiente. Su token único que cambia en cada solicitud es importante .
• Recuerde, si vuelve a generar el token en cada solicitud HTTP y realiza una solicitud POST antes de intentar conectarse a través de websockets, tendrá que devolver el token generado a su JavaScript antes de intentar conectarse (de lo contrario, su token). será inválido).
• Registrar todo Mantenga un registro de todas las personas que se conectan, preguntan qué tema y cuáles se desconectan. Monolog es genial para esto.

Answer 2

Para enviar a usuarios específicos, necesita un patrón ROUTER-DEALER en lugar de PUB-SUB. Esto se explica en la Guía, en el capítulo 3. La seguridad, si está utilizando ZMQ v4.0, se maneja a nivel de cable, por lo que no lo ve en la aplicación. Todavía requiere algo de trabajo, a menos que use el enlace CZMQ, que proporciona un marco de autenticación (zauth).

Básicamente, para autenticarse, instale un controlador en inproc: //zeromq.zap.01 y responda a las solicitudes a través de ese socket. Google ZeroMQ ZAP para el RFC; También hay un caso de prueba en el programa principal libzmq / tests / test_security_curve.cpp.