Making plugin empty
This commit is contained in:
godmodegalactus
parent
0698406657
commit
94ba418f84
File diff suppressed because it is too large
Load Diff
14
Cargo.toml
14
Cargo.toml
|
|
@ -1,7 +1,6 @@
|
|||
[workspace]
|
||||
members = [
|
||||
"plugin",
|
||||
"client"
|
||||
"plugin"
|
||||
]
|
||||
|
||||
[profile.release]
|
||||
|
|
@ -9,7 +8,10 @@ debug = true
|
|||
lto = true
|
||||
codegen-units = 1
|
||||
|
||||
[patch.crates-io]
|
||||
solana-geyser-plugin-interface = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-sdk = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-streamer = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
[workspace.dependencies]
|
||||
solana-geyser-plugin-interface = "1.17.28"
|
||||
solana-sdk = "1.17.28"
|
||||
solana-streamer = "1.17.28"
|
||||
solana-quic-client = "1.17.28"
|
||||
solana-net-utils = "1.17.28"
|
||||
solana-connection-cache = "1.17.28"
|
||||
|
|
|
|||
|
|
@ -1,51 +0,0 @@
|
|||
[package]
|
||||
name = "client"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros", "time", "fs"] }
|
||||
solana-sdk = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-geyser-plugin-interface = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-streamer = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-quic-client = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-net-utils = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-connection-cache = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
|
||||
geyser-quic-plugin = { path = "../plugin" }
|
||||
|
||||
itertools = "0.10.5"
|
||||
serde = { version = "1.0.160", features = ["derive"] }
|
||||
serde_json = "1.0.96"
|
||||
bincode = "1.3.3"
|
||||
bs58 = "0.4.0"
|
||||
base64 = "0.21.0"
|
||||
thiserror = "1.0.40"
|
||||
futures = "0.3.28"
|
||||
bytes = "1.4.0"
|
||||
anyhow = "1.0.70"
|
||||
log = "0.4.17"
|
||||
dashmap = "5.4.0"
|
||||
const_env = "0.1.2"
|
||||
jsonrpsee = { version = "0.17.0", features = ["macros", "full"] }
|
||||
tracing = "0.1.37"
|
||||
tracing-subscriber = "0.3.16"
|
||||
chrono = "0.4.24"
|
||||
native-tls = "0.2.11"
|
||||
postgres-native-tls = "0.5.0"
|
||||
prometheus = "0.13.3"
|
||||
lazy_static = "1.4.0"
|
||||
dotenv = "0.15.0"
|
||||
async-channel = "1.8.0"
|
||||
quinn = "0.10.2"
|
||||
rustls = { version = "0.21.8", default-features = false, features = ["quic"] }
|
||||
rcgen = "0.10.0"
|
||||
pkcs8 = "0.8.0"
|
||||
pem = "1.1.1"
|
||||
|
||||
clap = { version = "=4.3.24", features = ["cargo", "derive"] }
|
||||
anstyle = "=1.0.0"
|
||||
anstyle-parse = "=0.2.0"
|
||||
clap_lex = "=0.5.0"
|
||||
|
|
@ -1,11 +0,0 @@
|
|||
use clap::Parser;
|
||||
|
||||
#[derive(Parser, Debug, Clone)]
|
||||
#[command(author, version, about, long_about = None)]
|
||||
pub struct Args {
|
||||
#[arg(short, long, default_value_t = String::from("127.0.0.1:11000"))]
|
||||
pub geyser_quic_address: String,
|
||||
|
||||
#[arg(short, long, default_value_t = String::from("connection_identity.json"))]
|
||||
pub identity: String,
|
||||
}
|
||||
|
|
@ -1,99 +0,0 @@
|
|||
use std::{time::Duration, net::{IpAddr, Ipv4Addr, SocketAddr}, sync::Arc};
|
||||
|
||||
use cli::Args;
|
||||
use geyser_quic_plugin::{TransactionResults, ALPN_GEYSER_PROTOCOL_ID, tls_certificate::new_self_signed_tls_certificate};
|
||||
use quinn::{TokioRuntime, EndpointConfig, Endpoint, ClientConfig, TransportConfig, IdleTimeout};
|
||||
use skip_server_verification::SkipServerVerification;
|
||||
use solana_sdk::signature::Keypair;
|
||||
use clap::Parser;
|
||||
|
||||
mod cli;
|
||||
mod skip_server_verification;
|
||||
|
||||
pub const PACKET_DATA_SIZE: usize = 1280 - 40 - 8;
|
||||
|
||||
pub async fn load_identity_keypair(identity_file: &String) -> Option<Keypair> {
|
||||
let identity_file = tokio::fs::read_to_string(identity_file.as_str())
|
||||
.await
|
||||
.expect("Cannot find the identity file provided");
|
||||
let identity_bytes: Vec<u8> = serde_json::from_str(&identity_file).unwrap();
|
||||
Some(Keypair::from_bytes(identity_bytes.as_slice()).unwrap())
|
||||
}
|
||||
|
||||
pub fn create_endpoint(certificate: rustls::Certificate, key: rustls::PrivateKey) -> Endpoint {
|
||||
let mut endpoint = {
|
||||
let client_socket =
|
||||
solana_net_utils::bind_in_range(IpAddr::V4(Ipv4Addr::UNSPECIFIED), (8000, 10000))
|
||||
.expect("create_endpoint bind_in_range")
|
||||
.1;
|
||||
let config = EndpointConfig::default();
|
||||
quinn::Endpoint::new(config, None, client_socket, Arc::new(TokioRuntime))
|
||||
.expect("create_endpoint quinn::Endpoint::new")
|
||||
};
|
||||
|
||||
let mut crypto = rustls::ClientConfig::builder()
|
||||
.with_safe_defaults()
|
||||
.with_custom_certificate_verifier(Arc::new(SkipServerVerification {}))
|
||||
.with_client_auth_cert(vec![certificate], key)
|
||||
.expect("Failed to set QUIC client certificates");
|
||||
|
||||
crypto.enable_early_data = true;
|
||||
crypto.alpn_protocols = vec![ALPN_GEYSER_PROTOCOL_ID.to_vec()];
|
||||
|
||||
let mut config = ClientConfig::new(Arc::new(crypto));
|
||||
let mut transport_config = TransportConfig::default();
|
||||
|
||||
let timeout = IdleTimeout::try_from(Duration::from_secs(3600 * 48)).unwrap();
|
||||
transport_config.max_idle_timeout(Some(timeout));
|
||||
transport_config.keep_alive_interval(Some(Duration::from_millis(500)));
|
||||
config.transport_config(Arc::new(transport_config));
|
||||
|
||||
endpoint.set_default_client_config(config);
|
||||
|
||||
endpoint
|
||||
}
|
||||
|
||||
#[tokio::main()]
|
||||
pub async fn main() -> anyhow::Result<()> {
|
||||
let args = Args::parse();
|
||||
let address: SocketAddr = args.geyser_quic_address.parse().expect("should be valid socket address");
|
||||
|
||||
let keypair = load_identity_keypair(&args.identity).await.expect("Identity file should be valid");
|
||||
|
||||
let (certificate, key) = new_self_signed_tls_certificate(
|
||||
&keypair,
|
||||
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
|
||||
)
|
||||
.expect("Failed to initialize QUIC client certificates");
|
||||
|
||||
let endpoint = create_endpoint(certificate, key);
|
||||
let connection = endpoint.connect(address, "quic_geyser_plugin").expect("Should be connecting").await.expect("Should be able to connect to the plugin");
|
||||
let (mut send_stream, recv_stream) = connection.open_bi().await.expect("Should be able to create a bi directional connection");
|
||||
send_stream.write_all(b"connect").await.unwrap();
|
||||
// let jh = tokio::spawn(async move {
|
||||
// wait for 10 s max
|
||||
let mut buffer: [u8; PACKET_DATA_SIZE] = [0; PACKET_DATA_SIZE];
|
||||
let mut recv_stream = recv_stream;
|
||||
loop {
|
||||
let res = recv_stream.read(&mut buffer ).await;
|
||||
match res
|
||||
{
|
||||
Ok(Some(size)) => {
|
||||
let data = &buffer[0..size];
|
||||
if let Ok(result) = bincode::deserialize::<TransactionResults>(&data) {
|
||||
println!("Transaction Result \n s:{} e:{} slt:{}", result.signature, result.error.map(|x| x.to_string()).unwrap_or_default(), result.slot);
|
||||
}
|
||||
},
|
||||
Ok(None) => {
|
||||
log::warn!("got ok none");
|
||||
},
|
||||
Err(e) => {
|
||||
log::error!("got error {e:?}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
// });
|
||||
// jh.await.unwrap();
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,15 +0,0 @@
|
|||
pub struct SkipServerVerification;
|
||||
|
||||
impl rustls::client::ServerCertVerifier for SkipServerVerification {
|
||||
fn verify_server_cert(
|
||||
&self,
|
||||
_end_entity: &rustls::Certificate,
|
||||
_intermediates: &[rustls::Certificate],
|
||||
_server_name: &rustls::ServerName,
|
||||
_scts: &mut dyn Iterator<Item = &[u8]>,
|
||||
_ocsp_response: &[u8],
|
||||
_now: std::time::SystemTime,
|
||||
) -> Result<rustls::client::ServerCertVerified, rustls::Error> {
|
||||
Ok(rustls::client::ServerCertVerified::assertion())
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
[package]
|
||||
name = "geyser-quic-plugin"
|
||||
name = "geyser-empty-plugin"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
authors = ["Godmode Galactus"]
|
||||
|
|
@ -13,45 +13,17 @@ name = "config-check"
|
|||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
solana-sdk = { workspace = true }
|
||||
solana-geyser-plugin-interface = { workspace = true }
|
||||
|
||||
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros", "time", "fs"] }
|
||||
solana-sdk = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
solana-geyser-plugin-interface = { git = "https://github.com/blockworks-foundation/solana.git", branch = "geyser_send_transaction_results_v1.16.18" }
|
||||
itertools = "0.10.5"
|
||||
serde = { version = "1.0.160", features = ["derive"] }
|
||||
serde_json = "1.0.96"
|
||||
bincode = "1.3.3"
|
||||
bs58 = "0.4.0"
|
||||
base64 = "0.21.0"
|
||||
thiserror = "1.0.40"
|
||||
futures = "0.3.28"
|
||||
bytes = "1.4.0"
|
||||
anyhow = "1.0.70"
|
||||
log = "0.4.17"
|
||||
dashmap = "5.4.0"
|
||||
const_env = "0.1.2"
|
||||
jsonrpsee = { version = "0.17.0", features = ["macros", "full"] }
|
||||
tracing = "0.1.37"
|
||||
tracing-subscriber = "0.3.16"
|
||||
chrono = "0.4.24"
|
||||
native-tls = "0.2.11"
|
||||
postgres-native-tls = "0.5.0"
|
||||
prometheus = "0.13.3"
|
||||
lazy_static = "1.4.0"
|
||||
dotenv = "0.15.0"
|
||||
async-channel = "1.8.0"
|
||||
quinn = "0.10.2"
|
||||
rustls = { version = "0.21.8", default-features = false, features = ["quic", "dangerous_configuration"] }
|
||||
rcgen = "0.10.0"
|
||||
pkcs8 = "0.8.0"
|
||||
pem = "1.1.1"
|
||||
x509-parser = "0.14.0"
|
||||
clap = { version = "=4.3.24", features = ["cargo", "derive"] }
|
||||
anstyle = "=1.0.0"
|
||||
anstyle-parse = "=0.2.0"
|
||||
clap_lex = "=0.5.0"
|
||||
anyhow = "1.0.70"
|
||||
|
||||
[build-dependencies]
|
||||
anyhow = "1.0.62"
|
||||
cargo-lock = "9.0.0"
|
||||
git-version = "0.3.5"
|
||||
vergen = { version = "8.2.1", features = ["build", "rustc"] }
|
||||
vergen = { version = "8.2.1", features = ["build", "rustc"] }
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
use {clap::Parser, geyser_quic_plugin::config::Config};
|
||||
use {clap::Parser, geyser_empty_plugin::config::Config};
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[clap(author, version, about)]
|
||||
|
|
|
|||
|
|
@ -1,22 +1,15 @@
|
|||
use std::{net::SocketAddr, fs::read_to_string, path::Path};
|
||||
use std::{fs::read_to_string, path::Path};
|
||||
|
||||
use serde::Deserialize;
|
||||
use solana_geyser_plugin_interface::geyser_plugin_interface::GeyserPluginError;
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct Config {
|
||||
pub libpath: String,
|
||||
pub quic_plugin: ConfigQuicPlugin,
|
||||
}
|
||||
|
||||
pub struct Config {}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct ConfigQuicPlugin {
|
||||
/// Address of Grpc service.
|
||||
pub address: SocketAddr,
|
||||
}
|
||||
pub struct ConfigQuicPlugin {}
|
||||
|
||||
impl Config {
|
||||
fn load_from_str(config: &str) -> std::result::Result<Self, GeyserPluginError> {
|
||||
|
|
@ -29,4 +22,4 @@ impl Config {
|
|||
let config = read_to_string(file).map_err(GeyserPluginError::ConfigFileOpenError)?;
|
||||
Self::load_from_str(&config)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,266 +1,28 @@
|
|||
use std::{
|
||||
net::{IpAddr, Ipv4Addr, UdpSocket},
|
||||
str::FromStr,
|
||||
sync::{Arc, atomic::AtomicBool},
|
||||
};
|
||||
use solana_geyser_plugin_interface::geyser_plugin_interface::GeyserPlugin;
|
||||
|
||||
use crate::{config::Config, tls_certificate::new_self_signed_tls_certificate};
|
||||
use itertools::Itertools;
|
||||
use pem::Pem;
|
||||
use quinn::{Endpoint, EndpointConfig, IdleTimeout, ServerConfig, TokioRuntime};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use solana_geyser_plugin_interface::geyser_plugin_interface::{
|
||||
GeyserPlugin, GeyserPluginError, Result as PluginResult,
|
||||
};
|
||||
use solana_sdk::{
|
||||
packet::PACKET_DATA_SIZE,
|
||||
pubkey::Pubkey,
|
||||
quic::QUIC_MAX_TIMEOUT,
|
||||
signature::{Keypair, Signature},
|
||||
slot_history::Slot,
|
||||
transaction::{SanitizedTransaction, TransactionError}, compute_budget::{self, ComputeBudgetInstruction}, borsh0_10::try_from_slice_unchecked,
|
||||
};
|
||||
use tls_certificate::get_pubkey_from_tls_certificate;
|
||||
use tokio::{runtime::Runtime, sync::mpsc::UnboundedSender, task::JoinHandle};
|
||||
|
||||
use crate::skip_client_verification::SkipClientVerification;
|
||||
|
||||
pub mod skip_client_verification;
|
||||
pub mod config;
|
||||
pub mod tls_certificate;
|
||||
|
||||
pub const ALPN_GEYSER_PROTOCOL_ID: &[u8] = b"solana-geyser";
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
pub struct TransactionResults {
|
||||
pub signature: Signature,
|
||||
pub error: Option<TransactionError>,
|
||||
pub slot: Slot,
|
||||
pub writable_accounts: Vec<Pubkey>,
|
||||
pub readable_accounts: Vec<Pubkey>,
|
||||
pub cu_requested: u64,
|
||||
pub prioritization_fees : u64,
|
||||
}
|
||||
|
||||
fn decode_cu_requested_and_prioritization_fees(transaction: &SanitizedTransaction,) -> (u64, u64) {
|
||||
let mut cu_requested:u64 = 200_000;
|
||||
let mut prioritization_fees: u64 = 0;
|
||||
let accounts = transaction.message().account_keys().iter().map(|x| *x).collect_vec();
|
||||
for ix in transaction.message().instructions() {
|
||||
if ix.program_id(accounts.as_slice())
|
||||
.eq(&compute_budget::id())
|
||||
{
|
||||
let cb_ix = try_from_slice_unchecked::<ComputeBudgetInstruction>(ix.data.as_slice());
|
||||
if let Ok(ComputeBudgetInstruction::RequestUnitsDeprecated {
|
||||
units,
|
||||
additional_fee,
|
||||
}) = cb_ix
|
||||
{
|
||||
if additional_fee > 0 {
|
||||
return (units as u64, ((units * 1000) / additional_fee) as u64);
|
||||
} else {
|
||||
return (units as u64, 0);
|
||||
}
|
||||
} else if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(units)) = cb_ix {
|
||||
cu_requested = units as u64;
|
||||
} else if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) = cb_ix {
|
||||
prioritization_fees = price;
|
||||
}
|
||||
}
|
||||
}
|
||||
(cu_requested, prioritization_fees)
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct PluginInner {
|
||||
pub runtime: Runtime,
|
||||
pub handle: JoinHandle<()>,
|
||||
pub sender: Arc<UnboundedSender<TransactionResults>>,
|
||||
pub start_sending: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Plugin {
|
||||
inner: Option<PluginInner>,
|
||||
}
|
||||
|
||||
impl GeyserPlugin for Plugin {
|
||||
fn name(&self) -> &'static str {
|
||||
"geyser_quic_banking_transactions_result_sender"
|
||||
}
|
||||
|
||||
fn banking_transaction_results_notifications_enabled(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
#[allow(unused_variables)]
|
||||
fn notify_banking_stage_transaction_results(
|
||||
&self,
|
||||
transaction: &SanitizedTransaction,
|
||||
error: Option<TransactionError>,
|
||||
slot: Slot,
|
||||
) -> PluginResult<()> {
|
||||
if let Some(inner) = &self.inner {
|
||||
if !inner.start_sending.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
return Ok(())
|
||||
}
|
||||
if transaction.is_simple_vote_transaction() {
|
||||
return Ok(())
|
||||
}
|
||||
let message = transaction.message();
|
||||
|
||||
let accounts = message.account_keys();
|
||||
let is_writable = accounts.iter().enumerate().map(|(index, _)| {
|
||||
transaction.message().is_writable(index)
|
||||
}).collect_vec();
|
||||
let mut writable_accounts = is_writable.iter().enumerate().filter(|(_, v)| **v).map(|(index, get_mut)| accounts[index]).collect_vec();
|
||||
let mut readable_accounts = is_writable.iter().enumerate().filter(|(_, v)| !**v).map(|(index, get_mut)| accounts[index]).collect_vec();
|
||||
writable_accounts.truncate(32);
|
||||
readable_accounts.truncate(32);
|
||||
|
||||
let (cu_requested, prioritization_fees) = decode_cu_requested_and_prioritization_fees(transaction);
|
||||
|
||||
if let Err(e) = inner.sender.send(TransactionResults {
|
||||
signature: transaction.signature().clone(),
|
||||
error,
|
||||
slot,
|
||||
writable_accounts,
|
||||
readable_accounts,
|
||||
cu_requested,
|
||||
prioritization_fees,
|
||||
}) {
|
||||
log::error!("error sending on the channel {e:?}");
|
||||
}
|
||||
Ok(())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
"geyser_empty_plugin"
|
||||
}
|
||||
|
||||
fn on_load(
|
||||
&mut self,
|
||||
config_file: &str,
|
||||
_config_file: &str,
|
||||
_: bool,
|
||||
) -> solana_geyser_plugin_interface::geyser_plugin_interface::Result<()> {
|
||||
let plugin_config = Config::load_from_file(config_file)?;
|
||||
let runtime = Runtime::new().map_err(|error| GeyserPluginError::Custom(Box::new(error)))?;
|
||||
let res = configure_server(&Keypair::new(), IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)));
|
||||
|
||||
let (config, _) = res.map_err(|_| GeyserPluginError::TransactionUpdateError { msg: "error configuring server".to_string() })?;
|
||||
let sock = UdpSocket::bind(plugin_config.quic_plugin.address).expect("couldn't bind to address");
|
||||
|
||||
let (sender, reciever) = tokio::sync::mpsc::unbounded_channel::<TransactionResults>();
|
||||
|
||||
let allowed_connection =
|
||||
Pubkey::from_str("G8pLuvzarejjLuuPNVNR1gk9xiFKmAcs9J5LL3GZGM6F").unwrap();
|
||||
let start_sending = Arc::new(AtomicBool::new(false));
|
||||
let start_sending_cp = start_sending.clone();
|
||||
|
||||
let handle = runtime.block_on(async move {
|
||||
let mut reciever = reciever;
|
||||
let endpoint = Endpoint::new(EndpointConfig::default(), Some(config), sock, Arc::new(TokioRuntime)).expect("Should be able to create endpoint");
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let connecting = endpoint.accept().await;
|
||||
if let Some(connecting) = connecting {
|
||||
let connected = connecting.await;
|
||||
let connection = match connected {
|
||||
Ok(connection) => connection,
|
||||
Err(e) => {
|
||||
log::error!("geyser plugin connecting {} error", e);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let connection_identity = get_remote_pubkey(&connection);
|
||||
if let Some(connection_identity) = connection_identity {
|
||||
if !allowed_connection.eq(&connection_identity) {
|
||||
// not an authorized connection
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
continue;
|
||||
}
|
||||
let (mut send_stream, _) = match connection.accept_bi().await {
|
||||
Ok(res) => res,
|
||||
Err(e) => {
|
||||
log::error!("geyser plugin accepting bi-channel {} error", e);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
start_sending_cp.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
while let Some(msg) = reciever.recv().await {
|
||||
let bytes = bincode::serialize(&msg).unwrap_or(vec![]);
|
||||
if !bytes.is_empty() {
|
||||
if let Err(e) = send_stream.write_all(&bytes).await {
|
||||
log::error!("error writing on stream channel {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
start_sending_cp.store(false, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
})
|
||||
});
|
||||
|
||||
self.inner = Some(PluginInner {
|
||||
runtime,
|
||||
handle,
|
||||
sender: Arc::new(sender),
|
||||
start_sending,
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn on_unload(&mut self) {}
|
||||
}
|
||||
|
||||
pub(crate) fn configure_server(
|
||||
identity_keypair: &Keypair,
|
||||
host: IpAddr,
|
||||
) -> anyhow::Result<(ServerConfig, String)> {
|
||||
let (cert, priv_key) = new_self_signed_tls_certificate(identity_keypair, host)?;
|
||||
let cert_chain_pem_parts = vec![Pem {
|
||||
tag: "CERTIFICATE".to_string(),
|
||||
contents: cert.0.clone(),
|
||||
}];
|
||||
let cert_chain_pem = pem::encode_many(&cert_chain_pem_parts);
|
||||
|
||||
let mut server_tls_config = rustls::ServerConfig::builder()
|
||||
.with_safe_defaults()
|
||||
.with_client_cert_verifier(SkipClientVerification::new())
|
||||
.with_single_cert(vec![cert], priv_key)?;
|
||||
server_tls_config.alpn_protocols = vec![ALPN_GEYSER_PROTOCOL_ID.to_vec()];
|
||||
|
||||
let mut server_config = ServerConfig::with_crypto(Arc::new(server_tls_config));
|
||||
server_config.use_retry(true);
|
||||
let config = Arc::get_mut(&mut server_config.transport).unwrap();
|
||||
|
||||
config.max_concurrent_uni_streams((1 as u32).into());
|
||||
let recv_size = (PACKET_DATA_SIZE as u32 * 100).into();
|
||||
config.stream_receive_window(recv_size);
|
||||
config.receive_window(recv_size);
|
||||
let timeout = IdleTimeout::try_from(QUIC_MAX_TIMEOUT).unwrap();
|
||||
config.max_idle_timeout(Some(timeout));
|
||||
|
||||
// disable bidi & datagrams
|
||||
const MAX_CONCURRENT_BIDI_STREAMS: u32 = 10;
|
||||
config.max_concurrent_bidi_streams(MAX_CONCURRENT_BIDI_STREAMS.into());
|
||||
config.datagram_receive_buffer_size(None);
|
||||
|
||||
Ok((server_config, cert_chain_pem))
|
||||
}
|
||||
|
||||
pub fn get_remote_pubkey(connection: &quinn::Connection) -> Option<Pubkey> {
|
||||
// Use the client cert only if it is self signed and the chain length is 1.
|
||||
connection
|
||||
.peer_identity()?
|
||||
.downcast::<Vec<rustls::Certificate>>()
|
||||
.ok()
|
||||
.filter(|certs| certs.len() == 1)?
|
||||
.first()
|
||||
.and_then(get_pubkey_from_tls_certificate)
|
||||
}
|
||||
|
||||
#[no_mangle]
|
||||
#[allow(improper_ctypes_definitions)]
|
||||
pub unsafe extern "C" fn _create_plugin() -> *mut dyn GeyserPlugin {
|
||||
|
|
|
|||
|
|
@ -1,26 +0,0 @@
|
|||
use std::{sync::Arc, time::SystemTime};
|
||||
|
||||
use rustls::{server::ClientCertVerified, Certificate, DistinguishedName};
|
||||
|
||||
pub struct SkipClientVerification;
|
||||
|
||||
impl SkipClientVerification {
|
||||
pub fn new() -> Arc<Self> {
|
||||
Arc::new(Self)
|
||||
}
|
||||
}
|
||||
|
||||
impl rustls::server::ClientCertVerifier for SkipClientVerification {
|
||||
fn client_auth_root_subjects(&self) -> &[DistinguishedName] {
|
||||
&[]
|
||||
}
|
||||
|
||||
fn verify_client_cert(
|
||||
&self,
|
||||
_end_entity: &Certificate,
|
||||
_intermediates: &[Certificate],
|
||||
_now: SystemTime,
|
||||
) -> Result<ClientCertVerified, rustls::Error> {
|
||||
Ok(rustls::server::ClientCertVerified::assertion())
|
||||
}
|
||||
}
|
||||
|
|
@ -1,53 +0,0 @@
|
|||
use std::net::IpAddr;
|
||||
|
||||
use pkcs8::{AlgorithmIdentifier, ObjectIdentifier, der::Document};
|
||||
use rcgen::{RcgenError, CertificateParams, SanType, DistinguishedName, DnType};
|
||||
use solana_sdk::{signature::Keypair, pubkey::Pubkey};
|
||||
use x509_parser::{prelude::{X509Certificate, FromDer}, public_key::PublicKey};
|
||||
|
||||
pub fn new_self_signed_tls_certificate(
|
||||
keypair: &Keypair,
|
||||
san: IpAddr,
|
||||
) -> Result<(rustls::Certificate, rustls::PrivateKey), RcgenError> {
|
||||
const ED25519_IDENTIFIER: [u32; 4] = [1, 3, 101, 112];
|
||||
let mut private_key = Vec::<u8>::with_capacity(34);
|
||||
private_key.extend_from_slice(&[0x04, 0x20]); // ASN.1 OCTET STRING
|
||||
private_key.extend_from_slice(keypair.secret().as_bytes());
|
||||
let key_pkcs8 = pkcs8::PrivateKeyInfo {
|
||||
algorithm: AlgorithmIdentifier {
|
||||
oid: ObjectIdentifier::from_arcs(&ED25519_IDENTIFIER).expect("Failed to convert OID"),
|
||||
parameters: None,
|
||||
},
|
||||
private_key: &private_key,
|
||||
public_key: None,
|
||||
};
|
||||
let key_pkcs8_der = key_pkcs8
|
||||
.to_der()
|
||||
.expect("Failed to convert keypair to DER")
|
||||
.to_der();
|
||||
|
||||
let rcgen_keypair = rcgen::KeyPair::from_der(&key_pkcs8_der)?;
|
||||
|
||||
let mut cert_params = CertificateParams::default();
|
||||
cert_params.subject_alt_names = vec![SanType::IpAddress(san)];
|
||||
cert_params.alg = &rcgen::PKCS_ED25519;
|
||||
cert_params.key_pair = Some(rcgen_keypair);
|
||||
cert_params.distinguished_name = DistinguishedName::new();
|
||||
cert_params
|
||||
.distinguished_name
|
||||
.push(DnType::CommonName, "Solana node");
|
||||
|
||||
let cert = rcgen::Certificate::from_params(cert_params)?;
|
||||
let cert_der = cert.serialize_der().unwrap();
|
||||
let priv_key = cert.serialize_private_key_der();
|
||||
let priv_key = rustls::PrivateKey(priv_key);
|
||||
Ok((rustls::Certificate(cert_der), priv_key))
|
||||
}
|
||||
|
||||
pub fn get_pubkey_from_tls_certificate(der_cert: &rustls::Certificate) -> Option<Pubkey> {
|
||||
let (_, cert) = X509Certificate::from_der(der_cert.as_ref()).ok()?;
|
||||
match cert.public_key().parsed().ok()? {
|
||||
PublicKey::Unknown(key) => Pubkey::try_from(key).ok(),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
[toolchain]
|
||||
channel = "1.69.0"
|
||||
channel = "1.75.0"
|
||||
components = ["clippy", "rustfmt"]
|
||||
targets = []
|
||||
profile = "minimal"
|
||||
Loading…
Reference in New Issue