WebMessages system

dragonion_core/proto/web/webmessage.py

@@ -1,157 +0,0 @@
-from typing import Literal, Final, Union
-from dataclasses_json import dataclass_json
-from dataclasses import dataclass, field
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import padding
-from cryptography.hazmat.primitives import hashes
-import base64
-
-from ..encryption.identity import Identity
-
-webmessage_type_literal = Literal[
-    "connect", "message", "disconnect", "error", "notification"
-]
-webmessage_error_message_literal = Literal[
-    "unknown", "username_exists", "invalid_webmessage"
-]
-
-
-@dataclass_json
-@dataclass
-class _WebAnyMessage:
-    username: str | None = None
-    type: webmessage_type_literal = "message"
-    message: str | None = None
-    error_message: webmessage_error_message_literal | None = None
-
-
-@dataclass_json
-@dataclass
-class WebMessageMessage:
-    """
-    Sent as regular message
-    :param username: From user
-    :param message: Encrypted b64-encoded message
-    """
-    username: str
-    message: bytes
-    type: Final = "message"
-
-    def decrypt(self, identity: Identity):
-        return identity.decrypt(self.message)
-
-
-@dataclass_json
-@dataclass
-class WebErrorMessage:
-    """
-    Sent when error on server occurs
-    :param error_message: See webmessage_error_message_literal
-    """
-    error_message: webmessage_error_message_literal
-    type: Final = "error"
-
-
-@dataclass_json
-@dataclass
-class WebConnectionMessage:
-    """
-    Sent when user is connected (sent by user)
-    :param username: Username of connected
-    :param public_key: b64-encoded rsa public key
-    """
-    username: str
-    public_key: bytes
-    type = "connect"
-
-
-@dataclass_json
-@dataclass
-class WebDisconnectMessage:
-    """
-    Sent when user is disconnected
-    :param username: Username of disconnected
-    """
-    username: str
-    type = "disconnect"
-
-
-@dataclass_json
-@dataclass
-class WebNotificationMessage:
-    """
-    Sent from server name as unencrypted notification
-    :param message: Message content, not encrypted
-    """
-    message: str
-    type: Final = "notification"
-
-
-webmessages_union = Union[
-    WebMessageMessage,
-    WebErrorMessage,
-    WebConnectionMessage,
-    WebDisconnectMessage,
-    WebNotificationMessage
-]
-
-
-class WebMessage:
-    """
-    Class for handling incoming webmessages
-    """
-    @staticmethod
-    def from_json(data) -> webmessages_union:
-        """
-        Restores webmessage object from json
-        :param data: Valid json data
-        :return: One of types from webmessages_union
-        """
-        return {
-            "connect": WebConnectionMessage.from_json,
-            "disconnect": WebDisconnectMessage.from_json,
-            "message": WebMessageMessage.from_json,
-            "error": WebErrorMessage.from_json,
-            "notification": WebNotificationMessage.from_json
-        }[_WebAnyMessage.from_json(data).type](data)
-
-
-@dataclass
-class WebBroadcastableMessage:
-    """
-    Class for creating outcoming messages
-    :param from_user: User, that send message
-    :param message_content: Text of message
-    :param keys: Dict with public keys in format username:public_key
-    """
-    from_user: str
-    message_content: str
-    keys: dict[str, bytes]
-
-    encrypted_messages: dict[str, webmessages_union] = field(default_factory=dict)
-
-    def __post_init__(self):
-        for username in self.keys.keys():
-            public_key = serialization.load_der_public_key(
-                base64.urlsafe_b64decode(self.keys[username])
-            )
-            self.encrypted_messages[username] = WebMessageMessage(
-                username=self.from_user,
-                message=base64.urlsafe_b64encode(public_key.encrypt(
-                    self.message_content.encode(),
-                    padding=padding.OAEP(
-                        mgf=padding.MGF1(algorithm=hashes.SHA256()),
-                        algorithm=hashes.SHA256(),
-                        label=None
-                    )
-                ))
-            )
-
-    def json(self):
-        return dict(
-            map(
-                lambda i, j: (i, j),
-                self.encrypted_messages.keys(),
-                [item.to_json() for item in self.encrypted_messages.values()]
-            )
-        )

dragonion_core/proto/web/webmessage/__init__.py

@@ -0,0 +1,38 @@
+from .webmessage import (
+    webmessage_type_literal,
+    webmessages_union,
+    WebMessage
+)
+from .server import (
+    webmessage_error_message_literal,
+    WebErrorMessage,
+    WebNotificationMessage
+)
+from .connection import (
+    WebDisconnectMessage,
+    WebConnectionMessage,
+    WebConnectionResultMessage
+)
+from .message import (
+    WebMessageMessage,
+    WebBroadcastableMessage,
+    WebBroadcastableBuilder
+)
+
+__all__ = [
+    'webmessage_type_literal',
+    'webmessages_union',
+    'WebMessage',
+
+    'WebMessageMessage',
+    'WebBroadcastableMessage',
+    'WebBroadcastableBuilder',
+
+    'webmessage_error_message_literal',
+    'WebErrorMessage',
+    'WebNotificationMessage',
+
+    'WebDisconnectMessage',
+    'WebConnectionMessage',
+    'WebConnectionResultMessage'
+]

dragonion_core/proto/web/webmessage/connection.py

@@ -0,0 +1,39 @@
+from dataclasses import dataclass
+from dataclasses_json import dataclass_json
+
+from typing import Final
+
+
+@dataclass_json
+@dataclass
+class WebConnectionMessage:
+    """
+    Sent when user is connected (sent by user)
+    :param username: Username of connected
+    :param public_key: b64-encoded rsa public key
+    """
+    username: str
+    public_key: bytes
+    type: Final = "connect"
+
+
+@dataclass_json
+@dataclass
+class WebConnectionResultMessage:
+    """
+    Sent by server as answer to user connected
+    :param connected_users: Dict with public keys in format username:public_key
+    """
+    connected_users: dict[str, bytes] = None
+    type: Final = "connect_answer"
+
+
+@dataclass_json
+@dataclass
+class WebDisconnectMessage:
+    """
+    Sent when user is disconnected
+    :param username: Username of disconnected
+    """
+    username: str
+    type: Final = "disconnect"

dragonion_core/proto/web/webmessage/message.py

@@ -0,0 +1,78 @@
+from dataclasses import dataclass, field
+from dataclasses_json import dataclass_json
+from typing import Final
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives import hashes
+import base64
+
+from ...encryption.identity import Identity
+
+
+@dataclass_json
+@dataclass
+class WebMessageMessage:
+    """
+    Sent as regular message
+    :param username: From user
+    :param message: Encrypted b64-encoded message
+    """
+    username: str
+    message: bytes
+    type: Final = "message"
+
+    def decrypt(self, identity: Identity):
+        return identity.decrypt(self.message)
+
+
+@dataclass_json
+@dataclass
+class WebBroadcastableMessage:
+    """
+    Out-coming message, generated by WebBroadcastableBuilder
+    :param messages: Dict
+    """
+
+    messages: dict[str, WebMessageMessage] = field(default_factory=dict)
+    type: Final = "broadcastable"
+
+
+@dataclass
+class WebBroadcastableBuilder:
+    """
+    Class for creating outcoming messages
+    :param from_user: User, that send message
+    :param message_content: Text of message
+    :param keys: Dict with public keys in format username:public_key
+    """
+    from_user: str
+    message_content: str
+    keys: dict[str, bytes]
+
+    broadcastable: WebBroadcastableMessage = WebBroadcastableMessage()
+
+    def __post_init__(self):
+        for username in self.keys.keys():
+            public_key = serialization.load_der_public_key(
+                base64.urlsafe_b64decode(self.keys[username])
+            )
+
+            self.broadcastable.messages[username] = WebMessageMessage(
+                username=self.from_user,
+                message=base64.urlsafe_b64encode(public_key.encrypt(
+                    self.message_content.encode(),
+                    padding=padding.OAEP(
+                        mgf=padding.MGF1(algorithm=hashes.SHA256()),
+                        algorithm=hashes.SHA256(),
+                        label=None
+                    )
+                ))
+            )
+
+    def to_json(self):
+        """
+        Convert to json
+        :return: Only encrypted data, that can be sent to server
+        """
+        return self.broadcastable.to_json()

dragonion_core/proto/web/webmessage/server.py

@@ -0,0 +1,30 @@
+from dataclasses import dataclass
+from dataclasses_json import dataclass_json
+from typing import Final, Literal
+
+
+webmessage_error_message_literal = Literal[
+    "unknown", "username_exists", "invalid_webmessage"
+]
+
+
+@dataclass_json
+@dataclass
+class WebErrorMessage:
+    """
+    Sent when error on server occurs
+    :param error_message: See webmessage_error_message_literal
+    """
+    error_message: webmessage_error_message_literal
+    type: Final = "error"
+
+
+@dataclass_json
+@dataclass
+class WebNotificationMessage:
+    """
+    Sent from server name as unencrypted notification
+    :param message: Message content, not encrypted
+    """
+    message: str
+    type: Final = "notification"

dragonion_core/proto/web/webmessage/webmessage.py

@@ -0,0 +1,85 @@
+from typing import Literal, Union
+from dataclasses_json import dataclass_json
+from dataclasses import dataclass
+
+from .server import (
+    webmessage_error_message_literal,
+    WebErrorMessage,
+    WebNotificationMessage
+)
+from .connection import (
+    WebDisconnectMessage,
+    WebConnectionMessage,
+    WebConnectionResultMessage
+)
+from .message import (
+    WebMessageMessage,
+    WebBroadcastableMessage
+)
+
+__all__ = [
+    'webmessage_type_literal',
+    'webmessages_union',
+    'WebMessage',
+
+    'WebMessageMessage',
+    'WebBroadcastableMessage',
+
+    'webmessage_error_message_literal',
+    'WebErrorMessage',
+    'WebNotificationMessage',
+
+    'WebDisconnectMessage',
+    'WebConnectionMessage',
+    'WebConnectionResultMessage'
+]
+
+webmessage_type_literal = Literal[
+    "connect", "connect_answer", "disconnect",
+    "broadcastable", "message",
+    "error", "notification",
+]
+
+
+@dataclass_json
+@dataclass
+class _WebAnyMessage:
+    username: str | None = None
+    type: webmessage_type_literal = "message"
+    message: str | None = None
+    messages: dict[str, WebMessageMessage] = None
+    error_message: webmessage_error_message_literal | None = None
+
+
+webmessages_union = Union[
+    WebMessageMessage,
+    WebBroadcastableMessage,
+    WebErrorMessage,
+    WebConnectionMessage,
+    WebDisconnectMessage,
+    WebNotificationMessage,
+    WebConnectionResultMessage
+]
+
+
+class WebMessage:
+    """
+    Class for handling incoming webmessages
+    """
+
+    @staticmethod
+    def from_json(data) -> webmessages_union:
+        """
+        Restores webmessage object from json
+        :param data: Valid json data
+        :return: One of types from webmessages_union
+        """
+        return {
+            "connect": WebConnectionMessage.from_json,
+            "disconnect": WebDisconnectMessage.from_json,
+            "message": WebMessageMessage.from_json,
+            "error": WebErrorMessage.from_json,
+            "notification": WebNotificationMessage.from_json,
+            "connect_answer": WebConnectionResultMessage.from_json,
+            "broadcastable": WebBroadcastableMessage.from_json
+        }[_WebAnyMessage.from_json(data).type](data)

pyproject.toml

@@ -7,8 +7,8 @@ readme = "README.md"
 packages = [{include = "dragonion_core"}]
 
 [tool.poetry.dependencies]
-python = "^3.10"
+python = ">=3.7,<3.12"
-dataclasses-json = "^0.5.9"
+dataclasses-json = "^0.5.13"
 sqlitedict = "^2.1.0"
 cryptography = "^41.0.2"
 attrs = "^23.1.0"