Added REST and WebSocket API endpoints to support peer management, messaging, identity handling, and service control.

Author: sumanjeet0012

py-peer/API_REFERENCE.md

@@ -0,0 +1,3 @@
+"""
+Tornado REST + WebSocket API package for py-peer Universal Connectivity DApp.
+"""

py-peer/api/__init__.py

@@ -0,0 +1,107 @@
+"""
+Base handler for all Tornado API endpoints.
+
+Provides:
+- JSON response helpers
+- CORS headers
+- Service readiness check
+- Uniform error envelope
+"""
+
+import json
+import time
+import logging
+import traceback
+
+import tornado.web
+
+logger = logging.getLogger("api.base")
+
+
+class BaseHandler(tornado.web.RequestHandler):
+    """Base class for all REST API handlers."""
+
+    def initialize(self, service):
+        """Inject the HeadlessService instance."""
+        self.service = service
+
+    # ------------------------------------------------------------------ #
+    #  CORS                                                                #
+    # ------------------------------------------------------------------ #
+    def set_default_headers(self):
+        self.set_header("Access-Control-Allow-Origin", "*")
+        self.set_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
+        self.set_header("Access-Control-Allow-Headers", "Content-Type, X-API-Key")
+        self.set_header("Content-Type", "application/json")
+
+    def options(self, *args, **kwargs):
+        """Handle CORS preflight."""
+        self.set_status(204)
+        self.finish()
+
+    # ------------------------------------------------------------------ #
+    #  JSON helpers                                                        #
+    # ------------------------------------------------------------------ #
+    def send_success(self, data=None, status=200):
+        self.set_status(status)
+        self.finish(json.dumps({
+            "success": True,
+            "data": data,
+            "error": None,
+            "timestamp": time.time(),
+        }))
+
+    def send_error_response(self, message, status=400, detail=None):
+        self.set_status(status)
+        self.finish(json.dumps({
+            "success": False,
+            "data": None,
+            "error": {
+                "code": status,
+                "message": message,
+                "detail": detail,
+            },
+            "timestamp": time.time(),
+        }))
+
+    # ------------------------------------------------------------------ #
+    #  Request body helpers                                                #
+    # ------------------------------------------------------------------ #
+    def get_json_body(self):
+        try:
+            return json.loads(self.request.body)
+        except (json.JSONDecodeError, Exception):
+            return {}
+
+    # ------------------------------------------------------------------ #
+    #  Service readiness guard                                             #
+    # ------------------------------------------------------------------ #
+    def require_ready(self):
+        """Return False and send 503 if the service is not ready yet."""
+        if not self.service or not self.service.ready:
+            self.send_error_response(
+                "Service not ready yet — HeadlessService is still initialising.",
+                status=503,
+            )
+            return False
+        return True
+
+    # ------------------------------------------------------------------ #
+    #  Global exception handler                                            #
+    # ------------------------------------------------------------------ #
+    def write_error(self, status_code, **kwargs):
+        exc_info = kwargs.get("exc_info")
+        detail = None
+        if exc_info:
+            detail = traceback.format_exception(*exc_info)[-1].strip()
+        self.set_header("Content-Type", "application/json")
+        self.finish(json.dumps({
+            "success": False,
+            "data": None,
+            "error": {
+                "code": status_code,
+                "message": self._reason,
+                "detail": detail,
+            },
+            "timestamp": time.time(),
+        }))

py-peer/api/base.py

@@ -0,0 +1,74 @@
+"""
+DHT endpoints.
+
+GET /api/v1/dht/status         - mode, routing table size, random walk
+GET /api/v1/dht/peers          - peer IDs in DHT routing table
+GET /api/v1/dht/routing-table  - concise routing table dump
+"""
+
+from .base import BaseHandler
+
+
+class DHTStatusHandler(BaseHandler):
+    """GET /api/v1/dht/status"""
+
+    def get(self):
+        if not self.require_ready():
+            return
+        dht = self.service.dht
+        if not dht:
+            self.send_error_response("DHT is not initialised.", status=503)
+            return
+        try:
+            rt_size = len(list(dht.routing_table.get_peer_ids()))
+        except Exception:
+            rt_size = -1
+
+        # DHTMode enum → string
+        mode_val = getattr(dht, "mode", None)
+        mode_str = mode_val.name if hasattr(mode_val, "name") else str(mode_val)
+
+        self.send_success({
+            "mode": mode_str,
+            "random_walk_enabled": getattr(dht, "enable_random_walk", False),
+            "routing_table_size": rt_size,
+        })
+
+
+class DHTPeersHandler(BaseHandler):
+    """GET /api/v1/dht/peers"""
+
+    def get(self):
+        if not self.require_ready():
+            return
+        dht = self.service.dht
+        if not dht:
+            self.send_error_response("DHT is not initialised.", status=503)
+            return
+        try:
+            peers = [str(p) for p in dht.routing_table.get_peer_ids()]
+        except Exception as e:
+            self.send_error_response(f"Could not read routing table: {e}", status=500)
+            return
+        self.send_success({"peers": peers, "count": len(peers)})
+
+
+class DHTRoutingTableHandler(BaseHandler):
+    """GET /api/v1/dht/routing-table"""
+
+    def get(self):
+        if not self.require_ready():
+            return
+        dht = self.service.dht
+        if not dht:
+            self.send_error_response("DHT is not initialised.", status=503)
+            return
+        try:
+            peers = [str(p) for p in dht.routing_table.get_peer_ids()]
+        except Exception as e:
+            self.send_error_response(f"Could not read routing table: {e}", status=500)
+            return
+        self.send_success({
+            "routing_table": peers,
+            "total_peers": len(peers),
+        })

py-peer/api/dht.py

@@ -0,0 +1,163 @@
+"""
+File sharing endpoints (Bitswap / MerkleDag).
+
+GET  /api/v1/files/shared          - list files this node has shared
+GET  /api/v1/files/shared/{cid}    - metadata for a specific shared file
+POST /api/v1/files/share           - share a local file to a topic
+POST /api/v1/files/download        - download a file by CID hex
+POST /api/v1/files/upload          - upload via multipart and share to a topic
+"""
+
+import os
+from .base import BaseHandler
+
+
+class SharedFilesHandler(BaseHandler):
+    """GET /api/v1/files/shared"""
+
+    def get(self):
+        if not self.require_ready():
+            return
+        files = [
+            {"cid": cid, **meta}
+            for cid, meta in self.service.shared_files.items()
+        ]
+        self.send_success({"shared_files": files, "count": len(files)})
+
+
+class SharedFileDetailHandler(BaseHandler):
+    """GET /api/v1/files/shared/{cid}"""
+
+    def get(self, cid):
+        if not self.require_ready():
+            return
+        meta = self.service.shared_files.get(cid)
+        if not meta:
+            self.send_error_response(f"No shared file with CID '{cid}'.", status=404)
+            return
+        self.send_success({"cid": cid, **meta})
+
+
+class ShareFileHandler(BaseHandler):
+    """POST /api/v1/files/share — share a file that already exists on disk"""
+
+    def post(self):
+        if not self.require_ready():
+            return
+        body = self.get_json_body()
+        file_path = body.get("file_path", "").strip()
+        topic = body.get("topic", "").strip()
+
+        if not file_path:
+            self.send_error_response("'file_path' is required.")
+            return
+        if not topic:
+            self.send_error_response("'topic' is required.")
+            return
+        if not os.path.exists(file_path):
+            self.send_error_response(f"File not found: {file_path}", status=400)
+            return
+
+        subscribed = self.service.get_subscribed_topics()
+        if topic not in subscribed:
+            self.send_error_response(
+                f"Not subscribed to topic '{topic}'. Subscribe first via POST /api/v1/topics.",
+                status=400,
+            )
+            return
+
+        queued = self.service.share_file(file_path, topic)
+        if queued:
+            filename = os.path.basename(file_path)
+            self.send_success(
+                {"message": "File share request queued", "filename": filename, "topic": topic},
+                status=202,
+            )
+        else:
+            self.send_error_response("Failed to queue file share — service not ready.", status=503)
+
+
+class DownloadFileHandler(BaseHandler):
+    """POST /api/v1/files/download — download a file by CID hex"""
+
+    def post(self):
+        if not self.require_ready():
+            return
+        body = self.get_json_body()
+        cid = body.get("file_cid", "").strip()
+        name = body.get("file_name", "unknown").strip()
+
+        if not cid:
+            self.send_error_response("'file_cid' is required.")
+            return
+
+        queued = self.service.download_file(cid, name)
+        if queued:
+            self.send_success(
+                {"message": "Download request queued", "file_cid": cid, "file_name": name},
+                status=202,
+            )
+        else:
+            self.send_error_response("Failed to queue download — service not ready.", status=503)
+
+
+class UploadAndShareHandler(BaseHandler):
+    """
+    POST /api/v1/files/upload
+    Accepts multipart/form-data with fields:
+      - file   : the file bytes
+      - topic  : the topic to share to
+    Saves the file to the service's download_dir and queues a share.
+    """
+
+    def post(self):
+        if not self.require_ready():
+            return
+
+        topic = self.get_argument("topic", "").strip()
+        if not topic:
+            self.send_error_response("'topic' form field is required.")
+            return
+
+        if "file" not in self.request.files:
+            self.send_error_response("'file' form-data field is required.")
+            return
+
+        file_info = self.request.files["file"][0]
+        filename = file_info["filename"] or "upload"
+        file_data = file_info["body"]
+
+        # Save to download_dir
+        save_path = os.path.join(self.service.download_dir, filename)
+        # Handle name collisions
+        counter = 1
+        base, ext = os.path.splitext(filename)
+        while os.path.exists(save_path):
+            save_path = os.path.join(self.service.download_dir, f"{base}_{counter}{ext}")
+            counter += 1
+
+        with open(save_path, "wb") as f:
+            f.write(file_data)
+
+        subscribed = self.service.get_subscribed_topics()
+        if topic not in subscribed:
+            self.send_error_response(
+                f"Not subscribed to topic '{topic}'. Subscribe first via POST /api/v1/topics.",
+                status=400,
+            )
+            return
+
+        queued = self.service.share_file(save_path, topic)
+        if queued:
+            self.send_success(
+                {
+                    "message": "File uploaded and share request queued",
+                    "filename": os.path.basename(save_path),
+                    "size": len(file_data),
+                    "topic": topic,
+                    "saved_path": save_path,
+                },
+                status=202,
+            )
+        else:
+            self.send_error_response("Failed to queue file share — service not ready.", status=503)