2024-04-29 01:09:06 +12:00
#!/usr/bin/env python
2024-04-29 15:16:44 +12:00
import argparse
2024-04-29 01:09:06 +12:00
import json
2024-04-29 22:06:06 +12:00
import os
2024-05-24 01:35:30 +12:00
import sys
2024-06-02 11:15:09 +12:00
import threading
2024-05-19 21:41:34 +12:00
import time
2024-05-04 00:30:09 +12:00
from datetime import datetime , timezone
2024-05-04 02:42:18 +12:00
from typing import Callable , List
2024-04-29 01:09:06 +12:00
import RNS
2024-09-19 17:20:49 +12:00
import RNS . vendor . umsgpack as msgpack
2024-04-29 01:09:06 +12:00
import LXMF
2024-09-18 18:19:02 +12:00
from LXMF import LXMRouter
2024-05-04 02:42:18 +12:00
from aiohttp import web , WSMessage , WSMsgType , WSCloseCode
2024-04-29 01:09:06 +12:00
import asyncio
import base64
2024-05-19 00:48:45 +12:00
import webbrowser
2024-04-29 01:09:06 +12:00
2024-05-04 20:22:10 +12:00
from peewee import SqliteDatabase
2024-05-29 11:55:04 +12:00
from serial . tools import list_ports
2024-05-04 20:22:10 +12:00
2024-05-03 23:48:15 +12:00
import database
2024-09-17 15:46:45 +12:00
from src . backend . announce_handler import AnnounceHandler
2024-08-16 23:19:22 +12:00
from src . backend . lxmf_message_fields import LxmfImageField , LxmfFileAttachmentsField , LxmfFileAttachment , LxmfAudioField
from src . backend . audio_call_manager import AudioCall , AudioCallManager
2024-05-03 23:48:15 +12:00
2024-04-29 14:51:47 +12:00
2024-05-24 01:35:30 +12:00
# NOTE: this is required to be able to pack our app with cxfreeze as an exe, otherwise it can't access bundled assets
2024-07-09 17:38:45 +12:00
# this returns a file path based on if we are running meshchat.py directly, or if we have packed it as an exe with cxfreeze
2024-05-24 01:35:30 +12:00
# https://cx-freeze.readthedocs.io/en/latest/faq.html#using-data-files
def get_file_path ( filename ) :
if getattr ( sys , " frozen " , False ) :
datadir = os . path . dirname ( sys . executable )
else :
datadir = os . path . dirname ( __file__ )
return os . path . join ( datadir , filename )
2024-06-02 14:42:19 +12:00
class ReticulumMeshChat :
2024-04-29 01:09:06 +12:00
2024-05-04 20:57:20 +12:00
def __init__ ( self , identity : RNS . Identity , storage_dir , reticulum_config_dir ) :
2024-04-29 15:31:39 +12:00
2024-05-04 20:57:20 +12:00
# when providing a custom storage_dir, files will be saved as
# <storage_dir>/identities/<identity_hex>/
# <storage_dir>/identities/<identity_hex>/database.db
# if storage_dir is not provided, we will use ./storage instead
2024-05-04 20:46:42 +12:00
# ./storage/identities/<identity_hex>/
# ./storage/identities/<identity_hex>/database.db
2024-05-04 20:57:20 +12:00
# ensure a storage path exists for the loaded identity
2024-06-02 13:56:04 +12:00
self . storage_dir = storage_dir or os . path . join ( " storage " )
self . storage_path = os . path . join ( self . storage_dir , " identities " , identity . hash . hex ( ) )
print ( " Using Storage Path: {} " . format ( self . storage_path ) )
os . makedirs ( self . storage_path , exist_ok = True )
2024-05-04 20:46:42 +12:00
# define path to files based on storage path
2024-06-02 13:56:04 +12:00
self . database_path = os . path . join ( self . storage_path , " database.db " )
lxmf_router_path = os . path . join ( self . storage_path , " lxmf_router " )
2024-05-04 20:46:42 +12:00
2024-07-08 16:22:24 +12:00
# check if database already exists, before initialization
database_already_exists = os . path . exists ( self . database_path )
2024-05-03 23:48:15 +12:00
# init database
2024-06-02 13:56:04 +12:00
sqlite_database = SqliteDatabase ( self . database_path )
2024-05-05 02:50:36 +12:00
database . database . initialize ( sqlite_database )
2024-05-03 23:48:15 +12:00
self . db = database . database
self . db . connect ( )
self . db . create_tables ( [
2024-05-04 21:31:12 +12:00
database . Config ,
2024-05-05 21:11:22 +12:00
database . Announce ,
2024-09-19 23:46:08 +12:00
database . CustomDestinationDisplayName ,
2024-05-03 23:48:15 +12:00
database . LxmfMessage ,
2024-06-01 19:21:42 +12:00
database . LxmfConversationReadState ,
2024-12-04 01:19:47 +13:00
database . LxmfUserIcon ,
2024-05-03 23:48:15 +12:00
] )
2024-07-08 16:22:24 +12:00
# init config
self . config = Config ( )
# if database already existed before init, and we don't have a previous version set, we are on version 1
if database_already_exists and self . config . database_version . get ( ) is None :
self . config . database_version . set ( 1 )
# if database didn't already exist, it was just fully migrated when it was created, so set the current version
if not database_already_exists :
self . config . database_version . set ( database . latest_version )
# migrate database
current_database_version = self . config . database_version . get ( )
migrated_database_version = database . migrate ( current_version = current_database_version )
self . config . database_version . set ( migrated_database_version )
2024-05-05 02:50:36 +12:00
# vacuum database on start to shrink its file size
sqlite_database . execute_sql ( " VACUUM " )
2024-05-19 20:36:20 +12:00
# lxmf messages in outbound or sending state should be marked as failed when app starts as they are no longer being processed
( database . LxmfMessage . update ( state = " failed " )
. where ( database . LxmfMessage . state == " outbound " )
2024-09-19 13:06:35 +12:00
. orwhere ( ( database . LxmfMessage . state == " sent " ) & ( database . LxmfMessage . method == " opportunistic " ) )
2024-05-19 20:36:20 +12:00
. orwhere ( database . LxmfMessage . state == " sending " ) . execute ( ) )
2024-04-29 20:26:45 +12:00
# init reticulum
2024-04-30 21:25:27 +12:00
self . reticulum = RNS . Reticulum ( reticulum_config_dir )
2024-04-29 20:26:45 +12:00
self . identity = identity
2024-04-29 01:09:06 +12:00
2024-04-29 20:26:45 +12:00
# init lxmf router
2024-05-04 20:46:42 +12:00
self . message_router = LXMF . LXMRouter ( identity = self . identity , storagepath = lxmf_router_path )
2024-07-08 21:52:18 +12:00
self . message_router . PROCESSING_INTERVAL = 1
2024-04-29 01:09:06 +12:00
2024-08-03 00:17:33 +12:00
# increase limit for incoming lxmf messages (received over a resource), to allow receiving larger attachments
# the lxmf router expects delivery_per_transfer_limit to be provided in kilobytes, so we will do that...
self . message_router . delivery_per_transfer_limit = self . config . lxmf_delivery_transfer_limit_in_bytes . get ( ) / 1000
2024-04-29 20:26:45 +12:00
# register lxmf identity
2024-09-11 12:58:19 +12:00
self . local_lxmf_destination = self . message_router . register_delivery_identity (
identity = self . identity ,
display_name = self . config . display_name . get ( ) ,
)
2024-04-29 16:01:42 +12:00
2024-04-29 20:26:45 +12:00
# set a callback for when an lxmf message is received
self . message_router . register_delivery_callback ( self . on_lxmf_delivery )
2024-04-29 17:54:39 +12:00
2024-09-18 17:42:25 +12:00
# update active propagation node
self . set_active_propagation_node ( self . config . lxmf_preferred_propagation_node_destination_hash . get ( ) )
2024-09-19 15:59:04 +12:00
# enable propagation node (we don't call with false if disabled, as no need to announce disabled state every launch)
if self . config . lxmf_local_propagation_node_enabled . get ( ) :
self . enable_local_propagation_node ( )
2024-09-17 16:32:00 +12:00
# handle received announces based on aspect
2024-09-17 16:29:53 +12:00
RNS . Transport . register_announce_handler ( AnnounceHandler ( " call.audio " , self . on_audio_call_announce_received ) )
2024-09-17 15:40:41 +12:00
RNS . Transport . register_announce_handler ( AnnounceHandler ( " lxmf.delivery " , self . on_lxmf_announce_received ) )
RNS . Transport . register_announce_handler ( AnnounceHandler ( " lxmf.propagation " , self . on_lxmf_propagation_announce_received ) )
RNS . Transport . register_announce_handler ( AnnounceHandler ( " nomadnetwork.node " , self . on_nomadnet_node_announce_received ) )
2024-04-29 17:54:39 +12:00
2024-04-29 20:26:45 +12:00
# remember websocket clients
2024-05-04 02:42:18 +12:00
self . websocket_clients : List [ web . WebSocketResponse ] = [ ]
2024-04-29 15:31:39 +12:00
2024-05-20 23:18:17 +12:00
# register audio call identity
self . audio_call_manager = AudioCallManager ( identity = self . identity )
self . audio_call_manager . register_incoming_call_callback ( self . on_incoming_audio_call )
2024-06-02 11:15:09 +12:00
# start background thread for auto announce loop
2024-06-02 12:07:59 +12:00
thread = threading . Thread ( target = asyncio . run , args = ( self . announce_loop ( ) , ) )
2024-06-02 11:15:09 +12:00
thread . daemon = True
thread . start ( )
2024-09-18 21:06:03 +12:00
# start background thread for auto syncing propagation nodes
thread = threading . Thread ( target = asyncio . run , args = ( self . announce_sync_propagation_nodes ( ) , ) )
thread . daemon = True
thread . start ( )
2024-06-02 13:56:04 +12:00
# gets app version from package.json
def get_app_version ( self ) - > str :
with open ( get_file_path ( " package.json " ) ) as f :
package_json = json . load ( f )
return package_json [ " version " ]
2024-06-02 11:15:09 +12:00
# automatically announces based on user config
2024-06-02 12:07:59 +12:00
async def announce_loop ( self ) :
2024-06-02 11:15:09 +12:00
while True :
should_announce = False
# check if auto announce is enabled
if self . config . auto_announce_enabled . get ( ) :
# check if we have announced recently
last_announced_at = self . config . last_announced_at . get ( )
if last_announced_at is not None :
# determine when next announce should be sent
auto_announce_interval_seconds = self . config . auto_announce_interval_seconds . get ( )
next_announce_at = last_announced_at + auto_announce_interval_seconds
# we should announce if current time has passed next announce at timestamp
if time . time ( ) > next_announce_at :
should_announce = True
else :
# last announced at is null, so we have never announced, lets do it now
should_announce = True
# announce
if should_announce :
2024-06-02 12:07:59 +12:00
await self . announce ( )
2024-06-02 11:15:09 +12:00
# wait 1 second before next loop
2024-06-02 12:07:59 +12:00
await asyncio . sleep ( 1 )
2024-06-02 11:15:09 +12:00
2024-09-18 21:06:03 +12:00
# automatically syncs propagation nodes based on user config
async def announce_sync_propagation_nodes ( self ) :
while True :
should_sync = False
# check if auto sync is enabled
auto_sync_interval_seconds = self . config . lxmf_preferred_propagation_node_auto_sync_interval_seconds . get ( )
if auto_sync_interval_seconds > 0 :
# check if we have synced recently
last_synced_at = self . config . lxmf_preferred_propagation_node_last_synced_at . get ( )
if last_synced_at is not None :
# determine when next sync should happen
next_sync_at = last_synced_at + auto_sync_interval_seconds
# we should sync if current time has passed next sync at timestamp
if time . time ( ) > next_sync_at :
should_sync = True
else :
# last synced at is null, so we have never synced, lets do it now
should_sync = True
# sync
if should_sync :
await self . sync_propagation_nodes ( )
# wait 1 second before next loop
await asyncio . sleep ( 1 )
2024-09-18 17:42:25 +12:00
# uses the provided destination hash as the active propagation node
def set_active_propagation_node ( self , destination_hash : str | None ) :
# set outbound propagation node
if destination_hash is not None and destination_hash != " " :
try :
self . message_router . set_outbound_propagation_node ( bytes . fromhex ( destination_hash ) )
except :
# failed to set propagation node, clear it to ensure we don't use an old one by mistake
2024-09-19 01:45:23 +12:00
self . remove_active_propagation_node ( )
2024-09-18 17:42:25 +12:00
pass
# stop using propagation node
else :
2024-09-19 01:45:23 +12:00
self . remove_active_propagation_node ( )
# stops the in progress propagation node sync
def stop_propagation_node_sync ( self ) :
self . message_router . cancel_propagation_node_requests ( )
# stops and removes the active propagation node
def remove_active_propagation_node ( self ) :
# fixme: it's possible for internal transfer state to get stuck if we change propagation node during a sync
# this still happens even if we cancel the propagation node requests
# for now, the user can just manually cancel syncing in the ui if they think it's stuck...
self . stop_propagation_node_sync ( )
self . message_router . outbound_propagation_node = None
2024-09-18 17:42:25 +12:00
2024-09-19 15:59:04 +12:00
# enables or disables the local lxmf propagation node
def enable_local_propagation_node ( self , enabled : bool = True ) :
try :
if enabled :
self . message_router . enable_propagation ( )
else :
self . message_router . disable_propagation ( )
except :
print ( " failed to enable or disable propagation node " )
pass
2024-05-20 23:18:17 +12:00
# handle receiving a new audio call
def on_incoming_audio_call ( self , audio_call : AudioCall ) :
print ( " on_incoming_audio_call: {} " . format ( audio_call . link . hash . hex ( ) ) )
2024-07-31 15:56:10 +12:00
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
" type " : " incoming_audio_call " ,
} ) ) )
2024-05-20 17:48:55 +12:00
2024-05-04 02:42:18 +12:00
# web server has shutdown, likely ctrl+c, but if we don't do the following, the script never exits
async def shutdown ( self , app ) :
2024-04-29 17:54:39 +12:00
2024-05-04 02:42:18 +12:00
# force close websocket clients
for websocket_client in self . websocket_clients :
await websocket_client . close ( code = WSCloseCode . GOING_AWAY )
# stop reticulum
RNS . Transport . detach_interfaces ( )
self . reticulum . exit_handler ( )
RNS . exit ( )
2024-05-19 00:48:45 +12:00
def run ( self , host , port , launch_browser : bool ) :
2024-04-29 01:09:06 +12:00
2024-05-04 02:42:18 +12:00
# create route table
routes = web . RouteTableDef ( )
2024-04-29 22:06:06 +12:00
2024-04-29 20:26:45 +12:00
# serve index.html
2024-05-04 02:42:18 +12:00
@routes.get ( " / " )
async def index ( request ) :
2024-08-18 22:33:04 +12:00
return web . FileResponse ( path = get_file_path ( " public/index.html " ) , headers = {
# don't allow browser to store page in cache, otherwise new app versions may get stale ui
" Cache-Control " : " no-cache, no-store " ,
} )
2024-05-04 02:42:18 +12:00
2024-05-27 17:31:20 +12:00
# serve ping
@routes.get ( " /api/v1/status " )
async def index ( request ) :
return web . json_response ( {
" status " : " ok " ,
} )
2024-05-29 11:55:04 +12:00
# fetch com ports
@routes.get ( " /api/v1/comports " )
async def index ( request ) :
comports = [ ]
for comport in list_ports . comports ( ) :
comports . append ( {
" device " : comport . device ,
" product " : comport . product ,
" serial_number " : comport . serial_number ,
} )
return web . json_response ( {
" comports " : comports ,
} )
2024-05-29 12:36:42 +12:00
# fetch reticulum interfaces
@routes.get ( " /api/v1/reticulum/interfaces " )
async def index ( request ) :
interfaces = { }
if " interfaces " in self . reticulum . config :
interfaces = self . reticulum . config [ " interfaces " ]
return web . json_response ( {
" interfaces " : interfaces ,
} )
2024-05-29 21:37:03 +12:00
# enable reticulum interface
@routes.post ( " /api/v1/reticulum/interfaces/enable " )
async def index ( request ) :
# get request data
data = await request . json ( )
interface_name = data . get ( ' name ' )
# enable interface
if " interfaces " in self . reticulum . config :
interface = self . reticulum . config [ " interfaces " ] [ interface_name ]
if " enabled " in interface :
interface [ " enabled " ] = " true "
if " interface_enabled " in interface :
interface [ " interface_enabled " ] = " true "
# save config
self . reticulum . config . write ( )
return web . json_response ( {
2024-05-30 00:03:13 +12:00
" message " : " Interface is now enabled " ,
2024-05-29 21:37:03 +12:00
} )
# disable reticulum interface
@routes.post ( " /api/v1/reticulum/interfaces/disable " )
async def index ( request ) :
# get request data
data = await request . json ( )
interface_name = data . get ( ' name ' )
# disable interface
if " interfaces " in self . reticulum . config :
interface = self . reticulum . config [ " interfaces " ] [ interface_name ]
if " enabled " in interface :
interface [ " enabled " ] = " false "
if " interface_enabled " in interface :
interface [ " interface_enabled " ] = " false "
# save config
self . reticulum . config . write ( )
return web . json_response ( {
2024-05-30 00:03:13 +12:00
" message " : " Interface is now disabled " ,
2024-05-29 21:37:03 +12:00
} )
2024-05-29 21:20:35 +12:00
# delete reticulum interface
@routes.post ( " /api/v1/reticulum/interfaces/delete " )
async def index ( request ) :
# get request data
data = await request . json ( )
interface_name = data . get ( ' name ' )
# delete interface
if " interfaces " in self . reticulum . config :
del self . reticulum . config [ " interfaces " ] [ interface_name ]
# save config
self . reticulum . config . write ( )
return web . json_response ( {
2024-05-30 00:03:13 +12:00
" message " : " Interface has been deleted " ,
} )
# add reticulum interface
@routes.post ( " /api/v1/reticulum/interfaces/add " )
async def index ( request ) :
# get request data
data = await request . json ( )
interface_name = data . get ( ' name ' )
interface_type = data . get ( ' type ' )
2024-05-30 01:16:46 +12:00
allow_overwriting_interface = data . get ( ' allow_overwriting_interface ' , False )
2024-05-30 00:03:13 +12:00
# ensure name is provided
if interface_name is None or interface_name == " " :
return web . json_response ( {
" message " : " Name is required " ,
} , status = 422 )
# ensure type name provided
if interface_type is None or interface_type == " " :
return web . json_response ( {
" message " : " Type is required " ,
} , status = 422 )
# get existing interfaces
interfaces = { }
if " interfaces " in self . reticulum . config :
interfaces = self . reticulum . config [ " interfaces " ]
2024-05-30 00:12:31 +12:00
# ensure name is not for an existing interface, to prevent overwriting
2024-05-30 01:16:46 +12:00
if allow_overwriting_interface is False and interface_name in interfaces :
2024-05-30 00:12:31 +12:00
return web . json_response ( {
" message " : " Name is already in use by another interface " ,
} , status = 422 )
2024-05-30 01:16:46 +12:00
# get existing interface details if available
interface_details = { }
if interface_name in interfaces :
interface_details = interfaces [ interface_name ]
# update interface details
interface_details [ " type " ] = interface_type
2024-05-30 00:03:13 +12:00
2024-05-30 01:35:53 +12:00
# if interface doesn't have enabled or interface_enabled setting already, enable it by default
if " enabled " not in interface_details and " interface_enabled " not in interface_details :
interface_details [ " interface_enabled " ] = " true "
2024-05-30 00:03:13 +12:00
# handle tcp client interface
if interface_type == " TCPClientInterface " :
interface_target_host = data . get ( ' target_host ' )
interface_target_port = data . get ( ' target_port ' )
# ensure target host provided
if interface_target_host is None or interface_target_host == " " :
return web . json_response ( {
" message " : " Target Host is required " ,
} , status = 422 )
# ensure target port provided
if interface_target_port is None or interface_target_port == " " :
return web . json_response ( {
" message " : " Target Port is required " ,
} , status = 422 )
interface_details [ " target_host " ] = data . get ( ' target_host ' )
interface_details [ " target_port " ] = data . get ( ' target_port ' )
2024-07-12 11:41:28 +12:00
# handle tcp server interface
if interface_type == " TCPServerInterface " :
interface_listen_ip = data . get ( ' listen_ip ' )
interface_listen_port = data . get ( ' listen_port ' )
# ensure listen ip provided
if interface_listen_ip is None or interface_listen_ip == " " :
return web . json_response ( {
" message " : " Listen IP is required " ,
} , status = 422 )
# ensure listen port provided
if interface_listen_port is None or interface_listen_port == " " :
return web . json_response ( {
" message " : " Listen Port is required " ,
} , status = 422 )
interface_details [ " listen_ip " ] = data . get ( ' listen_ip ' )
interface_details [ " listen_port " ] = data . get ( ' listen_port ' )
2024-07-12 22:45:38 +12:00
# handle udp interface
if interface_type == " UDPInterface " :
interface_listen_ip = data . get ( ' listen_ip ' )
interface_listen_port = data . get ( ' listen_port ' )
interface_forward_ip = data . get ( ' forward_ip ' )
interface_forward_port = data . get ( ' forward_port ' )
# ensure listen ip provided
if interface_listen_ip is None or interface_listen_ip == " " :
return web . json_response ( {
" message " : " Listen IP is required " ,
} , status = 422 )
# ensure listen port provided
if interface_listen_port is None or interface_listen_port == " " :
return web . json_response ( {
" message " : " Listen Port is required " ,
} , status = 422 )
# ensure forward ip provided
if interface_forward_ip is None or interface_forward_ip == " " :
return web . json_response ( {
" message " : " Forward IP is required " ,
} , status = 422 )
# ensure forward port provided
if interface_forward_port is None or interface_forward_port == " " :
return web . json_response ( {
" message " : " Forward Port is required " ,
} , status = 422 )
interface_details [ " listen_ip " ] = data . get ( ' listen_ip ' )
interface_details [ " listen_port " ] = data . get ( ' listen_port ' )
interface_details [ " forward_ip " ] = data . get ( ' forward_ip ' )
interface_details [ " forward_port " ] = data . get ( ' forward_port ' )
2024-05-30 00:41:03 +12:00
# handle rnode interface
if interface_type == " RNodeInterface " :
interface_port = data . get ( ' port ' )
interface_frequency = data . get ( ' frequency ' )
interface_bandwidth = data . get ( ' bandwidth ' )
interface_txpower = data . get ( ' txpower ' )
interface_spreadingfactor = data . get ( ' spreadingfactor ' )
interface_codingrate = data . get ( ' codingrate ' )
# ensure port provided
if interface_port is None or interface_port == " " :
return web . json_response ( {
" message " : " Port is required " ,
} , status = 422 )
# ensure frequency provided
if interface_frequency is None or interface_frequency == " " :
return web . json_response ( {
" message " : " Frequency is required " ,
} , status = 422 )
# ensure bandwidth provided
if interface_bandwidth is None or interface_bandwidth == " " :
return web . json_response ( {
" message " : " Bandwidth is required " ,
} , status = 422 )
# ensure txpower provided
if interface_txpower is None or interface_txpower == " " :
return web . json_response ( {
" message " : " TX power is required " ,
} , status = 422 )
# ensure spreading factor provided
if interface_spreadingfactor is None or interface_spreadingfactor == " " :
return web . json_response ( {
" message " : " Spreading Factor is required " ,
} , status = 422 )
# ensure coding rate provided
if interface_codingrate is None or interface_codingrate == " " :
return web . json_response ( {
" message " : " Coding Rate is required " ,
} , status = 422 )
interface_details [ " port " ] = interface_port
interface_details [ " frequency " ] = interface_frequency
interface_details [ " bandwidth " ] = interface_bandwidth
interface_details [ " txpower " ] = interface_txpower
interface_details [ " spreadingfactor " ] = interface_spreadingfactor
interface_details [ " codingrate " ] = interface_codingrate
2024-05-30 00:03:13 +12:00
# merge new interface into existing interfaces
interfaces [ interface_name ] = interface_details
self . reticulum . config [ " interfaces " ] = interfaces
# save config
self . reticulum . config . write ( )
2024-05-30 01:16:46 +12:00
if allow_overwriting_interface :
return web . json_response ( {
" message " : " Interface has been saved " ,
} )
else :
return web . json_response ( {
" message " : " Interface has been added " ,
} )
2024-05-04 02:42:18 +12:00
# handle websocket clients
@routes.get ( " /ws " )
async def ws ( request ) :
# prepare websocket response
2024-05-05 00:49:09 +12:00
websocket_response = web . WebSocketResponse (
# set max message size accepted by server to 50 megabytes
max_msg_size = 50 * 1024 * 1024 ,
)
2024-05-04 02:42:18 +12:00
await websocket_response . prepare ( request )
# add client to connected clients list
self . websocket_clients . append ( websocket_response )
# send config to all clients
await self . send_config_to_websocket_clients ( )
# handle websocket messages until disconnected
async for msg in websocket_response :
msg : WSMessage = msg
if msg . type == WSMsgType . TEXT :
try :
data = json . loads ( msg . data )
await self . on_websocket_data_received ( websocket_response , data )
except Exception as e :
# ignore errors while handling message
print ( " failed to process client message " )
print ( e )
elif msg . type == WSMsgType . ERROR :
# ignore errors while handling message
print ( ' ws connection error %s ' % websocket_response . exception ( ) )
# websocket closed
self . websocket_clients . remove ( websocket_response )
return websocket_response
2024-05-04 00:54:16 +12:00
2024-06-02 13:56:04 +12:00
# get app info
@routes.get ( " /api/v1/app/info " )
async def index ( request ) :
return web . json_response ( {
" app_info " : {
" version " : self . get_app_version ( ) ,
2024-09-11 20:28:11 +12:00
" lxmf_version " : LXMF . __version__ ,
" rns_version " : RNS . __version__ ,
2024-06-02 13:56:04 +12:00
" storage_path " : self . storage_path ,
" database_path " : self . database_path ,
" database_file_size " : os . path . getsize ( self . database_path ) ,
" reticulum_config_path " : self . reticulum . configpath ,
2024-08-12 12:00:08 +12:00
" is_connected_to_shared_instance " : self . reticulum . is_connected_to_shared_instance ,
2024-08-12 12:07:24 +12:00
" is_transport_enabled " : self . reticulum . transport_enabled ( ) ,
2024-06-02 13:56:04 +12:00
} ,
} )
2024-05-21 00:09:03 +12:00
# get config
@routes.get ( " /api/v1/config " )
async def index ( request ) :
return web . json_response ( {
" config " : self . get_config_dict ( ) ,
} )
2024-08-02 18:44:41 +12:00
# update config
@routes.patch ( " /api/v1/config " )
async def index ( request ) :
# get request body as json
data = await request . json ( )
# update config
await self . update_config ( data )
return web . json_response ( {
" config " : self . get_config_dict ( ) ,
} )
2024-05-20 23:18:17 +12:00
# get calls
@routes.get ( " /api/v1/calls " )
async def index ( request ) :
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
# get audio calls
audio_calls = [ ]
for audio_call in self . audio_call_manager . audio_calls :
2024-05-21 02:42:03 +12:00
audio_calls . append ( self . convert_audio_call_to_dict ( audio_call ) )
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
return web . json_response ( {
" audio_calls " : audio_calls ,
} )
2024-05-20 16:06:30 +12:00
2024-05-23 23:34:33 +12:00
# clear call history
@routes.post ( " /api/v1/calls/clear-call-history " )
async def index ( request ) :
# delete inactive calls, which are classed as call history
for audio_call in self . audio_call_manager . audio_calls :
if audio_call . is_active ( ) is False :
self . audio_call_manager . delete_audio_call ( audio_call )
return web . json_response ( {
" message " : " Call history has been cleared " ,
} )
2024-05-23 23:24:26 +12:00
# hangup all calls
@routes.get ( " /api/v1/calls/hangup-all " )
async def index ( request ) :
self . audio_call_manager . hangup_all ( )
return web . json_response ( {
" message " : " All calls have been hungup " ,
} )
2024-05-21 03:22:45 +12:00
# get call
2024-05-21 02:42:03 +12:00
@routes.get ( " /api/v1/calls/ {audio_call_link_hash} " )
async def index ( request ) :
# get path params
audio_call_link_hash = request . match_info . get ( " audio_call_link_hash " , " " )
# convert hash to bytes
audio_call_link_hash = bytes . fromhex ( audio_call_link_hash )
# find audio call
audio_call = self . audio_call_manager . find_audio_call_by_link_hash ( audio_call_link_hash )
if audio_call is None :
return web . json_response ( {
" message " : " audio call not found " ,
} , status = 404 )
return web . json_response ( {
" audio_call " : self . convert_audio_call_to_dict ( audio_call ) ,
} )
2024-05-21 03:22:45 +12:00
# delete call
@routes.delete ( " /api/v1/calls/ {audio_call_link_hash} " )
async def index ( request ) :
# get path params
audio_call_link_hash = request . match_info . get ( " audio_call_link_hash " , " " )
# convert hash to bytes
audio_call_link_hash = bytes . fromhex ( audio_call_link_hash )
# delete audio call
self . audio_call_manager . delete_audio_call_by_link_hash ( audio_call_link_hash )
return web . json_response ( {
" message " : " audio call deleted " ,
} )
2024-05-20 23:18:17 +12:00
# initiate a call to the provided destination
@routes.get ( " /api/v1/calls/initiate/ {destination_hash} " )
async def index ( request ) :
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
2024-05-21 18:09:05 +12:00
timeout_seconds = int ( request . query . get ( " timeout " , 15 ) )
2024-05-20 23:18:17 +12:00
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
# initiate audio call
2024-05-23 22:28:34 +12:00
try :
2024-05-25 23:52:41 +12:00
audio_call = await self . audio_call_manager . initiate ( destination_hash , timeout_seconds )
2024-05-21 18:09:05 +12:00
return web . json_response ( {
2024-05-25 23:52:41 +12:00
" audio_call " : self . convert_audio_call_to_dict ( audio_call ) ,
2024-05-23 22:28:34 +12:00
} )
2024-05-20 23:18:17 +12:00
2024-05-23 22:28:34 +12:00
except Exception as e :
return web . json_response ( {
" message " : " Call Failed: {} " . format ( str ( e ) ) ,
} , status = 503 )
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
# handle websocket client for sending and receiving audio packets in a call
@routes.get ( " /api/v1/calls/ {audio_call_link_hash} /audio " )
async def ws ( request ) :
2024-05-20 16:06:30 +12:00
2024-05-20 23:18:17 +12:00
# get path params
audio_call_link_hash = request . match_info . get ( " audio_call_link_hash " , " " )
# convert hash to bytes
audio_call_link_hash = bytes . fromhex ( audio_call_link_hash )
2024-05-20 16:06:30 +12:00
2024-05-20 23:59:16 +12:00
# find audio call, this will be null until the link is established
2024-05-20 23:18:17 +12:00
audio_call = self . audio_call_manager . find_audio_call_by_link_hash ( audio_call_link_hash )
if audio_call is None :
# fixme: web browser expects websocket, so this won't be useful
return web . json_response ( {
" message " : " audio call not found " ,
} , status = 404 )
# send audio received from call initiator to call receiver websocket
def on_audio_packet ( data ) :
if websocket_response . closed is False :
try :
asyncio . run ( websocket_response . send_bytes ( data ) )
except :
# ignore errors sending audio packets to websocket
pass
2024-05-21 00:39:26 +12:00
# close websocket when call is hungup
def on_hangup ( ) :
if websocket_response . closed is False :
try :
asyncio . run ( websocket_response . close ( code = WSCloseCode . GOING_AWAY ) )
except :
# ignore errors closing websocket
pass
2024-05-20 23:18:17 +12:00
# register audio packet listener
audio_call . register_audio_packet_listener ( on_audio_packet )
2024-05-21 00:39:26 +12:00
audio_call . register_hangup_listener ( on_hangup )
2024-05-20 23:18:17 +12:00
# prepare websocket response
websocket_response = web . WebSocketResponse ( )
await websocket_response . prepare ( request )
2024-05-20 16:06:30 +12:00
# handle websocket messages until disconnected
2024-05-20 23:18:17 +12:00
# FIXME: we should send a type with the message, so we can send other data as well
2024-05-20 16:06:30 +12:00
async for msg in websocket_response :
msg : WSMessage = msg
2024-05-20 17:48:55 +12:00
if msg . type == WSMsgType . BINARY :
try :
2024-05-20 23:18:17 +12:00
audio_call . send_audio_packet ( msg . data )
2024-05-20 17:48:55 +12:00
except Exception as e :
# ignore errors while handling message
print ( " failed to process client message " )
print ( e )
elif msg . type == WSMsgType . ERROR :
2024-05-20 16:06:30 +12:00
# ignore errors while handling message
print ( ' ws connection error %s ' % websocket_response . exception ( ) )
2024-05-20 23:18:17 +12:00
# unregister audio packet handler now that the websocket has been closed
audio_call . register_audio_packet_listener ( on_audio_packet )
2024-05-20 16:06:30 +12:00
return websocket_response
2024-05-20 23:18:17 +12:00
# hangup calls
@routes.get ( " /api/v1/calls/ {audio_call_link_hash} /hangup " )
async def index ( request ) :
# get path params
audio_call_link_hash = request . match_info . get ( " audio_call_link_hash " , " " )
# convert hash to bytes
audio_call_link_hash = bytes . fromhex ( audio_call_link_hash )
# find audio call
audio_call = self . audio_call_manager . find_audio_call_by_link_hash ( audio_call_link_hash )
if audio_call is None :
return web . json_response ( {
" message " : " audio call not found " ,
} , status = 404 )
# hangup the call
audio_call . hangup ( )
return web . json_response ( {
2024-05-23 23:24:26 +12:00
" message " : " Call has been hungup " ,
2024-05-20 23:18:17 +12:00
} )
2024-05-21 04:01:38 +12:00
# announce
@routes.get ( " /api/v1/announce " )
async def index ( request ) :
2024-07-09 00:45:27 +12:00
await self . announce ( )
2024-05-21 04:01:38 +12:00
return web . json_response ( {
" message " : " announcing " ,
} )
2024-05-05 22:12:45 +12:00
# serve announces
@routes.get ( " /api/v1/announces " )
async def index ( request ) :
# get query params
aspect = request . query . get ( " aspect " , None )
2024-09-17 16:35:38 +12:00
identity_hash = request . query . get ( " identity_hash " , None )
2024-08-13 23:00:42 +12:00
limit = request . query . get ( " limit " , None )
2024-05-05 22:12:45 +12:00
# build announces database query
query = database . Announce . select ( )
# filter by provided aspect
if aspect is not None :
query = query . where ( database . Announce . aspect == aspect )
2024-09-17 16:35:38 +12:00
# filter by provided identity hash
if identity_hash is not None :
query = query . where ( database . Announce . identity_hash == identity_hash )
2024-05-05 22:12:45 +12:00
2024-08-13 23:00:42 +12:00
# limit results
if limit is not None :
query = query . limit ( limit )
# order announces latest to oldest
query_results = query . order_by ( database . Announce . updated_at . desc ( ) )
2024-05-05 22:12:45 +12:00
# process announces
announces = [ ]
for announce in query_results :
2024-05-05 23:51:15 +12:00
announces . append ( self . convert_db_announce_to_dict ( announce ) )
2024-05-05 22:12:45 +12:00
return web . json_response ( {
" announces " : announces ,
} )
2024-09-18 18:19:02 +12:00
# propagation node status
@routes.get ( " /api/v1/lxmf/propagation-node/status " )
async def index ( request ) :
return web . json_response ( {
" propagation_node_status " : {
" state " : self . convert_propagation_node_state_to_string ( self . message_router . propagation_transfer_state ) ,
" progress " : self . message_router . propagation_transfer_progress * 100 , # convert to percentage
" messages_received " : self . message_router . propagation_transfer_last_result ,
} ,
} )
# sync propagation node
@routes.get ( " /api/v1/lxmf/propagation-node/sync " )
async def index ( request ) :
2024-09-18 19:46:56 +12:00
# ensure propagation node is configured before attempting to sync
if self . message_router . get_outbound_propagation_node ( ) is None :
return web . json_response ( {
" message " : " A propagation node must be configured to sync messages. " ,
} , status = 400 )
2024-09-18 18:19:02 +12:00
# request messages from propagation node
2024-09-18 21:06:03 +12:00
await self . sync_propagation_nodes ( )
2024-09-18 18:19:02 +12:00
return web . json_response ( {
" message " : " Sync is starting " ,
} )
2024-09-19 01:45:23 +12:00
# stop syncing propagation node
@routes.get ( " /api/v1/lxmf/propagation-node/stop-sync " )
async def index ( request ) :
self . stop_propagation_node_sync ( )
return web . json_response ( {
" message " : " Sync is stopping " ,
} )
2024-09-18 23:26:16 +12:00
# serve propagation nodes
@routes.get ( " /api/v1/lxmf/propagation-nodes " )
async def index ( request ) :
# get query params
limit = request . query . get ( " limit " , None )
# get lxmf.propagation announces
query = database . Announce . select ( ) . where ( database . Announce . aspect == " lxmf.propagation " )
# limit results
if limit is not None :
query = query . limit ( limit )
# order announces latest to oldest
query_results = query . order_by ( database . Announce . updated_at . desc ( ) )
# process announces
lxmf_propagation_nodes = [ ]
for announce in query_results :
# find an lxmf.delivery announce for the same identity hash, so we can use that as an "operater by" name
lxmf_delivery_announce = ( database . Announce . select ( )
. where ( database . Announce . aspect == " lxmf.delivery " )
. where ( database . Announce . identity_hash == announce . identity_hash )
. get_or_none ( ) )
# find a nomadnetwork.node announce for the same identity hash, so we can use that as an "operated by" name
nomadnetwork_node_announce = ( database . Announce . select ( )
. where ( database . Announce . aspect == " nomadnetwork.node " )
. where ( database . Announce . identity_hash == announce . identity_hash )
. get_or_none ( ) )
# get a display name from other announces belonging to the propagation nodes identity
operator_display_name = None
if lxmf_delivery_announce is not None and lxmf_delivery_announce . app_data is not None :
operator_display_name = self . parse_lxmf_display_name ( lxmf_delivery_announce . app_data , None )
elif nomadnetwork_node_announce is not None and nomadnetwork_node_announce . app_data is not None :
operator_display_name = self . parse_nomadnetwork_node_display_name ( nomadnetwork_node_announce . app_data , None )
2024-09-19 17:20:49 +12:00
# parse app_data so we can see if propagation is enabled or disabled for this node
2024-09-19 17:21:54 +12:00
is_propagation_enabled = None
2024-09-19 17:20:49 +12:00
per_transfer_limit = None
propagation_node_data = self . parse_lxmf_propagation_node_app_data ( announce . app_data )
if propagation_node_data is not None :
is_propagation_enabled = propagation_node_data [ " enabled " ]
per_transfer_limit = propagation_node_data [ " per_transfer_limit " ]
2024-09-19 16:07:14 +12:00
2024-09-18 23:26:16 +12:00
lxmf_propagation_nodes . append ( {
" destination_hash " : announce . destination_hash ,
" identity_hash " : announce . identity_hash ,
" operator_display_name " : operator_display_name ,
2024-09-19 17:20:49 +12:00
" is_propagation_enabled " : is_propagation_enabled ,
" per_transfer_limit " : per_transfer_limit ,
2024-09-18 23:26:16 +12:00
" created_at " : announce . created_at ,
" updated_at " : announce . updated_at ,
} )
return web . json_response ( {
" lxmf_propagation_nodes " : lxmf_propagation_nodes ,
} )
2024-05-18 21:18:33 +12:00
# get path to destination
@routes.get ( " /api/v1/destination/ {destination_hash} /path " )
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
2024-05-19 21:41:34 +12:00
# check if user wants to request the path from the network right now
request_query_param = request . query . get ( " request " , " false " )
should_request_now = request_query_param == " true " or request_query_param == " 1 "
if should_request_now :
# determine how long we should wait for a path response
timeout_seconds = int ( request . query . get ( " timeout " , 15 ) )
timeout_after_seconds = time . time ( ) + timeout_seconds
# request path if we don't have it
if not RNS . Transport . has_path ( destination_hash ) :
RNS . Transport . request_path ( destination_hash )
# wait until we have a path, or give up after the configured timeout
while not RNS . Transport . has_path ( destination_hash ) and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
2024-05-18 21:18:33 +12:00
# ensure path is known
if not RNS . Transport . has_path ( destination_hash ) :
return web . json_response ( {
" path " : None ,
} )
# determine next hop and hop count
hops = RNS . Transport . hops_to ( destination_hash )
next_hop_bytes = self . reticulum . get_next_hop ( destination_hash )
# ensure next hop provided
if next_hop_bytes is None :
return web . json_response ( {
" path " : None ,
} )
next_hop = next_hop_bytes . hex ( )
next_hop_interface = self . reticulum . get_next_hop_if_name ( destination_hash )
return web . json_response ( {
" path " : {
" hops " : hops ,
" next_hop " : next_hop ,
" next_hop_interface " : next_hop_interface ,
} ,
} )
2024-10-24 01:04:50 +13:00
# pings an lxmf.delivery destination by sending empty data and waiting for the recipient to send a proof back
# the lxmf router proves all received packets, then drops them if they can't be decoded as lxmf messages
# this allows us to ping/probe any active lxmf.delivery destination and get rtt/snr/rssi data on demand
# https://github.com/markqvist/LXMF/blob/9ff76c0473e9d4107e079f266dd08144bb74c7c8/LXMF/LXMRouter.py#L234
# https://github.com/markqvist/LXMF/blob/9ff76c0473e9d4107e079f266dd08144bb74c7c8/LXMF/LXMRouter.py#L1374
@routes.get ( " /api/v1/ping/ {destination_hash} /lxmf.delivery " )
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
# determine how long until we should time out
timeout_seconds = int ( request . query . get ( " timeout " , 15 ) )
timeout_after_seconds = time . time ( ) + timeout_seconds
# request path if we don't have it
if not RNS . Transport . has_path ( destination_hash ) :
RNS . Transport . request_path ( destination_hash )
# wait until we have a path, or give up after the configured timeout
while not RNS . Transport . has_path ( destination_hash ) and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
# find destination identity
destination_identity = RNS . Identity . recall ( destination_hash )
if destination_identity is None :
return web . json_response ( {
" message " : " Ping failed. Could not find path to destination. " ,
} , status = 503 )
# create outbound destination
request_destination = RNS . Destination (
destination_identity ,
RNS . Destination . OUT ,
RNS . Destination . SINGLE ,
" lxmf " ,
" delivery " ,
)
# send empty packet to destination
packet = RNS . Packet ( request_destination , b " " )
receipt = packet . send ( )
# wait until delivered, or give up after time out
while receipt . status != RNS . PacketReceipt . DELIVERED and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
# ping failed if not delivered
if receipt . status != RNS . PacketReceipt . DELIVERED :
return web . json_response ( {
" message " : f " Ping failed. Timed out after { timeout_seconds } seconds. " ,
} , status = 503 )
# get number of hops to destination
hops = RNS . Transport . hops_to ( destination_hash )
# get rssi
rssi = receipt . proof_packet . rssi
if rssi is None :
rssi = self . reticulum . get_packet_rssi ( receipt . proof_packet . packet_hash )
# get snr
snr = receipt . proof_packet . snr
if snr is None :
snr = self . reticulum . get_packet_snr ( receipt . proof_packet . packet_hash )
# get signal quality
quality = receipt . proof_packet . q
if quality is None :
quality = self . reticulum . get_packet_q ( receipt . proof_packet . packet_hash )
# get and format round trip time
rtt = receipt . get_rtt ( )
rtt_milliseconds = round ( rtt * 1000 , 3 )
rtt_duration_string = f " { rtt_milliseconds } ms "
return web . json_response ( {
" message " : f " Valid reply from { receipt . destination . hash . hex ( ) } : hops= { hops } time= { rtt_duration_string } " ,
" ping_result " : {
" rtt " : rtt ,
" hops " : hops ,
" rssi " : rssi ,
" snr " : snr ,
" quality " : quality ,
} ,
} )
2024-09-19 23:46:08 +12:00
# get custom destination display name
@routes.get ( " /api/v1/destination/ {destination_hash} /custom-display-name " )
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
return web . json_response ( {
" custom_display_name " : self . get_custom_destination_display_name ( destination_hash ) ,
} )
# set custom destination display name
@routes.post ( " /api/v1/destination/ {destination_hash} /custom-display-name/update " )
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
# get request data
data = await request . json ( )
display_name = data . get ( ' display_name ' )
# update display name if provided
if len ( display_name ) > 0 :
self . db_upsert_custom_destination_display_name ( destination_hash , display_name )
return web . json_response ( {
" message " : " Custom display name has been updated " ,
} )
# otherwise remove display name
else :
database . CustomDestinationDisplayName . delete ( ) . where ( database . CustomDestinationDisplayName . destination_hash == destination_hash ) . execute ( )
return web . json_response ( {
" message " : " Custom display name has been removed " ,
} )
2024-09-25 22:19:12 +12:00
# get lxmf stamp cost for the provided lxmf.delivery destination hash
2024-09-26 16:42:13 +12:00
@routes.get ( " /api/v1/destination/ {destination_hash} /lxmf-stamp-info " )
2024-09-25 22:19:12 +12:00
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
2024-09-26 16:42:13 +12:00
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
2024-09-25 22:19:12 +12:00
# get lxmf stamp cost from announce in database
lxmf_stamp_cost = None
2024-09-26 16:42:13 +12:00
announce = database . Announce . get_or_none ( database . Announce . destination_hash == destination_hash . hex ( ) )
2024-09-25 22:19:12 +12:00
if announce is not None :
lxmf_stamp_cost = self . parse_lxmf_stamp_cost ( announce . app_data )
2024-09-26 16:42:13 +12:00
# get outbound ticket expiry for this lxmf destination
lxmf_outbound_ticket_expiry = self . message_router . get_outbound_ticket_expiry ( destination_hash )
2024-09-25 22:19:12 +12:00
return web . json_response ( {
2024-09-26 16:42:13 +12:00
" lxmf_stamp_info " : {
" stamp_cost " : lxmf_stamp_cost ,
" outbound_ticket_expiry " : lxmf_outbound_ticket_expiry ,
} ,
2024-09-25 22:19:12 +12:00
} )
2024-05-18 21:54:49 +12:00
# get interface stats
@routes.get ( " /api/v1/interface-stats " )
async def index ( request ) :
# get interface stats
interface_stats = self . reticulum . get_interface_stats ( )
# ensure transport_id is hex as json_response can't serialize bytes
if " transport_id " in interface_stats :
interface_stats [ " transport_id " ] = interface_stats [ " transport_id " ] . hex ( )
2024-09-24 20:35:51 +02:00
# ensure probe_responder is hex as json_response can't serialize bytes
2024-09-26 00:25:46 +12:00
if " probe_responder " in interface_stats and interface_stats [ " probe_responder " ] is not None :
2024-09-24 20:35:51 +02:00
interface_stats [ " probe_responder " ] = interface_stats [ " probe_responder " ] . hex ( )
2024-05-30 02:16:05 +12:00
# ensure ifac_signature is hex as json_response can't serialize bytes
for interface in interface_stats [ " interfaces " ] :
2024-09-26 00:25:46 +12:00
if " short_name " in interface :
interface [ " interface_name " ] = interface [ " short_name " ]
2024-08-12 10:35:08 +12:00
2024-09-26 00:26:07 +12:00
if " parent_interface_name " in interface and interface [ " parent_interface_name " ] is not None :
2024-09-24 20:35:51 +02:00
interface [ " parent_interface_hash " ] = interface [ " parent_interface_hash " ] . hex ( )
2024-08-12 10:35:08 +12:00
2024-05-30 02:16:05 +12:00
if " ifac_signature " in interface and interface [ " ifac_signature " ] :
interface [ " ifac_signature " ] = interface [ " ifac_signature " ] . hex ( )
2024-09-24 20:35:51 +02:00
if " hash " in interface and interface [ " hash " ] :
interface [ " hash " ] = interface [ " hash " ] . hex ( )
2024-05-18 21:54:49 +12:00
return web . json_response ( {
" interface_stats " : interface_stats ,
} )
2024-05-18 22:05:37 +12:00
# get path table
@routes.get ( " /api/v1/path-table " )
async def index ( request ) :
# get path table, making sure hash and via are in hex as json_response can't serialize bytes
path_table = [ ]
for path in self . reticulum . get_path_table ( ) :
path [ " hash " ] = path [ " hash " ] . hex ( )
path [ " via " ] = path [ " via " ] . hex ( )
path_table . append ( path )
return web . json_response ( {
" path_table " : path_table ,
} )
2024-07-08 23:54:51 +12:00
# send lxmf message
@routes.post ( " /api/v1/lxmf-messages/send " )
async def index ( request ) :
# get request body as json
data = await request . json ( )
2024-11-04 18:37:41 +13:00
# get delivery method
delivery_method = None
if " delivery_method " in data :
delivery_method = data [ " delivery_method " ]
2024-07-08 23:54:51 +12:00
# get data from json
destination_hash = data [ " lxmf_message " ] [ " destination_hash " ]
content = data [ " lxmf_message " ] [ " content " ]
fields = { }
if " fields " in data [ " lxmf_message " ] :
fields = data [ " lxmf_message " ] [ " fields " ]
# parse image field
image_field = None
if " image " in fields :
image_type = data [ " lxmf_message " ] [ " fields " ] [ " image " ] [ " image_type " ]
image_bytes = base64 . b64decode ( data [ " lxmf_message " ] [ " fields " ] [ " image " ] [ " image_bytes " ] )
image_field = LxmfImageField ( image_type , image_bytes )
# parse audio field
audio_field = None
if " audio " in fields :
audio_mode = data [ " lxmf_message " ] [ " fields " ] [ " audio " ] [ " audio_mode " ]
audio_bytes = base64 . b64decode ( data [ " lxmf_message " ] [ " fields " ] [ " audio " ] [ " audio_bytes " ] )
audio_field = LxmfAudioField ( audio_mode , audio_bytes )
# parse file attachments field
file_attachments_field = None
if " file_attachments " in fields :
file_attachments = [ ]
for file_attachment in data [ " lxmf_message " ] [ " fields " ] [ " file_attachments " ] :
file_name = file_attachment [ " file_name " ]
file_bytes = base64 . b64decode ( file_attachment [ " file_bytes " ] )
file_attachments . append ( LxmfFileAttachment ( file_name , file_bytes ) )
file_attachments_field = LxmfFileAttachmentsField ( file_attachments )
try :
# send lxmf message to destination
2024-09-17 16:05:58 +12:00
lxmf_message = await self . send_message (
destination_hash = destination_hash ,
content = content ,
image_field = image_field ,
audio_field = audio_field ,
2024-11-04 18:37:41 +13:00
file_attachments_field = file_attachments_field ,
delivery_method = delivery_method
2024-09-17 16:05:58 +12:00
)
2024-07-08 23:54:51 +12:00
return web . json_response ( {
" lxmf_message " : self . convert_lxmf_message_to_dict ( lxmf_message ) ,
} )
except Exception as e :
return web . json_response ( {
" message " : " Sending Failed: {} " . format ( str ( e ) ) ,
} , status = 503 )
2024-05-05 23:51:15 +12:00
# delete lxmf message
2024-07-08 23:10:21 +12:00
@routes.delete ( " /api/v1/lxmf-messages/ {hash} " )
2024-05-04 02:42:18 +12:00
async def index ( request ) :
2024-05-04 00:54:16 +12:00
2024-05-05 20:23:48 +12:00
# get path params
2024-07-08 23:10:21 +12:00
hash = request . match_info . get ( " hash " , None )
2024-05-04 03:10:06 +12:00
2024-07-08 23:10:21 +12:00
# hash is required
if hash is None :
2024-05-04 03:10:06 +12:00
return web . json_response ( {
2024-07-08 23:10:21 +12:00
" message " : " hash is required " ,
2024-05-04 03:10:06 +12:00
} , status = 422 )
2024-07-08 23:10:21 +12:00
# delete lxmf messages from db where hash matches
database . LxmfMessage . delete ( ) . where ( ( database . LxmfMessage . hash == hash ) ) . execute ( )
2024-05-05 20:23:48 +12:00
return web . json_response ( {
" message " : " ok " ,
} )
2024-05-05 23:51:15 +12:00
# serve lxmf messages for conversation
2024-05-05 20:23:48 +12:00
@routes.get ( " /api/v1/lxmf-messages/conversation/ {destination_hash} " )
async def index ( request ) :
# get path params
2024-05-05 21:53:13 +12:00
destination_hash = request . match_info . get ( " destination_hash " , " " )
2024-07-12 19:54:30 +12:00
order = request . query . get ( " order " , " asc " )
count = request . query . get ( " count " )
after_id = request . query . get ( " after_id " )
2024-05-05 20:23:48 +12:00
# get source hash from local lxmf destination
source_hash = self . local_lxmf_destination . hash . hex ( )
2024-05-04 03:57:57 +12:00
# get lxmf messages from db where "source to destination" or "destination to source" and ordered by oldest to newest
2024-05-04 03:10:06 +12:00
db_lxmf_messages = ( database . LxmfMessage . select ( )
2024-07-12 19:54:30 +12:00
. where ( ( database . LxmfMessage . source_hash == source_hash ) & ( database . LxmfMessage . destination_hash == destination_hash ) )
. orwhere ( ( database . LxmfMessage . destination_hash == source_hash ) & ( database . LxmfMessage . source_hash == destination_hash ) )
. order_by ( database . LxmfMessage . id . asc ( ) if order == " asc " else database . LxmfMessage . id . desc ( ) ) )
# limit how many messages to return
if count is not None :
db_lxmf_messages = db_lxmf_messages . limit ( count )
# only get records after provided id, based on query order
if after_id is not None :
if order == " asc " :
db_lxmf_messages = db_lxmf_messages . where ( ( database . LxmfMessage . id > after_id ) )
else :
db_lxmf_messages = db_lxmf_messages . where ( ( database . LxmfMessage . id < after_id ) )
2024-05-04 03:10:06 +12:00
# convert to response json
2024-05-04 00:54:16 +12:00
lxmf_messages = [ ]
for db_lxmf_message in db_lxmf_messages :
2024-09-18 22:00:59 +12:00
lxmf_messages . append ( self . convert_db_lxmf_message_to_dict ( db_lxmf_message ) )
2024-05-04 00:54:16 +12:00
2024-05-04 02:42:18 +12:00
return web . json_response ( {
2024-05-04 00:54:16 +12:00
" lxmf_messages " : lxmf_messages ,
} )
2024-05-05 02:40:03 +12:00
# delete lxmf messages for conversation
@routes.delete ( " /api/v1/lxmf-messages/conversation/ {destination_hash} " )
async def index ( request ) :
# get path params
2024-05-05 21:53:13 +12:00
destination_hash = request . match_info . get ( " destination_hash " , " " )
2024-05-05 02:40:03 +12:00
# get source hash from local lxmf destination
source_hash = self . local_lxmf_destination . hash . hex ( )
# delete lxmf messages from db where "source to destination" or "destination to source"
( database . LxmfMessage . delete ( )
. where ( ( database . LxmfMessage . source_hash == source_hash ) & ( database . LxmfMessage . destination_hash == destination_hash ) )
. orwhere ( ( database . LxmfMessage . destination_hash == source_hash ) & ( database . LxmfMessage . source_hash == destination_hash ) )
. execute ( ) )
return web . json_response ( {
" message " : " ok " ,
} )
2024-06-01 17:36:43 +12:00
# get unqiue lxmf conversations
@routes.get ( " /api/v1/lxmf/conversations " )
async def index ( request ) :
# sql query to fetch unique source/destination hash pairs ordered by the most recently updated message
query = """
WITH NormalizedMessages AS (
SELECT
CASE WHEN source_hash < destination_hash THEN source_hash ELSE destination_hash END AS normalized_source ,
CASE WHEN source_hash < destination_hash THEN destination_hash ELSE source_hash END AS normalized_destination ,
2024-06-01 20:38:06 +12:00
MAX ( created_at ) AS most_recent_created_at
2024-06-01 17:36:43 +12:00
FROM lxmf_messages
GROUP BY normalized_source , normalized_destination
)
SELECT
normalized_source AS source_hash ,
normalized_destination AS destination_hash ,
2024-06-01 20:38:06 +12:00
most_recent_created_at
2024-06-01 17:36:43 +12:00
FROM NormalizedMessages
2024-06-01 20:38:06 +12:00
ORDER BY most_recent_created_at DESC ;
2024-06-01 17:36:43 +12:00
"""
# execute sql query
cursor = database . database . execute_sql ( query )
# parse results to get a list of conversations we have sent or received a message from
conversations = [ ]
for row in cursor . fetchall ( ) :
# get data from row
source_hash = row [ 0 ]
destination_hash = row [ 1 ]
2024-06-01 20:38:06 +12:00
created_at = row [ 2 ]
2024-06-01 17:36:43 +12:00
# determine destination hash of other user
if source_hash == self . local_lxmf_destination . hexhash :
other_user_hash = destination_hash
else :
other_user_hash = source_hash
2024-12-04 01:19:47 +13:00
# find lxmf user icon from database
lxmf_user_icon = None
db_lxmf_user_icon = database . LxmfUserIcon . get_or_none ( database . LxmfUserIcon . destination_hash == other_user_hash )
if db_lxmf_user_icon is not None :
lxmf_user_icon = {
" icon_name " : db_lxmf_user_icon . icon_name ,
" foreground_colour " : db_lxmf_user_icon . foreground_colour ,
" background_colour " : db_lxmf_user_icon . background_colour ,
}
2024-06-01 17:36:43 +12:00
# add to conversations
conversations . append ( {
2024-09-11 21:35:43 +12:00
" display_name " : self . get_lxmf_conversation_name ( other_user_hash ) ,
2024-09-19 23:46:08 +12:00
" custom_display_name " : self . get_custom_destination_display_name ( other_user_hash ) ,
2024-06-01 17:36:43 +12:00
" destination_hash " : other_user_hash ,
2024-06-01 19:45:19 +12:00
" is_unread " : self . is_lxmf_conversation_unread ( other_user_hash ) ,
2024-07-28 16:52:57 +12:00
" failed_messages_count " : self . lxmf_conversation_failed_messages_count ( other_user_hash ) ,
2024-12-04 01:19:47 +13:00
" lxmf_user_icon " : lxmf_user_icon ,
2024-06-01 20:38:06 +12:00
# we say the conversation was updated when the latest message was created
# otherwise this will go crazy when sending a message, as the updated_at on the latest message changes very frequently
" updated_at " : created_at ,
2024-06-01 17:36:43 +12:00
} )
return web . json_response ( {
" conversations " : conversations ,
} )
2024-06-01 19:21:42 +12:00
# mark lxmf conversation as read
@routes.get ( " /api/v1/lxmf/conversations/ {destination_hash} /mark-as-read " )
async def index ( request ) :
# get path params
destination_hash = request . match_info . get ( " destination_hash " , " " )
# mark lxmf conversation as read
self . db_mark_lxmf_conversation_as_read ( destination_hash )
return web . json_response ( {
" message " : " ok " ,
} )
2024-05-19 00:48:45 +12:00
# called when web app has started
async def on_startup ( app ) :
# auto launch web browser
if launch_browser :
try :
webbrowser . open ( " http://127.0.0.1: {} " . format ( port ) )
except :
print ( " failed to launch web browser " )
2024-05-04 02:42:18 +12:00
# create and run web app
2024-08-03 00:21:55 +12:00
app = web . Application ( client_max_size = 1024 * 1024 * 50 ) # allow uploading files up to 50mb
2024-05-04 02:42:18 +12:00
app . add_routes ( routes )
2024-05-24 01:35:30 +12:00
app . add_routes ( [ web . static ( ' / ' , get_file_path ( " public/ " ) ) ] ) # serve anything in public folder
2024-05-04 02:42:18 +12:00
app . on_shutdown . append ( self . shutdown ) # need to force close websockets and stop reticulum now
2024-05-19 00:48:45 +12:00
app . on_startup . append ( on_startup )
2024-05-04 02:42:18 +12:00
web . run_app ( app , host = host , port = port )
2024-04-29 01:09:06 +12:00
2024-05-21 04:01:38 +12:00
# handle announcing
2024-06-02 12:07:59 +12:00
async def announce ( self ) :
2024-05-21 04:01:38 +12:00
2024-06-02 11:16:57 +12:00
# update last announced at timestamp
self . config . last_announced_at . set ( int ( time . time ( ) ) )
2024-09-11 12:58:19 +12:00
# send announce for lxmf (ensuring name is updated before announcing)
self . local_lxmf_destination . display_name = self . config . display_name . get ( )
self . message_router . announce ( destination_hash = self . local_lxmf_destination . hash )
2024-05-21 04:01:38 +12:00
2024-09-19 15:59:04 +12:00
# send announce for local propagation node (if enabled)
if self . config . lxmf_local_propagation_node_enabled . get ( ) :
self . message_router . announce_propagation_node ( )
2024-05-21 04:01:38 +12:00
# send announce for audio call
self . audio_call_manager . announce ( app_data = self . config . display_name . get ( ) . encode ( " utf-8 " ) )
2024-06-02 12:07:59 +12:00
# tell websocket clients we just announced
await self . send_announced_to_websocket_clients ( )
2024-09-18 21:06:03 +12:00
# handle syncing propagation nodes
async def sync_propagation_nodes ( self ) :
# update last synced at timestamp
self . config . lxmf_preferred_propagation_node_last_synced_at . set ( int ( time . time ( ) ) )
# request messages from propagation node
self . message_router . request_messages_from_propagation_node ( self . identity )
# send config to websocket clients (used to tell ui last synced at)
await self . send_config_to_websocket_clients ( )
2024-08-02 18:44:41 +12:00
async def update_config ( self , data ) :
2024-04-29 01:09:06 +12:00
2024-08-02 18:44:41 +12:00
# update display name in config
if " display_name " in data and data [ " display_name " ] != " " :
self . config . display_name . set ( data [ " display_name " ] )
2024-04-29 01:09:06 +12:00
2024-12-16 12:17:11 +13:00
# update theme in config
if " theme " in data and data [ " theme " ] != " " :
self . config . theme . set ( data [ " theme " ] )
2024-08-02 18:44:41 +12:00
# update auto announce interval
if " auto_announce_interval_seconds " in data :
2024-04-29 23:16:54 +12:00
2024-08-02 18:44:41 +12:00
# auto auto announce interval
auto_announce_interval_seconds = int ( data [ " auto_announce_interval_seconds " ] )
self . config . auto_announce_interval_seconds . set ( data [ " auto_announce_interval_seconds " ] )
2024-04-29 23:16:54 +12:00
2024-08-02 18:44:41 +12:00
# enable or disable auto announce based on interval
if auto_announce_interval_seconds > 0 :
self . config . auto_announce_enabled . set ( True )
else :
self . config . auto_announce_enabled . set ( False )
2024-04-29 23:16:54 +12:00
2024-08-02 18:44:41 +12:00
if " auto_resend_failed_messages_when_announce_received " in data :
value = bool ( data [ " auto_resend_failed_messages_when_announce_received " ] )
self . config . auto_resend_failed_messages_when_announce_received . set ( value )
2024-06-02 11:50:58 +12:00
2024-08-02 18:44:41 +12:00
if " allow_auto_resending_failed_messages_with_attachments " in data :
value = bool ( data [ " allow_auto_resending_failed_messages_with_attachments " ] )
self . config . allow_auto_resending_failed_messages_with_attachments . set ( value )
2024-06-02 11:50:58 +12:00
2024-09-18 21:27:32 +12:00
if " auto_send_failed_messages_to_propagation_node " in data :
value = bool ( data [ " auto_send_failed_messages_to_propagation_node " ] )
self . config . auto_send_failed_messages_to_propagation_node . set ( value )
2024-08-02 18:44:41 +12:00
if " show_suggested_community_interfaces " in data :
value = bool ( data [ " show_suggested_community_interfaces " ] )
self . config . show_suggested_community_interfaces . set ( value )
2024-06-02 11:50:58 +12:00
2024-09-18 17:42:25 +12:00
if " lxmf_preferred_propagation_node_destination_hash " in data :
# update config value
value = data [ " lxmf_preferred_propagation_node_destination_hash " ]
self . config . lxmf_preferred_propagation_node_destination_hash . set ( value )
# update active propagation node
self . set_active_propagation_node ( value )
2024-09-18 21:06:03 +12:00
# update auto sync interval
if " lxmf_preferred_propagation_node_auto_sync_interval_seconds " in data :
value = int ( data [ " lxmf_preferred_propagation_node_auto_sync_interval_seconds " ] )
self . config . lxmf_preferred_propagation_node_auto_sync_interval_seconds . set ( value )
2024-09-18 17:42:25 +12:00
2024-09-19 15:59:04 +12:00
if " lxmf_local_propagation_node_enabled " in data :
# update config value
value = bool ( data [ " lxmf_local_propagation_node_enabled " ] )
self . config . lxmf_local_propagation_node_enabled . set ( value )
# enable or disable local propagation node
self . enable_local_propagation_node ( value )
2024-08-02 18:44:41 +12:00
# send config to websocket clients
await self . send_config_to_websocket_clients ( )
2024-07-28 15:36:37 +12:00
2024-09-22 22:58:19 +12:00
# converts nomadnetwork page variables from a string to a map
# converts: "field1=123|field2=456"
# to the following map:
# - var_field1: 123
# - var_field2: 456
def convert_nomadnet_string_data_to_map ( self , path_data : str | None ) :
data = { }
if path_data is not None :
for field in path_data . split ( " | " ) :
if " = " in field :
variable_name , variable_value = field . split ( " = " )
data [ f ' var_ { variable_name } ' ] = variable_value
else :
print ( f " unhandled field: { field } " )
return data
2024-12-07 13:27:20 -05:00
def convert_nomadnet_field_data_to_map ( self , field_data ) :
data = { }
if field_data is not None or " {} " :
try :
json_data = field_data
if isinstance ( json_data , dict ) :
# add the prefixed keys to the result dictionary
data = { f " field_ { key } " : value for key , value in json_data . items ( ) }
else :
return None
except Exception as e :
print ( f " skipping invalid field data: { e } " )
return data
2024-09-22 22:58:19 +12:00
2024-08-02 18:44:41 +12:00
# handle data received from websocket client
async def on_websocket_data_received ( self , client , data ) :
2024-07-28 15:36:37 +12:00
2024-08-02 18:44:41 +12:00
# get type from client data
_type = data [ " type " ]
2024-07-28 22:30:22 +12:00
2024-08-02 18:44:41 +12:00
# handle updating config
if _type == " config.set " :
# get config from websocket
config = data [ " config " ]
# update config
await self . update_config ( config )
2024-04-29 23:16:54 +12:00
2024-05-01 23:36:14 +12:00
# handle downloading a file from a nomadnet node
elif _type == " nomadnet.file.download " :
# get data from websocket client
destination_hash = data [ " nomadnet_file_download " ] [ " destination_hash " ]
file_path = data [ " nomadnet_file_download " ] [ " file_path " ]
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
# handle successful file download
def on_file_download_success ( file_name , file_bytes ) :
2024-05-06 00:37:12 +12:00
asyncio . run ( client . send_str ( json . dumps ( {
2024-05-01 23:36:14 +12:00
" type " : " nomadnet.file.download " ,
" nomadnet_file_download " : {
" status " : " success " ,
" destination_hash " : destination_hash . hex ( ) ,
" file_path " : file_path ,
" file_name " : file_name ,
" file_bytes " : base64 . b64encode ( file_bytes ) . decode ( " utf-8 " ) ,
} ,
} ) ) )
# handle file download failure
def on_file_download_failure ( failure_reason ) :
2024-05-28 04:05:41 +12:00
asyncio . create_task ( client . send_str ( json . dumps ( {
2024-05-01 23:36:14 +12:00
" type " : " nomadnet.file.download " ,
" nomadnet_file_download " : {
2024-05-28 04:05:41 +12:00
" status " : " failure " ,
2024-05-01 23:36:14 +12:00
" failure_reason " : failure_reason ,
" destination_hash " : destination_hash . hex ( ) ,
" file_path " : file_path ,
} ,
} ) ) )
2024-05-02 00:14:13 +12:00
# handle file download progress
def on_file_download_progress ( progress ) :
2024-05-06 00:37:12 +12:00
asyncio . run ( client . send_str ( json . dumps ( {
2024-05-02 00:14:13 +12:00
" type " : " nomadnet.file.download " ,
" nomadnet_file_download " : {
" status " : " progress " ,
" progress " : progress ,
" destination_hash " : destination_hash . hex ( ) ,
" file_path " : file_path ,
} ,
} ) ) )
2024-05-01 23:36:14 +12:00
# todo: handle file download progress
# download the file
2024-05-28 03:56:07 +12:00
downloader = NomadnetFileDownloader ( destination_hash , file_path , on_file_download_success , on_file_download_failure , on_file_download_progress )
await downloader . download ( )
2024-05-01 23:36:14 +12:00
# handle downloading a page from a nomadnet node
elif _type == " nomadnet.page.download " :
# get data from websocket client
destination_hash = data [ " nomadnet_page_download " ] [ " destination_hash " ]
page_path = data [ " nomadnet_page_download " ] [ " page_path " ]
2024-12-07 13:27:20 -05:00
field_data = data [ " nomadnet_page_download " ] [ " field_data " ]
combined_data = { }
2024-09-22 22:58:19 +12:00
# parse data from page path
# example: hash:/page/index.mu`field1=123|field2=456
page_data = None
page_path_to_download = page_path
if " ` " in page_path :
page_path_parts = page_path . split ( " ` " )
page_path_to_download = page_path_parts [ 0 ]
page_data = self . convert_nomadnet_string_data_to_map ( page_path_parts [ 1 ] )
2024-12-07 13:27:20 -05:00
# Field data
field_data = self . convert_nomadnet_field_data_to_map ( field_data )
# Combine page data and field data
if page_data is not None :
combined_data . update ( page_data )
if field_data is not None :
combined_data . update ( field_data )
2024-05-01 23:36:14 +12:00
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
2024-12-07 13:27:20 -05:00
2024-05-01 23:36:14 +12:00
# handle successful page download
def on_page_download_success ( page_content ) :
2024-05-06 00:37:12 +12:00
asyncio . run ( client . send_str ( json . dumps ( {
2024-05-01 23:36:14 +12:00
" type " : " nomadnet.page.download " ,
" nomadnet_page_download " : {
" status " : " success " ,
" destination_hash " : destination_hash . hex ( ) ,
" page_path " : page_path ,
" page_content " : page_content ,
} ,
} ) ) )
# handle page download failure
def on_page_download_failure ( failure_reason ) :
2024-05-28 04:05:41 +12:00
asyncio . create_task ( client . send_str ( json . dumps ( {
2024-05-01 23:36:14 +12:00
" type " : " nomadnet.page.download " ,
" nomadnet_page_download " : {
2024-05-28 04:05:41 +12:00
" status " : " failure " ,
2024-05-01 23:36:14 +12:00
" failure_reason " : failure_reason ,
" destination_hash " : destination_hash . hex ( ) ,
" page_path " : page_path ,
} ,
} ) ) )
2024-05-02 00:14:13 +12:00
# handle page download progress
def on_page_download_progress ( progress ) :
2024-05-06 00:37:12 +12:00
asyncio . run ( client . send_str ( json . dumps ( {
2024-05-02 00:14:13 +12:00
" type " : " nomadnet.page.download " ,
" nomadnet_page_download " : {
" status " : " progress " ,
" progress " : progress ,
" destination_hash " : destination_hash . hex ( ) ,
" page_path " : page_path ,
} ,
} ) ) )
2024-05-01 23:36:14 +12:00
# todo: handle page download progress
# download the page
2024-12-07 13:27:20 -05:00
downloader = NomadnetPageDownloader ( destination_hash , page_path_to_download , combined_data , on_page_download_success , on_page_download_failure , on_page_download_progress )
2024-05-28 03:56:07 +12:00
await downloader . download ( )
2024-05-01 23:36:14 +12:00
2024-04-29 20:26:45 +12:00
# unhandled type
else :
print ( " unhandled client message type: " + _type )
2024-04-29 01:09:06 +12:00
# broadcast provided data to all connected websocket clients
2024-04-29 23:43:28 +12:00
async def websocket_broadcast ( self , data ) :
2024-04-29 20:26:45 +12:00
for websocket_client in self . websocket_clients :
2024-05-05 01:11:41 +12:00
try :
await websocket_client . send_str ( data )
except :
# do nothing if failed to broadcast to a specific websocket client
pass
2024-04-29 22:28:34 +12:00
# broadcasts config to all websocket clients
2024-04-29 23:43:28 +12:00
async def send_config_to_websocket_clients ( self ) :
await self . websocket_broadcast ( json . dumps ( {
2024-04-29 22:28:34 +12:00
" type " : " config " ,
2024-05-21 00:09:03 +12:00
" config " : self . get_config_dict ( ) ,
2024-04-29 22:28:34 +12:00
} ) )
2024-04-29 01:09:06 +12:00
2024-06-02 12:07:59 +12:00
# broadcasts to all websocket clients that we just announced
async def send_announced_to_websocket_clients ( self ) :
await self . websocket_broadcast ( json . dumps ( {
" type " : " announced " ,
} ) )
2024-05-21 00:09:03 +12:00
# returns a dictionary of config
def get_config_dict ( self ) :
return {
" display_name " : self . config . display_name . get ( ) ,
" identity_hash " : self . identity . hexhash ,
" lxmf_address_hash " : self . local_lxmf_destination . hexhash ,
" audio_call_address_hash " : self . audio_call_manager . audio_call_receiver . destination . hexhash ,
2024-06-02 11:50:58 +12:00
" auto_announce_enabled " : self . config . auto_announce_enabled . get ( ) ,
" auto_announce_interval_seconds " : self . config . auto_announce_interval_seconds . get ( ) ,
" last_announced_at " : self . config . last_announced_at . get ( ) ,
2024-12-16 12:17:11 +13:00
" theme " : self . config . theme . get ( ) ,
2024-07-28 15:36:37 +12:00
" auto_resend_failed_messages_when_announce_received " : self . config . auto_resend_failed_messages_when_announce_received . get ( ) ,
" allow_auto_resending_failed_messages_with_attachments " : self . config . allow_auto_resending_failed_messages_with_attachments . get ( ) ,
2024-09-18 21:27:32 +12:00
" auto_send_failed_messages_to_propagation_node " : self . config . auto_send_failed_messages_to_propagation_node . get ( ) ,
2024-07-28 22:30:22 +12:00
" show_suggested_community_interfaces " : self . config . show_suggested_community_interfaces . get ( ) ,
2024-09-19 15:59:04 +12:00
" lxmf_local_propagation_node_enabled " : self . config . lxmf_local_propagation_node_enabled . get ( ) ,
" lxmf_local_propagation_node_address_hash " : self . message_router . propagation_destination . hexhash ,
2024-09-18 17:42:25 +12:00
" lxmf_preferred_propagation_node_destination_hash " : self . config . lxmf_preferred_propagation_node_destination_hash . get ( ) ,
2024-09-18 21:06:03 +12:00
" lxmf_preferred_propagation_node_auto_sync_interval_seconds " : self . config . lxmf_preferred_propagation_node_auto_sync_interval_seconds . get ( ) ,
" lxmf_preferred_propagation_node_last_synced_at " : self . config . lxmf_preferred_propagation_node_last_synced_at . get ( ) ,
2024-05-21 00:09:03 +12:00
}
2024-05-21 02:42:03 +12:00
# convert audio call to dict
def convert_audio_call_to_dict ( self , audio_call : AudioCall ) :
# get remote identity hash
remote_identity_hash = None
remote_identity = audio_call . get_remote_identity ( )
if remote_identity is not None :
remote_identity_hash = remote_identity . hash . hex ( )
# get remote destination hash
# we need to know the remote identity to determine their destination hash
remote_destination_hash = None
2024-05-23 22:42:42 +12:00
remote_destination_hash_hex = None
2024-05-21 02:42:03 +12:00
if remote_identity is not None :
remote_destination_hash = RNS . Destination . hash ( remote_identity , " call " , " audio " )
2024-05-23 22:42:42 +12:00
remote_destination_hash_hex = remote_destination_hash . hex ( )
2024-05-21 02:42:03 +12:00
# determine path to remote destination
path = None
if remote_destination_hash is not None :
# determine next hop and hop count
hops = RNS . Transport . hops_to ( remote_destination_hash )
next_hop_bytes = self . reticulum . get_next_hop ( remote_destination_hash )
# ensure next hop provided
if next_hop_bytes is not None :
next_hop = next_hop_bytes . hex ( )
next_hop_interface = self . reticulum . get_next_hop_if_name ( remote_destination_hash )
path = {
" hops " : hops ,
" next_hop " : next_hop ,
" next_hop_interface " : next_hop_interface ,
}
return {
" hash " : audio_call . link . hash . hex ( ) ,
2024-05-23 22:42:42 +12:00
" remote_destination_hash " : remote_destination_hash_hex ,
2024-05-21 02:42:03 +12:00
" remote_identity_hash " : remote_identity_hash ,
" is_active " : audio_call . is_active ( ) ,
" is_outbound " : audio_call . is_outbound ,
" path " : path ,
}
2024-04-30 03:15:42 +12:00
# convert an lxmf message to a dictionary, for sending over websocket
def convert_lxmf_message_to_dict ( self , lxmf_message : LXMF . LXMessage ) :
# handle fields
fields = { }
message_fields = lxmf_message . get_fields ( )
for field_type in message_fields :
value = message_fields [ field_type ]
# handle file attachments field
if field_type == LXMF . FIELD_FILE_ATTACHMENTS :
# process file attachments
file_attachments = [ ]
for file_attachment in value :
file_name = file_attachment [ 0 ]
file_bytes = base64 . b64encode ( file_attachment [ 1 ] ) . decode ( " utf-8 " )
file_attachments . append ( {
" file_name " : file_name ,
" file_bytes " : file_bytes ,
} )
# add to fields
fields [ " file_attachments " ] = file_attachments
# handle image field
if field_type == LXMF . FIELD_IMAGE :
image_type = value [ 0 ]
image_bytes = base64 . b64encode ( value [ 1 ] ) . decode ( " utf-8 " )
fields [ " image " ] = {
" image_type " : image_type ,
" image_bytes " : image_bytes ,
}
2024-06-02 22:44:23 +12:00
# handle audio field
if field_type == LXMF . FIELD_AUDIO :
audio_mode = value [ 0 ]
audio_bytes = base64 . b64encode ( value [ 1 ] ) . decode ( " utf-8 " )
fields [ " audio " ] = {
" audio_mode " : audio_mode ,
" audio_bytes " : audio_bytes ,
}
2024-05-03 23:48:15 +12:00
# convert 0.0-1.0 progress to 0.00-100 percentage
progress_percentage = round ( lxmf_message . progress * 100 , 2 )
2024-08-10 15:41:13 +12:00
# get rssi
rssi = lxmf_message . rssi
if rssi is None :
rssi = self . reticulum . get_packet_rssi ( lxmf_message . hash )
# get snr
snr = lxmf_message . snr
if snr is None :
snr = self . reticulum . get_packet_snr ( lxmf_message . hash )
# get quality
quality = lxmf_message . q
if quality is None :
quality = self . reticulum . get_packet_q ( lxmf_message . hash )
2024-04-30 03:15:42 +12:00
return {
" hash " : lxmf_message . hash . hex ( ) ,
" source_hash " : lxmf_message . source_hash . hex ( ) ,
" destination_hash " : lxmf_message . destination_hash . hex ( ) ,
2024-05-04 04:58:15 +12:00
" is_incoming " : lxmf_message . incoming ,
2024-04-30 03:15:42 +12:00
" state " : self . convert_lxmf_state_to_string ( lxmf_message ) ,
2024-05-03 23:48:15 +12:00
" progress " : progress_percentage ,
2024-09-18 01:26:40 +12:00
" method " : self . convert_lxmf_method_to_string ( lxmf_message ) ,
2024-07-08 16:22:24 +12:00
" delivery_attempts " : lxmf_message . delivery_attempts ,
" next_delivery_attempt_at " : getattr ( lxmf_message , " next_delivery_attempt " , None ) , # attribute may not exist yet
2024-05-04 05:04:37 +12:00
" title " : lxmf_message . title . decode ( ' utf-8 ' ) ,
2024-04-30 03:15:42 +12:00
" content " : lxmf_message . content . decode ( ' utf-8 ' ) ,
" fields " : fields ,
2024-05-04 05:14:48 +12:00
" timestamp " : lxmf_message . timestamp ,
2024-08-10 15:41:13 +12:00
" rssi " : rssi ,
" snr " : snr ,
" quality " : quality ,
2024-04-30 03:15:42 +12:00
}
# convert lxmf state to a human friendly string
def convert_lxmf_state_to_string ( self , lxmf_message : LXMF . LXMessage ) :
# convert state to string
lxmf_message_state = " unknown "
2024-09-11 20:49:51 +12:00
if lxmf_message . state == LXMF . LXMessage . GENERATING :
lxmf_message_state = " generating "
2024-04-30 03:15:42 +12:00
elif lxmf_message . state == LXMF . LXMessage . OUTBOUND :
lxmf_message_state = " outbound "
elif lxmf_message . state == LXMF . LXMessage . SENDING :
lxmf_message_state = " sending "
elif lxmf_message . state == LXMF . LXMessage . SENT :
lxmf_message_state = " sent "
elif lxmf_message . state == LXMF . LXMessage . DELIVERED :
lxmf_message_state = " delivered "
elif lxmf_message . state == LXMF . LXMessage . FAILED :
lxmf_message_state = " failed "
2024-08-02 18:44:41 +12:00
2024-04-30 03:15:42 +12:00
return lxmf_message_state
2024-09-18 01:26:40 +12:00
# convert lxmf method to a human friendly string
def convert_lxmf_method_to_string ( self , lxmf_message : LXMF . LXMessage ) :
# convert method to string
lxmf_message_method = " unknown "
if lxmf_message . method == LXMF . LXMessage . OPPORTUNISTIC :
lxmf_message_method = " opportunistic "
elif lxmf_message . method == LXMF . LXMessage . DIRECT :
lxmf_message_method = " direct "
elif lxmf_message . method == LXMF . LXMessage . PROPAGATED :
lxmf_message_method = " propagated "
elif lxmf_message . method == LXMF . LXMessage . PAPER :
lxmf_message_method = " paper "
return lxmf_message_method
2024-09-18 18:19:02 +12:00
def convert_propagation_node_state_to_string ( self , state ) :
# map states to strings
state_map = {
LXMRouter . PR_IDLE : " idle " ,
LXMRouter . PR_PATH_REQUESTED : " path_requested " ,
LXMRouter . PR_LINK_ESTABLISHING : " link_establishing " ,
LXMRouter . PR_LINK_ESTABLISHED : " link_established " ,
LXMRouter . PR_REQUEST_SENT : " request_sent " ,
LXMRouter . PR_RECEIVING : " receiving " ,
LXMRouter . PR_RESPONSE_RECEIVED : " response_received " ,
LXMRouter . PR_COMPLETE : " complete " ,
LXMRouter . PR_NO_PATH : " no_path " ,
LXMRouter . PR_LINK_FAILED : " link_failed " ,
LXMRouter . PR_TRANSFER_FAILED : " transfer_failed " ,
LXMRouter . PR_NO_IDENTITY_RCVD : " no_identity_received " ,
LXMRouter . PR_NO_ACCESS : " no_access " ,
LXMRouter . PR_FAILED : " failed " ,
}
# return string for state, or fallback to unknown
if state in state_map :
return state_map [ state ]
else :
return " unknown "
2024-05-05 23:51:15 +12:00
# convert database announce to a dictionary
def convert_db_announce_to_dict ( self , announce : database . Announce ) :
2024-09-11 21:23:26 +12:00
2024-09-17 16:55:49 +12:00
# parse display name from announce
2024-09-11 21:23:26 +12:00
display_name = None
if announce . aspect == " lxmf.delivery " :
display_name = self . parse_lxmf_display_name ( announce . app_data )
2024-09-17 16:55:49 +12:00
elif announce . aspect == " nomadnetwork.node " :
display_name = self . parse_nomadnetwork_node_display_name ( announce . app_data )
2024-09-11 21:23:26 +12:00
2024-05-05 23:51:15 +12:00
return {
" id " : announce . id ,
" destination_hash " : announce . destination_hash ,
" aspect " : announce . aspect ,
" identity_hash " : announce . identity_hash ,
" identity_public_key " : announce . identity_public_key ,
" app_data " : announce . app_data ,
2024-09-11 21:23:26 +12:00
" display_name " : display_name ,
2024-09-19 23:46:08 +12:00
" custom_display_name " : self . get_custom_destination_display_name ( announce . destination_hash ) ,
2024-05-05 23:51:15 +12:00
" created_at " : announce . created_at ,
" updated_at " : announce . updated_at ,
}
2024-09-18 22:00:59 +12:00
# convert database lxmf message to a dictionary
def convert_db_lxmf_message_to_dict ( self , db_lxmf_message : database . LxmfMessage ) :
return {
" id " : db_lxmf_message . id ,
" hash " : db_lxmf_message . hash ,
" source_hash " : db_lxmf_message . source_hash ,
" destination_hash " : db_lxmf_message . destination_hash ,
" is_incoming " : db_lxmf_message . is_incoming ,
" state " : db_lxmf_message . state ,
" progress " : db_lxmf_message . progress ,
" method " : db_lxmf_message . method ,
" delivery_attempts " : db_lxmf_message . delivery_attempts ,
" next_delivery_attempt_at " : db_lxmf_message . next_delivery_attempt_at ,
" title " : db_lxmf_message . title ,
" content " : db_lxmf_message . content ,
" fields " : json . loads ( db_lxmf_message . fields ) ,
" timestamp " : db_lxmf_message . timestamp ,
" rssi " : db_lxmf_message . rssi ,
" snr " : db_lxmf_message . snr ,
" quality " : db_lxmf_message . quality ,
" created_at " : db_lxmf_message . created_at ,
" updated_at " : db_lxmf_message . updated_at ,
}
2024-12-04 01:19:47 +13:00
# updates the lxmf user icon for the provided destination hash
def update_lxmf_user_icon ( self , destination_hash : str , icon_name : str , foreground_colour : str , background_colour : str ) :
# log
print ( f " updating lxmf user icon for { destination_hash } to icon_name= { icon_name } , foreground_colour= { foreground_colour } , background_colour= { background_colour } " )
# prepare data to insert or update
data = {
" destination_hash " : destination_hash ,
" icon_name " : icon_name ,
" foreground_colour " : foreground_colour ,
" background_colour " : background_colour ,
" updated_at " : datetime . now ( timezone . utc ) ,
}
# upsert to database
query = database . LxmfUserIcon . insert ( data )
query = query . on_conflict ( conflict_target = [ database . LxmfUserIcon . destination_hash ] , update = data )
query . execute ( )
2024-04-29 20:26:45 +12:00
# handle an lxmf delivery from reticulum
2024-04-29 23:43:28 +12:00
# NOTE: cant be async, as Reticulum doesn't await it
2024-08-10 15:41:13 +12:00
def on_lxmf_delivery ( self , lxmf_message : LXMF . LXMessage ) :
2024-04-29 20:26:45 +12:00
try :
2024-04-29 01:09:06 +12:00
2024-05-04 21:41:55 +12:00
# upsert lxmf message to database
self . db_upsert_lxmf_message ( lxmf_message )
2024-05-03 23:48:15 +12:00
2024-12-04 01:26:18 +13:00
# update lxmf user icon if icon appearance field is available
2024-12-04 01:19:47 +13:00
try :
message_fields = lxmf_message . get_fields ( )
if LXMF . FIELD_ICON_APPEARANCE in message_fields :
icon_appearance = message_fields [ LXMF . FIELD_ICON_APPEARANCE ]
icon_name = icon_appearance [ 0 ]
foreground_colour = " # " + icon_appearance [ 1 ] . hex ( )
background_colour = " # " + icon_appearance [ 2 ] . hex ( )
self . update_lxmf_user_icon ( lxmf_message . source_hash . hex ( ) , icon_name , foreground_colour , background_colour )
except Exception as e :
print ( " failed to update lxmf user icon from lxmf message " )
print ( e )
pass
2024-09-18 22:00:59 +12:00
# find message from database
db_lxmf_message = database . LxmfMessage . get_or_none ( database . LxmfMessage . hash == lxmf_message . hash . hex ( ) )
if db_lxmf_message is None :
return
2024-04-29 20:26:45 +12:00
# send received lxmf message data to all websocket clients
2024-04-29 23:43:28 +12:00
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
2024-04-29 20:26:45 +12:00
" type " : " lxmf.delivery " ,
2024-09-18 22:00:59 +12:00
" lxmf_message " : self . convert_db_lxmf_message_to_dict ( db_lxmf_message ) ,
2024-04-29 23:43:28 +12:00
} ) ) )
2024-04-29 01:09:06 +12:00
2024-04-29 20:26:45 +12:00
except Exception as e :
# do nothing on error
print ( " lxmf_delivery error: {} " . format ( e ) )
2024-04-29 01:09:06 +12:00
2024-04-30 03:15:42 +12:00
# handle delivery status update for an outbound lxmf message
def on_lxmf_sending_state_updated ( self , lxmf_message ) :
2024-05-04 00:30:09 +12:00
# upsert lxmf message to database
self . db_upsert_lxmf_message ( lxmf_message )
2024-04-30 03:15:42 +12:00
# send lxmf message state to all websocket clients
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
" type " : " lxmf_message_state_updated " ,
" lxmf_message " : self . convert_lxmf_message_to_dict ( lxmf_message ) ,
} ) ) )
# handle delivery failed for an outbound lxmf message
def on_lxmf_sending_failed ( self , lxmf_message ) :
2024-09-18 01:26:40 +12:00
# check if this failed message should fall back to sending via a propagation node
if lxmf_message . state == LXMF . LXMessage . FAILED and hasattr ( lxmf_message , " try_propagation_on_fail " ) and lxmf_message . try_propagation_on_fail :
self . send_failed_message_via_propagation_node ( lxmf_message )
# update state
2024-04-30 03:15:42 +12:00
self . on_lxmf_sending_state_updated ( lxmf_message )
2024-09-18 01:26:40 +12:00
# sends a previously failed message via a propagation node
def send_failed_message_via_propagation_node ( self , lxmf_message : LXMF . LXMessage ) :
# reset internal message state
lxmf_message . packed = None
lxmf_message . delivery_attempts = 0
if hasattr ( lxmf_message , " next_delivery_attempt " ) :
del lxmf_message . next_delivery_attempt
# this message should now be sent via a propagation node
lxmf_message . desired_method = LXMF . LXMessage . PROPAGATED
lxmf_message . try_propagation_on_fail = False
# resend message
self . message_router . handle_outbound ( lxmf_message )
2024-05-04 00:30:09 +12:00
# upserts the provided lxmf message to the database
def db_upsert_lxmf_message ( self , lxmf_message : LXMF . LXMessage ) :
# convert lxmf message to dict
lxmf_message_dict = self . convert_lxmf_message_to_dict ( lxmf_message )
# prepare data to insert or update
data = {
" hash " : lxmf_message_dict [ " hash " ] ,
" source_hash " : lxmf_message_dict [ " source_hash " ] ,
" destination_hash " : lxmf_message_dict [ " destination_hash " ] ,
2024-05-04 04:58:15 +12:00
" is_incoming " : lxmf_message_dict [ " is_incoming " ] ,
2024-05-04 00:30:09 +12:00
" state " : lxmf_message_dict [ " state " ] ,
" progress " : lxmf_message_dict [ " progress " ] ,
2024-09-18 01:26:40 +12:00
" method " : lxmf_message_dict [ " method " ] ,
2024-07-08 16:22:24 +12:00
" delivery_attempts " : lxmf_message_dict [ " delivery_attempts " ] ,
" next_delivery_attempt_at " : lxmf_message_dict [ " next_delivery_attempt_at " ] ,
2024-05-04 05:04:37 +12:00
" title " : lxmf_message_dict [ " title " ] ,
2024-05-04 00:30:09 +12:00
" content " : lxmf_message_dict [ " content " ] ,
" fields " : json . dumps ( lxmf_message_dict [ " fields " ] ) ,
2024-05-04 05:14:48 +12:00
" timestamp " : lxmf_message_dict [ " timestamp " ] ,
2024-08-10 15:41:13 +12:00
" rssi " : lxmf_message_dict [ " rssi " ] ,
" snr " : lxmf_message_dict [ " snr " ] ,
" quality " : lxmf_message_dict [ " quality " ] ,
2024-05-05 21:49:27 +12:00
" updated_at " : datetime . now ( timezone . utc ) ,
2024-05-04 00:30:09 +12:00
}
# upsert to database
query = database . LxmfMessage . insert ( data )
query = query . on_conflict ( conflict_target = [ database . LxmfMessage . hash ] , update = data )
query . execute ( )
2024-05-05 21:11:22 +12:00
# upserts the provided announce to the database
def db_upsert_announce ( self , identity : RNS . Identity , destination_hash : bytes , aspect : str , app_data : bytes ) :
# prepare data to insert or update
data = {
" destination_hash " : destination_hash . hex ( ) ,
" aspect " : aspect ,
" identity_hash " : identity . hash . hex ( ) ,
" identity_public_key " : base64 . b64encode ( identity . get_public_key ( ) ) . decode ( " utf-8 " ) ,
2024-05-05 21:49:27 +12:00
" updated_at " : datetime . now ( timezone . utc ) ,
2024-05-05 21:11:22 +12:00
}
2024-08-13 23:25:39 +12:00
# only set app data if provided, as we don't want to wipe existing data when we request keys from the network
if app_data is not None :
2024-08-13 23:28:43 +12:00
# save app data as base64 string
2024-08-13 23:25:39 +12:00
data [ " app_data " ] = base64 . b64encode ( app_data ) . decode ( " utf-8 " )
2024-05-05 21:11:22 +12:00
# upsert to database
query = database . Announce . insert ( data )
query = query . on_conflict ( conflict_target = [ database . Announce . destination_hash ] , update = data )
query . execute ( )
2024-09-19 23:46:08 +12:00
# upserts a custom destination display name to the database
def db_upsert_custom_destination_display_name ( self , destination_hash : str , display_name : str ) :
# prepare data to insert or update
data = {
" destination_hash " : destination_hash ,
" display_name " : display_name ,
" updated_at " : datetime . now ( timezone . utc ) ,
}
# upsert to database
query = database . CustomDestinationDisplayName . insert ( data )
query = query . on_conflict ( conflict_target = [ database . CustomDestinationDisplayName . destination_hash ] , update = data )
query . execute ( )
2024-06-01 19:21:42 +12:00
# upserts lxmf conversation read state to the database
def db_mark_lxmf_conversation_as_read ( self , destination_hash : str ) :
# prepare data to insert or update
data = {
" destination_hash " : destination_hash ,
" last_read_at " : datetime . now ( timezone . utc ) ,
" updated_at " : datetime . now ( timezone . utc ) ,
}
# upsert to database
query = database . LxmfConversationReadState . insert ( data )
query = query . on_conflict ( conflict_target = [ database . LxmfConversationReadState . destination_hash ] , update = data )
query . execute ( )
2024-04-29 20:26:45 +12:00
# handle sending an lxmf message to reticulum
2024-07-28 14:28:11 +12:00
async def send_message ( self , destination_hash : str , content : str ,
2024-05-04 22:45:16 +12:00
image_field : LxmfImageField = None ,
2024-06-02 22:44:23 +12:00
audio_field : LxmfAudioField = None ,
2024-11-04 18:37:41 +13:00
file_attachments_field : LxmfFileAttachmentsField = None ,
delivery_method : str = None ) - > LXMF . LXMessage :
2024-04-29 01:09:06 +12:00
2024-07-08 23:54:51 +12:00
# convert destination hash to bytes
destination_hash = bytes . fromhex ( destination_hash )
2024-04-29 01:09:06 +12:00
2024-07-12 20:53:13 +12:00
# determine when to timeout finding path
timeout_after_seconds = time . time ( ) + 10
# check if we have a path to the destination
if not RNS . Transport . has_path ( destination_hash ) :
# we don't have a path, so we need to request it
RNS . Transport . request_path ( destination_hash )
# wait until we have a path, or give up after the configured timeout
while not RNS . Transport . has_path ( destination_hash ) and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
2024-07-08 23:54:51 +12:00
# find destination identity from hash
destination_identity = RNS . Identity . recall ( destination_hash )
if destination_identity is None :
2024-04-29 01:09:06 +12:00
2024-07-12 20:53:13 +12:00
# we have to bail out of sending, since we don't have the identity/path yet
2024-07-18 20:40:24 +12:00
raise Exception ( " Could not find path to destination. Try again later. " )
2024-04-29 01:09:06 +12:00
2024-07-08 23:54:51 +12:00
# create destination for recipients lxmf delivery address
lxmf_destination = RNS . Destination ( destination_identity , RNS . Destination . OUT , RNS . Destination . SINGLE , " lxmf " , " delivery " )
2024-04-29 01:09:06 +12:00
2024-11-04 18:37:41 +13:00
# determine how the user wants to send the message
desired_delivery_method = None
if delivery_method == " direct " :
desired_delivery_method = LXMF . LXMessage . DIRECT
elif delivery_method == " opportunistic " :
2024-09-20 18:00:49 +12:00
desired_delivery_method = LXMF . LXMessage . OPPORTUNISTIC
2024-11-04 18:37:41 +13:00
elif delivery_method == " propagated " :
desired_delivery_method = LXMF . LXMessage . PROPAGATED
# determine how to send the message if the user didn't provide a method
if desired_delivery_method is None :
# send messages over a direct link by default
desired_delivery_method = LXMF . LXMessage . DIRECT
if not self . message_router . delivery_link_available ( destination_hash ) and RNS . Identity . current_ratchet_id ( destination_hash ) != None :
# since there's no link established to the destination, it's faster to send opportunistically
# this is because it takes several packets to establish a link, and then we still have to send the message over it
# oppotunistic mode will send the message in a single packet (if the message is small enough, otherwise it falls back to a direct link)
# we will only do this if an encryption ratchet is available, so single packet delivery is more secure
desired_delivery_method = LXMF . LXMessage . OPPORTUNISTIC
2024-09-20 18:00:49 +12:00
2024-07-08 23:54:51 +12:00
# create lxmf message
2024-09-20 18:00:49 +12:00
lxmf_message = LXMF . LXMessage ( lxmf_destination , self . local_lxmf_destination , content , desired_method = desired_delivery_method )
2024-09-18 21:27:32 +12:00
lxmf_message . try_propagation_on_fail = self . config . auto_send_failed_messages_to_propagation_node . get ( )
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
lxmf_message . fields = { }
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
# add file attachments field
if file_attachments_field is not None :
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
# create array of [[file_name, file_bytes], [file_name, file_bytes], ...]
file_attachments = [ ]
for file_attachment in file_attachments_field . file_attachments :
file_attachments . append ( [ file_attachment . file_name , file_attachment . file_bytes ] )
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
# set field attachments field
lxmf_message . fields [ LXMF . FIELD_FILE_ATTACHMENTS ] = file_attachments
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
# add image field
if image_field is not None :
lxmf_message . fields [ LXMF . FIELD_IMAGE ] = [
image_field . image_type ,
image_field . image_bytes ,
]
2024-05-04 22:45:16 +12:00
2024-07-08 23:54:51 +12:00
# add audio field
if audio_field is not None :
lxmf_message . fields [ LXMF . FIELD_AUDIO ] = [
audio_field . audio_mode ,
audio_field . audio_bytes ,
]
2024-06-02 22:44:23 +12:00
2024-07-08 23:54:51 +12:00
# register delivery callbacks
lxmf_message . register_delivery_callback ( self . on_lxmf_sending_state_updated )
lxmf_message . register_failed_callback ( self . on_lxmf_sending_failed )
2024-04-29 01:09:06 +12:00
2024-07-08 23:54:51 +12:00
# send lxmf message to be routed to destination
self . message_router . handle_outbound ( lxmf_message )
2024-05-04 00:30:09 +12:00
2024-07-08 23:54:51 +12:00
# upsert lxmf message to database
self . db_upsert_lxmf_message ( lxmf_message )
2024-04-29 01:09:06 +12:00
2024-07-28 14:28:11 +12:00
# tell all websocket clients that old failed message was deleted so it can remove from ui
await self . websocket_broadcast ( json . dumps ( {
" type " : " lxmf_message_created " ,
" lxmf_message " : self . convert_lxmf_message_to_dict ( lxmf_message ) ,
} ) )
2024-07-08 23:54:51 +12:00
# handle lxmf message progress loop without blocking or awaiting
# otherwise other incoming websocket packets will not be processed until sending is complete
# which results in the next message not showing up until the first message is finished
asyncio . create_task ( self . handle_lxmf_message_progress ( lxmf_message ) )
2024-05-05 01:11:41 +12:00
2024-07-08 23:54:51 +12:00
return lxmf_message
2024-04-29 01:09:06 +12:00
2024-05-19 20:22:34 +12:00
# updates lxmf message in database and broadcasts to websocket until it's delivered, or it fails
async def handle_lxmf_message_progress ( self , lxmf_message ) :
2024-09-19 13:54:01 +12:00
# FIXME: there's no register_progress_callback on the lxmf message, so manually send progress until delivered, propagated or failed
2024-05-19 20:22:34 +12:00
# we also can't use on_lxmf_sending_state_updated method to do this, because of async/await issues...
2024-09-19 13:54:01 +12:00
should_update_message = True
while should_update_message :
2024-05-19 20:22:34 +12:00
# wait 1 second between sending updates
await asyncio . sleep ( 1 )
# upsert lxmf message to database (as we want to update the progress in database too)
self . db_upsert_lxmf_message ( lxmf_message )
# send update to websocket clients
await self . websocket_broadcast ( json . dumps ( {
" type " : " lxmf_message_state_updated " ,
" lxmf_message " : self . convert_lxmf_message_to_dict ( lxmf_message ) ,
} ) )
2024-09-19 13:54:01 +12:00
# check message state
has_delivered = lxmf_message . state == LXMF . LXMessage . DELIVERED
has_propagated = lxmf_message . state == LXMF . LXMessage . SENT and lxmf_message . method == LXMF . LXMessage . PROPAGATED
has_failed = lxmf_message . state == LXMF . LXMessage . FAILED
# check if we should stop updating
if has_delivered or has_propagated or has_failed :
should_update_message = False
2024-09-17 16:29:53 +12:00
# handle an announce received from reticulum, for an audio call address
# NOTE: cant be async, as Reticulum doesn't await it
def on_audio_call_announce_received ( self , aspect , destination_hash , announced_identity , app_data ) :
# log received announce
print ( " Received an announce from " + RNS . prettyhexrep ( destination_hash ) + " for [call.audio] " )
# upsert announce to database
self . db_upsert_announce ( announced_identity , destination_hash , aspect , app_data )
# find announce from database
announce = database . Announce . get_or_none ( database . Announce . destination_hash == destination_hash . hex ( ) )
if announce is None :
return
# send database announce to all websocket clients
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
" type " : " announce " ,
" announce " : self . convert_db_announce_to_dict ( announce ) ,
} ) ) )
2024-05-19 20:22:34 +12:00
2024-04-29 20:26:45 +12:00
# handle an announce received from reticulum, for an lxmf address
2024-04-29 23:43:28 +12:00
# NOTE: cant be async, as Reticulum doesn't await it
2024-09-17 15:40:41 +12:00
def on_lxmf_announce_received ( self , aspect , destination_hash , announced_identity , app_data ) :
2024-04-29 17:12:03 +12:00
2024-04-29 20:26:45 +12:00
# log received announce
2024-05-05 21:11:22 +12:00
print ( " Received an announce from " + RNS . prettyhexrep ( destination_hash ) + " for [lxmf.delivery] " )
# upsert announce to database
2024-09-17 15:40:41 +12:00
self . db_upsert_announce ( announced_identity , destination_hash , aspect , app_data )
2024-04-29 17:12:03 +12:00
2024-05-05 23:51:15 +12:00
# find announce from database
announce = database . Announce . get_or_none ( database . Announce . destination_hash == destination_hash . hex ( ) )
if announce is None :
return
2024-04-29 17:12:03 +12:00
2024-05-05 23:51:15 +12:00
# send database announce to all websocket clients
2024-04-29 23:43:28 +12:00
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
2024-04-29 20:26:45 +12:00
" type " : " announce " ,
2024-05-05 23:51:15 +12:00
" announce " : self . convert_db_announce_to_dict ( announce ) ,
2024-04-29 23:43:28 +12:00
} ) ) )
2024-04-29 17:12:03 +12:00
2024-07-28 14:28:11 +12:00
# resend all failed messages that were intended for this destination
if self . config . auto_resend_failed_messages_when_announce_received . get ( ) :
asyncio . run ( self . resend_failed_messages_for_destination ( destination_hash . hex ( ) ) )
2024-09-17 15:24:31 +12:00
# handle an announce received from reticulum, for an lxmf propagation node address
# NOTE: cant be async, as Reticulum doesn't await it
2024-09-17 15:40:41 +12:00
def on_lxmf_propagation_announce_received ( self , aspect , destination_hash , announced_identity , app_data ) :
2024-09-17 15:24:31 +12:00
# log received announce
print ( " Received an announce from " + RNS . prettyhexrep ( destination_hash ) + " for [lxmf.propagation] " )
# upsert announce to database
2024-09-17 15:40:41 +12:00
self . db_upsert_announce ( announced_identity , destination_hash , aspect , app_data )
2024-09-17 15:24:31 +12:00
# find announce from database
announce = database . Announce . get_or_none ( database . Announce . destination_hash == destination_hash . hex ( ) )
if announce is None :
return
# send database announce to all websocket clients
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
" type " : " announce " ,
" announce " : self . convert_db_announce_to_dict ( announce ) ,
} ) ) )
2024-07-28 14:28:11 +12:00
# resends all messages that previously failed to send to the provided destination hash
async def resend_failed_messages_for_destination ( self , destination_hash : str ) :
# get messages that failed to send to this destination
failed_messages = ( database . LxmfMessage . select ( )
. where ( database . LxmfMessage . state == " failed " )
. where ( database . LxmfMessage . destination_hash == destination_hash )
. order_by ( database . LxmfMessage . id . asc ( ) ) )
# resend failed messages
for failed_message in failed_messages :
try :
# parse fields as json
fields = json . loads ( failed_message . fields )
# parse image field
image_field = None
if " image " in fields :
image_field = LxmfImageField ( fields [ " image " ] [ " image_type " ] , base64 . b64decode ( fields [ " image " ] [ " image_bytes " ] ) )
# parse audio field
audio_field = None
if " audio " in fields :
audio_field = LxmfAudioField ( fields [ " audio " ] [ " audio_mode " ] , base64 . b64decode ( fields [ " audio " ] [ " audio_bytes " ] ) )
# parse file attachments field
file_attachments_field = None
if " file_attachments " in fields :
file_attachments = [ ]
for file_attachment in fields [ " file_attachments " ] :
file_attachments . append ( LxmfFileAttachment ( file_attachment [ " file_name " ] , base64 . b64decode ( file_attachment [ " file_bytes " ] ) ) )
file_attachments_field = LxmfFileAttachmentsField ( file_attachments )
2024-07-28 14:53:03 +12:00
# don't resend message with attachments if not allowed
if not self . config . allow_auto_resending_failed_messages_with_attachments . get ( ) :
if image_field is not None or audio_field is not None or file_attachments_field is not None :
print ( " Not resending failed message with attachments, as setting is disabled " )
continue
2024-07-28 14:28:11 +12:00
# send new message with failed message content
await self . send_message (
failed_message . destination_hash ,
failed_message . content ,
image_field ,
audio_field ,
file_attachments_field ,
)
# remove original failed message from database
database . LxmfMessage . delete ( ) . where ( ( database . LxmfMessage . hash == failed_message . hash ) ) . execute ( )
# tell all websocket clients that old failed message was deleted so it can remove from ui
await self . websocket_broadcast ( json . dumps ( {
" type " : " lxmf_message_deleted " ,
" hash " : failed_message . hash ,
} ) )
except Exception as e :
print ( " Error resending failed message: " + str ( e ) )
pass
2024-05-05 21:11:22 +12:00
# handle an announce received from reticulum, for a nomadnet node
# NOTE: cant be async, as Reticulum doesn't await it
2024-09-17 15:40:41 +12:00
def on_nomadnet_node_announce_received ( self , aspect , destination_hash , announced_identity , app_data ) :
2024-05-05 21:11:22 +12:00
# log received announce
print ( " Received an announce from " + RNS . prettyhexrep ( destination_hash ) + " for [nomadnetwork.node] " )
# upsert announce to database
2024-09-17 15:40:41 +12:00
self . db_upsert_announce ( announced_identity , destination_hash , aspect , app_data )
2024-05-05 21:11:22 +12:00
2024-05-05 23:51:15 +12:00
# find announce from database
announce = database . Announce . get_or_none ( database . Announce . destination_hash == destination_hash . hex ( ) )
if announce is None :
return
# send database announce to all websocket clients
asyncio . run ( self . websocket_broadcast ( json . dumps ( {
" type " : " announce " ,
" announce " : self . convert_db_announce_to_dict ( announce ) ,
} ) ) )
2024-09-19 23:46:08 +12:00
# gets the custom display name a user has set for the provided destination hash
def get_custom_destination_display_name ( self , destination_hash : str ) :
# get display name from database
db_destination_display_name = database . CustomDestinationDisplayName . get_or_none ( database . CustomDestinationDisplayName . destination_hash == destination_hash )
if db_destination_display_name is not None :
return db_destination_display_name . display_name
return None
2024-06-01 17:56:22 +12:00
# get name to show for an lxmf conversation
# currently, this will use the app data from the most recent announce
# TODO: we should fetch this from our contacts database, when it gets implemented, and if not found, fallback to app data
def get_lxmf_conversation_name ( self , destination_hash ) :
# get lxmf.delivery announce from database for the provided destination hash
lxmf_announce = ( database . Announce . select ( )
. where ( database . Announce . aspect == " lxmf.delivery " )
. where ( database . Announce . destination_hash == destination_hash )
. get_or_none ( ) )
# if app data is available in database, it should be base64 encoded text that was announced
2024-09-11 21:23:26 +12:00
# we will return the parsed lxmf display name as the conversation name
2024-06-01 17:56:22 +12:00
if lxmf_announce is not None and lxmf_announce . app_data is not None :
2024-09-11 21:23:26 +12:00
return self . parse_lxmf_display_name ( app_data_base64 = lxmf_announce . app_data )
# announce did not have app data, so provide a fallback name
return " Anonymous Peer "
2024-06-01 17:56:22 +12:00
2024-09-11 21:23:26 +12:00
# reads the lxmf display name from the provided base64 app data
2024-09-18 23:26:16 +12:00
def parse_lxmf_display_name ( self , app_data_base64 : str , default_value : str | None = " Anonymous Peer " ) :
2024-10-12 17:09:37 +13:00
2024-09-11 21:23:26 +12:00
try :
app_data_bytes = base64 . b64decode ( app_data_base64 )
2024-10-12 17:09:37 +13:00
display_name = LXMF . display_name_from_app_data ( app_data_bytes )
if display_name is not None :
return display_name
2024-09-11 21:23:26 +12:00
except :
2024-10-12 17:09:37 +13:00
pass
return default_value
2024-04-29 17:12:03 +12:00
2024-09-25 22:19:12 +12:00
# reads the lxmf stamp cost from the provided base64 app data
def parse_lxmf_stamp_cost ( self , app_data_base64 : str ) :
try :
app_data_bytes = base64 . b64decode ( app_data_base64 )
return LXMF . stamp_cost_from_app_data ( app_data_bytes )
except :
return None
2024-09-17 16:55:49 +12:00
# reads the nomadnetwork node display name from the provided base64 app data
2024-09-18 23:26:16 +12:00
def parse_nomadnetwork_node_display_name ( self , app_data_base64 : str , default_value : str | None = " Anonymous Node " ) :
2024-09-17 16:55:49 +12:00
try :
app_data_bytes = base64 . b64decode ( app_data_base64 )
return app_data_bytes . decode ( " utf-8 " )
except :
2024-09-18 23:26:16 +12:00
return default_value
2024-09-17 16:55:49 +12:00
2024-09-19 17:20:49 +12:00
# parses lxmf propagation node app data
def parse_lxmf_propagation_node_app_data ( self , app_data_base64 : str ) :
try :
app_data_bytes = base64 . b64decode ( app_data_base64 )
data = msgpack . unpackb ( app_data_bytes )
return {
" enabled " : bool ( data [ 0 ] ) ,
" timebase " : int ( data [ 1 ] ) ,
" per_transfer_limit " : int ( data [ 2 ] ) ,
}
except :
return None
2024-06-01 19:45:19 +12:00
# returns true if the conversation has messages newer than the last read at timestamp
def is_lxmf_conversation_unread ( self , destination_hash ) :
2024-08-12 11:20:47 +12:00
# get lxmf conversation read state from database for the provided destination hash
2024-06-01 19:45:19 +12:00
lxmf_conversation_read_state = ( database . LxmfConversationReadState . select ( )
. where ( database . LxmfConversationReadState . destination_hash == destination_hash )
. get_or_none ( ) )
# get most recent incoming message from destination hash
latest_incoming_lxmf_message = ( database . LxmfMessage . select ( )
. where ( database . LxmfMessage . source_hash == destination_hash )
. order_by ( database . LxmfMessage . created_at . desc ( ) )
. get_or_none ( ) )
# there's no incoming message, so it can't be unread
if latest_incoming_lxmf_message is None :
return False
2024-08-12 11:20:47 +12:00
# user has never read this conversation, so it's unread
if lxmf_conversation_read_state is None :
return True
2024-06-01 19:45:19 +12:00
# conversation is unread if last read at is before the latest incoming message creation date
conversation_last_read_at = datetime . strptime ( lxmf_conversation_read_state . last_read_at , " % Y- % m- %d % H: % M: % S. %f % z " )
conversation_latest_message_at = datetime . strptime ( latest_incoming_lxmf_message . created_at , " % Y- % m- %d % H: % M: % S. %f % z " )
return conversation_last_read_at < conversation_latest_message_at
2024-07-28 16:52:57 +12:00
# returns number of messages that failed to send in a conversation
def lxmf_conversation_failed_messages_count ( self , destination_hash : str ) :
return ( database . LxmfMessage . select ( )
. where ( database . LxmfMessage . state == " failed " )
. where ( database . LxmfMessage . destination_hash == destination_hash )
. count ( ) )
2024-08-12 10:35:08 +12:00
# find an interface by name
def find_interface_by_name ( self , name : str ) :
for interface in RNS . Transport . interfaces :
interface_name = str ( interface )
if name == interface_name :
return interface
return None
2024-07-28 16:52:57 +12:00
2024-05-04 21:31:12 +12:00
# class to manage config stored in database
class Config :
@staticmethod
2024-06-02 11:15:09 +12:00
def get ( key : str , default_value = None ) - > str | None :
2024-05-04 21:31:12 +12:00
# get config value from database
config_item = database . Config . get_or_none ( database . Config . key == key )
# return value if available
if config_item is not None :
return config_item . value
# fallback to returning default value
return default_value
@staticmethod
2024-09-19 00:47:30 +12:00
def set ( key : str , value : str | None ) :
# if none, delete the config entry
if value is None :
database . Config . delete ( ) . where ( database . Config . key == key ) . execute ( )
return
2024-05-04 21:31:12 +12:00
# prepare data to insert or update
data = {
" key " : key ,
" value " : value ,
2024-05-05 21:49:27 +12:00
" updated_at " : datetime . now ( timezone . utc ) ,
2024-05-04 21:31:12 +12:00
}
# upsert to database
query = database . Config . insert ( data )
query = query . on_conflict ( conflict_target = [ database . Config . key ] , update = data )
query . execute ( )
# handle config values that should be strings
class StringConfig :
2024-09-18 17:42:25 +12:00
def __init__ ( self , key : str , default_value : str | None = None ) :
2024-05-04 21:31:12 +12:00
self . key = key
self . default_value = default_value
2024-06-02 11:15:09 +12:00
def get ( self , default_value : str = None ) - > str | None :
2024-05-04 21:31:12 +12:00
_default_value = default_value or self . default_value
return Config . get ( self . key , default_value = _default_value )
2024-09-19 00:47:30 +12:00
def set ( self , value : str | None ) :
2024-06-02 11:15:09 +12:00
Config . set ( self . key , value )
# handle config values that should be bools
class BoolConfig :
def __init__ ( self , key : str , default_value : bool = False ) :
self . key = key
self . default_value = default_value
def get ( self ) - > bool :
# get string value, or return default
config_value = Config . get ( self . key , default_value = None )
if config_value is None :
return self . default_value
return config_value == " true "
def set ( self , value : bool ) :
# determine string value for bool
if value is True :
config_value = " true "
else :
config_value = " false "
Config . set ( self . key , config_value )
# handle config values that should be integers
class IntConfig :
def __init__ ( self , key : str , default_value : int | None = 0 ) :
self . key = key
self . default_value = default_value
def get ( self ) - > int | None :
# get string value, or return default
config_value = Config . get ( self . key , default_value = None )
if config_value is None :
return self . default_value
return int ( config_value )
def set ( self , value : int ) :
Config . set ( self . key , str ( value ) )
2024-05-04 21:31:12 +12:00
# all possible config items
2024-07-08 16:22:24 +12:00
database_version = IntConfig ( " database_version " , None )
2024-05-04 21:31:12 +12:00
display_name = StringConfig ( " display_name " , " Anonymous Peer " )
2024-06-02 11:15:09 +12:00
auto_announce_enabled = BoolConfig ( " auto_announce_enabled " , False )
2024-06-02 12:01:51 +12:00
auto_announce_interval_seconds = IntConfig ( " auto_announce_interval_seconds " , 0 )
2024-06-02 11:15:09 +12:00
last_announced_at = IntConfig ( " last_announced_at " , None )
2024-12-16 12:17:11 +13:00
theme = StringConfig ( " theme " , " light " )
2024-07-28 14:28:11 +12:00
auto_resend_failed_messages_when_announce_received = BoolConfig ( " auto_resend_failed_messages_when_announce_received " , True )
2024-07-28 14:53:03 +12:00
allow_auto_resending_failed_messages_with_attachments = BoolConfig ( " allow_auto_resending_failed_messages_with_attachments " , False )
2024-09-18 21:27:32 +12:00
auto_send_failed_messages_to_propagation_node = BoolConfig ( " auto_send_failed_messages_to_propagation_node " , False )
2024-07-28 22:30:22 +12:00
show_suggested_community_interfaces = BoolConfig ( " show_suggested_community_interfaces " , True )
2024-08-03 00:20:59 +12:00
lxmf_delivery_transfer_limit_in_bytes = IntConfig ( " lxmf_delivery_transfer_limit_in_bytes " , 1000 * 1000 * 10 ) # 10MB
2024-09-18 17:42:25 +12:00
lxmf_preferred_propagation_node_destination_hash = StringConfig ( " lxmf_preferred_propagation_node_destination_hash " , None )
2024-09-18 21:06:03 +12:00
lxmf_preferred_propagation_node_auto_sync_interval_seconds = IntConfig ( " lxmf_preferred_propagation_node_auto_sync_interval_seconds " , 0 )
lxmf_preferred_propagation_node_last_synced_at = IntConfig ( " lxmf_preferred_propagation_node_last_synced_at " , None )
2024-09-19 15:59:04 +12:00
lxmf_local_propagation_node_enabled = BoolConfig ( " lxmf_local_propagation_node_enabled " , False )
2024-05-04 21:31:12 +12:00
2024-12-11 01:29:22 +13:00
# FIXME: we should probably set this as an instance variable of ReticulumMeshChat so it has a proper home, and pass it in to the constructor?
nomadnet_cached_links = { }
2024-05-01 23:36:14 +12:00
class NomadnetDownloader :
2024-09-22 22:58:19 +12:00
def __init__ ( self , destination_hash : bytes , path : str , data : str | None , on_download_success : Callable [ [ bytes ] , None ] , on_download_failure : Callable [ [ str ] , None ] , on_progress_update : Callable [ [ float ] , None ] , timeout : int | None = None ) :
2024-05-01 23:36:14 +12:00
self . app_name = " nomadnetwork "
self . aspects = " node "
self . destination_hash = destination_hash
self . path = path
2024-09-22 22:58:19 +12:00
self . data = data
2024-05-01 23:36:14 +12:00
self . timeout = timeout
self . on_download_success = on_download_success
self . on_download_failure = on_download_failure
2024-05-02 00:14:13 +12:00
self . on_progress_update = on_progress_update
2024-05-01 23:36:14 +12:00
# setup link to destination and request download
2024-05-29 03:37:44 +12:00
async def download ( self , path_lookup_timeout : int = 15 , link_establishment_timeout : int = 15 ) :
2024-05-01 23:36:14 +12:00
2024-12-11 01:29:22 +13:00
# use existing established link if it's active
if self . destination_hash in nomadnet_cached_links :
link = nomadnet_cached_links [ self . destination_hash ]
if link . status is RNS . Link . ACTIVE :
print ( " [NomadnetDownloader] using existing link for request " )
self . link_established ( link )
return
2024-05-28 03:56:07 +12:00
# determine when to timeout
timeout_after_seconds = time . time ( ) + path_lookup_timeout
2024-05-01 23:36:14 +12:00
2024-05-28 03:56:07 +12:00
# check if we have a path to the destination
if not RNS . Transport . has_path ( self . destination_hash ) :
# we don't have a path, so we need to request it
RNS . Transport . request_path ( self . destination_hash )
# wait until we have a path, or give up after the configured timeout
while not RNS . Transport . has_path ( self . destination_hash ) and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
# if we still don't have a path, we can't establish a link, so bail out
if not RNS . Transport . has_path ( self . destination_hash ) :
self . on_download_failure ( " Could not find path to destination. " )
2024-05-01 23:36:14 +12:00
return
# create destination to nomadnet node
2024-05-28 03:56:07 +12:00
identity = RNS . Identity . recall ( self . destination_hash )
2024-05-01 23:36:14 +12:00
destination = RNS . Destination (
identity ,
RNS . Destination . OUT ,
RNS . Destination . SINGLE ,
self . app_name ,
self . aspects ,
)
# create link to destination
2024-12-11 01:29:22 +13:00
print ( " [NomadnetDownloader] establishing new link for request " )
2024-05-29 03:37:44 +12:00
link = RNS . Link ( destination , established_callback = self . link_established )
# determine when to timeout
timeout_after_seconds = time . time ( ) + link_establishment_timeout
# wait until we have established a link, or give up after the configured timeout
while link . status is not RNS . Link . ACTIVE and time . time ( ) < timeout_after_seconds :
await asyncio . sleep ( 0.1 )
# if we still haven't established a link, bail out
if link . status is not RNS . Link . ACTIVE :
self . on_download_failure ( " Could not establish link to destination. " )
2024-05-01 23:36:14 +12:00
# link to destination was established, we should now request the download
def link_established ( self , link ) :
2024-12-11 01:29:22 +13:00
# cache link for using in future requests
nomadnet_cached_links [ self . destination_hash ] = link
2024-05-01 23:36:14 +12:00
# request download over link
link . request (
self . path ,
2024-09-22 22:58:19 +12:00
data = self . data ,
2024-05-01 23:36:14 +12:00
response_callback = self . on_response ,
failed_callback = self . on_failed ,
progress_callback = self . on_progress ,
timeout = self . timeout ,
)
# handle successful download
def on_response ( self , request_receipt ) :
self . on_download_success ( request_receipt . response )
# handle failure
def on_failed ( self , request_receipt = None ) :
self . on_download_failure ( " request_failed " )
# handle download progress
def on_progress ( self , request_receipt ) :
2024-05-02 00:14:13 +12:00
self . on_progress_update ( request_receipt . progress )
2024-05-01 23:36:14 +12:00
class NomadnetPageDownloader ( NomadnetDownloader ) :
2024-09-22 22:58:19 +12:00
def __init__ ( self , destination_hash : bytes , page_path : str , data : str | None , on_page_download_success : Callable [ [ str ] , None ] , on_page_download_failure : Callable [ [ str ] , None ] , on_progress_update : Callable [ [ float ] , None ] , timeout : int | None = None ) :
2024-05-01 23:36:14 +12:00
self . on_page_download_success = on_page_download_success
self . on_page_download_failure = on_page_download_failure
2024-09-22 22:58:19 +12:00
super ( ) . __init__ ( destination_hash , page_path , data , self . on_download_success , self . on_download_failure , on_progress_update , timeout )
2024-05-01 23:36:14 +12:00
# page download was successful, decode the response and send to provided callback
def on_download_success ( self , response_bytes ) :
micron_markup_response = response_bytes . decode ( " utf-8 " )
self . on_page_download_success ( micron_markup_response )
# page download failed, send error to provided callback
def on_download_failure ( self , failure_reason ) :
self . on_page_download_failure ( failure_reason )
class NomadnetFileDownloader ( NomadnetDownloader ) :
2024-05-28 03:56:07 +12:00
def __init__ ( self , destination_hash : bytes , page_path : str , on_file_download_success : Callable [ [ str , bytes ] , None ] , on_file_download_failure : Callable [ [ str ] , None ] , on_progress_update : Callable [ [ float ] , None ] , timeout : int | None = None ) :
2024-05-01 23:36:14 +12:00
self . on_file_download_success = on_file_download_success
self . on_file_download_failure = on_file_download_failure
2024-09-22 22:58:19 +12:00
super ( ) . __init__ ( destination_hash , page_path , None , self . on_download_success , self . on_download_failure , on_progress_update , timeout )
2024-05-01 23:36:14 +12:00
# file download was successful, decode the response and send to provided callback
def on_download_success ( self , response ) :
file_name : str = response [ 0 ]
file_data : bytes = response [ 1 ]
self . on_file_download_success ( file_name , file_data )
# page download failed, send error to provided callback
def on_download_failure ( self , failure_reason ) :
self . on_file_download_failure ( failure_reason )
2024-05-04 02:42:18 +12:00
def main ( ) :
2024-04-29 20:26:45 +12:00
# parse command line args
2024-06-02 14:42:19 +12:00
parser = argparse . ArgumentParser ( description = " ReticulumMeshChat " )
2024-05-24 02:42:54 +12:00
parser . add_argument ( " --host " , nargs = ' ? ' , default = " 127.0.0.1 " , type = str , help = " The address the web server should listen on. " )
2024-04-29 20:26:45 +12:00
parser . add_argument ( " --port " , nargs = ' ? ' , default = " 8000 " , type = int , help = " The port the web server should listen on. " )
2024-05-19 00:48:45 +12:00
parser . add_argument ( " --headless " , action = ' store_true ' , help = " Web browser will not automatically launch when this flag is passed. " )
2024-04-29 20:26:45 +12:00
parser . add_argument ( " --identity-file " , type = str , help = " Path to a Reticulum Identity file to use as your LXMF address. " )
parser . add_argument ( " --identity-base64 " , type = str , help = " A base64 encoded Reticulum Identity to use as your LXMF address. " )
2024-04-30 21:38:34 +12:00
parser . add_argument ( " --generate-identity-file " , type = str , help = " Generates and saves a new Reticulum Identity to the provided file path and then exits. " )
parser . add_argument ( " --generate-identity-base64 " , action = ' store_true ' , help = " Outputs a randomly generated Reticulum Identity as base64 and then exits. " )
2024-04-30 21:25:27 +12:00
parser . add_argument ( " --reticulum-config-dir " , type = str , help = " Path to a Reticulum config directory for the RNS stack to use (e.g: ~/.reticulum) " )
2024-05-04 20:57:20 +12:00
parser . add_argument ( " --storage-dir " , type = str , help = " Path to a directory for storing databases and config files (default: ./storage) " )
2024-07-30 21:39:37 +12:00
parser . add_argument ( " --test-exception-message " , type = str , help = " Throws an exception. Used for testing the electron error dialog " )
2024-08-16 21:31:18 +12:00
parser . add_argument ( ' args ' , nargs = argparse . REMAINDER ) # allow unknown command line args
2024-04-29 20:26:45 +12:00
args = parser . parse_args ( )
2024-04-29 17:12:03 +12:00
2024-07-30 21:39:37 +12:00
# check if we want to test exception messages
if args . test_exception_message is not None :
raise Exception ( args . test_exception_message )
2024-04-30 21:38:34 +12:00
# util to generate reticulum identity and save to file without using rnid
if args . generate_identity_file is not None :
2024-04-30 21:42:30 +12:00
# do not overwrite existing files, otherwise user could lose existing keys
if os . path . exists ( args . generate_identity_file ) :
print ( " DANGER: the provided identity file path already exists, not overwriting! " )
return
# generate a new identity and save to provided file path
2024-04-30 21:38:34 +12:00
identity = RNS . Identity ( create_keys = True )
with open ( args . generate_identity_file , " wb " ) as file :
file . write ( identity . get_private_key ( ) )
2024-04-30 21:42:30 +12:00
2024-04-30 21:38:34 +12:00
print ( " A new Reticulum Identity has been saved to: {} " . format ( args . generate_identity_file ) )
return
# util to generate reticulum identity as base64 without using rnid
2024-04-30 21:48:15 +12:00
if args . generate_identity_base64 is True :
2024-04-30 21:38:34 +12:00
identity = RNS . Identity ( create_keys = True )
print ( base64 . b64encode ( identity . get_private_key ( ) ) . decode ( " utf-8 " ) )
return
2024-04-29 20:26:45 +12:00
# use provided identity, or fallback to a random one
if args . identity_file is not None :
identity = RNS . Identity ( create_keys = False )
identity . load ( args . identity_file )
2024-05-24 02:02:23 +12:00
print ( " Reticulum Identity < {} > has been loaded from file {} . " . format ( identity . hash . hex ( ) , args . identity_file ) )
2024-04-29 20:26:45 +12:00
elif args . identity_base64 is not None :
identity = RNS . Identity ( create_keys = False )
identity . load_private_key ( base64 . b64decode ( args . identity_base64 ) )
2024-05-18 22:16:22 +12:00
print ( " Reticulum Identity < {} > has been loaded from base64. " . format ( identity . hash . hex ( ) ) )
2024-04-29 20:26:45 +12:00
else :
2024-05-24 02:02:23 +12:00
2024-05-24 02:19:21 +12:00
# ensure provided storage dir exists, or the default storage dir exists
base_storage_dir = args . storage_dir or os . path . join ( " storage " )
os . makedirs ( base_storage_dir , exist_ok = True )
2024-05-24 02:02:23 +12:00
# configure path to default identity file
2024-05-24 02:19:21 +12:00
default_identity_file = os . path . join ( base_storage_dir , " identity " )
2024-05-24 02:02:23 +12:00
# if default identity file does not exist, generate a new identity and save it
if not os . path . exists ( default_identity_file ) :
identity = RNS . Identity ( create_keys = True )
with open ( default_identity_file , " wb " ) as file :
file . write ( identity . get_private_key ( ) )
print ( " Reticulum Identity < {} > has been randomly generated and saved to {} . " . format ( identity . hash . hex ( ) , default_identity_file ) )
# default identity file exists, load it
identity = RNS . Identity ( create_keys = False )
identity . load ( default_identity_file )
2024-05-25 03:42:15 +12:00
print ( " Reticulum Identity < {} > has been loaded from file {} . " . format ( identity . hash . hex ( ) , default_identity_file ) )
2024-04-29 17:12:03 +12:00
2024-04-29 20:26:45 +12:00
# init app
2024-06-02 14:42:19 +12:00
reticulum_meshchat = ReticulumMeshChat ( identity , args . storage_dir , args . reticulum_config_dir )
reticulum_meshchat . run ( args . host , args . port , launch_browser = args . headless is False )
2024-08-02 18:44:41 +12:00
2024-04-29 01:09:06 +12:00
if __name__ == " __main__ " :
2024-05-04 02:42:18 +12:00
main ( )