Decentralized Instant Messaging Protocol (Python)
Handshake Command Protocol
0. (C-S) handshake start
(S-C) handshake again with new session
(C-S) handshake restart with new session
(S-C) handshake success
from abc import ABC , abstractmethod
from enum import IntEnum
from typing import Optional , Dict
from dimp import Command , BaseCommand
class HandshakeState (IntEnum ):
Start = 0 # C -> S, without session key(or session expired)
Again = 1 # S -> C, with new session key
Restart = 2 # C -> S, with new session key
Success = 3 # S -> C, handshake accepted
def handshake_state (title : str , session : str = None ) -> HandshakeState :
# Server -> Client
if title == 'DIM!' : # or title == 'OK!':
return HandshakeState .Success
if title == 'DIM?' :
return HandshakeState .Again
# Client -> Server: "Hello world!"
if session is None or len (session ) == 0 :
return HandshakeState .Start
else :
return HandshakeState .Restart
class HandshakeCommand (Command , ABC ):
"""
Handshake Command
~~~~~~~~~~~~~~~~~
data format: {
type : 0x88,
sn : 123,
command : "handshake", // command name
title : "Hello world!", // "DIM?", "DIM!"
session : "{SESSION_ID}", // session key
}
"""
HANDSHAKE = 'handshake'
@property
@abstractmethod
def title (self ) -> str :
raise NotImplemented
@property
@abstractmethod
def session (self ) -> Optional [str ]:
raise NotImplemented
@property
@abstractmethod
def state (self ) -> HandshakeState :
raise NotImplemented
#
# Factories
#
@classmethod
def offer (cls , session : str = None ) -> Command :
"""
Create client-station handshake offer
:param session: Old session key
:return: HandshakeCommand object
"""
return BaseHandshakeCommand (title = 'Hello world!' , session = session )
@classmethod
def ask (cls , session : str ) -> Command :
"""
Create station-client handshake again with new session
:param session: New session key
:return: HandshakeCommand object
"""
return BaseHandshakeCommand (title = 'DIM?' , session = session )
@classmethod
def accepted (cls , session : str = None ) -> Command :
"""
Create station-client handshake success notice
:return: HandshakeCommand object
"""
return BaseHandshakeCommand (title = 'DIM!' , session = session )
start = offer # (1. C->S) first handshake, without session
again = ask # (2. S->C) ask client to handshake with new session key
restart = offer # (3. C->S) handshake with new session key
success = accepted # (4. S->C) notice the client that handshake accepted
class BaseHandshakeCommand (BaseCommand , HandshakeCommand ):
def __init__ (self , content : Dict = None , title : str = None , session : str = None ):
if content is None :
# 1. new command with title & session key
assert title is not None , 'handshake command error: %s' % session
cmd = self .HANDSHAKE
super ().__init__ (cmd = cmd )
self ['title' ] = title
self ['message' ] = title # TODO: remove after all clients upgraded
if session is not None :
self ['session' ] = session
else :
# 2. command info from network
assert title is None and session is None , 'params error: %s, %s, %s' % (content , title , session )
super ().__init__ (content )
@property
def title (self ) -> str :
return self .get_str (key = 'title' , default = '' )
@property
def session (self ) -> Optional [str ]:
return self .get_str (key = 'session' )
@property
def state (self ) -> HandshakeState :
return handshake_state (title = self .title , session = self .session )
from typing import Any , Dict
from dimp import *
class ApplicationContent (BaseContent , AppContent ):
"""
Application Customized message
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
data format: {
type : i2s(0xA0),
sn : 123,
app : "{APP_ID}", // application (e.g.: "chat.dim.sechat")
extra : info // action parameters
}
"""
def __init__ (self , content : Dict [str , Any ] = None ,
msg_type : str = None , app : str = None ):
if content is None :
# 1. new content with type, app_id
assert app is not None , 'customized content error: %s, %s' % (msg_type , app )
if msg_type is None :
msg_type = ContentType .APPLICATION
super ().__init__ (None , msg_type )
self ['app' ] = app
else :
# 2. content info from network
assert msg_type is None and app is None , 'params error: %s, %s, %s' % (content , msg_type , app )
super ().__init__ (content )
@property # Override
def application (self ) -> str :
return self .get_str (key = 'app' , default = '' )
Copyright © 2018-2025 Albert Moky