diff --git a/api/dmrapi.py b/api/dmrapi.py new file mode 100755 index 0000000..35af974 --- /dev/null +++ b/api/dmrapi.py @@ -0,0 +1,100 @@ +import uvicorn + +from datetime import datetime +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles + +from dmrtools.dispatcher import Dispatcher + + +class DMRApiHelper: + CALLS_MAX = 50 + + dispatcher: Dispatcher|None = None + + @classmethod + def get_peers(cls) -> list: + if cls.dispatcher is None: + return list() + + peers = sorted(cls.dispatcher.peer_keeper.peers, + key=lambda p: p.connect_time) + + peers_info = [{ + "name": p.name, + "peer_id": p.peer_id, + "addr": p.addr_str, + "status": p.status.name, + "connect_time": p.connect_time, + "active_time": p.active_time, + "units": list(p.units.keys()), + "config": p.config} for p in peers] + + return peers_info + + @classmethod + def get_calls(cls) -> list: + if cls.dispatcher is None: + return list() + + calls = sorted(cls.dispatcher.call_keeper.calls_log, + key=lambda c: c.start_time, reverse=True) + + # get first CALLS_MAX if more than + if len(calls) > cls.CALLS_MAX: + calls = calls[:cls.CALLS_MAX] + + calls_info = [{ + "call_id": c.call_id, + "dir": f"{c.src_id}->{c.dst_hr}", + "src_id": c.src_id, + "dst_id": c.dst_id, + "peer_id": c.peer_id, + "call_type": c.call_type.name, + "start_time": c.start_time, + "last_packet_time": c.last_packet_time, + "is_ended": c.is_ended, + "end_time": c.end_time, + "broadcast": c.route_to is None, + "route_to": (list(peer.name for peer in c.route_to) + if c.route_to is not None else []), + "time": f"{c.time:.1f}s"} for c in calls] + + return calls_info + + +# Define FastAPI app +app = FastAPI() + + +# Mount the "dashboard" directory at /dashboard +app.mount("/dashboard", StaticFiles(directory="dashboard"), name="dashboard") + + +@app.get("/api/dashboard") +async def get_dashboard(): + return {"success": True, + "peers": DMRApiHelper.get_peers(), + "calls": DMRApiHelper.get_calls()} + + +@app.get("/api/peers") +async def get_peers(): + return {"success": True, "peers": DMRApiHelper.get_peers()} + + +@app.get("/api/calls") +async def get_calls(): + return {"success": True, "calls": DMRApiHelper.get_calls()} + + +async def start_api(host: str, port: int, dispatcher: Dispatcher) -> None: + DMRApiHelper.dispatcher = dispatcher + config = uvicorn.Config(app, host=host, port=port, log_level="error", + loop="asyncio") + server = uvicorn.Server(config) + + try: + await server.serve() + except Exception as e: + print(f"start_api: Exception {e}") diff --git a/dashboard/app.js b/dashboard/app.js new file mode 100755 index 0000000..d7c07df --- /dev/null +++ b/dashboard/app.js @@ -0,0 +1,175 @@ +const peersContainer = document.getElementById('peers'); +const callsContainer = document.getElementById('calls'); +const baseURL = `${window.location.protocol}//${window.location.host}`; + +function formatTimestamp(ts) { + if (!ts) return "--" + + const date = new Date(ts * 1000); + const now = new Date(); + + const isToday = + date.getDate() === now.getDate() && + date.getMonth() === now.getMonth() && + date.getFullYear() === now.getFullYear(); + + const timeStr = date.toLocaleTimeString('en-US', { hour12: false }); + const dateStr = date.toLocaleDateString(); + + const secondsAgo = Math.floor((now - date) / 1000); + const rel = secondsAgo < 60 + ? `${secondsAgo}s ago` + : secondsAgo < 3600 + ? `${Math.floor(secondsAgo / 60)}m ago` + : secondsAgo < 86400 + ? `${Math.floor(secondsAgo / 3600)}h ago` + : `${Math.floor(secondsAgo / 86400)}d ago`; + + return `${isToday ? timeStr : `${dateStr} ${timeStr}`} (${rel})`; +} + +async function fetchData() { + try { + const dashboardRes = await fetch(`${baseURL}/api/dashboard`); + const dashboardData = await dashboardRes.json(); + + renderPeers(dashboardData.peers || []); + renderCallsTable(dashboardData.calls || []); + } catch (err) { + console.error('Error fetching data:', err); + } +} + +function getStatusColor(status) { + switch (status) { + case 'ACTIVE': return 'bg-green-900 text-green-200'; + case 'TIMEOUT': return 'bg-yellow-900 text-yellow-200'; + case 'DEAD': return 'bg-red-950 text-red-200'; + default: return 'bg-cyan-900 text-cyan-200'; + } +} + +function toggleVisibility(id) { + const el = document.getElementById(id); + if (el) { + el.classList.toggle('hidden'); + } +} + +let nextExpandedSectionId = 0; + +function renderPeers(peers) { + peersContainer.innerHTML = ''; + if (peers.length === 0) { + peersContainer.innerHTML = '

No peers connected.

'; + return; + } + + peers.forEach(peer => { + const div = document.createElement('div'); + div.className = 'bg-gray-800 rounded-md shadow overflow-hidden'; + + const cardTitleColor = getStatusColor(peer.status); + const unitsHTML = (peer.units || []) + .map(unit => `${unit}`) + .join(''); + + const callsign = peer.config.callsign || "--" + + const expandedSectionId = `expandable-${nextExpandedSectionId++}`; + + div.innerHTML = ` +

${peer.name}

+
+
Callsign: ${callsign}
+
Address: ${peer.addr}
+
Status: ${peer.status}
+
Connected: ${formatTimestamp(peer.connect_time)}
+
Last Active: ${formatTimestamp(peer.active_time)}
+ ${unitsHTML? `
${unitsHTML}
` : ''} +
+ `; + + peersContainer.appendChild(div); + }); +} + +let tableUpdateIntervals = []; + +function renderCallsTable(calls) { + tableUpdateIntervals.forEach(tui => { + clearInterval(tui) + }); + tableUpdateIntervals = [] + + const tableBody = document.getElementById('calls-table-body'); + tableBody.innerHTML = ''; + + if (!calls.length) { + tableBody.innerHTML = ` + + No calls at the moment. + + `; + return; + } + + calls.forEach(call => { + const row = document.createElement('tr'); + row.className = `border-b border-gray-700 ${ + call.is_ended ? '' : 'bg-yellow-900 animate-pulse' + }`; + + const durationCell = document.createElement('td'); + durationCell.className = 'px-4 py-2 text-center'; + + if (call.is_ended) { + durationCell.textContent = call.time; + } else { + // Dynamic live updating + const startMs = call.start_time * 1000; + function updateDuration() { + const now = Date.now(); + const delta = ((now - startMs) / 1000).toFixed(1); + durationCell.textContent = `${delta}s`; + // console.log(`Upd dur ${durationCell}`) + } + updateDuration(); + tableUpdateIntervals.push(setInterval(updateDuration, 330)); + } + + let dstBadgeStyle = call.call_type === 'GROUP' + ? 'bg-blue-900 text-blue-200' + : 'bg-cyan-900 text-cyan-200'; + + let dstBadgeText = call.call_type === 'GROUP' + ? "TG-" + call.dst_id + : call.dst_id; + + let routingHTML = ''; + if (call.broadcast) { + routingHTML = `Broadcast`; + } else if (call.route_to && call.route_to.length > 0) { + routingHTML = call.route_to + .map(r => `${r}`) + .join(''); + } + + row.innerHTML = ` + ${call.call_id} + ${formatTimestamp(call.start_time)} + + ${call.src_id} + ⇉ + ${dstBadgeText} + + ${routingHTML} + `; + + row.appendChild(durationCell); + tableBody.appendChild(row); + }); +} + +fetchData(); +setInterval(fetchData, 3000); diff --git a/dashboard/index.html b/dashboard/index.html new file mode 100755 index 0000000..220035f --- /dev/null +++ b/dashboard/index.html @@ -0,0 +1,39 @@ + + + + + DMR Network Dashboard + + + + +

DMR Network Dashboard

+ +
+ +
+

Connected Peers

+
+
+ + +

Active & Recent Calls

+
+ + + + + + + + + + + + + +
IDTimeDirectionRouteDur.
Loading calls...
+
+
+ + diff --git a/dmrmaster.py b/dmrmaster.py new file mode 100755 index 0000000..f085621 --- /dev/null +++ b/dmrmaster.py @@ -0,0 +1,140 @@ +import asyncio +import logging + +from argparse import ArgumentParser + +from api.dmrapi import start_api +from dmrtools import Dispatcher +from dmrtools.app import App +from dmrtools.asyncnetwork import AsyncDatagramServer +from dmrtools.auth import IPeerAuth + + +class DMRMaster: + def __init__(self) -> None: + self.interface: str = '0.0.0.0' + self.port: int = 62031 + self.web_interface: str = '0.0.0.0' + self.web_port: int = 8000 + self.dispatcher: Dispatcher|None = None + self.dg_server: AsyncDatagramServer|None = None + + def setup(self) -> None: + ap = ArgumentParser( + prog="dmrmaster.py", + description='0DMRMaster Server by Alexander Mokrov (UR6LKW)') + + ap.add_argument('-i', '--interface', type=str, + help='Interface to listen on. Defaults to 0.0.0.0.') + ap.add_argument('-p', '--port', type=int, + help='UDP port to listen on. Defaults to 62031.') + ap.add_argument('--web-interface', type=str, + help='Interface to run API on. Defaults to 0.0.0.0.') + ap.add_argument('--web-port', type=int, + help='TCP port to run API on. Defaults to 8000.') + ap.add_argument('-l', '--log-file', type=str, help='Log filename') + ap.add_argument('-d', '--ll-debug', action='store_true', + help='Log level (INFO/DEBUG)') + + args = ap.parse_args() + # print(args) + + if args.interface is not None: + self.interface = args.interface + if args.port is not None: + self.port = args.port + if args.web_interface is not None: + self.web_interface = args.web_interface + if args.web_port is not None: + self.web_port = args.web_port + + log_level = logging.DEBUG if args.ll_debug else logging.INFO + self.setup_log(log_level, args.log_file) + + def setup_log(self, log_level = logging.DEBUG, log_file = None) -> None: + handlers = [logging.StreamHandler()] + + if log_file: + file_handler = logging.FileHandler(log_file) + handlers.append(file_handler) # type: ignore[arg-type] + + logging.basicConfig( + level=log_level, + format='%(asctime)s [%(levelname)s] %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', + handlers=handlers + ) + + def set_peer_auth(self, peer_auth: IPeerAuth) -> None: + if self.dispatcher is None: + logging.error("Can't set auth: no dispatcher") + return + logging.debug(f"Peer auth set to {peer_auth}") + self.dispatcher.peer_auth = peer_auth + + def register_app(self, app: App) -> None: + if self.dispatcher is None: + logging.error("Can't register app: no dispatcher") + return + self.dispatcher.app_keeper.register(app) + + def config(self) -> None: + """ + Local config + """ + pass + + def run(self) -> None: + """ + Run, go async and handle KeyboardInterupt + """ + try: + asyncio.run(self.__async_run()) + except KeyboardInterrupt: + logging.info("Interrupted by user (Ctrl+C). Exiting gracefully.") + + async def __async_run(self) -> None: + await self.__start_udp() + + if not isinstance(self.dg_server, AsyncDatagramServer): + logging.critical(f"Can't listen on {self.interface}:{self.port}") + return + + self.dispatcher = Dispatcher(self.dg_server) + + self.config() # local config for apps + + logging.info(f"Starting API on {self.web_interface}:{self.web_port}") + + try: + await start_api(self.web_interface, self.web_port, self.dispatcher) + # await asyncio.Future() # Run forever + finally: + self.stop() + + async def __start_udp(self) -> None: + logging.info( + f"Starting server listening on {self.interface}:{self.port}") + + loop = asyncio.get_running_loop() + transport, self.dg_server = await loop.create_datagram_endpoint( + lambda: AsyncDatagramServer(), + local_addr=(self.interface, self.port)) + + def stop(self) -> None: + if self.dispatcher is not None: + self.dispatcher.shutdown() + self.dispatcher = None + if self.dg_server is not None: + self.dg_server.close() + self.dg_server = None + + +def main(): + master = DMRMaster(); + master.setup() + master.run() + + +if __name__ == '__main__': + main() diff --git a/dmrtools/__init__.py b/dmrtools/__init__.py index eb4d863..75a6ab8 100755 --- a/dmrtools/__init__.py +++ b/dmrtools/__init__.py @@ -1,2 +1,5 @@ +from .dispatcher import Dispatcher from .dmrproto import DMRPPacketFactory +from .network import IDatagramSender, IDatagramReceiver from .pphex import hexdump +from .auth import AllowAllPeerAuth, DenyAllPeerAuth, ListPeerAuth diff --git a/dmrtools/app.py b/dmrtools/app.py new file mode 100755 index 0000000..d3c16f4 --- /dev/null +++ b/dmrtools/app.py @@ -0,0 +1,54 @@ +import logging +import time + +from abc import ABC, abstractmethod + +from .call import Call +from .dmrproto import DMRPPacketData + + +class AppException(Exception): + pass + + +class IAppDispatcher(ABC): + """ + Dispatcher interface for app + """ + @abstractmethod + def inject_packet(self, p: DMRPPacketData) -> None: + pass + + +class IAppCallInterceptor(ABC): + @abstractmethod + def process_call_packet(self, call: Call, p: DMRPPacketData) -> None: + pass + + +class App(ABC): + def __init__(self) -> None: + self.dispatcher: IAppDispatcher|None = None + + @abstractmethod + def get_name(self) -> str: + pass + + +class AppKeeper: + def __init__(self, dispatcher: IAppDispatcher): + self.dispatcher: IAppDispatcher = dispatcher + self.apps: list[App] = [] + + def register(self, app: App) -> bool: + logging.info(f"Registered app '{app.get_name()}' ") + app.dispatcher = self.dispatcher + self.apps.append(app) + return True + + def process_call_packet(self, call: Call, p: DMRPPacketData) -> None: + for app in self.apps: + if isinstance(app, IAppCallInterceptor): + app.process_call_packet(call, p) + + diff --git a/dmrtools/asyncnetwork.py b/dmrtools/asyncnetwork.py new file mode 100755 index 0000000..3538e06 --- /dev/null +++ b/dmrtools/asyncnetwork.py @@ -0,0 +1,45 @@ +import asyncio +import logging + +from typing import Any + +from .network import IDatagramReceiver, IDatagramSender + + +class AsyncDatagramServer(asyncio.DatagramProtocol, IDatagramSender): + def __init__(self) -> None: + self.receiver: IDatagramReceiver|None = None + self.transport: Any|None = None + + def connection_made(self, transport: Any) -> None: + # bug in linux implementation - _SelectorDatagramTransport is + # not inherited from DatagramTransport. That's why hasattr is used + if (hasattr(transport, 'sendto') or + isinstance(transport, asyncio.DatagramTransport)): + self.transport = transport + else: + logging.error("AsyncDatagramServer.connection_made(): " + "Unexpected transport received") + + def set_receiver(self, receiver: IDatagramReceiver): + self.receiver = receiver + + def datagram_received(self, data: bytes, addr: tuple) -> None: + if self.receiver is not None: + self.receiver.recv_dg(data, addr) + + def send_dg(self, data: bytes, addr: tuple) -> None: + if self.transport is not None: + self.transport.sendto(data, addr) + + # def error_received(self, exc: Exception|None) -> None: + # logging.error(f"Client protocol error: {exc}") + + def connection_lost(self, exc: Exception|None) -> None: + logging.info(f"Client connection closed: {exc}") + self.transport = None + + def close(self) -> None: + if self.transport: + self.transport.close() + diff --git a/dmrtools/auth.py b/dmrtools/auth.py new file mode 100755 index 0000000..6b77ff5 --- /dev/null +++ b/dmrtools/auth.py @@ -0,0 +1,69 @@ +import logging + +from abc import ABC, abstractmethod + +from .dmrproto import calc_password_hash + + +class IPeerAuth(ABC): + @abstractmethod + def allow_peer_id(self, peer_id: int) -> bool: + pass + + @abstractmethod + def check_password(self, peer_id: int, salt: bytes, + pass_hash: bytes) -> bool: + pass + + +class AllowAllPeerAuth(IPeerAuth): + """ + Auth agent to allow any peer id and any password + """ + def allow_peer_id(self, peer_id: int) -> bool: + return True + + def check_password(self, peer_id: int, salt: bytes, + pass_hash: bytes) -> bool: + return True + + +class DenyAllPeerAuth(IPeerAuth): + """ + Auth agent to deny any peer id and any password + """ + def allow_peer_id(self, peer_id: int) -> bool: + logging.warning("Deny all policy active") + return False + + def check_password(self, peer_id: int, salt: bytes, + pass_hash: bytes) -> bool: + return False + + +class ListPeerAuth(IPeerAuth): + """ + Auth agent with list of allowed peers and their passwords + use empty password to accept any password + """ + def __init__(self, allowed_peers: dict[int, str]|None = None) -> None: + """ + allowed_peers in format peer_id -> password + """ + self.allowed_peers: dict[int, str] = ( + dict() if allowed_peers is None else allowed_peers) + + def allow_peer_id(self, peer_id: int) -> bool: + return peer_id in self.allowed_peers + + def check_password(self, peer_id: int, salt: bytes, + pass_hash: bytes) -> bool: + if peer_id not in self.allowed_peers: + return False + + valid_password = self.allowed_peers[peer_id] + if valid_password == '': + logging.debug(f"Any password accepted for {peer_id}") + return True # Accept any password if empty in config + + return pass_hash == calc_password_hash(salt, valid_password) diff --git a/dmrtools/call.py b/dmrtools/call.py new file mode 100755 index 0000000..190936a --- /dev/null +++ b/dmrtools/call.py @@ -0,0 +1,118 @@ +import logging + +from time import time + +from .dmrproto import DMRPPacketData +from .peer import Peer + + +class Call: + DEAD_TIMEOUT = 5 + CLEAN_TIMEOUT = 60 + CLEAN_LOG_TIMEOUT = 3600 * 6 # 6h + + def __init__(self, call_id: int, + src_id: int, dst_id: int, peer_id: int, + call_type: DMRPPacketData.CallType) -> None: + self.call_id: int = call_id + self.src_id: int = src_id + self.dst_id: int = dst_id + self.peer_id: int = peer_id + self.call_type: DMRPPacketData.CallType = call_type + self.start_time: float = time() + self.last_packet_time: float = time() + self.end_time: float|None = None + self.packets: int = 0 + self.route_to: set[Peer]|None = None + logging.info(f"Voice call beg {str(self)}") + + @property + def dst_hr(self) -> str: + dst: str = f"{self.dst_id}" + if self.call_type == DMRPPacketData.CallType.GROUP: + dst = f"TG-{self.dst_id}" + return dst + + @property + def is_ended(self) -> bool: + return self.end_time is not None + + @property + def is_dead(self) -> bool: + return (not self.is_ended and + not self.check_timeout(self.DEAD_TIMEOUT)) + + @property + def time(self) -> float: + if self.end_time is None: + return self.last_packet_time - self.start_time + return self.end_time - self.start_time + + @property + def to_be_cleaned(self) -> bool: + return self.is_ended and not self.check_timeout(self.CLEAN_TIMEOUT) + + @property + def to_be_cleaned_log(self) -> bool: + return self.is_ended and not self.check_timeout(self.CLEAN_LOG_TIMEOUT) + + def packet_received(self) -> None: + self.last_packet_time = time() + self.packets += 1 + + def check_timeout(self, timeout: float) -> bool: + return time() - self.last_packet_time < timeout + + def end(self, by_timeout: bool = False) -> None: + self.end_time = self.last_packet_time if by_timeout else time() + action = "t/o" if by_timeout else "end" + logging.info(f"Voice call {action} {str(self)}") + + def __str__(self) -> str: + duration = f"dur:{self.time:.1f}s" if self.is_ended else "running" + res = (f"id:{self.call_id} {self.src_id}->{self.dst_hr} " + f"peer:{self.peer_id} {duration}") + return res + + def __repr__(self) -> str: + return f"" + + +class CallKeeper: + def __init__(self) -> None: + self.calls: set[Call] = set() + self.calls_log: set[Call] = set() + + def maintain(self) -> None: + # end dead calls + dead_calls: set[Call] = set(call for call in self.calls + if call.is_dead) + + if len(dead_calls) > 0: + logging.debug(f"Ending dead calls {dead_calls}") + for call in dead_calls: + call.end(by_timeout=True) + + logging.debug( + "Upkeep calls" + + ("".join(["\n - " + str(call) for call in self.calls]))) + + clean_calls: set[Call] = set(call for call in self.calls + if call.to_be_cleaned) + + if len(clean_calls) > 0: + logging.debug(f"Removing ended calls {clean_calls}") + self.calls -= clean_calls + + clean_log: set[Call] = set(call for call in self.calls_log + if call.to_be_cleaned_log) + self.calls_log -= clean_log + + def by_call_id(self, call_id: int) -> Call|None: + call_id_map: dict[int, Call] = {call.call_id: call + for call in self.calls} + return call_id_map[call_id] if call_id in call_id_map else None + + def add(self, call: Call) -> None: + self.calls.add(call) + self.calls_log.add(call) diff --git a/dmrtools/dispatcher.py b/dmrtools/dispatcher.py new file mode 100755 index 0000000..7cf7b44 --- /dev/null +++ b/dmrtools/dispatcher.py @@ -0,0 +1,138 @@ +import asyncio +import logging +import traceback + +from time import time + +from .app import AppKeeper, IAppDispatcher +from .call import Call, CallKeeper +from .dmrproto import DMRPPacketData, DMRPPacketTalkerAlias +from .dmrproto import DMRPPacketFactory, DMRPBasePacket, DMRPBasePeerPacket +from .network import IDatagramReceiver, IDatagramSender +from .peer import Peer, PeerKeeper +from .peer_controller import IPCDispatcher, PeerController +from .auth import IPeerAuth, DenyAllPeerAuth +from .pphex import hexdump + + +class Dispatcher(IDatagramReceiver, IAppDispatcher, IPCDispatcher): + MAINTENANCE_PERIOD = 10 + + def __init__(self, sender: IDatagramSender) -> None: + self.sender: IDatagramSender = sender + self.peer_auth: IPeerAuth = DenyAllPeerAuth() # Default policy to deny + self.peer_keeper: PeerKeeper = PeerKeeper() + self.call_keeper: CallKeeper = CallKeeper() + self.app_keeper: AppKeeper = AppKeeper(self) + + sender.set_receiver(self) + + asyncio.create_task(self.maintain_task()) + + def maintain(self) -> None: + self.peer_keeper.maintain() + self.call_keeper.maintain() + + async def maintain_task(self) -> None: + logging.debug("Dispatcher: periodic maintenance task scheduled") + while True: + await asyncio.sleep(self.MAINTENANCE_PERIOD) + self.maintain() + + def shutdown(self) -> None: + logging.info("Dispatcher: shutting down") + for peer in self.peer_keeper.get_all(): + controller: PeerController = PeerController(peer, self) + controller.send_close() + + def dispatch_data_packet(self, p: DMRPPacketData, orig_addr: tuple) -> None: + logging.debug(f"Dispatching packet from {orig_addr}:\n{p}\n") + + call_id = p.stream_id + + if (call := self.call_keeper.by_call_id(call_id)) is None: + call = Call(p.stream_id, p.src_id, p.dst_id, + p.peer_id, p.call_type) + self.call_keeper.add(call) + + if call.call_type == DMRPPacketData.CallType.UNIT: + # get peers location and rout unit call + peers = self.peer_keeper.get_by_unit(p.dst_id) + if len(peers) > 0: + call.route_to = peers + + call.packet_received() + + if p.is_voice_term: + call.end() + + self.app_keeper.process_call_packet(call, p) + + if isinstance(p, DMRPBasePeerPacket): + self.distribute_by_peers(p, orig_addr, call.route_to) + + def dispatch_ta_packet(self, p: DMRPPacketTalkerAlias, + orig_addr: tuple) -> None: + self.distribute_by_peers(p, orig_addr) + + def distribute_by_peers(self, p: DMRPBasePeerPacket, + orig_addr: tuple, + peers: set[Peer]|None = None) -> None: + sendp = p.copy() + + if peers is None: + peers = self.peer_keeper.get_active() + for peer in peers: + if orig_addr == peer.addr: # skip myself + continue + sendp.peer_id = peer.peer_id + # logging.debug(f" - sending to {peer.logname}" + # f"\n{hexdump(sendp.get_data())}\n{sendp}\n") + self.sender.send_dg(sendp.get_data(), peer.addr) + + #------------------------------- + # IDatagramReceiver implementation + def recv_dg(self, data: bytes, addr: tuple) -> None: + peer: Peer = self.peer_keeper.get_by_addr(addr) + p: DMRPBasePacket|None = None + + try: + p = DMRPPacketFactory.fd(data) + logging.debug(f"Got packet " + f"from {peer.logname} | {len(data)} bytes:\n" + f"{hexdump(data)}\n" + f"{str(p)}\n") + except Exception as e: + logging.error(f"Exception with packet " + f"from {peer.logname} | {len(data)} bytes:\n" + f"{hexdump(data)}\n" + f"{traceback.format_exc()}") + return + + # process in context of the peer by controller + controller: PeerController = PeerController(peer, self) + if not controller.process_packet(p): + return + + if type(p) is DMRPPacketData: + self.dispatch_data_packet(p, addr) + + if type(p) is DMRPPacketTalkerAlias: + self.dispatch_ta_packet(p, addr) + + #------------------------------- + # IAppDispatcher implementation + def inject_packet(self, p: DMRPPacketData) -> None: + logging.debug(f"Injecting packet from app") + self.dispatch_data_packet(p, (None, None)) + + #------------------------------- + # IPCDispatcher implementation + def send_dg(self, data: bytes, addr: tuple) -> None: + self.sender.send_dg(data, addr) + + def get_peer_keeper(self) -> PeerKeeper: + return self.peer_keeper + + def get_peer_auth(self) -> IPeerAuth: + return self.peer_auth diff --git a/dmrtools/network.py b/dmrtools/network.py new file mode 100755 index 0000000..c997e29 --- /dev/null +++ b/dmrtools/network.py @@ -0,0 +1,18 @@ +from abc import ABC, abstractmethod + + +class IDatagramReceiver(ABC): + @abstractmethod + def recv_dg(self, data: bytes, addr: tuple) -> None: + pass + + +class IDatagramSender(ABC): + @abstractmethod + def send_dg(self, data: bytes, addr: tuple) -> None: + pass + + @abstractmethod + def set_receiver(self, receiver: IDatagramReceiver) -> None: + pass + diff --git a/dmrtools/parrot_app.py b/dmrtools/parrot_app.py new file mode 100755 index 0000000..88a48a6 --- /dev/null +++ b/dmrtools/parrot_app.py @@ -0,0 +1,102 @@ +import asyncio +import logging +import time + +from .app import App, IAppCallInterceptor, AppException +from .call import Call +from .dmrproto import DMRPPacketData + + +class ParrotApp(App, IAppCallInterceptor): + """ + Simple parrot application + """ + def __init__(self, parrot_id: int = 9990, repeat_delay: int = 5, + enable_unit: bool = True, enable_group: bool = True) -> None: + # settings + self.parrot_id: int = parrot_id + self.repeat_delay: int = repeat_delay + self.enable_unit: bool = enable_unit + self.enable_group: bool = enable_group + + super().__init__() + + # internal state + self._mycalls: set[int] = set() + self._records: dict[int, list[DMRPPacketData]] = dict() + + @property + def name(self) -> str: + return self.get_name() + + async def repeat(self, packets: list[DMRPPacketData]) -> None: + if self.dispatcher is None: + raise AppException( + "send_packets called before dispatcher had been set") + + if len(packets) == 0: + logging.debug(f"{self.name}: nothing to repeat, 0 packets in call") + return + + await asyncio.sleep(self.repeat_delay) + + logging.info(f"{self.name}: repeating call") + + stream_id = packets[0].stream_id + self._mycalls.add(stream_id) + + start_time = time.monotonic() + for i, p in enumerate(packets): + self.dispatcher.inject_packet(p) + if (delay := start_time + (i + 1) * 0.06 - time.monotonic()) > 0: + await asyncio.sleep(delay) + + self._mycalls.discard(stream_id) + + def record(self, call: Call, p: DMRPPacketData) -> None: + if p.src_id == self.parrot_id or p.dst_id != self.parrot_id: + return + + if (p.call_type == DMRPPacketData.CallType.UNIT + and not self.enable_unit): + logging.debug(f"{self.name}: unit calls disabled") + return + + if (p.call_type == DMRPPacketData.CallType.GROUP + and not self.enable_group): + logging.debug(f"{self.name}: group calls disabled") + return + + if p.stream_id in self._mycalls: + logging.debug(f"{self.name}: skipping my own call") + return + + if p.stream_id not in self._records: + logging.info(f"{self.name}: recording {call.call_id}") + self._records[p.stream_id] = [] + + ans_p = p.copy() + + # invert source and dest for unit call + if p.call_type == DMRPPacketData.CallType.UNIT: + ans_p.src_id = self.parrot_id + ans_p.dst_id = p.src_id + + ans_p.stream_id = p.stream_id + 1 + + self._records[p.stream_id].append(ans_p) + + if call.is_ended: + packets: list[DMRPPacketData] = self._records[p.stream_id] + asyncio.create_task(self.repeat(packets)) + del self._records[p.stream_id] + + #------------------------------------ + # App implementation + def get_name(self) -> str: + return f"Parrot {self.parrot_id}" + + #------------------------------------ + # IAppCallInterceptor implementation + def process_call_packet(self, call: Call, p: DMRPPacketData) -> None: + self.record(call, p) diff --git a/dmrtools/peer.py b/dmrtools/peer.py new file mode 100755 index 0000000..ae2ecf7 --- /dev/null +++ b/dmrtools/peer.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import enum +import logging + +from time import time + + +class Unit: + TIMEOUT = 3600 + + def __init__(self, unit_id: int) -> None: + self.unit_id: int = unit_id + self.active_time: float = time() + + def update_active(self) -> None: + self.active_time = time() + + def check_timeout(self) -> bool: + return time() - self.active_time < self.TIMEOUT + + +class Peer: + class Status(enum.IntEnum): + LOGIN = enum.auto() + AUTH = enum.auto() + CONFIG = enum.auto() + ACTIVE = enum.auto() + DEAD = enum.auto() + + def is_applicable(self, test_status: Peer.Status) -> bool: + """ + Loose check, if test_status is applicable for the current one, + like we can treat any status except DEAD as LOGIN and accept login + packet in it + """ + applicable_map: dict[Peer.Status, set[Peer.Status]] = { + # current status --> Applicable test statuses + Peer.Status.AUTH: {Peer.Status.LOGIN, + Peer.Status.AUTH}, + Peer.Status.CONFIG: {Peer.Status.LOGIN, + Peer.Status.AUTH, + Peer.Status.CONFIG}, + Peer.Status.ACTIVE: {Peer.Status.LOGIN, + Peer.Status.AUTH, + Peer.Status.CONFIG, + Peer.Status.ACTIVE}, + } + + # if not in map, return exact match + if self not in applicable_map: + return self == test_status + + # return if applicable + return test_status in applicable_map[self] + + + PING_TIMEOUT = 130 + + def __init__(self, addr: tuple) -> None: + self.addr: tuple = addr + self.auth_salt: bytes|None = None + self.status: Peer.Status = Peer.Status.LOGIN + self.peer_id: int = 0 + self.connect_time: float = time() + self.active_time: float = time() + self.config: dict[str, str] = dict() + self.units: dict[int, Unit] = dict() + + @property + def logname(self) -> str: + return (self.addr_str if self.peer_id == 0 + else f"{self.peer_id}/{self.addr_str}") + + @property + def name(self) -> str: + return str(self.peer_id) if self.peer_id != 0 else self.addr_str + + @property + def addr_str(self) -> str: + return f"{self.addr[0]}:{self.addr[1]}" + + def check_timeout(self) -> bool: + return time() - self.active_time < self.PING_TIMEOUT + + def update_active(self) -> None: + self.active_time = time() + + def update_unit(self, unit_id: int) -> None: + if unit_id not in self.units: + logging.info(f"Unit {unit_id} added to {self.logname}") + self.units[unit_id] = Unit(unit_id) + self.units[unit_id].update_active() + + def die(self) -> None: + self.status = Peer.Status.DEAD + self.peer_id = 0 + + def __str__(self) -> str: + return f"{self.logname} status:{self.status.name}" + + def __repr__(self) -> str: + return f"" + + +class PeerKeeper: + def __init__(self) -> None: + self.peers: set[Peer] = set() + + def maintain(self) -> None: + # Check timeouts + for peer in self.peers: + # check unit timeouts: + tout_unit_ids = [unit_id for unit_id, unit in peer.units.items() + if not unit.check_timeout()] + + for unit_id in tout_unit_ids: + logging.info(f"Unit {unit_id} removed from {peer.logname}") + del peer.units[unit_id] + + if not peer.check_timeout(): + logging.info(f"Peer {peer.logname} timed out") + peer.die() + + # remove dead peers + dead_peers = set(peer for peer in self.peers + if peer.status == Peer.Status.DEAD) + + if len(dead_peers) > 0: + logging.debug(f"Removing dead peers {dead_peers}") + self.peers -= dead_peers + + logging.debug( + "Upkeep peers" + + ("".join(["\n - " + str(peer) for peer in self.peers]))) + + def get_by_addr(self, addr: tuple) -> Peer: + addr_map: dict[tuple, Peer] = {peer.addr: peer for peer in self.peers} + if addr in addr_map: + return addr_map[addr] + + peer = Peer(addr) + logging.info(f"Peer {peer.logname} connected") + self.peers.add(peer) + return peer + + def get_by_id(self, peer_id: int) -> set[Peer]: + return {peer for peer in self.peers if peer.peer_id == peer_id} + + def get_by_unit(self, unit_id: int) -> set[Peer]: + return {peer for peer in self.peers if unit_id in peer.units} + + def get_all(self) -> set[Peer]: + return self.peers + + def get_active(self) -> set[Peer]: + return {peer for peer in self.peers + if peer.status == Peer.Status.ACTIVE} diff --git a/dmrtools/peer_controller.py b/dmrtools/peer_controller.py new file mode 100755 index 0000000..30425a2 --- /dev/null +++ b/dmrtools/peer_controller.py @@ -0,0 +1,210 @@ +import logging + +from abc import ABC, abstractmethod + +from .auth import IPeerAuth +from .dmrproto import DMRPBasePacket +from .dmrproto import DMRPPacketConfig, DMRPPacketPing, DMRPPacketPong +from .dmrproto import DMRPPacketData, DMRPPacketTalkerAlias +from .dmrproto import DMRPPacketLogin, DMRPPacketSalt, DMRPPacketAuth +from .dmrproto import DMRPPacketMasterClose, DMRPPacketRepeaterClose +from .dmrproto import DMRPPacketMasterNoAck, DMRPPacketAck +from .peer import Peer, PeerKeeper +from .pphex import hexdump + + +class IPCDispatcher(ABC): + """ + Dispatcher interface to pass into controller to call back + """ + @abstractmethod + def send_dg(self, data: bytes, addr: tuple) -> None: + pass + + @abstractmethod + def get_peer_keeper(self) -> PeerKeeper: + pass + + @abstractmethod + def get_peer_auth(self) -> IPeerAuth: + pass + + +class PeerController: + def __init__(self, peer: Peer, dispatcher: IPCDispatcher) -> None: + self.peer: Peer = peer + self.dispatcher: IPCDispatcher = dispatcher + self.peer_auth: IPeerAuth = dispatcher.get_peer_auth() + + def process_packet(self, p: DMRPBasePacket) -> bool: + """ + Process packet in context of the peer + + Returns: + True - continue processing with dispatcher (for data) + False - stop processing (peer cmds) + """ + # CLOSE packet + if type(p) is DMRPPacketRepeaterClose: + logging.info(f"Peer {self.peer.logname} disconnected") + self.peer.die() + return False + + # DATA packet + if type(p) is DMRPPacketData: + if not self.peer.status.is_applicable(Peer.Status.ACTIVE): + logging.debug( + f"Got DMRPPacketData " + f"while status is {self.peer.status.name}") + logging.error(f"Data from inactive peer {self.peer.logname}. " + f"Closing connection.") + self.peer.die() + self.send_close(p.peer_id) + return False + + self.peer.update_active() + self.peer.update_unit(p.src_id) + return True + + # PING packet + if type(p) is DMRPPacketPing: + if not self.peer.status.is_applicable(Peer.Status.ACTIVE): + logging.debug( + f"Got DMRPPacketPing " + f"while status is {self.peer.status.name}") + logging.error(f"Ping from inactive peer {self.peer.logname}. " + f"Closing connection.") + self.peer.die() + self.send_close(p.peer_id) + return False + + logging.debug(f"Ping-pong {self.peer.logname}.") + self.peer.update_active() + self.send_pong() + return False + + # LOGIN sequence + if type(p) is DMRPPacketLogin: + if not self.peer.status.is_applicable(Peer.Status.LOGIN): + logging.debug( + f"Got DMRPPacketLogin " + f"while status is {self.peer.status.name}") + logging.error(f"Bad login sequence from {self.peer.logname}. " + f"Closing connection.") + self.peer.die() + self.send_close(p.peer_id) + return False + + peer_keeper: PeerKeeper = self.dispatcher.get_peer_keeper() + if len(peer_keeper.get_by_id(p.peer_id)) > 0: + logging.error(f"Auth {self.peer.logname}: " + f"peer id {p.peer_id} is already connected") + self.peer.die() + self.send_close(p.peer_id) + return False + + if not self.peer_auth.allow_peer_id(p.peer_id): + logging.error(f"Auth {self.peer.logname}: " + f"peer id {p.peer_id} disabled") + self.peer.die() + self.send_close(p.peer_id) + return False + + self.peer.status = Peer.Status.AUTH + logging.info(f"Login request from {self.peer.logname}" + f" ({p.peer_id})") + self.send_salt() + return False + + if type(p) is DMRPPacketAuth: + if not self.peer.status.is_applicable(Peer.Status.AUTH): + logging.debug( + f"Got DMRPPacketAuth " + f"while status is {self.peer.status.name}") + logging.error(f"Bad login sequence from {self.peer.logname}. " + f"Closing connection.") + self.peer.die() + self.send_close(p.peer_id) + return False + + if (self.peer.auth_salt is None or + not self.peer_auth.check_password(p.peer_id, + self.peer.auth_salt, + p.pass_hash)): + logging.error( + f"Auth {self.peer.logname}: password incorrect") + self.peer.die() + self.send_close(p.peer_id) + return False + + self.peer.peer_id = p.peer_id + self.peer.status = Peer.Status.CONFIG + logging.info(f"Auth success {self.peer.logname}") + self.send_ack_ok() + return False + + if type(p) is DMRPPacketConfig: + if not self.peer.status.is_applicable(Peer.Status.CONFIG): + logging.debug( + f"Got DMRPPacketConfig " + f"while status is {self.peer.status.name}") + logging.error(f"Bad login sequence from {self.peer.logname}. " + f"Closing connection.") + self.peer.die() + self.send_close(p.peer_id) + return False + + config = { + 'callsign': p.callsign, + 'rx_freq': p.rx_freq, + 'tx_freq': p.tx_freq, + 'power': p.power, + 'color_code': p.color_code, + 'lat': p.lat, + 'lon': p.lon, + 'height': p.height, + 'location': p.location, + 'description': p.description, + 'slots': p.slots, + 'url': p.url, + 'software_id': p.software_id, + 'package_id': p.package_id, + } + logging.info(f"Config from {self.peer.logname}: " + + ", ".join(f"{k}={v}" for k, v in config.items())); + + self.peer.status = Peer.Status.ACTIVE + self.peer.config = config + self.send_ack_ok() + return False + + return True + + def send_data(self, p: DMRPBasePacket) -> None: + data: bytes = p.get_data() + logging.debug( + f"Sending packet to {self.peer.logname} | {len(data)} bytes:\n" + f"{hexdump(data)}\n" + f"{str(p)}\n") + self.dispatcher.send_dg(p.get_data(), self.peer.addr) + + def send_close(self, peer_id: int = 0) -> None: + p = DMRPPacketMasterClose() + p.peer_id = peer_id if peer_id != 0 else self.peer.peer_id + self.send_data(p) + + def send_salt(self) -> None: + p = DMRPPacketSalt() + p.set_random_salt() + self.peer.auth_salt = p.salt + self.send_data(p) + + def send_ack_ok(self) -> None: + p = DMRPPacketAck() + p.peer_id = self.peer.peer_id + self.send_data(p) + + def send_pong(self) -> None: + p = DMRPPacketPong() + p.peer_id = self.peer.peer_id + self.send_data(p) diff --git a/localmaster_EXAMPLE.py b/localmaster_EXAMPLE.py new file mode 100755 index 0000000..f938006 --- /dev/null +++ b/localmaster_EXAMPLE.py @@ -0,0 +1,24 @@ +import logging +import time + +from dmrmaster import DMRMaster +from dmrtools import AllowAllPeerAuth, ListPeerAuth +from dmrtools.parrot_app import ParrotApp + + +class DMRMasterLocal(DMRMaster): + def config(self): + # Local config + self.set_peer_auth(AllowAllPeerAuth()) + # self.set_peer_auth(ListPeerAuth({1: 'pass1', 2: ''})) + self.register_app(ParrotApp(9990)) + + +def main(): + master = DMRMasterLocal(); + master.setup() + master.run() + + +if __name__ == '__main__': + main() diff --git a/requirements.txt b/requirements.txt new file mode 100755 index 0000000..c855156 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,37 @@ +annotated-types==0.7.0 +anyio==4.9.0 +asyncio==3.4.3 +certifi==2025.4.26 +click==8.1.8 +colorama==0.4.6 +dnspython==2.7.0 +email_validator==2.2.0 +fastapi==0.115.12 +fastapi-cli==0.0.7 +h11==0.16.0 +httpcore==1.0.9 +httptools==0.6.4 +httpx==0.28.1 +idna==3.10 +Jinja2==3.1.6 +markdown-it-py==3.0.0 +MarkupSafe==3.0.2 +mdurl==0.1.2 +pydantic==2.11.4 +pydantic_core==2.33.2 +Pygments==2.19.1 +python-dotenv==1.1.0 +python-multipart==0.0.20 +PyYAML==6.0.2 +rich==14.0.0 +rich-toolkit==0.14.4 +shellingham==1.5.4 +sniffio==1.3.1 +starlette==0.46.2 +tomli==2.2.1 +typer==0.15.3 +typing-inspection==0.4.0 +typing_extensions==4.13.2 +uvicorn==0.34.2 +watchfiles==1.0.5 +websockets==15.0.1