[FEATURE-REQUEST]: Websocket via channels[daphne]

Author: amadolid

jaseci_serv/jaseci_serv/asgi.py

@@ -8,9 +8,16 @@
 """
 
 import os
-
+from .socket.routing import websocket_urlpatterns
 from django.core.asgi import get_asgi_application
+from channels.routing import ProtocolTypeRouter, URLRouter
+from channels.auth import AuthMiddlewareStack
 
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "jaseci_serv.settings")
 
-application = get_asgi_application()
+application = ProtocolTypeRouter(
+    {
+        "http": get_asgi_application(),
+        "websocket": AuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
+    }
+)

jaseci_serv/jaseci_serv/settings.py

@@ -37,6 +37,7 @@
 # Application definition
 
 INSTALLED_APPS = [
+    "daphne",
     "django.contrib.admin",
     "django.contrib.auth",
     "django.contrib.contenttypes",
@@ -96,7 +97,7 @@
     },
 ]
 
-WSGI_APPLICATION = "jaseci_serv.wsgi.application"
+ASGI_APPLICATION = "jaseci_serv.asgi.application"
 
 # Database
 # https://docs.djangoproject.com/en/3.0/ref/settings/#databases

jaseci_serv/jaseci_serv/socket/README.md

@@ -0,0 +1,43 @@
+# **HOW TO SETUP `CHANNEL_LAYER`**
+
+## **GLOBAL VARS:** `CHANNEL_LAYER`
+### **`IN MEMORY`** *(default)*
+```json
+{
+    "BACKEND": "channels.layers.InMemoryChannelLayer"
+}
+```
+- This config will only works with single jaseci instance. Notification from async walkers will also not work
+
+### **`REDIS`**
+```json
+{
+    "BACKEND": "channels_redis.core.RedisChannelLayer",
+    "CONFIG": {
+        "hosts": [
+            ["localhost", 6379]
+        ]
+    }
+}
+```
+- This should work on mutiple jaseci instance. Notification from async walker should also work
+---
+# **`NOTIFICATION FROM JAC`**
+## wb.**`notify`**
+> **`Arguments`:** \
+> **target**: str \
+> **data**: dict
+>
+> **`Return`:** \
+> None
+>
+> **`Usage`:** \
+> Send notification from jac to target group
+>
+> **`Remarks`:** \
+> if user is logged in, `target` can be master id without `urn:uuid:` \
+> else used thed `session_id` from client connection
+##### **`HOW TO TRIGGER`**
+```js
+wb.notify(target, {"test": 123456});
+```

jaseci_serv/jaseci_serv/socket/__init__.py

@@ -0,0 +1,62 @@
+import os
+from uuid import uuid4
+from json import loads, dumps
+
+from asgiref.sync import async_to_sync
+from channels.layers import settings
+from channels.generic.websocket import WebsocketConsumer
+
+from .event_action import authenticated_user
+from jaseci_serv.base.models import lookup_global_config
+
+
+class SocketConsumer(WebsocketConsumer):
+    def connect(self):
+        self.accept()
+        session_id = None
+        authenticated = False
+        target = self.scope["url_route"]["kwargs"]["target"]
+
+        if target == "anonymous":
+            self.target = session_id = str(uuid4())
+        else:
+            user = authenticated_user(target)
+            if user:
+                self.target = user.master.urn[9:]
+                authenticated = True
+            else:
+                self.target = session_id = str(uuid4())
+
+        async_to_sync(self.channel_layer.group_add)(self.target, self.channel_name)
+        self.send(
+            text_data=dumps(
+                {
+                    "type": "connect",
+                    "authenticated": authenticated,
+                    "session_id": session_id,
+                }
+            )
+        )
+
+    def receive(self, text_data=None, bytes_data=None):
+        data = loads(text_data)
+
+        async_to_sync(self.channel_layer.group_send)(
+            self.target, {"type": "notify", "data": data}
+        )
+
+    def notify(self, data):
+        self.send(text_data=dumps(data))
+
+
+setattr(
+    settings,
+    "CHANNEL_LAYERS",
+    {
+        "default": loads(
+            lookup_global_config(
+                "CHANNEL_LAYER", '{"BACKEND": "channels.layers.InMemoryChannelLayer"}'
+            )
+        )
+    },
+)

jaseci_serv/jaseci_serv/socket/consumer.py

@@ -0,0 +1,39 @@
+try:
+    from hmac import compare_digest
+except ImportError:
+
+    def compare_digest(a, b):
+        return a == b
+
+
+import binascii
+from knox.crypto import hash_token
+from knox.models import AuthToken
+from knox.settings import CONSTANTS
+from jaseci.jsorc.live_actions import jaseci_action
+from channels.layers import get_channel_layer
+from asgiref.sync import async_to_sync
+from django.utils import timezone
+
+
+def authenticated_user(token: str):
+    for auth_token in AuthToken.objects.filter(
+        token_key=token[: CONSTANTS.TOKEN_KEY_LENGTH]
+    ):
+        try:
+            digest = hash_token(token)
+            if (
+                compare_digest(digest, auth_token.digest)
+                and auth_token.expiry > timezone.now()
+            ):
+                return auth_token.user
+        except (TypeError, binascii.Error):
+            pass
+    return None
+
+
+@jaseci_action(act_group=["wb"])
+def notify(target: str, data: dict):
+    async_to_sync(get_channel_layer().group_send)(
+        target, {"type": "notify", "data": data}
+    )

jaseci_serv/jaseci_serv/socket/event_action.py

@@ -0,0 +1,6 @@
+from django.urls import path
+from . import consumer
+
+websocket_urlpatterns = [
+    path(r"ws/socket-server/<str:target>", consumer.SocketConsumer.as_asgi())
+]

jaseci_serv/jaseci_serv/socket/routing.py

@@ -0,0 +1,45 @@
+import pytest
+from uuid import UUID
+from json import loads
+from jaseci.utils.utils import TestCaseHelper
+from django.test import TestCase
+
+from channels.routing import URLRouter
+from channels.testing import WebsocketCommunicator
+
+
+class WebSocketTests(TestCaseHelper, TestCase):
+    """Test the publicly available node API"""
+
+    def setUp(self):
+        super().setUp()
+
+        from ..routing import websocket_urlpatterns
+
+        self.app = URLRouter(websocket_urlpatterns)
+
+    def tearDown(self):
+        super().tearDown()
+
+    def is_valid_uuid(self, uuid):
+        try:
+            _uuid = UUID(uuid, version=4)
+        except ValueError:
+            return False
+        return str(_uuid) == uuid
+
+    @pytest.mark.asyncio
+    async def test_websocket(self):
+        communicator = WebsocketCommunicator(self.app, "ws/socket-server/anonymous")
+
+        connected, subprotocol = await communicator.connect()
+        self.assertTrue(connected)
+        response: dict = loads(await communicator.receive_from())
+        self.assertTrue(self.is_valid_uuid(response.pop("session_id")))
+        self.assertEqual(response, {"type": "connect", "authenticated": False})
+
+        await communicator.send_to(text_data='{"test": true}')
+        response: dict = loads(await communicator.receive_from())
+        self.assertEqual(response, {"type": "notify", "data": {"test": True}})
+
+        await communicator.disconnect()

jaseci_serv/jaseci_serv/socket/tests/__init__.py

@@ -48,6 +48,8 @@ def get_ver():
         "dj-rest-auth[with_social]",
         "django-allauth>=0.52.0",
         "tzdata>=2022.7",
+        "channels[daphne]==4.0.0",
+        "channels-redis",
     ],
     package_data={
         "": [

jaseci_serv/jaseci_serv/socket/tests/test_websocket.py

@@ -145,4 +145,18 @@ <h1>Google Identity Services Authorization Token model</h1>
     });
     {% endif %}
 {% endif %}
+
+{% if provider == "google" %}
+socket = new WebSocket(`ws://${window.location.host}/ws/socket-server/276a40aec1dffc48a25463c3e2545473b45a663364adf3a2f523b903aa254c9f`)
+{% else %}
+socket = new WebSocket(`ws://${window.location.host}/ws/socket-server/anonymous`)
+{% endif %}
+socket.onmessage = (event) => {
+    console.log(event)
+}
+
+function notify-server() {
+    socket.send(JSON.stringify({"message": "test"}))
+}
+
 </script>