cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'api-alignment' into interop

Browse Source
This commit is contained in:
Luke Anderson 2019-09-16 09:36:47 +10:00
commit 43c110f228
No known key found for this signature in database
GPG Key ID: 44408169EC61E228
19 changed files with 1944 additions and 716 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
beacon_node/client/src/lib.rs
View File

@ -240,6 +240,7 @@ where
executor,
beacon_chain.clone(),
network.clone(),
network_send.clone(),
client_config.db_path().expect("unable to read datadir"),
eth2_config.clone(),
&log,

6
beacon_node/rest_api/Cargo.toml
View File

@ -25,8 +25,7 @@ types = { path = "../../eth2/types" }
clap = "2.32.0"
http = "^0.1.17"
prometheus = { version = "^0.6", features = ["process"] }
hyper = "0.12.32"
futures = "0.1"
hyper = "0.12.34"
exit-future = "0.1.3"
tokio = "0.1.17"
url = "2.0"
@ -35,3 +34,6 @@ eth2_config = { path = "../../eth2/utils/eth2_config" }
lighthouse_metrics = { path = "../../eth2/utils/lighthouse_metrics" }
slot_clock = { path = "../../eth2/utils/slot_clock" }
hex = "0.3.2"
parking_lot = "0.9"
futures = "0.1.25"

171
beacon_node/rest_api/src/beacon.rs
View File

@ -1,18 +1,25 @@
use super::{success_response, ApiResult, ResponseBuilder};
use crate::{helpers::*, ApiError, UrlQuery};
use crate::helpers::*;
use crate::response_builder::ResponseBuilder;
use crate::{ApiError, ApiResult, UrlQuery};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use hyper::{Body, Request};
use serde::Serialize;
use ssz_derive::Encode;
use std::sync::Arc;
use store::Store;
use types::{BeaconBlock, BeaconState, EthSpec, Hash256, Slot};
use types::{BeaconBlock, BeaconState, Epoch, EthSpec, Hash256, Slot, Validator};
#[derive(Serialize)]
#[derive(Serialize, Encode)]
pub struct HeadResponse {
pub slot: Slot,
pub block_root: Hash256,
pub state_root: Hash256,
pub finalized_slot: Slot,
pub finalized_block_root: Hash256,
pub justified_slot: Slot,
pub justified_block_root: Hash256,
pub previous_justified_slot: Slot,
pub previous_justified_block_root: Hash256,
}
/// HTTP handler to return a `BeaconBlock` at a given `root` or `slot`.
@ -22,16 +29,33 @@ pub fn get_head<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let chain_head = beacon_chain.head();
let head = HeadResponse {
slot: beacon_chain.head().beacon_state.slot,
block_root: beacon_chain.head().beacon_block_root,
state_root: beacon_chain.head().beacon_state_root,
slot: chain_head.beacon_state.slot,
block_root: chain_head.beacon_block_root,
state_root: chain_head.beacon_state_root,
finalized_slot: chain_head
.beacon_state
.finalized_checkpoint
.epoch
.start_slot(T::EthSpec::slots_per_epoch()),
finalized_block_root: chain_head.beacon_state.finalized_checkpoint.root,
justified_slot: chain_head
.beacon_state
.current_justified_checkpoint
.epoch
.start_slot(T::EthSpec::slots_per_epoch()),
justified_block_root: chain_head.beacon_state.current_justified_checkpoint.root,
previous_justified_slot: chain_head
.beacon_state
.previous_justified_checkpoint
.epoch
.start_slot(T::EthSpec::slots_per_epoch()),
previous_justified_block_root: chain_head.beacon_state.previous_justified_checkpoint.root,
};
let json: String = serde_json::to_string(&head)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize HeadResponse: {:?}", e)))?;
Ok(success_response(Body::from(json)))
ResponseBuilder::new(&req)?.body(&head)
}
#[derive(Serialize, Encode)]
@ -56,7 +80,7 @@ pub fn get_block<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
let target = parse_slot(&value)?;
block_root_at_slot(&beacon_chain, target).ok_or_else(|| {
ApiError::NotFound(format!("Unable to find BeaconBlock for slot {}", target))
ApiError::NotFound(format!("Unable to find BeaconBlock for slot {:?}", target))
})?
}
("root", value) => parse_root(&value)?,
@ -68,7 +92,7 @@ pub fn get_block<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
.get::<BeaconBlock<T::EthSpec>>(&block_root)?
.ok_or_else(|| {
ApiError::NotFound(format!(
"Unable to find BeaconBlock for root {}",
"Unable to find BeaconBlock for root {:?}",
block_root
))
})?;
@ -78,42 +102,59 @@ pub fn get_block<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
beacon_block: block,
};
ResponseBuilder::new(&req).body(&response)
ResponseBuilder::new(&req)?.body(&response)
}
/// HTTP handler to return a `BeaconBlock` root at a given `slot`.
pub fn get_block_root<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let slot_string = UrlQuery::from_request(&req)?.only_one("slot")?;
let target = parse_slot(&slot_string)?;
let root = block_root_at_slot(&beacon_chain, target).ok_or_else(|| {
ApiError::NotFound(format!("Unable to find BeaconBlock for slot {}", target))
ApiError::NotFound(format!("Unable to find BeaconBlock for slot {:?}", target))
})?;
let json: String = serde_json::to_string(&root)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize root: {:?}", e)))?;
Ok(success_response(Body::from(json)))
ResponseBuilder::new(&req)?.body(&root)
}
/// HTTP handler to return a `BeaconState` at a given `root` or `slot`.
/// HTTP handler to return the `Fork` of the current head.
pub fn get_fork<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
ResponseBuilder::new(&req)?.body(&beacon_chain.head().beacon_state.fork)
}
/// HTTP handler to return the set of validators for an `Epoch`
///
/// Will not return a state if the request slot is in the future. Will return states higher than
/// the current head by skipping slots.
pub fn get_genesis_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
/// The `Epoch` parameter can be any epoch number. If it is not specified,
/// the current epoch is assumed.
pub fn get_validators<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let (_root, state) = state_at_slot(&beacon_chain, Slot::new(0))?;
let epoch = match UrlQuery::from_request(&req) {
// We have some parameters, so make sure it's the epoch one and parse it
Ok(query) => query
.only_one("epoch")?
.parse::<u64>()
.map(Epoch::from)
.map_err(|e| {
ApiError::BadRequest(format!("Invalid epoch parameter, must be a u64. {:?}", e))
})?,
// In this case, our url query did not contain any parameters, so we take the default
Err(_) => beacon_chain.epoch().map_err(|e| {
ApiError::ServerError(format!("Unable to determine current epoch: {:?}", e))
})?,
};
ResponseBuilder::new(&req).body(&state)
let all_validators = &beacon_chain.head().beacon_state.validators;
let active_vals: Vec<Validator> = all_validators
.iter()
.filter(|v| v.is_active_at(epoch))
.cloned()
.collect();
ResponseBuilder::new(&req)?.body(&active_vals)
}
#[derive(Serialize, Encode)]
@ -128,13 +169,23 @@ pub struct StateResponse<T: EthSpec> {
/// Will not return a state if the request slot is in the future. Will return states higher than
/// the current head by skipping slots.
pub fn get_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let head_state = beacon_chain.head().beacon_state;
let query_params = ["root", "slot"];
let (key, value) = UrlQuery::from_request(&req)?.first_of(&query_params)?;
let (key, value) = match UrlQuery::from_request(&req) {
Ok(query) => {
// We have *some* parameters, just check them.
let query_params = ["root", "slot"];
query.first_of(&query_params)?
}
Err(ApiError::BadRequest(_)) => {
// No parameters provided at all, use current slot.
(String::from("slot"), head_state.slot.to_string())
}
Err(e) => {
return Err(e);
}
};
let (root, state): (Hash256, BeaconState<T::EthSpec>) = match (key.as_ref(), value) {
("slot", value) => state_at_slot(&beacon_chain, parse_slot(&value)?)?,
@ -144,7 +195,7 @@ pub fn get_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
let state = beacon_chain
.store
.get(root)?
.ok_or_else(|| ApiError::NotFound(format!("No state for root: {}", root)))?;
.ok_or_else(|| ApiError::NotFound(format!("No state for root: {:?}", root)))?;
(*root, state)
}
@ -156,7 +207,7 @@ pub fn get_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
beacon_state: state,
};
ResponseBuilder::new(&req).body(&response)
ResponseBuilder::new(&req)?.body(&response)
}
/// HTTP handler to return a `BeaconState` root at a given `slot`.
@ -164,39 +215,33 @@ pub fn get_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult
/// Will not return a state if the request slot is in the future. Will return states higher than
/// the current head by skipping slots.
pub fn get_state_root<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let slot_string = UrlQuery::from_request(&req)?.only_one("slot")?;
let slot = parse_slot(&slot_string)?;
let root = state_root_at_slot(&beacon_chain, slot)?;
let json: String = serde_json::to_string(&root)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize root: {:?}", e)))?;
Ok(success_response(Body::from(json)))
ResponseBuilder::new(&req)?.body(&root)
}
/// HTTP handler to return the highest finalized slot.
pub fn get_latest_finalized_checkpoint<T: BeaconChainTypes + 'static>(
pub fn get_current_finalized_checkpoint<T: BeaconChainTypes + 'static>(
req: Request<Body>,
) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let head_state = beacon_chain.head().beacon_state;
let checkpoint = beacon_chain
.head()
.beacon_state
.finalized_checkpoint
.clone();
let checkpoint = head_state.finalized_checkpoint.clone();
let json: String = serde_json::to_string(&checkpoint)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize checkpoint: {:?}", e)))?;
Ok(success_response(Body::from(json)))
ResponseBuilder::new(&req)?.body(&checkpoint)
}
/// HTTP handler to return a `BeaconState` at the genesis block.
pub fn get_genesis_state<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let (_root, state) = state_at_slot(&beacon_chain, Slot::new(0))?;
ResponseBuilder::new(&req)?.body(&state)
}

86
beacon_node/rest_api/src/error.rs Normal file
View File

@ -0,0 +1,86 @@
use crate::BoxFut;
use hyper::{Body, Response, StatusCode};
use std::error::Error as StdError;
#[derive(PartialEq, Debug, Clone)]
pub enum ApiError {
MethodNotAllowed(String),
ServerError(String),
NotImplemented(String),
BadRequest(String),
NotFound(String),
UnsupportedType(String),
ImATeapot(String), // Just in case.
ProcessingError(String), // A 202 error, for when a block/attestation cannot be processed, but still transmitted.
}
pub type ApiResult = Result<Response<Body>, ApiError>;
impl ApiError {
pub fn status_code(self) -> (StatusCode, String) {
match self {
ApiError::MethodNotAllowed(desc) => (StatusCode::METHOD_NOT_ALLOWED, desc),
ApiError::ServerError(desc) => (StatusCode::INTERNAL_SERVER_ERROR, desc),
ApiError::NotImplemented(desc) => (StatusCode::NOT_IMPLEMENTED, desc),
ApiError::BadRequest(desc) => (StatusCode::BAD_REQUEST, desc),
ApiError::NotFound(desc) => (StatusCode::NOT_FOUND, desc),
ApiError::UnsupportedType(desc) => (StatusCode::UNSUPPORTED_MEDIA_TYPE, desc),
ApiError::ImATeapot(desc) => (StatusCode::IM_A_TEAPOT, desc),
ApiError::ProcessingError(desc) => (StatusCode::ACCEPTED, desc),
}
}
}
impl Into<Response<Body>> for ApiError {
fn into(self) -> Response<Body> {
let status_code = self.status_code();
Response::builder()
.status(status_code.0)
.header("content-type", "text/plain; charset=utf-8")
.body(Body::from(status_code.1))
.expect("Response should always be created.")
}
}
impl Into<BoxFut> for ApiError {
fn into(self) -> BoxFut {
Box::new(futures::future::err(self))
}
}
impl From<store::Error> for ApiError {
fn from(e: store::Error) -> ApiError {
ApiError::ServerError(format!("Database error: {:?}", e))
}
}
impl From<types::BeaconStateError> for ApiError {
fn from(e: types::BeaconStateError) -> ApiError {
ApiError::ServerError(format!("BeaconState error: {:?}", e))
}
}
impl From<state_processing::per_slot_processing::Error> for ApiError {
fn from(e: state_processing::per_slot_processing::Error) -> ApiError {
ApiError::ServerError(format!("PerSlotProcessing error: {:?}", e))
}
}
impl From<hyper::error::Error> for ApiError {
fn from(e: hyper::error::Error) -> ApiError {
ApiError::ServerError(format!("Networking error: {:?}", e))
}
}
impl StdError for ApiError {
fn cause(&self) -> Option<&dyn StdError> {
None
}
}
impl std::fmt::Display for ApiError {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
let status = self.clone().status_code();
write!(f, "{:?}: {:?}", status.0, status.1)
}
}

113
beacon_node/rest_api/src/helpers.rs
View File

@ -1,10 +1,20 @@
use crate::{ApiError, ApiResult};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use bls::PublicKey;
use eth2_libp2p::{PubsubMessage, Topic};
use eth2_libp2p::{
BEACON_ATTESTATION_TOPIC, BEACON_BLOCK_TOPIC, TOPIC_ENCODING_POSTFIX, TOPIC_PREFIX,
};
use hex;
use http::header;
use hyper::{Body, Request};
use network::NetworkMessage;
use parking_lot::RwLock;
use ssz::Encode;
use std::sync::Arc;
use store::{iter::AncestorIter, Store};
use types::{BeaconState, EthSpec, Hash256, RelativeEpoch, Slot};
use tokio::sync::mpsc;
use types::{Attestation, BeaconBlock, BeaconState, EthSpec, Hash256, RelativeEpoch, Slot};
/// Parse a slot from a `0x` preixed string.
///
@ -13,7 +23,22 @@ pub fn parse_slot(string: &str) -> Result<Slot, ApiError> {
string
.parse::<u64>()
.map(Slot::from)
.map_err(|e| ApiError::InvalidQueryParams(format!("Unable to parse slot: {:?}", e)))
.map_err(|e| ApiError::BadRequest(format!("Unable to parse slot: {:?}", e)))
}
/// Checks the provided request to ensure that the `content-type` header.
///
/// The content-type header should either be omitted, in which case JSON is assumed, or it should
/// explicity specify `application/json`. If anything else is provided, an error is returned.
pub fn check_content_type_for_json(req: &Request<Body>) -> Result<(), ApiError> {
match req.headers().get(header::CONTENT_TYPE) {
Some(h) if h == "application/json" => Ok(()),
Some(h) => Err(ApiError::BadRequest(format!(
"The provided content-type {:?} is not available, this endpoint only supports json.",
h
))),
_ => Ok(()),
}
}
/// Parse a root from a `0x` preixed string.
@ -26,9 +51,9 @@ pub fn parse_root(string: &str) -> Result<Hash256, ApiError> {
let trimmed = string.trim_start_matches(PREFIX);
trimmed
.parse()
.map_err(|e| ApiError::InvalidQueryParams(format!("Unable to parse root: {:?}", e)))
.map_err(|e| ApiError::BadRequest(format!("Unable to parse root: {:?}", e)))
} else {
Err(ApiError::InvalidQueryParams(
Err(ApiError::BadRequest(
"Root must have a '0x' prefix".to_string(),
))
}
@ -39,13 +64,13 @@ pub fn parse_pubkey(string: &str) -> Result<PublicKey, ApiError> {
const PREFIX: &str = "0x";
if string.starts_with(PREFIX) {
let pubkey_bytes = hex::decode(string.trim_start_matches(PREFIX))
.map_err(|e| ApiError::InvalidQueryParams(format!("Invalid hex string: {:?}", e)))?;
.map_err(|e| ApiError::BadRequest(format!("Invalid hex string: {:?}", e)))?;
let pubkey = PublicKey::from_bytes(pubkey_bytes.as_slice()).map_err(|e| {
ApiError::InvalidQueryParams(format!("Unable to deserialize public key: {:?}.", e))
ApiError::BadRequest(format!("Unable to deserialize public key: {:?}.", e))
})?;
return Ok(pubkey);
} else {
return Err(ApiError::InvalidQueryParams(
return Err(ApiError::BadRequest(
"Public key must have a '0x' prefix".to_string(),
));
}
@ -122,7 +147,7 @@ pub fn state_root_at_slot<T: BeaconChainTypes>(
//
// We could actually speculate about future state roots by skipping slots, however that's
// likely to cause confusion for API users.
Err(ApiError::InvalidQueryParams(format!(
Err(ApiError::BadRequest(format!(
"Requested slot {} is past the current slot {}",
slot, current_slot
)))
@ -169,6 +194,78 @@ pub fn implementation_pending_response(_req: Request<Body>) -> ApiResult {
))
}
pub fn get_beacon_chain_from_request<T: BeaconChainTypes + 'static>(
req: &Request<Body>,
) -> Result<(Arc<BeaconChain<T>>), ApiError> {
// Get beacon state
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".into()))?;
Ok(beacon_chain.clone())
}
pub fn get_logger_from_request(req: &Request<Body>) -> slog::Logger {
let log = req
.extensions()
.get::<slog::Logger>()
.expect("Should always get the logger from the request, since we put it in there.");
log.to_owned()
}
pub fn publish_beacon_block_to_network<T: BeaconChainTypes + 'static>(
chan: Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>,
block: BeaconBlock<T::EthSpec>,
) -> Result<(), ApiError> {
// create the network topic to send on
let topic_string = format!(
"/{}/{}/{}",
TOPIC_PREFIX, BEACON_BLOCK_TOPIC, TOPIC_ENCODING_POSTFIX
);
let topic = Topic::new(topic_string);
let message = PubsubMessage::Block(block.as_ssz_bytes());
// Publish the block to the p2p network via gossipsub.
if let Err(e) = chan.write().try_send(NetworkMessage::Publish {
topics: vec![topic],
message,
}) {
return Err(ApiError::ServerError(format!(
"Unable to send new block to network: {:?}",
e
)));
}
Ok(())
}
pub fn publish_attestation_to_network<T: BeaconChainTypes + 'static>(
chan: Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>,
attestation: Attestation<T::EthSpec>,
) -> Result<(), ApiError> {
// create the network topic to send on
let topic_string = format!(
"/{}/{}/{}",
TOPIC_PREFIX, BEACON_ATTESTATION_TOPIC, TOPIC_ENCODING_POSTFIX
);
let topic = Topic::new(topic_string);
let message = PubsubMessage::Attestation(attestation.as_ssz_bytes());
// Publish the attestation to the p2p network via gossipsub.
if let Err(e) = chan.write().try_send(NetworkMessage::Publish {
topics: vec![topic],
message,
}) {
return Err(ApiError::ServerError(format!(
"Unable to send new attestation to network: {:?}",
e
)));
}
Ok(())
}
#[cfg(test)]
mod test {
use super::*;

323
beacon_node/rest_api/src/lib.rs
View File

@ -1,9 +1,12 @@
#[macro_use]
mod macros;
#[macro_use]
extern crate lazy_static;
extern crate network as client_network;
mod beacon;
mod config;
mod error;
mod helpers;
mod metrics;
mod network;
@ -14,66 +17,176 @@ mod url_query;
mod validator;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use client_network::NetworkMessage;
use client_network::Service as NetworkService;
use error::{ApiError, ApiResult};
use eth2_config::Eth2Config;
use futures::future::IntoFuture;
use hyper::rt::Future;
use hyper::service::service_fn_ok;
use hyper::{Body, Method, Response, Server, StatusCode};
use response_builder::ResponseBuilder;
use hyper::service::Service;
use hyper::{Body, Method, Request, Response, Server};
use parking_lot::RwLock;
use slog::{info, o, warn};
use std::ops::Deref;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
use tokio::sync::mpsc;
use url_query::UrlQuery;
pub use beacon::{BlockResponse, HeadResponse, StateResponse};
pub use config::Config as ApiConfig;
#[derive(PartialEq, Debug)]
pub enum ApiError {
MethodNotAllowed(String),
ServerError(String),
NotImplemented(String),
InvalidQueryParams(String),
NotFound(String),
ImATeapot(String), // Just in case.
type BoxFut = Box<dyn Future<Item = Response<Body>, Error = ApiError> + Send>;
pub struct ApiService<T: BeaconChainTypes + 'static> {
log: slog::Logger,
beacon_chain: Arc<BeaconChain<T>>,
db_path: DBPath,
network_service: Arc<NetworkService<T>>,
network_channel: Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>,
eth2_config: Arc<Eth2Config>,
}
pub type ApiResult = Result<Response<Body>, ApiError>;
fn into_boxfut<F: IntoFuture + 'static>(item: F) -> BoxFut
where
F: IntoFuture<Item = Response<Body>, Error = ApiError>,
F::Future: Send,
{
Box::new(item.into_future())
}
impl Into<Response<Body>> for ApiError {
fn into(self) -> Response<Body> {
let status_code: (StatusCode, String) = match self {
ApiError::MethodNotAllowed(desc) => (StatusCode::METHOD_NOT_ALLOWED, desc),
ApiError::ServerError(desc) => (StatusCode::INTERNAL_SERVER_ERROR, desc),
ApiError::NotImplemented(desc) => (StatusCode::NOT_IMPLEMENTED, desc),
ApiError::InvalidQueryParams(desc) => (StatusCode::BAD_REQUEST, desc),
ApiError::NotFound(desc) => (StatusCode::NOT_FOUND, desc),
ApiError::ImATeapot(desc) => (StatusCode::IM_A_TEAPOT, desc),
impl<T: BeaconChainTypes> Service for ApiService<T> {
type ReqBody = Body;
type ResBody = Body;
type Error = ApiError;
type Future = BoxFut;
fn call(&mut self, mut req: Request<Body>) -> Self::Future {
metrics::inc_counter(&metrics::REQUEST_COUNT);
let timer = metrics::start_timer(&metrics::REQUEST_RESPONSE_TIME);
// Add all the useful bits into the request, so that we can pull them out in the individual
// functions.
req.extensions_mut()
.insert::<slog::Logger>(self.log.clone());
req.extensions_mut()
.insert::<Arc<BeaconChain<T>>>(self.beacon_chain.clone());
req.extensions_mut().insert::<DBPath>(self.db_path.clone());
req.extensions_mut()
.insert::<Arc<NetworkService<T>>>(self.network_service.clone());
req.extensions_mut()
.insert::<Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>>(
self.network_channel.clone(),
);
req.extensions_mut()
.insert::<Arc<Eth2Config>>(self.eth2_config.clone());
let path = req.uri().path().to_string();
// Route the request to the correct handler.
let result = match (req.method(), path.as_ref()) {
// Methods for Client
(&Method::GET, "/node/version") => into_boxfut(node::get_version(req)),
(&Method::GET, "/node/genesis_time") => into_boxfut(node::get_genesis_time::<T>(req)),
(&Method::GET, "/node/syncing") => {
into_boxfut(helpers::implementation_pending_response(req))
}
// Methods for Network
(&Method::GET, "/network/enr") => into_boxfut(network::get_enr::<T>(req)),
(&Method::GET, "/network/peer_count") => into_boxfut(network::get_peer_count::<T>(req)),
(&Method::GET, "/network/peer_id") => into_boxfut(network::get_peer_id::<T>(req)),
(&Method::GET, "/network/peers") => into_boxfut(network::get_peer_list::<T>(req)),
(&Method::GET, "/network/listen_port") => {
into_boxfut(network::get_listen_port::<T>(req))
}
(&Method::GET, "/network/listen_addresses") => {
into_boxfut(network::get_listen_addresses::<T>(req))
}
// Methods for Beacon Node
(&Method::GET, "/beacon/head") => into_boxfut(beacon::get_head::<T>(req)),
(&Method::GET, "/beacon/block") => into_boxfut(beacon::get_block::<T>(req)),
(&Method::GET, "/beacon/block_root") => into_boxfut(beacon::get_block_root::<T>(req)),
(&Method::GET, "/beacon/blocks") => {
into_boxfut(helpers::implementation_pending_response(req))
}
(&Method::GET, "/beacon/fork") => into_boxfut(beacon::get_fork::<T>(req)),
(&Method::GET, "/beacon/attestations") => {
into_boxfut(helpers::implementation_pending_response(req))
}
(&Method::GET, "/beacon/attestations/pending") => {
into_boxfut(helpers::implementation_pending_response(req))
}
(&Method::GET, "/beacon/validators") => into_boxfut(beacon::get_validators::<T>(req)),
(&Method::GET, "/beacon/validators/indicies") => {
into_boxfut(helpers::implementation_pending_response(req))
}
(&Method::GET, "/beacon/validators/pubkeys") => {
into_boxfut(helpers::implementation_pending_response(req))
}
// Methods for Validator
(&Method::GET, "/beacon/validator/duties") => {
into_boxfut(validator::get_validator_duties::<T>(req))
}
(&Method::GET, "/beacon/validator/block") => {
into_boxfut(validator::get_new_beacon_block::<T>(req))
}
(&Method::POST, "/beacon/validator/block") => validator::publish_beacon_block::<T>(req),
(&Method::GET, "/beacon/validator/attestation") => {
into_boxfut(validator::get_new_attestation::<T>(req))
}
(&Method::POST, "/beacon/validator/attestation") => {
validator::publish_attestation::<T>(req)
}
(&Method::GET, "/beacon/state") => into_boxfut(beacon::get_state::<T>(req)),
(&Method::GET, "/beacon/state_root") => into_boxfut(beacon::get_state_root::<T>(req)),
(&Method::GET, "/beacon/state/current_finalized_checkpoint") => {
into_boxfut(beacon::get_current_finalized_checkpoint::<T>(req))
}
(&Method::GET, "/beacon/state/genesis") => {
into_boxfut(beacon::get_genesis_state::<T>(req))
}
//TODO: Add aggreggate/filtered state lookups here, e.g. /beacon/validators/balances
// Methods for bootstrap and checking configuration
(&Method::GET, "/spec") => into_boxfut(spec::get_spec::<T>(req)),
(&Method::GET, "/spec/slots_per_epoch") => {
into_boxfut(spec::get_slots_per_epoch::<T>(req))
}
(&Method::GET, "/spec/deposit_contract") => {
into_boxfut(helpers::implementation_pending_response(req))
}
(&Method::GET, "/spec/eth2_config") => into_boxfut(spec::get_eth2_config::<T>(req)),
(&Method::GET, "/metrics") => into_boxfut(metrics::get_prometheus::<T>(req)),
_ => Box::new(futures::future::err(ApiError::NotFound(
"Request path and/or method not found.".to_owned(),
))),
};
Response::builder()
.status(status_code.0)
.body(Body::from(status_code.1))
.expect("Response should always be created.")
}
}
impl From<store::Error> for ApiError {
fn from(e: store::Error) -> ApiError {
ApiError::ServerError(format!("Database error: {:?}", e))
}
}
let response = match result.wait() {
// Return the `hyper::Response`.
Ok(response) => {
metrics::inc_counter(&metrics::SUCCESS_COUNT);
slog::debug!(self.log, "Request successful: {:?}", path);
response
}
// Map the `ApiError` into `hyper::Response`.
Err(e) => {
slog::debug!(self.log, "Request failure: {:?}", path);
e.into()
}
};
impl From<types::BeaconStateError> for ApiError {
fn from(e: types::BeaconStateError) -> ApiError {
ApiError::ServerError(format!("BeaconState error: {:?}", e))
}
}
metrics::stop_timer(timer);
impl From<state_processing::per_slot_processing::Error> for ApiError {
fn from(e: state_processing::per_slot_processing::Error) -> ApiError {
ApiError::ServerError(format!("PerSlotProcessing error: {:?}", e))
Box::new(futures::future::ok(response))
}
}
@ -82,6 +195,7 @@ pub fn start_server<T: BeaconChainTypes>(
executor: &TaskExecutor,
beacon_chain: Arc<BeaconChain<T>>,
network_service: Arc<NetworkService<T>>,
network_chan: mpsc::UnboundedSender<NetworkMessage>,
db_path: PathBuf,
eth2_config: Eth2Config,
log: &slog::Logger,
@ -107,108 +221,14 @@ pub fn start_server<T: BeaconChainTypes>(
let server_bc = beacon_chain.clone();
let eth2_config = Arc::new(eth2_config);
let service = move || {
let log = server_log.clone();
let beacon_chain = server_bc.clone();
let db_path = db_path.clone();
let network_service = network_service.clone();
let eth2_config = eth2_config.clone();
// Create a simple handler for the router, inject our stateful objects into the request.
service_fn_ok(move |mut req| {
metrics::inc_counter(&metrics::REQUEST_COUNT);
let timer = metrics::start_timer(&metrics::REQUEST_RESPONSE_TIME);
req.extensions_mut().insert::<slog::Logger>(log.clone());
req.extensions_mut()
.insert::<Arc<BeaconChain<T>>>(beacon_chain.clone());
req.extensions_mut().insert::<DBPath>(db_path.clone());
req.extensions_mut()
.insert::<Arc<NetworkService<T>>>(network_service.clone());
req.extensions_mut()
.insert::<Arc<Eth2Config>>(eth2_config.clone());
let path = req.uri().path().to_string();
// Route the request to the correct handler.
let result = match (req.method(), path.as_ref()) {
// Methods for Beacon Node
//TODO: Remove?
//(&Method::GET, "/beacon/best_slot") => beacon::get_best_slot::<T>(req),
(&Method::GET, "/beacon/head") => beacon::get_head::<T>(req),
(&Method::GET, "/beacon/block") => beacon::get_block::<T>(req),
(&Method::GET, "/beacon/blocks") => helpers::implementation_pending_response(req),
//TODO Is the below replaced by finalized_checkpoint?
(&Method::GET, "/beacon/chainhead") => {
helpers::implementation_pending_response(req)
}
(&Method::GET, "/beacon/block_root") => beacon::get_block_root::<T>(req),
(&Method::GET, "/beacon/latest_finalized_checkpoint") => {
beacon::get_latest_finalized_checkpoint::<T>(req)
}
(&Method::GET, "/beacon/state") => beacon::get_state::<T>(req),
(&Method::GET, "/beacon/state/genesis") => beacon::get_genesis_state::<T>(req),
(&Method::GET, "/beacon/state_root") => beacon::get_state_root::<T>(req),
//TODO: Add aggreggate/filtered state lookups here, e.g. /beacon/validators/balances
// Methods for Client
(&Method::GET, "/metrics") => metrics::get_prometheus::<T>(req),
(&Method::GET, "/network/enr") => network::get_enr::<T>(req),
(&Method::GET, "/network/peer_count") => network::get_peer_count::<T>(req),
(&Method::GET, "/network/peer_id") => network::get_peer_id::<T>(req),
(&Method::GET, "/network/peers") => network::get_peer_list::<T>(req),
(&Method::GET, "/network/listen_port") => network::get_listen_port::<T>(req),
(&Method::GET, "/network/listen_addresses") => {
network::get_listen_addresses::<T>(req)
}
(&Method::GET, "/node/version") => node::get_version(req),
(&Method::GET, "/node/genesis_time") => node::get_genesis_time::<T>(req),
(&Method::GET, "/node/deposit_contract") => {
helpers::implementation_pending_response(req)
}
(&Method::GET, "/node/syncing") => helpers::implementation_pending_response(req),
(&Method::GET, "/node/fork") => helpers::implementation_pending_response(req),
// Methods for Validator
(&Method::GET, "/validator/duties") => validator::get_validator_duties::<T>(req),
(&Method::GET, "/validator/block") => helpers::implementation_pending_response(req),
(&Method::POST, "/validator/block") => {
helpers::implementation_pending_response(req)
}
(&Method::GET, "/validator/attestation") => {
helpers::implementation_pending_response(req)
}
(&Method::POST, "/validator/attestation") => {
helpers::implementation_pending_response(req)
}
(&Method::GET, "/spec") => spec::get_spec::<T>(req),
(&Method::GET, "/spec/slots_per_epoch") => spec::get_slots_per_epoch::<T>(req),
(&Method::GET, "/spec/eth2_config") => spec::get_eth2_config::<T>(req),
_ => Err(ApiError::NotFound(
"Request path and/or method not found.".to_owned(),
)),
};
let response = match result {
// Return the `hyper::Response`.
Ok(response) => {
metrics::inc_counter(&metrics::SUCCESS_COUNT);
slog::debug!(log, "Request successful: {:?}", path);
response
}
// Map the `ApiError` into `hyper::Response`.
Err(e) => {
slog::debug!(log, "Request failure: {:?}", path);
e.into()
}
};
metrics::stop_timer(timer);
response
let service = move || -> futures::future::FutureResult<ApiService<T>, String> {
futures::future::ok(ApiService {
log: server_log.clone(),
beacon_chain: server_bc.clone(),
db_path: db_path.clone(),
network_service: network_service.clone(),
network_channel: Arc::new(RwLock::new(network_chan.clone())),
eth2_config: eth2_config.clone(),
})
};
@ -218,16 +238,16 @@ pub fn start_server<T: BeaconChainTypes>(
.with_graceful_shutdown(server_exit)
.map_err(move |e| {
warn!(
log_clone,
"API failed to start, Unable to bind"; "address" => format!("{:?}", e)
log_clone,
"API failed to start, Unable to bind"; "address" => format!("{:?}", e)
)
});
info!(
log,
"REST API started";
"address" => format!("{}", config.listen_address),
"port" => config.port,
log,
"REST API started";
"address" => format!("{}", config.listen_address),
"port" => config.port,
);
executor.spawn(server);
@ -235,13 +255,6 @@ pub fn start_server<T: BeaconChainTypes>(
Ok(exit_signal)
}
fn success_response(body: Body) -> Response<Body> {
Response::builder()
.status(StatusCode::OK)
.body(body)
.expect("We should always be able to make response from the success body.")
}
#[derive(Clone)]
pub struct DBPath(PathBuf);

13
beacon_node/rest_api/src/macros.rs Normal file
View File

@ -0,0 +1,13 @@
macro_rules! try_future {
($expr:expr) => {
match $expr {
core::result::Result::Ok(val) => val,
core::result::Result::Err(err) => {
return Box::new(futures::future::err(std::convert::From::from(err)))
}
}
};
($expr:expr,) => {
$crate::try_future!($expr)
};
}

16
beacon_node/rest_api/src/metrics.rs
View File

@ -1,8 +1,9 @@
use crate::{success_response, ApiError, ApiResult, DBPath};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use crate::helpers::get_beacon_chain_from_request;
use crate::response_builder::ResponseBuilder;
use crate::{ApiError, ApiResult, DBPath};
use beacon_chain::BeaconChainTypes;
use hyper::{Body, Request};
use prometheus::{Encoder, TextEncoder};
use std::sync::Arc;
pub use lighthouse_metrics::*;
@ -30,10 +31,7 @@ pub fn get_prometheus<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiR
let mut buffer = vec![];
let encoder = TextEncoder::new();
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let db_path = req
.extensions()
.get::<DBPath>()
@ -64,6 +62,6 @@ pub fn get_prometheus<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiR
.unwrap();
String::from_utf8(buffer)
.map(|string| success_response(Body::from(string)))
.map_err(|e| ApiError::ServerError(format!("Failed to encode prometheus info: {:?}", e)))
.map(|string| ResponseBuilder::new(&req)?.body_text(string))
.map_err(|e| ApiError::ServerError(format!("Failed to encode prometheus info: {:?}", e)))?
}

78
beacon_node/rest_api/src/network.rs
View File

@ -1,108 +1,78 @@
use crate::{success_response, ApiError, ApiResult, NetworkService};
use crate::error::ApiResult;
use crate::response_builder::ResponseBuilder;
use crate::NetworkService;
use beacon_chain::BeaconChainTypes;
use eth2_libp2p::{Enr, Multiaddr, PeerId};
use eth2_libp2p::{Multiaddr, PeerId};
use hyper::{Body, Request};
use std::sync::Arc;
/// HTTP handle to return the list of libp2p multiaddr the client is listening on.
/// HTTP handler to return the list of libp2p multiaddr the client is listening on.
///
/// Returns a list of `Multiaddr`, serialized according to their `serde` impl.
pub fn get_listen_addresses<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
.expect("The network service should always be there, we put it there");
let multiaddresses: Vec<Multiaddr> = network.listen_multiaddrs();
Ok(success_response(Body::from(
serde_json::to_string(&multiaddresses)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize Enr: {:?}", e)))?,
)))
ResponseBuilder::new(&req)?.body_no_ssz(&multiaddresses)
}
/// HTTP handle to return the list of libp2p multiaddr the client is listening on.
/// HTTP handler to return the network port the client is listening on.
///
/// Returns a list of `Multiaddr`, serialized according to their `serde` impl.
/// Returns the TCP port number in its plain form (which is also valid JSON serialization)
pub fn get_listen_port<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
Ok(success_response(Body::from(
serde_json::to_string(&network.listen_port())
.map_err(|e| ApiError::ServerError(format!("Unable to serialize port: {:?}", e)))?,
)))
.expect("The network service should always be there, we put it there")
.clone();
ResponseBuilder::new(&req)?.body(&network.listen_port())
}
/// HTTP handle to return the Discv5 ENR from the client's libp2p service.
/// HTTP handler to return the Discv5 ENR from the client's libp2p service.
///
/// ENR is encoded as base64 string.
pub fn get_enr<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
let enr: Enr = network.local_enr();
Ok(success_response(Body::from(
serde_json::to_string(&enr.to_base64())
.map_err(|e| ApiError::ServerError(format!("Unable to serialize Enr: {:?}", e)))?,
)))
.expect("The network service should always be there, we put it there");
ResponseBuilder::new(&req)?.body_no_ssz(&network.local_enr().to_base64())
}
/// HTTP handle to return the `PeerId` from the client's libp2p service.
/// HTTP handler to return the `PeerId` from the client's libp2p service.
///
/// PeerId is encoded as base58 string.
pub fn get_peer_id<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
let peer_id: PeerId = network.local_peer_id();
Ok(success_response(Body::from(
serde_json::to_string(&peer_id.to_base58())
.map_err(|e| ApiError::ServerError(format!("Unable to serialize Enr: {:?}", e)))?,
)))
.expect("The network service should always be there, we put it there");
ResponseBuilder::new(&req)?.body_no_ssz(&network.local_peer_id().to_base58())
}
/// HTTP handle to return the number of peers connected in the client's libp2p service.
/// HTTP handler to return the number of peers connected in the client's libp2p service.
pub fn get_peer_count<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
let connected_peers: usize = network.connected_peers();
Ok(success_response(Body::from(
serde_json::to_string(&connected_peers)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize Enr: {:?}", e)))?,
)))
.expect("The network service should always be there, we put it there");
ResponseBuilder::new(&req)?.body(&network.connected_peers())
}
/// HTTP handle to return the list of peers connected to the client's libp2p service.
/// HTTP handler to return the list of peers connected to the client's libp2p service.
///
/// Peers are presented as a list of `PeerId::to_string()`.
pub fn get_peer_list<T: BeaconChainTypes>(req: Request<Body>) -> ApiResult {
let network = req
.extensions()
.get::<Arc<NetworkService<T>>>()
.ok_or_else(|| ApiError::ServerError("NetworkService extension missing".to_string()))?;
.expect("The network service should always be there, we put it there");
let connected_peers: Vec<String> = network
.connected_peer_set()
.iter()
.map(PeerId::to_string)
.collect();
Ok(success_response(Body::from(
serde_json::to_string(&connected_peers).map_err(|e| {
ApiError::ServerError(format!("Unable to serialize Vec<PeerId>: {:?}", e))
})?,
)))
ResponseBuilder::new(&req)?.body_no_ssz(&connected_peers)
}

24
beacon_node/rest_api/src/node.rs
View File

@ -1,25 +1,17 @@
use crate::{success_response, ApiResult};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use crate::helpers::get_beacon_chain_from_request;
use crate::response_builder::ResponseBuilder;
use crate::ApiResult;
use beacon_chain::BeaconChainTypes;
use hyper::{Body, Request};
use std::sync::Arc;
use version;
/// Read the version string from the current Lighthouse build.
pub fn get_version(_req: Request<Body>) -> ApiResult {
let body = Body::from(
serde_json::to_string(&version::version())
.expect("Version should always be serialializable as JSON."),
);
Ok(success_response(body))
pub fn get_version(req: Request<Body>) -> ApiResult {
ResponseBuilder::new(&req)?.body_no_ssz(&version::version())
}
/// Read the genesis time from the current beacon chain state.
pub fn get_genesis_time<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req.extensions().get::<Arc<BeaconChain<T>>>().unwrap();
let gen_time: u64 = beacon_chain.head().beacon_state.genesis_time;
let body = Body::from(
serde_json::to_string(&gen_time)
.expect("Genesis should time always have a valid JSON serialization."),
);
Ok(success_response(body))
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
ResponseBuilder::new(&req)?.body(&beacon_chain.head().beacon_state.genesis_time)
}

89
beacon_node/rest_api/src/response_builder.rs
View File

@ -8,6 +8,7 @@ pub enum Encoding {
JSON,
SSZ,
YAML,
TEXT,
}
pub struct ResponseBuilder {
@ -15,36 +16,84 @@ pub struct ResponseBuilder {
}
impl ResponseBuilder {
pub fn new(req: &Request<Body>) -> Self {
let encoding = match req.headers().get(header::CONTENT_TYPE) {
Some(h) if h == "application/ssz" => Encoding::SSZ,
Some(h) if h == "application/yaml" => Encoding::YAML,
pub fn new(req: &Request<Body>) -> Result<Self, ApiError> {
let content_header: String = req
.headers()
.get(header::CONTENT_TYPE)
.map_or(Ok(""), |h| h.to_str())
.map_err(|e| {
ApiError::BadRequest(format!(
"The content-type header contains invalid characters: {:?}",
e
))
})
.map(|h| String::from(h))?;
// JSON is our default encoding, unless something else is requested.
let encoding = match content_header {
ref h if h.starts_with("application/ssz") => Encoding::SSZ,
ref h if h.starts_with("application/yaml") => Encoding::YAML,
ref h if h.starts_with("text/") => Encoding::TEXT,
_ => Encoding::JSON,
};
Self { encoding }
Ok(Self { encoding })
}
pub fn body<T: Serialize + Encode>(self, item: &T) -> ApiResult {
let body: Body = match self.encoding {
Encoding::JSON => Body::from(serde_json::to_string(&item).map_err(|e| {
ApiError::ServerError(format!(
"Unable to serialize response body as JSON: {:?}",
e
))
})?),
Encoding::SSZ => Body::from(item.as_ssz_bytes()),
Encoding::YAML => Body::from(serde_yaml::to_string(&item).map_err(|e| {
ApiError::ServerError(format!(
"Unable to serialize response body as YAML: {:?}",
e
))
})?),
match self.encoding {
Encoding::SSZ => Response::builder()
.status(StatusCode::OK)
.header("content-type", "application/ssz")
.body(Body::from(item.as_ssz_bytes()))
.map_err(|e| ApiError::ServerError(format!("Failed to build response: {:?}", e))),
_ => self.body_no_ssz(item),
}
}
pub fn body_no_ssz<T: Serialize>(self, item: &T) -> ApiResult {
let (body, content_type) = match self.encoding {
Encoding::JSON => (
Body::from(serde_json::to_string(&item).map_err(|e| {
ApiError::ServerError(format!(
"Unable to serialize response body as JSON: {:?}",
e
))
})?),
"application/json",
),
Encoding::SSZ => {
return Err(ApiError::UnsupportedType(
"Response cannot be encoded as SSZ.".into(),
));
}
Encoding::YAML => (
Body::from(serde_yaml::to_string(&item).map_err(|e| {
ApiError::ServerError(format!(
"Unable to serialize response body as YAML: {:?}",
e
))
})?),
"application/yaml",
),
Encoding::TEXT => {
return Err(ApiError::UnsupportedType(
"Response cannot be encoded as plain text.".into(),
));
}
};
Response::builder()
.status(StatusCode::OK)
.header("content-type", content_type)
.body(Body::from(body))
.map_err(|e| ApiError::ServerError(format!("Failed to build response: {:?}", e)))
}
pub fn body_text(self, text: String) -> ApiResult {
Response::builder()
.status(StatusCode::OK)
.header("content-type", "text/plain; charset=utf-8")
.body(Body::from(text))
.map_err(|e| ApiError::ServerError(format!("Failed to build response: {:?}", e)))
}
}

29
beacon_node/rest_api/src/spec.rs
View File

@ -1,6 +1,8 @@
use super::{success_response, ApiResult};
use super::ApiResult;
use crate::helpers::get_beacon_chain_from_request;
use crate::response_builder::ResponseBuilder;
use crate::ApiError;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use beacon_chain::BeaconChainTypes;
use eth2_config::Eth2Config;
use hyper::{Body, Request};
use std::sync::Arc;
@ -8,15 +10,8 @@ use types::EthSpec;
/// HTTP handler to return the full spec object.
pub fn get_spec<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let json: String = serde_json::to_string(&beacon_chain.spec)
.map_err(|e| ApiError::ServerError(format!("Unable to serialize spec: {:?}", e)))?;
Ok(success_response(Body::from(json)))
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
ResponseBuilder::new(&req)?.body_no_ssz(&beacon_chain.spec)
}
/// HTTP handler to return the full Eth2Config object.
@ -26,16 +21,10 @@ pub fn get_eth2_config<T: BeaconChainTypes + 'static>(req: Request<Body>) -> Api
.get::<Arc<Eth2Config>>()
.ok_or_else(|| ApiError::ServerError("Eth2Config extension missing".to_string()))?;
let json: String = serde_json::to_string(eth2_config.as_ref())
.map_err(|e| ApiError::ServerError(format!("Unable to serialize Eth2Config: {:?}", e)))?;
Ok(success_response(Body::from(json)))
ResponseBuilder::new(&req)?.body_no_ssz(eth2_config.as_ref())
}
/// HTTP handler to return the full spec object.
pub fn get_slots_per_epoch<T: BeaconChainTypes + 'static>(_req: Request<Body>) -> ApiResult {
let json: String = serde_json::to_string(&T::EthSpec::slots_per_epoch())
.map_err(|e| ApiError::ServerError(format!("Unable to serialize epoch: {:?}", e)))?;
Ok(success_response(Body::from(json)))
pub fn get_slots_per_epoch<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
ResponseBuilder::new(&req)?.body(&T::EthSpec::slots_per_epoch())
}

8
beacon_node/rest_api/src/url_query.rs
View File

@ -12,7 +12,7 @@ impl<'a> UrlQuery<'a> {
/// Returns `Err` if `req` does not contain any query parameters.
pub fn from_request<T>(req: &'a Request<T>) -> Result<Self, ApiError> {
let query_str = req.uri().query().ok_or_else(|| {
ApiError::InvalidQueryParams(
ApiError::BadRequest(
"URL query must be valid and contain at least one key.".to_string(),
)
})?;
@ -28,7 +28,7 @@ impl<'a> UrlQuery<'a> {
.find(|(key, _value)| keys.contains(&&**key))
.map(|(key, value)| (key.into_owned(), value.into_owned()))
.ok_or_else(|| {
ApiError::InvalidQueryParams(format!(
ApiError::BadRequest(format!(
"URL query must contain at least one of the following keys: {:?}",
keys
))
@ -48,13 +48,13 @@ impl<'a> UrlQuery<'a> {
if first_key == key {
Ok(first_value.to_string())
} else {
Err(ApiError::InvalidQueryParams(format!(
Err(ApiError::BadRequest(format!(
"Only the {} query parameter is supported",
key
)))
}
} else {
Err(ApiError::InvalidQueryParams(format!(
Err(ApiError::BadRequest(format!(
"Only one query parameter is allowed, {} supplied",
queries.len()
)))

330
beacon_node/rest_api/src/validator.rs
View File

@ -1,12 +1,23 @@
use super::{success_response, ApiResult};
use crate::{helpers::*, ApiError, UrlQuery};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use bls::PublicKey;
use crate::helpers::{
check_content_type_for_json, get_beacon_chain_from_request, get_logger_from_request,
parse_pubkey, publish_attestation_to_network, publish_beacon_block_to_network,
};
use crate::response_builder::ResponseBuilder;
use crate::{ApiError, ApiResult, BoxFut, UrlQuery};
use beacon_chain::{AttestationProcessingOutcome, BeaconChainTypes, BlockProcessingOutcome};
use bls::{AggregateSignature, PublicKey, Signature};
use futures::future::Future;
use futures::stream::Stream;
use hyper::{Body, Request};
use network::NetworkMessage;
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use slog::{info, trace, warn};
use std::sync::Arc;
use tokio;
use tokio::sync::mpsc;
use types::beacon_state::EthSpec;
use types::{Epoch, RelativeEpoch, Shard, Slot};
use types::{Attestation, BeaconBlock, BitList, Epoch, RelativeEpoch, Shard, Slot};
#[derive(Debug, Serialize, Deserialize)]
pub struct ValidatorDuty {
@ -33,44 +44,47 @@ impl ValidatorDuty {
/// HTTP Handler to retrieve a the duties for a set of validators during a particular epoch
pub fn get_validator_duties<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
// Get beacon state
let beacon_chain = req
.extensions()
.get::<Arc<BeaconChain<T>>>()
.ok_or_else(|| ApiError::ServerError("Beacon chain extension missing".to_string()))?;
let head_state = &beacon_chain.head().beacon_state;
let log = get_logger_from_request(&req);
slog::trace!(log, "Validator duties requested of API: {:?}", &req);
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let mut head_state = beacon_chain.head().beacon_state;
slog::trace!(log, "Got head state from request.");
// Parse and check query parameters
let query = UrlQuery::from_request(&req)?;
let current_epoch = head_state.current_epoch();
let epoch = match query.first_of(&["epoch"]) {
Ok((_, v)) => Epoch::new(v.parse::<u64>().map_err(|e| {
ApiError::InvalidQueryParams(format!("Invalid epoch parameter, must be a u64. {:?}", e))
})?),
Ok((_, v)) => {
slog::trace!(log, "Requested epoch {:?}", v);
Epoch::new(v.parse::<u64>().map_err(|e| {
slog::info!(log, "Invalid epoch {:?}", e);
ApiError::BadRequest(format!("Invalid epoch parameter, must be a u64. {:?}", e))
})?)
}
Err(_) => {
// epoch not supplied, use the current epoch
slog::info!(log, "Using default epoch {:?}", current_epoch);
current_epoch
}
};
let relative_epoch = RelativeEpoch::from_epoch(current_epoch, epoch).map_err(|e| {
ApiError::InvalidQueryParams(format!(
slog::info!(log, "Requested epoch out of range.");
ApiError::BadRequest(format!(
"Cannot get RelativeEpoch, epoch out of range: {:?}",
e
))
})?;
//TODO: Handle an array of validators, currently only takes one
let validators: Vec<PublicKey> = match query.all_of("validator_pubkeys") {
Ok(v) => v
.iter()
.map(|pk| parse_pubkey(pk))
.collect::<Result<Vec<_>, _>>()?,
Err(e) => {
return Err(e);
}
};
let validators: Vec<PublicKey> = query
.all_of("validator_pubkeys")?
.iter()
.map(|pk| parse_pubkey(pk))
.collect::<Result<Vec<_>, _>>()?;
let mut duties: Vec<ValidatorDuty> = Vec::new();
// Build cache for the requested epoch
head_state
.build_committee_cache(relative_epoch, &beacon_chain.spec)
.map_err(|e| ApiError::ServerError(format!("Unable to build committee cache: {:?}", e)))?;
// Get a list of all validators for this epoch
let validator_proposers: Vec<usize> = epoch
.slot_iter(T::EthSpec::slots_per_epoch())
@ -135,9 +149,263 @@ pub fn get_validator_duties<T: BeaconChainTypes + 'static>(req: Request<Body>) -
duties.append(&mut vec![duty]);
}
let body = Body::from(
serde_json::to_string(&duties)
.expect("We should always be able to serialize the duties we created."),
);
Ok(success_response(body))
ResponseBuilder::new(&req)?.body_no_ssz(&duties)
}
/// HTTP Handler to produce a new BeaconBlock from the current state, ready to be signed by a validator.
pub fn get_new_beacon_block<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let query = UrlQuery::from_request(&req)?;
let slot = query
.first_of(&["slot"])
.map(|(_key, value)| value)?
.parse::<u64>()
.map(Slot::from)
.map_err(|e| {
ApiError::BadRequest(format!("Invalid slot parameter, must be a u64. {:?}", e))
})?;
let randao_bytes = query
.first_of(&["randao_reveal"])
.map(|(_key, value)| value)
.map(hex::decode)?
.map_err(|e| {
ApiError::BadRequest(format!("Invalid hex string for randao_reveal: {:?}", e))
})?;
let randao_reveal = Signature::from_bytes(randao_bytes.as_slice()).map_err(|e| {
ApiError::BadRequest(format!("randao_reveal is not a valid signature: {:?}", e))
})?;
let (new_block, _state) = beacon_chain
.produce_block(randao_reveal, slot)
.map_err(|e| {
ApiError::ServerError(format!(
"Beacon node is not able to produce a block: {:?}",
e
))
})?;
ResponseBuilder::new(&req)?.body(&new_block)
}
/// HTTP Handler to publish a BeaconBlock, which has been signed by a validator.
pub fn publish_beacon_block<T: BeaconChainTypes + 'static>(req: Request<Body>) -> BoxFut {
let _ = try_future!(check_content_type_for_json(&req));
let log = get_logger_from_request(&req);
let beacon_chain = try_future!(get_beacon_chain_from_request::<T>(&req));
// Get the network sending channel from the request, for later transmission
let network_chan = req
.extensions()
.get::<Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>>()
.expect("Should always get the network channel from the request, since we put it in there.")
.clone();
let response_builder = ResponseBuilder::new(&req);
let body = req.into_body();
trace!(
log,
"Got the request body, now going to parse it into a block."
);
Box::new(body
.concat2()
.map_err(|e| ApiError::ServerError(format!("Unable to get request body: {:?}",e)))
.map(|chunk| chunk.iter().cloned().collect::<Vec<u8>>())
.and_then(|chunks| {
serde_json::from_slice(&chunks.as_slice()).map_err(|e| {
ApiError::BadRequest(format!(
"Unable to deserialize JSON into a BeaconBlock: {:?}",
e
))
})
})
.and_then(move |block: BeaconBlock<T::EthSpec>| {
let slot = block.slot;
match beacon_chain.process_block(block.clone()) {
Ok(BlockProcessingOutcome::Processed { block_root }) => {
// Block was processed, publish via gossipsub
info!(log, "Processed valid block from API, transmitting to network."; "block_slot" => slot, "block_root" => format!("{}", block_root));
publish_beacon_block_to_network::<T>(network_chan, block)
}
Ok(outcome) => {
warn!(log, "BeaconBlock could not be processed, but is being sent to the network anyway."; "outcome" => format!("{:?}", outcome));
publish_beacon_block_to_network::<T>(network_chan, block)?;
Err(ApiError::ProcessingError(format!(
"The BeaconBlock could not be processed, but has still been published: {:?}",
outcome
)))
}
Err(e) => {
Err(ApiError::ServerError(format!(
"Error while processing block: {:?}",
e
)))
}
}
}).and_then(|_| {
response_builder?.body_no_ssz(&())
}))
}
/// HTTP Handler to produce a new Attestation from the current state, ready to be signed by a validator.
pub fn get_new_attestation<T: BeaconChainTypes + 'static>(req: Request<Body>) -> ApiResult {
let beacon_chain = get_beacon_chain_from_request::<T>(&req)?;
let mut head_state = beacon_chain.head().beacon_state;
let query = UrlQuery::from_request(&req)?;
let val_pk_str = query
.first_of(&["validator_pubkey"])
.map(|(_key, value)| value)?;
let val_pk = parse_pubkey(val_pk_str.as_str())?;
head_state
.update_pubkey_cache()
.map_err(|e| ApiError::ServerError(format!("Unable to build pubkey cache: {:?}", e)))?;
// Get the validator index from the supplied public key
// If it does not exist in the index, we cannot continue.
let val_index = head_state
.get_validator_index(&val_pk)
.map_err(|e| {
ApiError::ServerError(format!("Unable to read validator index cache. {:?}", e))
})?
.ok_or(ApiError::BadRequest(
"The provided validator public key does not correspond to a validator index.".into(),
))?;
// Build cache for the requested epoch
head_state
.build_committee_cache(RelativeEpoch::Current, &beacon_chain.spec)
.map_err(|e| ApiError::ServerError(format!("Unable to build committee cache: {:?}", e)))?;
// Get the duties of the validator, to make sure they match up.
// If they don't have duties this epoch, then return an error
let val_duty = head_state
.get_attestation_duties(val_index, RelativeEpoch::Current)
.map_err(|e| {
ApiError::ServerError(format!(
"unable to read cache for attestation duties: {:?}",
e
))
})?
.ok_or(ApiError::BadRequest("No validator duties could be found for the requested validator. Cannot provide valid attestation.".into()))?;
// Check that we are requesting an attestation during the slot where it is relevant.
let present_slot = beacon_chain.slot().map_err(|e| ApiError::ServerError(
format!("Beacon node is unable to determine present slot, either the state isn't generated or the chain hasn't begun. {:?}", e)
))?;
if val_duty.slot != present_slot {
return Err(ApiError::BadRequest(format!("Validator is only able to request an attestation during the slot they are allocated. Current slot: {:?}, allocated slot: {:?}", head_state.slot, val_duty.slot)));
}
// Parse the POC bit and insert it into the aggregation bits
let poc_bit = query
.first_of(&["poc_bit"])
.map(|(_key, value)| value)?
.parse::<bool>()
.map_err(|e| {
ApiError::BadRequest(format!("Invalid slot parameter, must be a u64. {:?}", e))
})?;
let mut aggregation_bits = BitList::with_capacity(val_duty.committee_len)
.expect("An empty BitList should always be created, or we have bigger problems.");
aggregation_bits
.set(val_duty.committee_index, poc_bit)
.map_err(|e| {
ApiError::ServerError(format!(
"Unable to set aggregation bits for the attestation: {:?}",
e
))
})?;
// Allow a provided slot parameter to check against the expected slot as a sanity check only.
// Presently, we don't support attestations at future or past slots.
let requested_slot = query
.first_of(&["slot"])
.map(|(_key, value)| value)?
.parse::<u64>()
.map(Slot::from)
.map_err(|e| {
ApiError::BadRequest(format!("Invalid slot parameter, must be a u64. {:?}", e))
})?;
let current_slot = beacon_chain.head().beacon_state.slot.as_u64();
if requested_slot != current_slot {
return Err(ApiError::BadRequest(format!("Attestation data can only be requested for the current slot ({:?}), not your requested slot ({:?})", current_slot, requested_slot)));
}
let shard = query
.first_of(&["shard"])
.map(|(_key, value)| value)?
.parse::<u64>()
.map_err(|e| ApiError::BadRequest(format!("Shard is not a valid u64 value: {:?}", e)))?;
let attestation_data = beacon_chain
.produce_attestation_data(shard, current_slot.into())
.map_err(|e| ApiError::ServerError(format!("Could not produce an attestation: {:?}", e)))?;
let attestation: Attestation<T::EthSpec> = Attestation {
aggregation_bits,
data: attestation_data,
custody_bits: BitList::with_capacity(val_duty.committee_len)
.expect("Should be able to create an empty BitList for the custody bits."),
signature: AggregateSignature::new(),
};
ResponseBuilder::new(&req)?.body(&attestation)
}
/// HTTP Handler to publish an Attestation, which has been signed by a validator.
pub fn publish_attestation<T: BeaconChainTypes + 'static>(req: Request<Body>) -> BoxFut {
let _ = try_future!(check_content_type_for_json(&req));
let log = get_logger_from_request(&req);
let beacon_chain = try_future!(get_beacon_chain_from_request::<T>(&req));
// Get the network sending channel from the request, for later transmission
let network_chan = req
.extensions()
.get::<Arc<RwLock<mpsc::UnboundedSender<NetworkMessage>>>>()
.expect("Should always get the network channel from the request, since we put it in there.")
.clone();
let response_builder = ResponseBuilder::new(&req);
let body = req.into_body();
trace!(
log,
"Got the request body, now going to parse it into an attesation."
);
Box::new(body
.concat2()
.map_err(|e| ApiError::ServerError(format!("Unable to get request body: {:?}",e)))
.map(|chunk| chunk.iter().cloned().collect::<Vec<u8>>())
.and_then(|chunks| {
serde_json::from_slice(&chunks.as_slice()).map_err(|e| {
ApiError::BadRequest(format!(
"Unable to deserialize JSON into a BeaconBlock: {:?}",
e
))
})
})
.and_then(move |attestation: Attestation<T::EthSpec>| {
match beacon_chain.process_attestation(attestation.clone()) {
Ok(AttestationProcessingOutcome::Processed) => {
// Block was processed, publish via gossipsub
info!(log, "Processed valid attestation from API, transmitting to network.");
publish_attestation_to_network::<T>(network_chan, attestation)
}
Ok(outcome) => {
warn!(log, "Attestation could not be processed, but is being sent to the network anyway."; "outcome" => format!("{:?}", outcome));
publish_attestation_to_network::<T>(network_chan, attestation)?;
Err(ApiError::ProcessingError(format!(
"The Attestation could not be processed, but has still been published: {:?}",
outcome
)))
}
Err(e) => {
Err(ApiError::ServerError(format!(
"Error while processing attestation: {:?}",
e
)))
}
}
}).and_then(|_| {
response_builder?.body_no_ssz(&())
}))
}

1329
docs/rest_oapi.yaml → docs/api_spec.yaml
View File

File diff suppressed because it is too large Load Diff

1
validator_client/Cargo.toml
View File

@ -26,6 +26,7 @@ slot_clock = { path = "../eth2/utils/slot_clock" }
types = { path = "../eth2/types" }
serde = "1.0"
serde_derive = "1.0"
serde_json = "^1.0"
slog = { version = "^2.2.3" , features = ["max_level_trace", "release_max_level_trace"] }
slog-async = "^2.3.0"
slog-json = "^2.3"

30
validator_client/src/block_producer/mod.rs
View File

@ -6,7 +6,8 @@ pub use self::beacon_node_block::{BeaconNodeError, PublishOutcome};
pub use self::grpc::BeaconBlockGrpcClient;
use crate::signer::Signer;
use core::marker::PhantomData;
use slog::{error, info, warn};
use serde_json;
use slog::{error, info, trace, warn};
use std::sync::Arc;
use tree_hash::{SignedRoot, TreeHash};
use types::{BeaconBlock, ChainSpec, Domain, EthSpec, Fork, Slot};
@ -53,30 +54,32 @@ pub struct BlockProducer<'a, B: BeaconNodeBlock, S: Signer, E: EthSpec> {
pub slots_per_epoch: u64,
/// Mere vessel for E.
pub _phantom: PhantomData<E>,
/// The logger, for logging
pub log: slog::Logger,
}
impl<'a, B: BeaconNodeBlock, S: Signer, E: EthSpec> BlockProducer<'a, B, S, E> {
/// Handle outputs and results from block production.
pub fn handle_produce_block(&mut self, log: slog::Logger) {
pub fn handle_produce_block(&mut self) {
match self.produce_block() {
Ok(ValidatorEvent::BlockProduced(slot)) => info!(
log,
self.log,
"Block produced";
"validator" => format!("{}", self.signer),
"slot" => slot,
),
Err(e) => error!(log, "Block production error"; "Error" => format!("{:?}", e)),
Err(e) => error!(self.log, "Block production error"; "Error" => format!("{:?}", e)),
Ok(ValidatorEvent::SignerRejection(_slot)) => {
error!(log, "Block production error"; "Error" => "Signer Could not sign the block".to_string())
error!(self.log, "Block production error"; "Error" => "Signer Could not sign the block".to_string())
}
Ok(ValidatorEvent::SlashableBlockNotProduced(_slot)) => {
error!(log, "Block production error"; "Error" => "Rejected the block as it could have been slashed".to_string())
error!(self.log, "Block production error"; "Error" => "Rejected the block as it could have been slashed".to_string())
}
Ok(ValidatorEvent::BeaconNodeUnableToProduceBlock(_slot)) => {
error!(log, "Block production error"; "Error" => "Beacon node was unable to produce a block".to_string())
error!(self.log, "Block production error"; "Error" => "Beacon node was unable to produce a block".to_string())
}
Ok(v) => {
warn!(log, "Unknown result for block production"; "Error" => format!("{:?}",v))
warn!(self.log, "Unknown result for block production"; "Error" => format!("{:?}",v))
}
}
}
@ -93,14 +96,21 @@ impl<'a, B: BeaconNodeBlock, S: Signer, E: EthSpec> BlockProducer<'a, B, S, E> {
/// slashing.
pub fn produce_block(&mut self) -> Result<ValidatorEvent, Error> {
let epoch = self.slot.epoch(self.slots_per_epoch);
trace!(self.log, "Producing block"; "epoch" => epoch);
let message = epoch.tree_hash_root();
let randao_reveal = match self.signer.sign_message(
&message,
self.spec.get_domain(epoch, Domain::Randao, &self.fork),
) {
None => return Ok(ValidatorEvent::SignerRejection(self.slot)),
Some(signature) => signature,
None => {
warn!(self.log, "Signing rejected"; "message" => format!("{:?}", message));
return Ok(ValidatorEvent::SignerRejection(self.slot));
}
Some(signature) => {
info!(self.log, "Signed tree_hash_root for randao_reveal"; "message" => format!("{:?}", message), "signature" => serde_json::to_string(&signature).expect("We should always be able to serialize a signature as JSON."));
signature
}
};
if let Some(block) = self

10
validator_client/src/main.rs
View File

@ -55,7 +55,6 @@ fn main() {
)
.arg(
Arg::with_name("spec")
.short("s")
.long("spec")
.value_name("TITLE")
.help("Specifies the default eth2 spec type.")
@ -143,6 +142,15 @@ fn main() {
.help("Path to a YAML file."))
)
)
.subcommand(SubCommand::with_name("sign_block")
.about("Connects to the beacon server, requests a new block (after providing reveal),\
and prints the signed block to standard out")
.arg(Arg::with_name("validator")
.value_name("VALIDATOR")
.required(true)
.help("The pubkey of the validator that should sign the block.")
)
)
.get_matches();
let drain = match matches.value_of("debug-level") {

3
validator_client/src/service.rs
View File

@ -392,8 +392,9 @@ impl<B: BeaconNodeDuties + 'static, S: Signer + 'static, E: EthSpec> Service<B,
signer,
slots_per_epoch,
_phantom: PhantomData::<E>,
log,
};
block_producer.handle_produce_block(log);
block_producer.handle_produce_block();
});
}
if work_type.attestation_duty.is_some() {