Node endpoints (#1778)

## Issue Addressed

`node` endpoints in #1434

## Proposed Changes

Implement these:
```
 /eth/v1/node/health
 /eth/v1/node/peers/{peer_id}
 /eth/v1/node/peers
```
- Add an `Option<Enr>` to `PeerInfo`
- Finish implementation of `/eth/v1/node/identity`

## Additional Info
- should update the `peers` endpoints when #1764 is resolved



Co-authored-by: realbigsean <seananderson33@gmail.com>

beacon_node/http_api/Cargo.toml

@@ -24,6 +24,7 @@ lighthouse_metrics = { path = "../../common/lighthouse_metrics" }
 lazy_static = "1.4.0"
 warp_utils = { path = "../../common/warp_utils" }
 slot_clock = { path = "../../common/slot_clock" }
+bs58 = "0.3.1"
 
 [dev-dependencies]
 store = { path = "../store" }

beacon_node/http_api/src/lib.rs

@@ -21,7 +21,7 @@ use eth2::{
     types::{self as api_types, ValidatorId},
     StatusCode,
 };
-use eth2_libp2p::{types::SyncState, NetworkGlobals, PubsubMessage};
+use eth2_libp2p::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
 use lighthouse_version::version_with_platform;
 use network::NetworkMessage;
 use parking_lot::Mutex;
@@ -109,7 +109,11 @@ pub fn slog_logging(
 ) -> warp::filters::log::Log<impl Fn(warp::filters::log::Info) + Clone> {
     warp::log::custom(move |info| {
         match info.status() {
-            status if status == StatusCode::OK || status == StatusCode::NOT_FOUND => {
+            status
+                if status == StatusCode::OK
+                    || status == StatusCode::NOT_FOUND
+                    || status == StatusCode::PARTIAL_CONTENT =>
+            {
                 trace!(
                     log,
                     "Processed HTTP API request";
@@ -1106,10 +1110,28 @@ pub fn serve<T: BeaconChainTypes>(
         .and(network_globals.clone())
         .and_then(|network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
             blocking_json_task(move || {
+                let enr = network_globals.local_enr();
+                let p2p_addresses = enr.multiaddr_p2p_tcp();
+                let discovery_addresses = enr.multiaddr_p2p_udp();
                 Ok(api_types::GenericResponse::from(api_types::IdentityData {
                     peer_id: network_globals.local_peer_id().to_base58(),
-                    enr: network_globals.local_enr(),
-                    p2p_addresses: network_globals.listen_multiaddrs(),
+                    enr,
+                    p2p_addresses,
+                    discovery_addresses,
+                    metadata: api_types::MetaData {
+                        seq_number: network_globals.local_metadata.read().seq_number,
+                        attnets: format!(
+                            "0x{}",
+                            hex::encode(
+                                network_globals
+                                    .local_metadata
+                                    .read()
+                                    .attnets
+                                    .clone()
+                                    .into_bytes()
+                            ),
+                        ),
+                    },
                 }))
             })
         });
@@ -1159,6 +1181,122 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // GET node/health
+    let get_node_health = eth1_v1
+        .and(warp::path("node"))
+        .and(warp::path("health"))
+        .and(warp::path::end())
+        .and(network_globals.clone())
+        .and_then(|network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
+            blocking_task(move || match *network_globals.sync_state.read() {
+                SyncState::SyncingFinalized { .. } | SyncState::SyncingHead { .. } => {
+                    Ok(warp::reply::with_status(
+                        warp::reply(),
+                        warp::http::StatusCode::PARTIAL_CONTENT,
+                    ))
+                }
+                SyncState::Synced => Ok(warp::reply::with_status(
+                    warp::reply(),
+                    warp::http::StatusCode::OK,
+                )),
+                SyncState::Stalled => Err(warp_utils::reject::not_synced(
+                    "sync stalled, beacon chain may not yet be initialized.".to_string(),
+                )),
+            })
+        });
+
+    // GET node/peers/{peer_id}
+    let get_node_peers_by_id = eth1_v1
+        .and(warp::path("node"))
+        .and(warp::path("peers"))
+        .and(warp::path::param::<String>())
+        .and(warp::path::end())
+        .and(network_globals.clone())
+        .and_then(
+            |requested_peer_id: String, network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
+                blocking_json_task(move || {
+                    let peer_id = PeerId::from_bytes(
+                        bs58::decode(requested_peer_id.as_str())
+                            .into_vec()
+                            .map_err(|e| {
+                                warp_utils::reject::custom_bad_request(format!(
+                                    "invalid peer id: {}",
+                                    e
+                                ))
+                            })?,
+                    )
+                    .map_err(|_| {
+                        warp_utils::reject::custom_bad_request("invalid peer id.".to_string())
+                    })?;
+
+                    if let Some(peer_info) = network_globals.peers.read().peer_info(&peer_id) {
+                        //TODO: update this to seen_addresses once #1764 is resolved
+                        let address = match peer_info.listening_addresses.get(0) {
+                            Some(addr) => addr.to_string(),
+                            None => "".to_string(), // this field is non-nullable in the eth2 API spec
+                        };
+
+                        // the eth2 API spec implies only peers we have been connected to at some point should be included.
+                        if let Some(dir) = peer_info.connection_direction.as_ref() {
+                            return Ok(api_types::GenericResponse::from(api_types::PeerData {
+                                peer_id: peer_id.to_string(),
+                                enr: peer_info.enr.as_ref().map(|enr| enr.to_base64()),
+                                last_seen_p2p_address: address,
+                                direction: api_types::PeerDirection::from_connection_direction(
+                                    &dir,
+                                ),
+                                state: api_types::PeerState::from_peer_connection_status(
+                                    &peer_info.connection_status,
+                                ),
+                            }));
+                        }
+                    }
+                    Err(warp_utils::reject::custom_not_found(
+                        "peer not found.".to_string(),
+                    ))
+                })
+            },
+        );
+
+    // GET node/peers
+    let get_node_peers = eth1_v1
+        .and(warp::path("node"))
+        .and(warp::path("peers"))
+        .and(warp::path::end())
+        .and(network_globals.clone())
+        .and_then(|network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
+            blocking_json_task(move || {
+                let mut peers: Vec<api_types::PeerData> = Vec::new();
+                network_globals
+                    .peers
+                    .read()
+                    .peers()
+                    // the eth2 API spec implies only peers we have been connected to at some point should be included.
+                    .filter(|(_, peer_info)| peer_info.connection_direction.is_some())
+                    .for_each(|(peer_id, peer_info)| {
+                        //TODO: update this to seen_addresses once #1764 is resolved
+                        let address = match peer_info.listening_addresses.get(0) {
+                            Some(addr) => addr.to_string(),
+                            None => "".to_string(), // this field is non-nullable in the eth2 API spec
+                        };
+                        if let Some(dir) = peer_info.connection_direction.as_ref() {
+                            peers.push(api_types::PeerData {
+                                peer_id: peer_id.to_string(),
+                                enr: peer_info.enr.as_ref().map(|enr| enr.to_base64()),
+                                last_seen_p2p_address: address,
+                                direction: api_types::PeerDirection::from_connection_direction(
+                                    &dir,
+                                ),
+                                state: api_types::PeerState::from_peer_connection_status(
+                                    &peer_info.connection_status,
+                                ),
+                            });
+                        }
+                    });
+                Ok(api_types::GenericResponse::from(peers))
+            })
+        });
+
     /*
      * validator
      */
@@ -1653,6 +1791,9 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_node_identity.boxed())
                 .or(get_node_version.boxed())
                 .or(get_node_syncing.boxed())
+                .or(get_node_health.boxed())
+                .or(get_node_peers_by_id.boxed())
+                .or(get_node_peers.boxed())
                 .or(get_validator_duties_attester.boxed())
                 .or(get_validator_duties_proposer.boxed())
                 .or(get_validator_blocks.boxed())

beacon_node/http_api/tests/tests.rs

@@ -8,7 +8,7 @@ use eth2::{types::*, BeaconNodeHttpClient, Url};
 use eth2_libp2p::{
     rpc::methods::MetaData,
     types::{EnrBitfield, SyncState},
-    NetworkGlobals,
+    Enr, EnrExt, NetworkGlobals, PeerId,
 };
 use http_api::{Config, Context};
 use network::NetworkMessage;
@@ -23,6 +23,7 @@ use types::{
     test_utils::generate_deterministic_keypairs, AggregateSignature, BeaconState, BitList, Domain,
     EthSpec, Hash256, Keypair, MainnetEthSpec, RelativeEpoch, SelectionProof, SignedRoot, Slot,
 };
+use warp::http::StatusCode;
 
 type E = MainnetEthSpec;
 
@@ -31,6 +32,10 @@ const VALIDATOR_COUNT: usize = SLOTS_PER_EPOCH as usize;
 const CHAIN_LENGTH: u64 = SLOTS_PER_EPOCH * 5;
 const JUSTIFIED_EPOCH: u64 = 4;
 const FINALIZED_EPOCH: u64 = 3;
+const TCP_PORT: u16 = 42;
+const UDP_PORT: u16 = 42;
+const SEQ_NUMBER: u64 = 0;
+const EXTERNAL_ADDR: &str = "/ip4/0.0.0.0";
 
 /// Skipping the slots around the epoch boundary allows us to check that we're obtaining states
 /// from skipped slots for the finalized and justified checkpoints (instead of the state from the
@@ -53,6 +58,8 @@ struct ApiTester {
     _server_shutdown: oneshot::Sender<()>,
     validator_keypairs: Vec<Keypair>,
     network_rx: mpsc::UnboundedReceiver<NetworkMessage<E>>,
+    local_enr: Enr,
+    external_peer_id: PeerId,
 }
 
 impl ApiTester {
@@ -139,12 +146,24 @@ impl ApiTester {
 
         // Default metadata
         let meta_data = MetaData {
-            seq_number: 0,
-            attnets: EnrBitfield::<MinimalEthSpec>::default(),
+            seq_number: SEQ_NUMBER,
+            attnets: EnrBitfield::<MainnetEthSpec>::default(),
         };
         let enr_key = CombinedKey::generate_secp256k1();
         let enr = EnrBuilder::new("v4").build(&enr_key).unwrap();
-        let network_globals = NetworkGlobals::new(enr, 42, 42, meta_data, vec![], &log);
+        let enr_clone = enr.clone();
+        let network_globals = NetworkGlobals::new(enr, TCP_PORT, UDP_PORT, meta_data, vec![], &log);
+
+        let peer_id = PeerId::random();
+        network_globals.peers.write().connect_ingoing(
+            &peer_id,
+            EXTERNAL_ADDR.parse().unwrap(),
+            None,
+        );
+        //TODO: have to update this once #1764 is resolved
+        if let Some(peer_info) = network_globals.peers.write().peer_info_mut(&peer_id) {
+            peer_info.listening_addresses = vec![EXTERNAL_ADDR.parse().unwrap()];
+        }
 
         *network_globals.sync_state.write() = SyncState::Synced;
 
@@ -190,6 +209,8 @@ impl ApiTester {
             _server_shutdown: shutdown_tx,
             validator_keypairs: harness.validator_keypairs,
             network_rx,
+            local_enr: enr_clone,
+            external_peer_id: peer_id,
         }
     }
 
@@ -1004,6 +1025,69 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_get_node_identity(self) -> Self {
+        let result = self.client.get_node_identity().await.unwrap().data;
+
+        let expected = IdentityData {
+            peer_id: self.local_enr.peer_id().to_string(),
+            enr: self.local_enr.clone(),
+            p2p_addresses: self.local_enr.multiaddr_p2p_tcp(),
+            discovery_addresses: self.local_enr.multiaddr_p2p_udp(),
+            metadata: eth2::types::MetaData {
+                seq_number: 0,
+                attnets: "0x0000000000000000".to_string(),
+            },
+        };
+
+        assert_eq!(result, expected);
+
+        self
+    }
+
+    pub async fn test_get_node_health(self) -> Self {
+        let status = self.client.get_node_health().await.unwrap();
+        assert_eq!(status, StatusCode::OK);
+
+        self
+    }
+
+    pub async fn test_get_node_peers_by_id(self) -> Self {
+        let result = self
+            .client
+            .get_node_peers_by_id(self.external_peer_id.clone())
+            .await
+            .unwrap()
+            .data;
+
+        let expected = PeerData {
+            peer_id: self.external_peer_id.to_string(),
+            enr: None,
+            last_seen_p2p_address: EXTERNAL_ADDR.to_string(),
+            state: PeerState::Connected,
+            direction: PeerDirection::Inbound,
+        };
+
+        assert_eq!(result, expected);
+
+        self
+    }
+
+    pub async fn test_get_node_peers(self) -> Self {
+        let result = self.client.get_node_peers().await.unwrap().data;
+
+        let expected = PeerData {
+            peer_id: self.external_peer_id.to_string(),
+            enr: None,
+            last_seen_p2p_address: EXTERNAL_ADDR.to_string(),
+            state: PeerState::Connected,
+            direction: PeerDirection::Inbound,
+        };
+
+        assert_eq!(result, vec![expected]);
+
+        self
+    }
+
     pub async fn test_get_debug_beacon_states(self) -> Self {
         for state_id in self.interesting_state_ids() {
             let result = self
@@ -1660,6 +1744,14 @@ async fn node_get() {
         .test_get_node_version()
         .await
         .test_get_node_syncing()
+        .await
+        .test_get_node_identity()
+        .await
+        .test_get_node_health()
+        .await
+        .test_get_node_peers_by_id()
+        .await
+        .test_get_node_peers()
         .await;
 }