feat: /admin rest api endpoint

apps/wakunode2/app.nim

@@ -38,6 +38,7 @@ import
   ../../waku/waku_api/rest/lightpush/handlers as rest_lightpush_api,
   ../../waku/waku_api/rest/store/handlers as rest_store_api,
   ../../waku/waku_api/rest/health/handlers as rest_health_api,
+  ../../waku/waku_api/rest/admin/handlers as rest_admin_api,
   ../../waku/waku_api/jsonrpc/admin/handlers as rpc_admin_api,
   ../../waku/waku_api/jsonrpc/debug/handlers as rpc_debug_api,
   ../../waku/waku_api/jsonrpc/filter/handlers as rpc_filter_api,
@@ -567,6 +568,9 @@ proc startApp*(app: App): Future[AppResult[void]] {.async.} =
 proc startRestServer(app: App, address: ValidIpAddress, port: Port, conf: WakuNodeConf): AppResult[RestServerRef] =
   let server = ? newRestHttpServer(address, port)
 
+  ## Admin REST API
+  installAdminApiHandlers(server.router, app.node)
+
   ## Debug REST API
   installDebugApiHandlers(server.router, app.node)
 

tests/all_tests_waku.nim

@@ -98,7 +98,8 @@ import
   ./wakunode_rest/test_rest_store,
   ./wakunode_rest/test_rest_filter,
   ./wakunode_rest/test_rest_legacy_filter,
-  ./wakunode_rest/test_rest_lightpush
+  ./wakunode_rest/test_rest_lightpush,
+  ./wakunode_rest/test_rest_admin
 
 import
   ./waku_rln_relay/test_waku_rln_relay,

tests/wakunode_rest/test_all.nim

@@ -9,4 +9,5 @@ import
   ./test_rest_relay_serdes,
   ./test_rest_relay,
   ./test_rest_serdes,
-  ./test_rest_store
+  ./test_rest_store,
+  ./test_rest_admin

tests/wakunode_rest/test_rest_admin.nim

@@ -0,0 +1,79 @@
+{.used.}
+
+import
+  std/sequtils,
+  stew/shims/net,
+  testutils/unittests,
+  presto, presto/client as presto_client,
+  libp2p/crypto/crypto
+
+import
+  ../../waku/waku_core,
+  ../../waku/waku_node,
+  ../../waku/node/peer_manager,
+  ../../waku/waku_api/rest/server,
+  ../../waku/waku_api/rest/client,
+  ../../waku/waku_api/rest/responses,
+  ../../waku/waku_api/rest/admin/types,
+  ../../waku/waku_api/rest/admin/handlers as admin_api,
+  ../../waku/waku_api/rest/admin/client as admin_api_client,
+  ../../waku/waku_relay,
+  ../testlib/wakucore,
+  ../testlib/wakunode,
+  ../testlib/testasync
+
+suite "Waku v2 Rest API - Admin":
+  var node1 {.threadvar.}: WakuNode
+  var node2 {.threadvar.}: WakuNode
+  var node3 {.threadvar.}: WakuNode
+  var peerInfo2 {.threadvar.}: RemotePeerInfo
+  var peerInfo3 {.threadvar.}: RemotePeerInfo
+  var restServer {.threadvar.}: RestServerRef
+  var client{.threadvar.}: RestClientRef
+
+  asyncSetup:
+    node1 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("127.0.0.1"), Port(60600))
+    node2 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("127.0.0.1"), Port(60602))
+    peerInfo2 = node2.switch.peerInfo
+    node3 = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("127.0.0.1"), Port(60604))
+    peerInfo3 = node3.switch.peerInfo
+
+    await allFutures(node1.start(), node2.start(), node3.start())
+    await allFutures(node1.mountRelay(), node2.mountRelay(), node3.mountRelay())
+
+    let restPort = Port(58011)
+    let restAddress = ValidIpAddress.init("127.0.0.1")
+    restServer = RestServerRef.init(restAddress, restPort).tryGet()
+
+    installAdminApiHandlers(restServer.router, node1)
+
+    restServer.start()
+
+    client = newRestHttpClient(initTAddress(restAddress, restPort))
+
+  # asyncTearDown:
+    # await restServer.stop()
+    # await restServer.closeWait()
+    # await allFutures(node1.stop(), node2.stop(), node3.stop())
+
+  asyncTest "Set and get remote peers":
+    # Connect to nodes 2 and 3 using the Admin API
+    let postRes = await client.postPeers(@[constructMultiaddrStr(peerInfo2),
+                                           constructMultiaddrStr(peerInfo3)])
+
+    check:
+      postRes.status == 200
+
+    # Verify that newly connected peers are being managed
+    let getRes = await client.getPeers()
+
+    check:
+      getRes.status == 200
+      $getRes.contentType == $MIMETYPE_JSON
+      getRes.data.len() == 2
+      # Check peer 2
+      getRes.data.anyIt(it.protocol == WakuRelayCodec and
+                   it.multiaddr == constructMultiaddrStr(peerInfo2))
+      # Check peer 3
+      getRes.data.anyIt(it.protocol == WakuRelayCodec and
+                   it.multiaddr == constructMultiaddrStr(peerInfo3))

waku/waku_api/jsonrpc/admin/handlers.nim

@@ -14,30 +14,13 @@ import
   ../../../waku_relay,
   ../../../waku_node,
   ../../../node/peer_manager,
+  ../../../waku_core,
   ./types
 
 
 logScope:
   topics = "waku node jsonrpc admin_api"
 
-
-
-proc constructMultiaddrStr*(wireaddr: MultiAddress, peerId: PeerId): string =
-  # Constructs a multiaddress with both wire address and p2p identity
-  $wireaddr & "/p2p/" & $peerId
-
-proc constructMultiaddrStr*(peerInfo: PeerInfo): string =
-  # Constructs a multiaddress with both location (wire) address and p2p identity
-  if peerInfo.listenAddrs.len == 0:
-    return ""
-  constructMultiaddrStr(peerInfo.listenAddrs[0], peerInfo.peerId)
-
-proc constructMultiaddrStr*(remotePeerInfo: RemotePeerInfo): string =
-  # Constructs a multiaddress with both location (wire) address and p2p identity
-  if remotePeerInfo.addrs.len == 0:
-    return ""
-  constructMultiaddrStr(remotePeerInfo.addrs[0], remotePeerInfo.peerId)
-
 proc installAdminApiHandlers*(node: WakuNode, rpcsrv: RpcServer) =
 
   rpcsrv.rpc("post_waku_v2_admin_v1_peers") do (peers: seq[string]) -> bool:

waku/waku_api/rest/admin/client.nim

@@ -0,0 +1,74 @@
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import
+  chronicles,
+  json_serialization,
+  json_serialization/std/options,
+  presto/[route, client],
+  stew/byteutils
+
+import
+  ../serdes,
+  ../responses,
+  ./types
+
+export types
+
+
+logScope:
+  topics = "waku node rest admin api"
+
+proc decodeBytes*(t: typedesc[seq[WakuPeer]], data: openArray[byte],
+                  contentType: Opt[ContentTypeData]): RestResult[seq[WakuPeer]] =
+  if MediaType.init($contentType) != MIMETYPE_JSON:
+    error "Unsupported response contentType value", contentType = contentType
+    return err("Unsupported response contentType")
+
+  let decoded = decodeFromJsonBytes(seq[WakuPeer], data)
+
+  if decoded.isErr():
+    # return err(fmt("Invalid response from server, could not decode. {decoded.error}"))
+    return err("Invalid response from server, could not decode.")
+
+  return ok(decoded.get())
+
+# proc decodeBytes*(t: typedesc[WakuPeer], data: openArray[byte],
+#                   contentType: Opt[ContentTypeData]): RestResult[WakuPeer] =
+#   if MediaType.init($contentType) != MIMETYPE_JSON:
+#     error "Unsupported response contentType value", contentType = contentType
+#     return err("Unsupported response contentType")
+
+#   let decoded = ?decodeFromJsonBytes(WakuPeer, data)
+#   return ok(decoded)
+
+proc decodeBytes*(t: typedesc[string], value: openArray[byte],
+                  contentType: Opt[ContentTypeData]): RestResult[string] =
+  if MediaType.init($contentType) != MIMETYPE_TEXT:
+    error "Unsupported contentType value", contentType = contentType
+    return err("Unsupported contentType")
+
+  var res: string
+  if len(value) > 0:
+    res = newString(len(value))
+    copyMem(addr res[0], unsafeAddr value[0], len(value))
+  return ok(res)
+
+proc encodeBytes*(value: seq[string],
+                  contentType: string): RestResult[seq[byte]] =
+  if MediaType.init(contentType) != MIMETYPE_JSON:
+    error "Unsupported contentType value", contentType = contentType
+    return err("Unsupported contentType")
+
+  let encoded = ?encodeIntoJsonBytes(value)
+  return ok(encoded)
+
+proc getPeers*():
+    RestResponse[seq[WakuPeer]]
+    {.rest, endpoint: "/admin/v1/peers", meth: HttpMethod.MethodGet.}
+
+proc postPeers*(body: seq[string]):
+      RestResponse[string]
+      {.rest, endpoint: "/admin/v1/peers", meth: HttpMethod.MethodPost.}

waku/waku_api/rest/admin/handlers.nim

@@ -0,0 +1,111 @@
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import
+  std/strformat,
+  std/sequtils,
+  stew/byteutils,
+  chronicles,
+  json_serialization,
+  presto/route,
+  libp2p/[peerinfo, switch]
+
+import
+  ../../../waku_core,
+  ../../../waku_store,
+  ../../../waku_filter,
+  ../../../waku_relay,
+  ../../../waku_node,
+  ../../../node/peer_manager,
+  ../responses,
+  ../serdes,
+  ./types
+
+export types
+
+logScope:
+  topics = "waku node rest admin api"
+
+const ROUTE_ADMIN_V1_PEERS* = "/admin/v1/peers"
+
+func decodeRequestBody[T](contentBody: Option[ContentBody]) : Result[T, RestApiResponse] =
+  if contentBody.isNone():
+    return err(RestApiResponse.badRequest("Missing content body"))
+
+  let reqBodyContentType = MediaType.init($contentBody.get().contentType)
+  if reqBodyContentType != MIMETYPE_JSON:
+    return err(RestApiResponse.badRequest("Wrong Content-Type, expected application/json"))
+
+  let reqBodyData = contentBody.get().data
+
+  let requestResult = decodeFromJsonBytes(T, reqBodyData)
+  if requestResult.isErr():
+    return err(RestApiResponse.badRequest("Invalid content body, could not decode. " &
+                                          $requestResult.error))
+
+  return ok(requestResult.get())
+
+proc installAdminV1GetPeersHandler(router: var RestRouter, node: WakuNode) =
+  router.api(MethodGet, ROUTE_ADMIN_V1_PEERS) do () -> RestApiResponse:
+    var peers : seq[WakuPeer] = newSeq[WakuPeer]()
+
+    if not node.wakuRelay.isNil():
+      # Map managed peers to WakuPeers and add to return list
+      let relayPeers = node.peerManager
+                            .peerStore.peers(WakuRelayCodec)
+                            .mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
+                                            protocol: WakuRelayCodec,
+                                            connected: it.connectedness == Connectedness.Connected))
+
+      peers.add(relayPeers)
+
+    if not node.wakuFilterLegacy.isNil():
+      # Map WakuFilter peers to WakuPeers and add to return list
+      let filterPeers = node.peerManager.peerStore.peers(WakuLegacyFilterCodec)
+          .mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
+                          protocol: WakuLegacyFilterCodec,
+                          connected: it.connectedness == Connectedness.Connected))
+      peers.add(filterPeers)
+
+    if not node.wakuStore.isNil():
+      # Map WakuStore peers to WakuPeers and add to return list
+      let storePeers = node.peerManager.peerStore
+                           .peers(WakuStoreCodec)
+                           .mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
+                                          protocol: WakuStoreCodec,
+                                          connected: it.connectedness == Connectedness.Connected))
+      peers.add(storePeers)
+
+    let resp = RestApiResponse.jsonResponse(peers, status=Http200)
+    if resp.isErr():
+      error "An error ocurred while building the json respose: ", error=resp.error
+      return RestApiResponse.internalServerError(fmt("An error ocurred while building the json respose: {resp.error}"))
+
+    return resp.get()
+
+proc installAdminV1PostPeersHandler(router: var RestRouter, node: WakuNode) =
+  router.api(MethodPost, ROUTE_ADMIN_V1_PEERS) do (contentBody: Option[ContentBody]) -> RestApiResponse:
+
+    let decodedBody = decodeRequestBody[seq[string]](contentBody)
+
+    if decodedBody.isErr():
+      return RestApiResponse.badRequest(fmt("Failed to decode request: {decodedBody.error}"))
+
+    let peers: seq[string] = decodedBody.value()
+
+    for i, peer in peers:
+      let peerInfo = parsePeerInfo(peer)
+      if peerInfo.isErr():
+        return RestApiResponse.badRequest(fmt("Couldn't parse remote peer info: {peerInfo.error}"))
+
+      let connOk = await node.peerManager.connectRelay(peerInfo.value, source="rest")
+      if not connOk:
+        return RestApiResponse.badRequest(fmt("Failed to connect to peer at index: {i} - {peer}"))
+
+    return RestApiResponse.ok()
+
+proc installAdminApiHandlers*(router: var RestRouter, node: WakuNode) =
+  installAdminV1GetPeersHandler(router, node)
+  installAdminV1PostPeersHandler(router, node)