tornadocontrib/web3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lint

Browse Source
This commit is contained in:
Bryan Stitt 2023-03-01 11:23:59 -08:00
parent 1e29a057ab
commit b87c988439
12 changed files with 198 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
entities/src/admin.rs
View File

@ -1,6 +1,5 @@
//! `SeaORM` Entity. Generated by sea-orm-codegen 0.10.6
use crate::serialization;
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};

21
migration/src/m20230130_124740_read_only_login_logic.rs
View File

@ -11,12 +11,10 @@ impl MigrationTrait for Migration {
.alter_table(
Table::alter()
.table(Login::Table)
.add_column(
ColumnDef::new(Login::ReadOnly)
.boolean()
.not_null()
).to_owned()
).await
.add_column(ColumnDef::new(Login::ReadOnly).boolean().not_null())
.to_owned(),
)
.await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
@ -26,8 +24,9 @@ impl MigrationTrait for Migration {
Table::alter()
.table(Login::Table)
.drop_column(Login::ReadOnly)
.to_owned()
).await
.to_owned(),
)
.await
}
}
@ -35,8 +34,8 @@ impl MigrationTrait for Migration {
#[derive(Iden)]
enum Login {
Table,
Id,
BearerToken,
// Id,
// BearerToken,
ReadOnly,
UserId,
// UserId,
}

38
migration/src/m20230130_165144_prepare_admin_imitation_pre_login.rs
View File

@ -10,19 +10,18 @@ impl MigrationTrait for Migration {
.alter_table(
Table::alter()
.table(PendingLogin::Table)
.add_column(
ColumnDef::new(PendingLogin::ImitatingUser)
.big_unsigned()
.add_column(ColumnDef::new(PendingLogin::ImitatingUser).big_unsigned())
.add_foreign_key(
TableForeignKey::new()
.name("fk-pending_login-imitating_user")
.from_tbl(PendingLogin::Table)
.to_tbl(User::Table)
.from_col(PendingLogin::ImitatingUser)
.to_col(User::Id),
)
.add_foreign_key(&TableForeignKey::new()
.name("fk-pending_login-imitating_user")
.from_tbl(PendingLogin::Table)
.to_tbl(User::Table)
.from_col(PendingLogin::ImitatingUser)
.to_col(User::Id)
)
.to_owned()
).await
.to_owned(),
)
.await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
@ -32,8 +31,9 @@ impl MigrationTrait for Migration {
.table(PendingLogin::Table)
.drop_foreign_key(Alias::new("fk-pending_login-imitating_user"))
.drop_column(PendingLogin::ImitatingUser)
.to_owned()
).await
.to_owned(),
)
.await
}
}
@ -41,10 +41,10 @@ impl MigrationTrait for Migration {
#[derive(Iden)]
enum PendingLogin {
Table,
Id,
Nonce,
Message,
ExpiresAt,
// Id,
// Nonce,
// Message,
// ExpiresAt,
ImitatingUser,
}
@ -52,5 +52,5 @@ enum PendingLogin {
#[derive(Iden)]
enum User {
Table,
Id
Id,
}

6
migration/src/m20230215_152254_admin_trail.rs
View File

@ -87,7 +87,7 @@ enum AdminTrail {
enum User {
Table,
Id,
Address,
Description,
Email,
// Address,
// Description,
// Email,
}

73
web3_proxy/src/admin_queries.rs
View File

@ -2,50 +2,58 @@ use crate::app::Web3ProxyApp;
use crate::frontend::errors::FrontendErrorResponse;
use crate::user_queries::get_user_id_from_params;
use anyhow::Context;
use axum::{
Json,
headers::{authorization::Bearer, Authorization},
TypedHeader,
};
use axum::response::{IntoResponse, Response};
use axum::{
headers::{authorization::Bearer, Authorization},
Json, TypedHeader,
};
use entities::{admin, login, user, user_tier};
use ethers::prelude::Address;
use ethers::types::Bytes;
use ethers::utils::keccak256;
use hashbrown::HashMap;
use http::StatusCode;
use migration::sea_orm::{self, ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter};
use log::{info, debug};
use redis_rate_limiter::redis::AsyncCommands;
use log::{debug, info};
use migration::sea_orm::{
self, ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter,
};
// TODO: Add some logic to check if the operating user is an admin
// If he is, return true
// If he is not, return false
// This function is used to give permission to certain users
pub async fn query_admin_modify_usertier<'a>(
app: &'a Web3ProxyApp,
bearer: Option<TypedHeader<Authorization<Bearer>>>,
params: &'a HashMap<String, String>
params: &'a HashMap<String, String>,
) -> Result<Response, FrontendErrorResponse> {
// Quickly return if any of the input tokens are bad
let user_address: Vec<u8> = params
.get("user_address")
.ok_or_else(|| FrontendErrorResponse::BadRequest("Unable to find user_address key in request".to_string()))?
.ok_or_else(|| {
FrontendErrorResponse::BadRequest(
"Unable to find user_address key in request".to_string(),
)
})?
.parse::<Address>()
.map_err(|_| FrontendErrorResponse::BadRequest("Unable to parse user_address as an Address".to_string()))?
.to_fixed_bytes().into();
let user_tier_title = params
.get("user_tier_title")
.ok_or_else(||FrontendErrorResponse::BadRequest("Unable to get the user_tier_title key from the request".to_string()))?;
.map_err(|_| {
FrontendErrorResponse::BadRequest(
"Unable to parse user_address as an Address".to_string(),
)
})?
.to_fixed_bytes()
.into();
let user_tier_title = params.get("user_tier_title").ok_or_else(|| {
FrontendErrorResponse::BadRequest(
"Unable to get the user_tier_title key from the request".to_string(),
)
})?;
// Prepare output body
let mut response_body = HashMap::new();
// Establish connections
let db_conn = app.db_conn().context("query_admin_modify_user needs a db")?;
let db_conn = app
.db_conn()
.context("query_admin_modify_user needs a db")?;
let db_replica = app
.db_replica()
.context("query_user_stats needs a db replica")?;
@ -57,16 +65,16 @@ pub async fn query_admin_modify_usertier<'a>(
// Will modify logic here
// Try to get the user who is calling from redis (if existent) / else from the database
// TODO: Make a single query, where you retrieve the user, and directly from it the secondary user (otherwise we do two jumpy, which is unnecessary)
// get the user id first. if it is 0, we should use a cache on the app
let caller_id = get_user_id_from_params(&mut redis_conn, &db_conn, &db_replica, bearer, &params).await?;
let caller_id =
get_user_id_from_params(&mut redis_conn, &db_conn, &db_replica, bearer, params).await?;
debug!("Caller id is: {:?}", caller_id);
// Check if the caller is an admin (i.e. if he is in an admin table)
let admin: admin::Model = admin::Entity::find()
let _admin: admin::Model = admin::Entity::find()
.filter(admin::Column::UserId.eq(caller_id))
.one(&db_conn)
.await?
@ -79,7 +87,9 @@ pub async fn query_admin_modify_usertier<'a>(
.filter(user::Column::Address.eq(user_address))
.one(&db_conn)
.await?
.ok_or(FrontendErrorResponse::BadRequest("No user with this id found".to_string()))?;
.ok_or(FrontendErrorResponse::BadRequest(
"No user with this id found".to_string(),
))?;
// Return early if the target user_tier_id is the same as the original user_tier_id
response_body.insert(
"user_tier_title",
@ -91,7 +101,9 @@ pub async fn query_admin_modify_usertier<'a>(
.filter(user_tier::Column::Title.eq(user_tier_title.clone()))
.one(&db_conn)
.await?
.ok_or(FrontendErrorResponse::BadRequest("User Tier name was not found".to_string()))?;
.ok_or(FrontendErrorResponse::BadRequest(
"User Tier name was not found".to_string(),
))?;
if user.user_tier_id == new_user_tier.id {
info!("user already has that tier");
@ -105,18 +117,11 @@ pub async fn query_admin_modify_usertier<'a>(
info!("user's tier changed");
}
// Query the login table, and get all bearer tokens by this user
let bearer_tokens = login::Entity::find()
.filter(login::Column::UserId.eq(user.id))
.all(&db_conn)
.await?;
// Now delete these tokens ...
// Now delete all bearer tokens of this user
login::Entity::delete_many()
.filter(login::Column::UserId.eq(user.id))
.exec(&db_conn)
.await?;
Ok(Json(&response_body).into_response())
}

19
web3_proxy/src/bin/web3_proxy_cli/change_user_admin_status.rs
View File

@ -1,15 +1,11 @@
use anyhow::Context;
use argh::FromArgs;
use entities::{admin, login, user};
use ethers::types::{Address, Bytes};
use ethers::utils::keccak256;
use http::StatusCode;
use log::{debug, info};
use ethers::types::Address;
use log::debug;
use migration::sea_orm::{
self, ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, ModelTrait, IntoActiveModel,
QueryFilter,
self, ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, ModelTrait, QueryFilter,
};
use web3_proxy::frontend::errors::FrontendErrorResponse;
/// change a user's admin status. eiter they are an admin, or they aren't
#[derive(FromArgs, PartialEq, Eq, Debug)]
@ -44,7 +40,8 @@ impl ChangeUserAdminStatusSubCommand {
match admin::Entity::find()
.filter(admin::Column::UserId.eq(address))
.one(db_conn)
.await? {
.await?
{
Some(old_admin) if !should_be_admin => {
// User is already an admin, but shouldn't be
old_admin.delete(db_conn).await?;
@ -65,12 +62,6 @@ impl ChangeUserAdminStatusSubCommand {
}
}
// Get the bearer tokens of this user and delete them ...
let bearer_tokens = login::Entity::find()
.filter(login::Column::UserId.eq(user.id))
.all(db_conn)
.await?;
// Remove any user logins from the database (incl. bearer tokens)
let delete_result = login::Entity::delete_many()
.filter(login::Column::UserId.eq(user.id))

82
web3_proxy/src/frontend/admin.rs
View File

@ -1,16 +1,13 @@
//! Handle admin helper logic
use super::authorization::{login_is_authorized, RpcSecretKey};
use super::authorization::login_is_authorized;
use super::errors::FrontendResult;
use crate::admin_queries::query_admin_modify_usertier;
use crate::app::Web3ProxyApp;
use crate::user_queries::{get_page_from_params, get_user_id_from_params};
use crate::user_queries::{
get_chain_id_from_params, get_query_start_from_params, query_user_stats, StatResponse,
};
use entities::prelude::{User, SecondaryUser};
use crate::frontend::errors::FrontendErrorResponse;
use crate::user_token::UserBearerToken;
use crate::PostLogin;
use anyhow::Context;
use axum::headers::{Header, Origin, Referer, UserAgent};
use axum::{
extract::{Path, Query},
headers::{authorization::Bearer, Authorization},
@ -20,20 +17,15 @@ use axum::{
use axum_client_ip::InsecureClientIp;
use axum_macros::debug_handler;
use chrono::{TimeZone, Utc};
use entities::sea_orm_active_enums::{LogLevel, Role};
use entities::{admin, admin_trail, login, pending_login, revert_log, rpc_key, secondary_user, user, user_tier};
use entities::{admin_trail, login, pending_login, rpc_key, user};
use ethers::{abi::AbiEncode, prelude::Address, types::Bytes};
use hashbrown::HashMap;
use http::{HeaderValue, StatusCode};
use ipnet::IpNet;
use itertools::Itertools;
use log::{debug, info, warn};
use http::StatusCode;
use log::{debug, warn};
use migration::sea_orm::prelude::Uuid;
use migration::sea_orm::{
self, ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,
QueryOrder, TransactionTrait, TryIntoModel,
self, ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter,
};
use serde::Deserialize;
use serde_json::json;
use siwe::{Message, VerificationOpts};
use std::ops::Add;
@ -41,9 +33,6 @@ use std::str::FromStr;
use std::sync::Arc;
use time::{Duration, OffsetDateTime};
use ulid::Ulid;
use crate::admin_queries::query_admin_modify_usertier;
use crate::frontend::errors::FrontendErrorResponse;
use crate::{PostLogin, PostLoginQuery};
/// `GET /admin/modify_role` -- As an admin, modify a user's user-tier
///
@ -95,18 +84,34 @@ pub async fn admin_login_get(
// get the admin field ...
let admin_address: Address = params
.get("admin_address")
.ok_or_else(|| FrontendErrorResponse::BadRequest("Unable to find admin_address key in request".to_string()))?
.ok_or_else(|| {
FrontendErrorResponse::BadRequest(
"Unable to find admin_address key in request".to_string(),
)
})?
.parse::<Address>()
.map_err(|err| { FrontendErrorResponse::BadRequest("Unable to parse user_address as an Address".to_string()) })?;
.map_err(|_err| {
FrontendErrorResponse::BadRequest(
"Unable to parse admin_address as an Address".to_string(),
)
})?;
// Fetch the user_address parameter from the login string ... (as who we want to be logging in ...)
let user_address: Vec<u8> = params
.get("user_address")
.ok_or_else(|| FrontendErrorResponse::BadRequest("Unable to find user_address key in request".to_string()))?
.ok_or_else(|| {
FrontendErrorResponse::BadRequest(
"Unable to find user_address key in request".to_string(),
)
})?
.parse::<Address>()
.map_err(|err| { FrontendErrorResponse::BadRequest("Unable to parse user_address as an Address".to_string(), ) })?
.to_fixed_bytes().into();
.map_err(|_err| {
FrontendErrorResponse::BadRequest(
"Unable to parse user_address as an Address".to_string(),
)
})?
.to_fixed_bytes()
.into();
// We want to login to llamanodes.com
let login_domain = app
@ -141,7 +146,9 @@ pub async fn admin_login_get(
};
let db_conn = app.db_conn().context("login requires a database")?;
let db_replica = app.db_replica().context("login requires a replica database")?;
let db_replica = app
.db_replica()
.context("login requires a replica database")?;
// Get the user that we want to imitate from the read-only database (their id ...)
// TODO: Only get the id, not the whole user object ...
@ -149,13 +156,17 @@ pub async fn admin_login_get(
.filter(user::Column::Address.eq(user_address))
.one(db_replica.conn())
.await?
.ok_or(FrontendErrorResponse::BadRequest("Could not find user in db".to_string()))?;
.ok_or(FrontendErrorResponse::BadRequest(
"Could not find user in db".to_string(),
))?;
let admin = user::Entity::find()
.filter(user::Column::Address.eq(admin_address.encode()))
.one(db_replica.conn())
.await?
.ok_or(FrontendErrorResponse::BadRequest("Could not find admin in db".to_string()))?;
.ok_or(FrontendErrorResponse::BadRequest(
"Could not find admin in db".to_string(),
))?;
// Note that the admin is trying to log in as this user
let trail = admin_trail::ActiveModel {
@ -187,7 +198,7 @@ pub async fn admin_login_get(
nonce: sea_orm::Set(uuid),
message: sea_orm::Set(message.to_string()),
expires_at: sea_orm::Set(expires_at),
imitating_user: sea_orm::Set(Some(user.id))
imitating_user: sea_orm::Set(Some(user.id)),
};
user_pending_login
@ -221,7 +232,6 @@ pub async fn admin_login_get(
pub async fn admin_login_post(
Extension(app): Extension<Arc<Web3ProxyApp>>,
InsecureClientIp(ip): InsecureClientIp,
Query(query): Query<PostLoginQuery>,
Json(payload): Json<PostLogin>,
) -> FrontendResult {
login_is_authorized(&app, ip).await?;
@ -293,7 +303,6 @@ pub async fn admin_login_post(
.verify_eip191(&their_sig)
.context("verifying eip191 signature against our local message")
{
// delete ALL expired rows.
let now = Utc::now();
let delete_result = pending_login::Entity::delete_many()
@ -309,12 +318,12 @@ pub async fn admin_login_post(
err_1,
err_191
)
.into());
.into());
}
}
// TODO: Maybe add a context?
let imitating_user_id = user_pending_login.imitating_user
let imitating_user_id = user_pending_login
.imitating_user
.context("getting address of the imitating user")?;
// TODO: limit columns or load whole user?
@ -386,9 +395,9 @@ pub async fn admin_login_post(
let user_login = login::ActiveModel {
id: sea_orm::NotSet,
bearer_token: sea_orm::Set(user_bearer_token.uuid()),
user_id: sea_orm::Set(imitating_user.id), // Yes, this should be the user ... because the rest of the applications takes this item, from the initial user
user_id: sea_orm::Set(imitating_user.id), // Yes, this should be the user ... because the rest of the applications takes this item, from the initial user
expires_at: sea_orm::Set(expires_at),
read_only: sea_orm::Set(true)
read_only: sea_orm::Set(true),
};
user_login
@ -405,7 +414,6 @@ pub async fn admin_login_post(
}
Ok(response)
}
// TODO: This is basically an exact copy of the user endpoint, I should probabl refactor this code ...

2
web3_proxy/src/frontend/authorization.rs
View File

@ -9,7 +9,7 @@ use axum::headers::authorization::Bearer;
use axum::headers::{Header, Origin, Referer, UserAgent};
use chrono::Utc;
use deferred_rate_limiter::DeferredRateLimitResult;
use entities::{admin, login, rpc_key, user, user_tier};
use entities::{login, rpc_key, user, user_tier};
use ethers::types::Bytes;
use ethers::utils::keccak256;
use futures::TryFutureExt;

72
web3_proxy/src/frontend/rpc_proxy_http.rs
View File

@ -74,13 +74,40 @@ async fn _proxy_web3_rpc(
// TODO: this might be slow. think about this more
// TODO: special string if no rpcs were used (cache hit)?
let rpcs: String = rpcs.into_iter().map(|x| x.name.clone()).join(",");
let mut backup_used = false;
let rpcs: String = rpcs
.into_iter()
.map(|x| {
if x.backup {
backup_used = true;
}
x.name.clone()
})
.join(",");
headers.insert(
"W3P-BACKEND-RPCS",
"X-W3P-BACKEND-RPCS",
rpcs.parse().expect("W3P-BACKEND-RPCS should always parse"),
);
headers.insert(
"X-W3P-BACKUP-RPC",
backup_used
.to_string()
.parse()
.expect("W3P-BACKEND-RPCS should always parse"),
);
// TODO: add a header if a backend rpc was used
headers.insert(
"X-W3P-CLIENT-IP",
ip.to_string()
.parse()
.expect("X-CLIENT-IP should always parse"),
);
Ok(response)
}
@ -184,6 +211,8 @@ async fn _proxy_web3_rpc_with_key(
let authorization = Arc::new(authorization);
let rpc_secret_key_id = authorization.checks.rpc_secret_key_id;
let (response, rpcs, _semaphore) = app
.proxy_web3_rpc(authorization, payload, proxy_mode)
.await
@ -193,13 +222,48 @@ async fn _proxy_web3_rpc_with_key(
let headers = response.headers_mut();
let mut backup_used = false;
// TODO: special string if no rpcs were used (cache hit)? or is an empty string fine? maybe the rpc name + "cached"
let rpcs: String = rpcs.into_iter().map(|x| x.name.clone()).join(",");
let rpcs: String = rpcs
.into_iter()
.map(|x| {
if x.backup {
backup_used = true;
}
x.name.clone()
})
.join(",");
headers.insert(
"W3P-BACKEND-RPCs",
"X-W3P-BACKEND-RPCs",
rpcs.parse().expect("W3P-BACKEND-RPCS should always parse"),
);
headers.insert(
"X-W3P-BACKUP-RPC",
backup_used
.to_string()
.parse()
.expect("W3P-BACKEND-RPCS should always parse"),
);
headers.insert(
"X-W3P-CLIENT-IP",
ip.to_string()
.parse()
.expect("X-CLIENT-IP should always parse"),
);
if let Some(rpc_secret_key_id) = rpc_secret_key_id {
headers.insert(
"X-W3P-KEY-ID",
rpc_secret_key_id
.to_string()
.parse()
.expect("X-CLIENT-IP should always parse"),
);
}
Ok(response)
}

21
web3_proxy/src/frontend/users.rs
View File

@ -1,14 +1,13 @@
//! Handle registration, logins, and managing account data.
use super::authorization::{login_is_authorized, RpcSecretKey};
use super::errors::FrontendResult;
use crate::app::Web3ProxyApp;
use crate::user_queries::{get_page_from_params, get_user_id_from_params};
use crate::user_queries::get_page_from_params;
use crate::user_queries::{
get_chain_id_from_params, get_query_start_from_params, query_user_stats, StatResponse,
};
use entities::prelude::{User, SecondaryUser};
use crate::user_token::UserBearerToken;
use crate::{PostLogin, PostLoginQuery};
use anyhow::Context;
use axum::headers::{Header, Origin, Referer, UserAgent};
use axum::{
@ -20,14 +19,14 @@ use axum::{
use axum_client_ip::InsecureClientIp;
use axum_macros::debug_handler;
use chrono::{TimeZone, Utc};
use entities::sea_orm_active_enums::{LogLevel, Role};
use entities::{login, pending_login, revert_log, rpc_key, secondary_user, user, user_tier};
use entities::sea_orm_active_enums::LogLevel;
use entities::{login, pending_login, revert_log, rpc_key, user};
use ethers::{prelude::Address, types::Bytes};
use hashbrown::HashMap;
use http::{HeaderValue, StatusCode};
use ipnet::IpNet;
use itertools::Itertools;
use log::{debug, info, warn};
use log::{debug, warn};
use migration::sea_orm::prelude::Uuid;
use migration::sea_orm::{
self, ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,
@ -41,12 +40,6 @@ use std::str::FromStr;
use std::sync::Arc;
use time::{Duration, OffsetDateTime};
use ulid::Ulid;
use entities::user::Relation::UserTier;
use migration::extension::postgres::Type;
use thread_fast_rng::rand;
use crate::admin_queries::query_admin_modify_usertier;
use crate::frontend::errors::FrontendErrorResponse;
use crate::{PostLogin, PostLoginQuery};
/// `GET /user/login/:user_address` or `GET /user/login/:user_address/:message_eip` -- Start the "Sign In with Ethereum" (siwe) login flow.
///
@ -135,7 +128,7 @@ pub async fn user_login_get(
nonce: sea_orm::Set(uuid),
message: sea_orm::Set(message.to_string()),
expires_at: sea_orm::Set(expires_at),
imitating_user: sea_orm::Set(None)
imitating_user: sea_orm::Set(None),
};
user_pending_login
@ -355,7 +348,7 @@ pub async fn user_login_post(
bearer_token: sea_orm::Set(user_bearer_token.uuid()),
user_id: sea_orm::Set(u.id),
expires_at: sea_orm::Set(expires_at),
read_only: sea_orm::Set(false)
read_only: sea_orm::Set(false),
};
user_login

2
web3_proxy/src/rpcs/one.rs
View File

@ -117,7 +117,7 @@ pub struct Web3Rpc {
/// use web3 queries to find the block data limit for archive/pruned nodes
pub(super) automatic_block_limit: bool,
/// only use this rpc if everything else is lagging too far. this allows us to ignore fast but very low limit rpcs
pub(super) backup: bool,
pub backup: bool,
/// TODO: have an enum for this so that "no limit" prints pretty?
pub(super) block_data_limit: AtomicU64,
/// Lower tiers are higher priority when sending requests

1
web3_proxy/src/user_queries.rs
View File

@ -9,7 +9,6 @@ use axum::{
TypedHeader,
};
use chrono::{NaiveDateTime, Utc};
use ethers::prelude::Address;
use entities::{login, rpc_accounting, rpc_key};
use hashbrown::HashMap;
use http::StatusCode;