rns-page-node/rns_page_node/main.py

#!/usr/bin/env python3
"""
Minimal Reticulum Page Node
Serves .mu pages and files over RNS.
"""

import os
import time
import threading
import subprocess
import RNS
import argparse

DEFAULT_INDEX = '''>Default Home Page

This node is serving pages using page node, but the home page file (index.mu) was not found in the pages directory. Please add an index.mu file to customize the home page.
'''

DEFAULT_NOTALLOWED = '''>Request Not Allowed

You are not authorised to carry out the request.
'''

class PageNode:
    def __init__(self, identity, pagespath, filespath, announce_interval=360, name=None):
        self.identity = identity
        self.name = name
        self.pagespath = pagespath
        self.filespath = filespath
        self.destination = RNS.Destination(
            identity,
            RNS.Destination.IN,
            RNS.Destination.SINGLE,
            "nomadnetwork",
            "node"
        )
        self.announce_interval = announce_interval
        self.last_announce = 0

        self.register_pages()
        self.register_files()

        self.destination.set_link_established_callback(self.on_connect)

        threading.Thread(target=self._announce_loop, daemon=True).start()

    def register_pages(self):
        self.servedpages = []
        self._scan_pages(self.pagespath)

        if not os.path.isfile(os.path.join(self.pagespath, "index.mu")):
            self.destination.register_request_handler(
                "/page/index.mu",
                response_generator=self.serve_default_index,
                allow=RNS.Destination.ALLOW_ALL
            )

        for full_path in self.servedpages:
            rel = full_path[len(self.pagespath):]
            request_path = f"/page{rel}"
            self.destination.register_request_handler(
                request_path,
                response_generator=self.serve_page,
                allow=RNS.Destination.ALLOW_ALL
            )

    def register_files(self):
        self.servedfiles = []
        self._scan_files(self.filespath)

        for full_path in self.servedfiles:
            rel = full_path[len(self.filespath):]
            request_path = f"/file{rel}"
            self.destination.register_request_handler(
                request_path,
                response_generator=self.serve_file,
                allow=RNS.Destination.ALLOW_ALL,
                auto_compress=32_000_000
            )

    def _scan_pages(self, base):
        for entry in os.listdir(base):
            if entry.startswith('.'):
                continue
            path = os.path.join(base, entry)
            if os.path.isdir(path):
                self._scan_pages(path)
            elif os.path.isfile(path) and not entry.endswith(".allowed"):
                self.servedpages.append(path)

    def _scan_files(self, base):
        for entry in os.listdir(base):
            if entry.startswith('.'):
                continue
            path = os.path.join(base, entry)
            if os.path.isdir(path):
                self._scan_files(path)
            elif os.path.isfile(path):
                self.servedfiles.append(path)

    def serve_default_index(self, path, data, request_id, link_id, remote_identity, requested_at):
        return DEFAULT_INDEX.encode('utf-8')

    def serve_page(self, path, data, request_id, link_id, remote_identity, requested_at):
        file_path = path.replace("/page", self.pagespath, 1)
        if os.access(file_path, os.X_OK):
            result = subprocess.run([file_path], stdout=subprocess.PIPE)
            return result.stdout
        else:
            with open(file_path, 'rb') as f:
                return f.read()

    def serve_file(self, path, data, request_id, link_id, remote_identity, requested_at):
        file_path = path.replace("/file", self.filespath, 1)
        return [open(file_path, 'rb'), {"name": os.path.basename(file_path).encode('utf-8')}]

    def on_connect(self, link):
        pass

    def _announce_loop(self):
        while True:
            if time.time() - self.last_announce > self.announce_interval:
                if self.name:
                    self.destination.announce(app_data=self.name.encode('utf-8'))
                else:
                    self.destination.announce()
                self.last_announce = time.time()
            time.sleep(1)


def main():
    parser = argparse.ArgumentParser(description="Minimal Reticulum Page Node")
    parser.add_argument('-c', '--config', dest='configpath', help='Reticulum config path', default=None)
    parser.add_argument('-p', '--pages-dir', dest='pages_dir', help='Pages directory', default=os.path.join(os.getcwd(), 'pages'))
    parser.add_argument('-f', '--files-dir', dest='files_dir', help='Files directory', default=os.path.join(os.getcwd(), 'files'))
    parser.add_argument('-n', '--node-name', dest='node_name', help='Node display name', default=None)
    parser.add_argument('-a', '--announce-interval', dest='announce_interval', type=int, help='Announce interval in seconds', default=360)
    parser.add_argument('-i', '--identity-dir', dest='identity_dir', help='Directory to store node identity', default=os.path.join(os.getcwd(), 'node-config'))
    args = parser.parse_args()

    configpath = args.configpath
    pages_dir = args.pages_dir
    files_dir = args.files_dir
    node_name = args.node_name
    announce_interval = args.announce_interval
    identity_dir = args.identity_dir

    RNS.Reticulum(configpath)
    os.makedirs(identity_dir, exist_ok=True)
    identity_file = os.path.join(identity_dir, 'identity')
    if os.path.isfile(identity_file):
        identity = RNS.Identity.from_file(identity_file)
    else:
        identity = RNS.Identity()
        identity.to_file(identity_file)

    os.makedirs(pages_dir, exist_ok=True)
    os.makedirs(files_dir, exist_ok=True)

    node = PageNode(identity, pages_dir, files_dir, announce_interval, node_name)
    print("Page node running. Press Ctrl-C to exit.")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Shutting down.")

if __name__ == '__main__':
    main()