web3-proxy/web3_proxy/src/frontend/authorization.rs

1324 lines
47 KiB
Rust
Raw Normal View History

2022-10-18 00:47:58 +03:00
//! Utilities for authorization of logged in and anonymous users.
use super::errors::{Web3ProxyError, Web3ProxyErrorContext, Web3ProxyResult};
2023-03-03 04:39:50 +03:00
use super::rpc_proxy_ws::ProxyMode;
use crate::app::{AuthorizationChecks, Web3ProxyApp, APP_USER_AGENT};
use crate::jsonrpc::{JsonRpcForwardedResponse, JsonRpcRequest};
use crate::rpcs::one::Web3Rpc;
use crate::stats::{AppStat, BackendRequests, RpcQueryStats};
2022-10-31 23:05:58 +03:00
use crate::user_token::UserBearerToken;
2022-10-26 00:10:05 +03:00
use axum::headers::authorization::Bearer;
2022-10-27 00:39:26 +03:00
use axum::headers::{Header, Origin, Referer, UserAgent};
2022-10-11 08:13:00 +03:00
use chrono::Utc;
use core::fmt;
2022-09-23 00:03:37 +03:00
use deferred_rate_limiter::DeferredRateLimitResult;
use derive_more::From;
use entities::sea_orm_active_enums::TrackingLevel;
User Balance + Referral Logic (#44) * will implement balance topup endpoint * will quickly fix other PR reviews * merging from master * will finish up godmoe * will finish up login * added logic to top up balance (first iteration) * should implement additional columns soon (currency, amount, tx-hash), as well as a new table for spend * updated migrations, will account for spend next * get back to this later * will merge PR from stats-v2 * stats v2 rebased all my commits and squashed them down to one * cargo upgrade * added migrtation for spend in accounting table. will run test-deposit next * trying to get request from polygon * first iteration /user/balance/:tx_hash works, needs to add accepted tokens next * creating the referral code seems to work * will now check if spending enough credits will lead to both parties receiving credits * rpcstats takes care of accounting for spend data * removed track spend from table * Revert "removed track spend from table" This reverts commit a50802d6ae75f786864c5ec42d0ceb2cb27124ed. * Revert "rpcstats takes care of accounting for spend data" This reverts commit 1cec728bf241e4cfd24351134637ed81c1a5a10b. * removed rpc request table entity * updated referral code to use ulid s * credits used are aggregated * added a bunch of fields to referrer * added database logic whenever an aggregate stats is added. will have to iterate over this a couple times i think. go to (1) detecting accepted stables next, (2) fix influxdb bug and (3) start to write test * removed track spend as this will occur in the database * will first work on "balance", then referral. these should really be treated as two separate PRs (although already convoluted) * balance logic initial commit * breaking WIP, changing the RPC call logic functions * will start testing next * got rid of warnings & lint * will proceed with subtracting / adding to balance * added decimal points, balance tracking seems to work * will beautify code a bit * removed deprecated dependency, and added topic + deposit contract to app.yaml * brownie test suite does not rely on local contract files it pulls all from polygonscan * will continue with referral * should perhaps (in a future revision) recordhow much the referees got for free. marking referrals seems to work rn * user is upgraded to premium if they deposit more than 10$. we dont accept more than $10M in a single tx * will start PR, referral seems to be fine so far, perhaps up to some numbers that still may need tweaking * will start PR * removed rogue comments, cleaned up payments a bit * changes before PR * apply stats * added unique constraint * some refactoring such that the user file is not too bloated * compiling * progress with subusers, creating a table entry seems to work * good response type is there as well now, will work on getters from primary user and secondary user next * subuser logic also seems fine now * downgrade logic * fixed bug influxdb does not support different types in same query (which makes sense) * WIP temporary commit * merging with PR * Delete daemon.rs there are multiple daemons now, so this was moved to `proxyd` * will remove request clone to &mut * multiple request handles for payment * making requests still seem fine * removed redundant commented out bits * added deposit endpoint, added deposit amount and deposit user, untested yet * small bug with downgrade tier id * will add authorization so balance can be received for users * balance history should be set now too * will check balance over time again * subususer can see rpc key balance if admin or owner * stats also seems to work fine now with historical balance * things seem to be building and working * removed clone from OpenRequestHandle * removed influxdb from workspace members * changed config files * reran sea-orm generate entities, added a foreign key, should be proper now * removed contract from commit * made deposit contract optional * added topic in polygon dev * changed deposit contract to deposit factory contract * added selfrelation on user_tier * added payment required * changed chain id to u64 * add wss in polygon llamarpc * removed origin and method from the table * added onchain transactions naming (and forgot to add a migration before) * changed foreign key to be the referrer (id), not the code itself * forgot to add id as the target foreign key * WIP adding cache to update role * fixed merge conflicts --------- Co-authored-by: Bryan Stitt <bryan@llamanodes.com> Co-authored-by: Bryan Stitt <bryan@stitthappens.com>
2023-05-12 19:45:15 +03:00
use entities::{balance, login, rpc_key, user, user_tier};
use ethers::types::{Bytes, U64};
2022-12-28 09:11:18 +03:00
use ethers::utils::keccak256;
use futures::TryFutureExt;
use hashbrown::HashMap;
2022-10-27 00:39:26 +03:00
use http::HeaderValue;
use ipnet::IpNet;
use log::{error, trace, warn};
2022-11-14 21:24:52 +03:00
use migration::sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter};
use rdkafka::message::{Header as KafkaHeader, OwnedHeaders as KafkaOwnedHeaders, OwnedMessage};
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::util::Timeout as KafkaTimeout;
2022-12-28 09:11:18 +03:00
use redis_rate_limiter::redis::AsyncCommands;
2022-09-24 06:59:21 +03:00
use redis_rate_limiter::RedisRateLimitResult;
2022-09-24 08:53:45 +03:00
use std::fmt::Display;
use std::mem;
use std::sync::atomic::{self, AtomicBool, AtomicI64, AtomicU64, AtomicUsize};
use std::time::Duration;
2022-09-24 08:53:45 +03:00
use std::{net::IpAddr, str::FromStr, sync::Arc};
2023-05-13 02:02:43 +03:00
use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore};
use tokio::task::JoinHandle;
2022-09-23 00:03:37 +03:00
use tokio::time::Instant;
2022-09-24 08:53:45 +03:00
use ulid::Ulid;
2022-09-23 00:03:37 +03:00
use uuid::Uuid;
2022-09-24 08:53:45 +03:00
/// This lets us use UUID and ULID while we transition to only ULIDs
2022-10-18 00:47:58 +03:00
/// TODO: include the key's description.
2022-10-26 03:22:58 +03:00
#[derive(Copy, Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
2022-11-01 21:54:39 +03:00
pub enum RpcSecretKey {
2022-09-24 08:53:45 +03:00
Ulid(Ulid),
Uuid(Uuid),
}
/// TODO: should this have IpAddr and Origin or AuthorizationChecks?
2022-10-10 07:15:07 +03:00
#[derive(Debug)]
pub enum RateLimitResult {
Allowed(Authorization, Option<OwnedSemaphorePermit>),
RateLimited(
Authorization,
/// when their rate limit resets and they can try more requests
Option<Instant>,
),
2022-10-10 07:15:07 +03:00
/// This key is not in our database. Deny access!
UnknownKey,
}
#[derive(Clone, Debug)]
2022-12-12 07:39:54 +03:00
pub enum AuthorizationType {
Internal,
Frontend,
}
/// TODO: include the authorization checks in this?
2022-10-27 00:39:26 +03:00
#[derive(Clone, Debug)]
pub struct Authorization {
pub checks: AuthorizationChecks,
// TODO: instead of the conn, have a channel?
pub db_conn: Option<DatabaseConnection>,
2022-10-10 07:15:07 +03:00
pub ip: IpAddr,
2022-10-27 00:39:26 +03:00
pub origin: Option<Origin>,
pub referer: Option<Referer>,
pub user_agent: Option<UserAgent>,
2022-12-12 07:39:54 +03:00
pub authorization_type: AuthorizationType,
2022-10-10 07:15:07 +03:00
}
pub struct KafkaDebugLogger {
topic: String,
key: Vec<u8>,
headers: KafkaOwnedHeaders,
producer: FutureProducer,
num_requests: AtomicUsize,
num_responses: AtomicUsize,
}
impl fmt::Debug for KafkaDebugLogger {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("KafkaDebugLogger")
.field("topic", &self.topic)
.finish_non_exhaustive()
}
}
type KafkaLogResult = Result<(i32, i64), (rdkafka::error::KafkaError, OwnedMessage)>;
impl KafkaDebugLogger {
fn try_new(
app: &Web3ProxyApp,
authorization: Arc<Authorization>,
head_block_num: Option<&U64>,
kafka_topic: &str,
request_ulid: Ulid,
) -> Option<Arc<Self>> {
let kafka_producer = app.kafka_producer.clone()?;
let kafka_topic = kafka_topic.to_string();
let rpc_secret_key_id = authorization
.checks
.rpc_secret_key_id
.map(|x| x.get())
.unwrap_or_default();
let kafka_key =
rmp_serde::to_vec(&rpc_secret_key_id).expect("ids should always serialize with rmp");
let chain_id = app.config.chain_id;
let head_block_num = head_block_num
.copied()
.or_else(|| app.balanced_rpcs.head_block_num());
// TODO: would be nice to have the block hash too
// another item is added with the response, so initial_capacity is +1 what is needed here
let kafka_headers = KafkaOwnedHeaders::new_with_capacity(6)
.insert(KafkaHeader {
key: "rpc_secret_key_id",
value: authorization
.checks
.rpc_secret_key_id
.map(|x| x.to_string())
.as_ref(),
})
.insert(KafkaHeader {
key: "ip",
value: Some(&authorization.ip.to_string()),
})
.insert(KafkaHeader {
key: "request_ulid",
value: Some(&request_ulid.to_string()),
})
.insert(KafkaHeader {
key: "head_block_num",
value: head_block_num.map(|x| x.to_string()).as_ref(),
})
.insert(KafkaHeader {
key: "chain_id",
value: Some(&chain_id.to_le_bytes()),
});
// save the key and headers for when we log the response
let x = Self {
topic: kafka_topic,
key: kafka_key,
headers: kafka_headers,
producer: kafka_producer,
num_requests: 0.into(),
num_responses: 0.into(),
};
let x = Arc::new(x);
Some(x)
}
fn background_log(&self, payload: Vec<u8>) -> JoinHandle<KafkaLogResult> {
let topic = self.topic.clone();
let key = self.key.clone();
let producer = self.producer.clone();
let headers = self.headers.clone();
let f = async move {
let record = FutureRecord::to(&topic)
.key(&key)
.payload(&payload)
.headers(headers);
let produce_future =
producer.send(record, KafkaTimeout::After(Duration::from_secs(5 * 60)));
let kafka_response = produce_future.await;
if let Err((err, msg)) = kafka_response.as_ref() {
error!("produce kafka request: {} - {:?}", err, msg);
// TODO: re-queue the msg? log somewhere else like a file on disk?
// TODO: this is bad and should probably trigger an alarm
};
kafka_response
};
tokio::spawn(f)
}
/// for opt-in debug usage, log the request to kafka
/// TODO: generic type for request
pub fn log_debug_request(&self, request: &JsonRpcRequest) -> JoinHandle<KafkaLogResult> {
// TODO: is rust message pack a good choice? try rkyv instead
let payload =
rmp_serde::to_vec(&request).expect("requests should always serialize with rmp");
self.num_requests.fetch_add(1, atomic::Ordering::SeqCst);
self.background_log(payload)
}
pub fn log_debug_response<R>(&self, response: &R) -> JoinHandle<KafkaLogResult>
where
R: serde::Serialize,
{
let payload =
rmp_serde::to_vec(&response).expect("requests should always serialize with rmp");
self.num_responses.fetch_add(1, atomic::Ordering::SeqCst);
self.background_log(payload)
}
}
#[derive(Debug)]
2022-10-10 07:15:07 +03:00
pub struct RequestMetadata {
/// TODO: set archive_request during the new instead of after
/// TODO: this is more complex than "requires a block older than X height". different types of data can be pruned differently
2022-11-03 02:14:16 +03:00
pub archive_request: AtomicBool,
pub authorization: Option<Arc<Authorization>>,
pub request_ulid: Ulid,
/// Size of the JSON request. Does not include headers or things like that.
pub request_bytes: usize,
/// users can opt out of method tracking for their personal dashboads
/// but we still have to store the method at least temporarily for cost calculations
pub method: Option<String>,
/// Instant that the request was received (or at least close to it)
/// We use Instant and not timestamps to avoid problems with leap seconds and similar issues
pub start_instant: tokio::time::Instant,
2022-12-20 02:59:01 +03:00
/// if this is empty, there was a cache_hit
/// otherwise, it is populated with any rpc servers that were used by this request
pub backend_requests: BackendRequests,
/// The number of times the request got stuck waiting because no servers were synced
2022-10-25 06:41:59 +03:00
pub no_servers: AtomicU64,
/// If handling the request hit an application error
/// This does not count things like a transcation reverting or a malformed request
pub error_response: AtomicBool,
/// Size in bytes of the JSON response. Does not include headers or things like that.
pub response_bytes: AtomicU64,
/// How many milliseconds it took to respond to the request
pub response_millis: AtomicU64,
/// What time the (first) response was proxied.
/// TODO: think about how to store response times for ProxyMode::Versus
pub response_timestamp: AtomicI64,
/// True if the response required querying a backup RPC
/// RPC aggregators that query multiple providers to compare response may use this header to ignore our response.
pub response_from_backup_rpc: AtomicBool,
2022-10-10 07:15:07 +03:00
/// ProxyMode::Debug logs requests and responses with Kafka
/// TODO: maybe this shouldn't be determined by ProxyMode. A request param should probably enable this
pub kafka_debug_logger: Option<Arc<KafkaDebugLogger>>,
2023-05-13 02:02:43 +03:00
/// Cancel-safe channel to send stats to
pub stat_sender: Option<mpsc::UnboundedSender<AppStat>>,
}
2022-10-10 07:15:07 +03:00
impl Default for RequestMetadata {
fn default() -> Self {
Self {
archive_request: Default::default(),
authorization: Default::default(),
backend_requests: Default::default(),
error_response: Default::default(),
kafka_debug_logger: Default::default(),
method: Default::default(),
no_servers: Default::default(),
request_bytes: Default::default(),
request_ulid: Default::default(),
response_bytes: Default::default(),
response_from_backup_rpc: Default::default(),
response_millis: Default::default(),
response_timestamp: Default::default(),
start_instant: Instant::now(),
stat_sender: Default::default(),
}
}
}
#[derive(From)]
pub enum RequestOrMethod<'a> {
Request(&'a JsonRpcRequest),
/// jsonrpc method (or similar label) and the size that the request should count as (sometimes 0)
Method(&'a str, usize),
RequestSize(usize),
}
impl<'a> RequestOrMethod<'a> {
fn method(&self) -> Option<&str> {
match self {
Self::Request(x) => Some(&x.method),
Self::Method(x, _) => Some(x),
_ => None,
}
}
fn jsonrpc_request(&self) -> Option<&JsonRpcRequest> {
match self {
Self::Request(x) => Some(x),
_ => None,
}
}
fn num_bytes(&self) -> usize {
match self {
RequestOrMethod::Method(_, num_bytes) => *num_bytes,
RequestOrMethod::Request(x) => x.num_bytes(),
RequestOrMethod::RequestSize(num_bytes) => *num_bytes,
}
}
}
impl<'a> From<&'a str> for RequestOrMethod<'a> {
fn from(value: &'a str) -> Self {
if value.is_empty() {
Self::RequestSize(0)
} else {
Self::Method(value, 0)
}
}
}
// TODO: i think a trait is actually the right thing to use here
#[derive(From)]
pub enum ResponseOrBytes<'a> {
Json(&'a serde_json::Value),
Response(&'a JsonRpcForwardedResponse),
Bytes(usize),
}
impl<'a> From<u64> for ResponseOrBytes<'a> {
fn from(value: u64) -> Self {
Self::Bytes(value as usize)
}
}
impl ResponseOrBytes<'_> {
pub fn num_bytes(&self) -> usize {
match self {
Self::Json(x) => serde_json::to_string(x)
.expect("this should always serialize")
.len(),
Self::Response(x) => x.num_bytes(),
Self::Bytes(num_bytes) => *num_bytes,
}
}
}
impl RequestMetadata {
pub async fn new<'a, R: Into<RequestOrMethod<'a>>>(
app: &Web3ProxyApp,
authorization: Arc<Authorization>,
request: R,
head_block_num: Option<&U64>,
) -> Arc<Self> {
let request = request.into();
let method = request.method().map(|x| x.to_string());
let request_bytes = request.num_bytes();
// TODO: modify the request here? I don't really like that very much. but its a sure way to get archive_request set correctly
// TODO: add the Ulid at the haproxy or amazon load balancer level? investigate OpenTelemetry
let request_ulid = Ulid::new();
let kafka_debug_logger = if matches!(authorization.checks.proxy_mode, ProxyMode::Debug) {
KafkaDebugLogger::try_new(
app,
authorization.clone(),
head_block_num,
"web3_proxy:rpc",
request_ulid,
)
} else {
None
};
if let Some(ref kafka_debug_logger) = kafka_debug_logger {
if let Some(request) = request.jsonrpc_request() {
// TODO: channels might be more ergonomic than spawned futures
// spawned things run in parallel easier but generally need more Arcs
kafka_debug_logger.log_debug_request(request);
} else {
// there probably isn't a new request attached to this metadata.
// this happens with websocket subscriptions
}
}
let x = Self {
2022-11-03 02:14:16 +03:00
archive_request: false.into(),
2022-12-20 02:59:01 +03:00
backend_requests: Default::default(),
error_response: false.into(),
kafka_debug_logger,
no_servers: 0.into(),
authorization: Some(authorization),
request_bytes,
method,
response_bytes: 0.into(),
response_from_backup_rpc: false.into(),
response_millis: 0.into(),
request_ulid,
response_timestamp: 0.into(),
start_instant: Instant::now(),
stat_sender: app.stat_sender.clone(),
};
Arc::new(x)
}
pub fn backend_rpcs_used(&self) -> Vec<Arc<Web3Rpc>> {
self.backend_requests.lock().clone()
}
pub fn tracking_level(&self) -> TrackingLevel {
if let Some(authorization) = self.authorization.as_ref() {
authorization.checks.tracking_level.clone()
} else {
TrackingLevel::None
}
}
pub fn opt_in_method(&self) -> Option<String> {
match self.tracking_level() {
TrackingLevel::None | TrackingLevel::Aggregated => None,
TrackingLevel::Detailed => self.method.clone(),
}
}
pub fn take_opt_in_method(&mut self) -> Option<String> {
match self.tracking_level() {
TrackingLevel::None | TrackingLevel::Aggregated => None,
TrackingLevel::Detailed => self.method.take(),
}
}
pub fn try_send_stat(mut self) -> Web3ProxyResult<Option<Self>> {
if let Some(stat_sender) = self.stat_sender.take() {
trace!("sending stat! {:?}", self);
let stat: RpcQueryStats = self.try_into()?;
let stat: AppStat = stat.into();
if let Err(err) = stat_sender.send(stat) {
2023-05-13 02:02:43 +03:00
error!("failed sending stat {:?}: {:?}", err.0, err);
// TODO: return it? that seems like it might cause an infinite loop
2023-05-13 02:02:43 +03:00
// TODO: but dropping stats is bad... hmm... i guess better to undercharge customers than overcharge
};
Ok(None)
} else {
Ok(Some(self))
}
}
pub fn add_response<'a, R: Into<ResponseOrBytes<'a>>>(&'a self, response: R) {
// TODO: fetch? set? should it be None in a Mutex? or a OnceCell?
let response = response.into();
let num_bytes = response.num_bytes() as u64;
self.response_bytes
.fetch_add(num_bytes, atomic::Ordering::AcqRel);
self.response_millis.fetch_add(
self.start_instant.elapsed().as_millis() as u64,
atomic::Ordering::AcqRel,
);
// TODO: record first or last timestamp? really, we need multiple
self.response_timestamp
.store(Utc::now().timestamp(), atomic::Ordering::Release);
if let Some(kafka_debug_logger) = self.kafka_debug_logger.as_ref() {
if let ResponseOrBytes::Response(response) = response {
kafka_debug_logger.log_debug_response(response);
}
}
}
pub fn try_send_arc_stat(self: Arc<Self>) -> anyhow::Result<Option<Arc<Self>>> {
match Arc::try_unwrap(self) {
Ok(x) => {
let not_sent = x.try_send_stat()?.map(Arc::new);
Ok(not_sent)
}
Err(not_sent) => {
trace!(
"could not send stat while {} arcs are active",
Arc::strong_count(&not_sent)
);
Ok(Some(not_sent))
}
}
}
// TODO: helper function to duplicate? needs to clear request_bytes, and all the atomics tho...
}
// TODO: is this where the panic comes from?
impl Drop for RequestMetadata {
fn drop(&mut self) {
if self.stat_sender.is_some() {
// turn `&mut self` into `self`
let x = mem::take(self);
// warn!("request metadata dropped without stat send! {:?}", self);
let _ = x.try_send_stat();
}
2022-10-10 07:15:07 +03:00
}
}
2022-11-01 21:54:39 +03:00
impl RpcSecretKey {
2022-09-24 08:53:45 +03:00
pub fn new() -> Self {
Ulid::new().into()
}
}
2022-11-01 21:54:39 +03:00
impl Default for RpcSecretKey {
2022-10-26 00:10:05 +03:00
fn default() -> Self {
Self::new()
}
}
2022-11-01 21:54:39 +03:00
impl Display for RpcSecretKey {
2022-09-24 08:53:45 +03:00
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// TODO: do this without dereferencing
let ulid: Ulid = (*self).into();
ulid.fmt(f)
}
}
2022-11-01 21:54:39 +03:00
impl FromStr for RpcSecretKey {
type Err = Web3ProxyError;
2022-09-24 08:53:45 +03:00
fn from_str(s: &str) -> Result<Self, Self::Err> {
if let Ok(ulid) = s.parse::<Ulid>() {
Ok(ulid.into())
} else if let Ok(uuid) = s.parse::<Uuid>() {
Ok(uuid.into())
} else {
2022-10-10 07:15:07 +03:00
// TODO: custom error type so that this shows as a 400
Err(Web3ProxyError::InvalidUserKey)
2022-09-24 08:53:45 +03:00
}
}
}
2022-11-01 21:54:39 +03:00
impl From<Ulid> for RpcSecretKey {
2022-09-24 08:53:45 +03:00
fn from(x: Ulid) -> Self {
2022-11-01 21:54:39 +03:00
RpcSecretKey::Ulid(x)
2022-09-24 08:53:45 +03:00
}
}
2022-11-01 21:54:39 +03:00
impl From<Uuid> for RpcSecretKey {
2022-09-24 08:53:45 +03:00
fn from(x: Uuid) -> Self {
2022-11-01 21:54:39 +03:00
RpcSecretKey::Uuid(x)
2022-09-24 08:53:45 +03:00
}
}
2022-11-01 21:54:39 +03:00
impl From<RpcSecretKey> for Ulid {
fn from(x: RpcSecretKey) -> Self {
2022-09-24 08:53:45 +03:00
match x {
2022-11-01 21:54:39 +03:00
RpcSecretKey::Ulid(x) => x,
RpcSecretKey::Uuid(x) => Ulid::from(x.as_u128()),
2022-09-24 08:53:45 +03:00
}
}
}
2022-11-01 21:54:39 +03:00
impl From<RpcSecretKey> for Uuid {
fn from(x: RpcSecretKey) -> Self {
2022-09-24 08:53:45 +03:00
match x {
2022-11-01 21:54:39 +03:00
RpcSecretKey::Ulid(x) => Uuid::from_u128(x.0),
RpcSecretKey::Uuid(x) => x,
2022-09-24 08:53:45 +03:00
}
}
}
impl Authorization {
pub fn internal(db_conn: Option<DatabaseConnection>) -> Web3ProxyResult<Self> {
let authorization_checks = AuthorizationChecks {
// any error logs on a local (internal) query are likely problems. log them all
log_revert_chance: 1.0,
tracking_level: TrackingLevel::Detailed,
// default for everything else should be fine. we don't have a user_id or ip to give
..Default::default()
};
let ip: IpAddr = "127.0.0.1".parse().expect("localhost should always parse");
let user_agent = UserAgent::from_str(APP_USER_AGENT).ok();
Self::try_new(
authorization_checks,
db_conn,
ip,
None,
None,
user_agent,
2022-12-12 07:39:54 +03:00
AuthorizationType::Internal,
)
}
pub fn external(
allowed_origin_requests_per_period: &HashMap<String, u64>,
db_conn: Option<DatabaseConnection>,
ip: IpAddr,
origin: Option<Origin>,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
referer: Option<Referer>,
user_agent: Option<UserAgent>,
) -> Web3ProxyResult<Self> {
// some origins can override max_requests_per_period for anon users
let max_requests_per_period = origin
.as_ref()
.map(|origin| {
allowed_origin_requests_per_period
.get(&origin.to_string())
.cloned()
})
.unwrap_or_default();
let authorization_checks = AuthorizationChecks {
max_requests_per_period,
2023-03-03 04:39:50 +03:00
proxy_mode,
tracking_level: TrackingLevel::Detailed,
..Default::default()
};
Self::try_new(
authorization_checks,
db_conn,
ip,
origin,
referer,
user_agent,
2022-12-12 07:39:54 +03:00
AuthorizationType::Frontend,
)
}
2023-03-03 04:39:50 +03:00
#[allow(clippy::too_many_arguments)]
2022-09-23 00:03:37 +03:00
pub fn try_new(
authorization_checks: AuthorizationChecks,
db_conn: Option<DatabaseConnection>,
2022-09-23 00:03:37 +03:00
ip: IpAddr,
origin: Option<Origin>,
2022-09-23 00:03:37 +03:00
referer: Option<Referer>,
user_agent: Option<UserAgent>,
2022-12-12 07:39:54 +03:00
authorization_type: AuthorizationType,
) -> Web3ProxyResult<Self> {
// check ip
match &authorization_checks.allowed_ips {
None => {}
Some(allowed_ips) => {
if !allowed_ips.iter().any(|x| x.contains(&ip)) {
return Err(Web3ProxyError::IpNotAllowed(ip));
}
}
}
// check origin
match (&origin, &authorization_checks.allowed_origins) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(Web3ProxyError::OriginRequired),
(Some(origin), Some(allowed_origins)) => {
2022-10-27 00:39:26 +03:00
if !allowed_origins.contains(origin) {
return Err(Web3ProxyError::OriginNotAllowed(origin.clone()));
}
}
}
// check referer
match (&referer, &authorization_checks.allowed_referers) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(Web3ProxyError::RefererRequired),
(Some(referer), Some(allowed_referers)) => {
if !allowed_referers.contains(referer) {
return Err(Web3ProxyError::RefererNotAllowed(referer.clone()));
}
}
}
// check user_agent
match (&user_agent, &authorization_checks.allowed_user_agents) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(Web3ProxyError::UserAgentRequired),
(Some(user_agent), Some(allowed_user_agents)) => {
if !allowed_user_agents.contains(user_agent) {
return Err(Web3ProxyError::UserAgentNotAllowed(user_agent.clone()));
}
}
}
2022-09-23 00:03:37 +03:00
Ok(Self {
checks: authorization_checks,
db_conn,
2022-09-23 00:03:37 +03:00
ip,
origin,
referer,
user_agent,
authorization_type,
2022-09-23 00:03:37 +03:00
})
}
}
/// rate limit logins only by ip.
/// we want all origins and referers and user agents to count together
pub async fn login_is_authorized(app: &Web3ProxyApp, ip: IpAddr) -> Web3ProxyResult<Authorization> {
2023-03-03 04:39:50 +03:00
let authorization = match app.rate_limit_login(ip, ProxyMode::Best).await? {
RateLimitResult::Allowed(authorization, None) => authorization,
RateLimitResult::RateLimited(authorization, retry_at) => {
return Err(Web3ProxyError::RateLimited(authorization, retry_at));
2022-09-24 06:59:21 +03:00
}
// TODO: don't panic. give the user an error
x => unimplemented!("rate_limit_login shouldn't ever see these: {:?}", x),
};
Ok(authorization)
2022-09-24 06:59:21 +03:00
}
/// semaphore won't ever be None, but its easier if key auth and ip auth work the same way
2022-09-23 00:03:37 +03:00
pub async fn ip_is_authorized(
2022-12-28 09:11:18 +03:00
app: &Arc<Web3ProxyApp>,
2022-09-23 00:03:37 +03:00
ip: IpAddr,
origin: Option<Origin>,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
) -> Web3ProxyResult<(Authorization, Option<OwnedSemaphorePermit>)> {
2022-09-23 00:03:37 +03:00
// TODO: i think we could write an `impl From` for this
// TODO: move this to an AuthorizedUser extrator
let (authorization, semaphore) = match app
2023-03-03 04:39:50 +03:00
.rate_limit_by_ip(
&app.config.allowed_origin_requests_per_period,
ip,
origin,
proxy_mode,
)
.await?
{
RateLimitResult::Allowed(authorization, semaphore) => (authorization, semaphore),
RateLimitResult::RateLimited(authorization, retry_at) => {
2022-12-28 06:43:02 +03:00
// TODO: in the background, emit a stat (maybe simplest to use a channel?)
return Err(Web3ProxyError::RateLimited(authorization, retry_at));
2022-09-23 00:03:37 +03:00
}
// TODO: don't panic. give the user an error
x => unimplemented!("rate_limit_by_ip shouldn't ever see these: {:?}", x),
};
// in the background, add the ip to a recent_users map
2022-12-28 09:11:18 +03:00
if app.config.public_recent_ips_salt.is_some() {
let app = app.clone();
2022-12-28 09:11:18 +03:00
let f = async move {
let now = Utc::now().timestamp();
2022-12-29 09:21:09 +03:00
if let Some(mut redis_conn) = app.redis_conn().await? {
let salt = app
.config
.public_recent_ips_salt
.as_ref()
.expect("public_recent_ips_salt must exist in here");
2022-12-28 09:11:18 +03:00
2022-12-29 09:21:09 +03:00
let salted_ip = format!("{}:{}", salt, ip);
2022-12-28 09:11:18 +03:00
2022-12-29 09:21:09 +03:00
let hashed_ip = Bytes::from(keccak256(salted_ip.as_bytes()));
2022-12-28 09:11:18 +03:00
2022-12-29 09:21:09 +03:00
let recent_ip_key = format!("recent_users:ip:{}", app.config.chain_id);
2022-12-28 09:11:18 +03:00
2022-12-29 09:21:09 +03:00
redis_conn
.zadd(recent_ip_key, hashed_ip.to_string(), now)
.await?;
};
2022-12-28 09:11:18 +03:00
Ok::<_, Web3ProxyError>(())
2022-12-28 09:11:18 +03:00
}
.map_err(|err| {
warn!("background update of recent_users:ip failed: {}", err);
2022-12-28 09:11:18 +03:00
err
});
tokio::spawn(f);
}
2022-12-28 06:43:02 +03:00
Ok((authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
/// like app.rate_limit_by_rpc_key but converts to a Web3ProxyError;
2022-09-23 00:03:37 +03:00
pub async fn key_is_authorized(
app: &Arc<Web3ProxyApp>,
2022-11-01 21:54:39 +03:00
rpc_key: RpcSecretKey,
2022-09-23 00:03:37 +03:00
ip: IpAddr,
origin: Option<Origin>,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
2022-09-23 00:03:37 +03:00
referer: Option<Referer>,
user_agent: Option<UserAgent>,
) -> Web3ProxyResult<(Authorization, Option<OwnedSemaphorePermit>)> {
2022-09-23 00:03:37 +03:00
// check the rate limits. error if over the limit
// TODO: i think this should be in an "impl From" or "impl Into"
let (authorization, semaphore) = match app
2023-03-03 04:39:50 +03:00
.rate_limit_by_rpc_key(ip, origin, proxy_mode, referer, rpc_key, user_agent)
.await?
{
RateLimitResult::Allowed(authorization, semaphore) => (authorization, semaphore),
RateLimitResult::RateLimited(authorization, retry_at) => {
return Err(Web3ProxyError::RateLimited(authorization, retry_at));
2022-09-23 00:03:37 +03:00
}
RateLimitResult::UnknownKey => return Err(Web3ProxyError::UnknownKey),
2022-09-23 00:03:37 +03:00
};
// TODO: DRY and maybe optimize the hashing
// in the background, add the ip to a recent_users map
if app.config.public_recent_ips_salt.is_some() {
let app = app.clone();
let user_id = authorization.checks.user_id;
let f = async move {
let now = Utc::now().timestamp();
2022-12-29 09:21:09 +03:00
if let Some(mut redis_conn) = app.redis_conn().await? {
let salt = app
.config
.public_recent_ips_salt
.as_ref()
.expect("public_recent_ips_salt must exist in here");
2022-12-29 09:21:09 +03:00
let salted_user_id = format!("{}:{}", salt, user_id);
2022-12-29 09:21:09 +03:00
let hashed_user_id = Bytes::from(keccak256(salted_user_id.as_bytes()));
2022-12-29 10:16:35 +03:00
let recent_user_id_key = format!("recent_users:id:{}", app.config.chain_id);
2022-12-29 09:21:09 +03:00
redis_conn
.zadd(recent_user_id_key, hashed_user_id.to_string(), now)
.await?;
}
Ok::<_, Web3ProxyError>(())
}
.map_err(|err| {
2022-12-29 10:16:35 +03:00
warn!("background update of recent_users:id failed: {}", err);
err
});
tokio::spawn(f);
}
Ok((authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
impl Web3ProxyApp {
2022-10-27 03:12:42 +03:00
/// Limit the number of concurrent requests from the given ip address.
pub async fn ip_semaphore(&self, ip: &IpAddr) -> Web3ProxyResult<Option<OwnedSemaphorePermit>> {
2022-10-25 07:01:41 +03:00
if let Some(max_concurrent_requests) = self.config.public_max_concurrent_requests {
let semaphore = self
.ip_semaphores
.get_with_by_ref(ip, async move {
2022-10-25 07:01:41 +03:00
// TODO: set max_concurrent_requests dynamically based on load?
2022-10-25 07:31:18 +03:00
let s = Semaphore::new(max_concurrent_requests);
2022-10-25 07:01:41 +03:00
Arc::new(s)
})
.await;
// if semaphore.available_permits() == 0 {
// // TODO: concurrent limit hit! emit a stat? less important for anon users
// // TODO: there is probably a race here
// }
let semaphore_permit = semaphore.acquire_owned().await?;
Ok(Some(semaphore_permit))
} else {
Ok(None)
}
2022-09-28 06:35:55 +03:00
}
/// Limit the number of concurrent requests from the given rpc key.
pub async fn registered_user_semaphore(
2022-09-28 06:35:55 +03:00
&self,
authorization_checks: &AuthorizationChecks,
) -> Web3ProxyResult<Option<OwnedSemaphorePermit>> {
if let Some(max_concurrent_requests) = authorization_checks.max_concurrent_requests {
let user_id = authorization_checks
.user_id
.try_into()
.or(Err(Web3ProxyError::UserIdZero))
.web3_context("user ids should always be non-zero")?;
2022-09-28 06:35:55 +03:00
let semaphore = self
.registered_user_semaphores
.get_with(user_id, async move {
2022-10-25 07:31:18 +03:00
let s = Semaphore::new(max_concurrent_requests as usize);
// trace!("new semaphore for user_id {}", user_id);
Arc::new(s)
2022-09-28 06:35:55 +03:00
})
.await;
2022-09-28 06:35:55 +03:00
2022-10-10 07:15:07 +03:00
// if semaphore.available_permits() == 0 {
// // TODO: concurrent limit hit! emit a stat? this has a race condition though.
// // TODO: maybe have a stat on how long we wait to acquire the semaphore instead?
2022-10-10 07:15:07 +03:00
// }
2022-09-28 06:35:55 +03:00
let semaphore_permit = semaphore.acquire_owned().await?;
Ok(Some(semaphore_permit))
} else {
// unlimited requests allowed
2022-09-28 06:35:55 +03:00
Ok(None)
}
}
2022-10-26 00:10:05 +03:00
/// Verify that the given bearer token and address are allowed to take the specified action.
/// This includes concurrent request limiting.
pub async fn bearer_is_authorized(
&self,
bearer: Bearer,
) -> Web3ProxyResult<(user::Model, OwnedSemaphorePermit)> {
2022-12-14 05:13:23 +03:00
// get the user id for this bearer token
let user_bearer_token = UserBearerToken::try_from(bearer)?;
2022-10-26 00:10:05 +03:00
// limit concurrent requests
let semaphore = self
.bearer_token_semaphores
.get_with_by_ref(&user_bearer_token, async move {
2022-10-26 00:10:05 +03:00
let s = Semaphore::new(self.config.bearer_token_max_concurrent_requests as usize);
Arc::new(s)
})
.await;
let semaphore_permit = semaphore.acquire_owned().await?;
2022-12-14 05:13:23 +03:00
// get the attached address from the database for the given auth_token.
let db_replica = self
.db_replica()
.web3_context("checking if bearer token is authorized")?;
2022-10-26 00:10:05 +03:00
2022-12-14 05:13:23 +03:00
let user_bearer_uuid: Uuid = user_bearer_token.into();
2022-10-26 00:10:05 +03:00
2022-12-14 05:13:23 +03:00
let user = user::Entity::find()
.left_join(login::Entity)
.filter(login::Column::BearerToken.eq(user_bearer_uuid))
.one(db_replica.conn())
2022-10-26 00:10:05 +03:00
.await
.web3_context("fetching user from db by bearer token")?
.web3_context("unknown bearer token")?;
2022-10-26 00:10:05 +03:00
Ok((user, semaphore_permit))
}
2023-03-03 04:39:50 +03:00
pub async fn rate_limit_login(
&self,
ip: IpAddr,
proxy_mode: ProxyMode,
) -> Web3ProxyResult<RateLimitResult> {
// TODO: dry this up with rate_limit_by_rpc_key?
// we don't care about user agent or origin or referer
let authorization = Authorization::external(
&self.config.allowed_origin_requests_per_period,
self.db_conn(),
ip,
None,
2023-03-03 04:39:50 +03:00
proxy_mode,
None,
None,
)?;
// no semaphore is needed here because login rate limits are low
// TODO: are we sure do we want a semaphore here?
let semaphore = None;
2022-09-24 06:59:21 +03:00
if let Some(rate_limiter) = &self.login_rate_limiter {
match rate_limiter.throttle_label(&ip.to_string(), None, 1).await {
Ok(RedisRateLimitResult::Allowed(_)) => {
Ok(RateLimitResult::Allowed(authorization, semaphore))
}
2022-09-24 06:59:21 +03:00
Ok(RedisRateLimitResult::RetryAt(retry_at, _)) => {
// TODO: set headers so they know when they can retry
// TODO: debug or trace?
// this is too verbose, but a stat might be good
2022-11-12 11:24:32 +03:00
// // trace!(?ip, "login rate limit exceeded until {:?}", retry_at);
Ok(RateLimitResult::RateLimited(authorization, Some(retry_at)))
2022-09-24 06:59:21 +03:00
}
Ok(RedisRateLimitResult::RetryNever) => {
// TODO: i don't think we'll get here. maybe if we ban an IP forever? seems unlikely
2022-11-12 11:24:32 +03:00
// // trace!(?ip, "login rate limit is 0");
Ok(RateLimitResult::RateLimited(authorization, None))
2022-09-24 06:59:21 +03:00
}
Err(err) => {
// internal error, not rate limit being hit
// TODO: i really want axum to do this for us in a single place.
2022-11-12 11:24:32 +03:00
error!("login rate limiter is unhappy. allowing ip. err={:?}", err);
2022-09-27 05:01:45 +03:00
Ok(RateLimitResult::Allowed(authorization, None))
2022-09-24 06:59:21 +03:00
}
}
} else {
// TODO: if no redis, rate limit with a local cache? "warn!" probably isn't right
Ok(RateLimitResult::Allowed(authorization, None))
2022-09-24 06:59:21 +03:00
}
}
/// origin is included because it can override the default rate limits
2022-10-21 23:59:05 +03:00
pub async fn rate_limit_by_ip(
&self,
allowed_origin_requests_per_period: &HashMap<String, u64>,
2022-10-21 23:59:05 +03:00
ip: IpAddr,
origin: Option<Origin>,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
) -> Web3ProxyResult<RateLimitResult> {
// ip rate limits don't check referer or user agent
2023-03-31 14:43:41 +03:00
// they do check origin because we can override rate limits for some origins
let authorization = Authorization::external(
allowed_origin_requests_per_period,
self.db_conn.clone(),
ip,
origin,
2023-03-03 04:39:50 +03:00
proxy_mode,
None,
None,
)?;
2022-09-28 06:35:55 +03:00
2022-09-23 00:03:37 +03:00
if let Some(rate_limiter) = &self.frontend_ip_rate_limiter {
match rate_limiter
.throttle(ip, authorization.checks.max_requests_per_period, 1)
.await
{
2022-09-27 05:01:45 +03:00
Ok(DeferredRateLimitResult::Allowed) => {
// rate limit allowed us. check concurrent request limits
let semaphore = self.ip_semaphore(&ip).await?;
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-27 05:01:45 +03:00
}
2022-09-23 00:03:37 +03:00
Ok(DeferredRateLimitResult::RetryAt(retry_at)) => {
// TODO: set headers so they know when they can retry
2022-11-12 11:24:32 +03:00
// // trace!(?ip, "rate limit exceeded until {:?}", retry_at);
Ok(RateLimitResult::RateLimited(authorization, Some(retry_at)))
2022-09-23 00:03:37 +03:00
}
Ok(DeferredRateLimitResult::RetryNever) => {
// TODO: i don't think we'll get here. maybe if we ban an IP forever? seems unlikely
2022-11-12 11:24:32 +03:00
// // trace!(?ip, "rate limit is 0");
Ok(RateLimitResult::RateLimited(authorization, None))
2022-09-23 00:03:37 +03:00
}
Err(err) => {
// this an internal error of some kind, not the rate limit being hit
2022-09-23 00:03:37 +03:00
// TODO: i really want axum to do this for us in a single place.
2022-11-12 11:24:32 +03:00
error!("rate limiter is unhappy. allowing ip. err={:?}", err);
2022-09-27 05:01:45 +03:00
// at least we can still check the semaphore
let semaphore = self.ip_semaphore(&ip).await?;
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
}
} else {
// no redis, but we can still check the ip semaphore
let semaphore = self.ip_semaphore(&ip).await?;
2022-09-23 00:03:37 +03:00
// TODO: if no redis, rate limit with a local cache? "warn!" probably isn't right
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
}
// check the local cache for user data, or query the database
pub(crate) async fn authorization_checks(
2022-11-01 21:54:39 +03:00
&self,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
2022-11-01 21:54:39 +03:00
rpc_secret_key: RpcSecretKey,
) -> Web3ProxyResult<AuthorizationChecks> {
let authorization_checks: Result<_, Arc<Web3ProxyError>> = self
2022-11-01 21:54:39 +03:00
.rpc_secret_key_cache
.try_get_with(rpc_secret_key.into(), async move {
// trace!(?rpc_secret_key, "user cache miss");
2022-09-23 00:03:37 +03:00
let db_replica = self
.db_replica()
.web3_context("Getting database connection")?;
2022-09-23 00:03:37 +03:00
// TODO: join the user table to this to return the User? we don't always need it
2022-11-01 21:54:39 +03:00
// TODO: join on secondary users
// TODO: join on user tier
match rpc_key::Entity::find()
2023-01-19 03:17:43 +03:00
.filter(rpc_key::Column::SecretKey.eq(<Uuid>::from(rpc_secret_key)))
2022-11-01 21:54:39 +03:00
.filter(rpc_key::Column::Active.eq(true))
.one(db_replica.conn())
2022-09-23 00:03:37 +03:00
.await?
{
2022-10-27 03:12:42 +03:00
Some(rpc_key_model) => {
2022-10-27 00:39:26 +03:00
// TODO: move these splits into helper functions
// TODO: can we have sea orm handle this for us?
2022-11-01 22:12:57 +03:00
let user_model = user::Entity::find_by_id(rpc_key_model.user_id)
.one(db_replica.conn())
2022-11-01 22:12:57 +03:00
.await?
.expect("related user");
User Balance + Referral Logic (#44) * will implement balance topup endpoint * will quickly fix other PR reviews * merging from master * will finish up godmoe * will finish up login * added logic to top up balance (first iteration) * should implement additional columns soon (currency, amount, tx-hash), as well as a new table for spend * updated migrations, will account for spend next * get back to this later * will merge PR from stats-v2 * stats v2 rebased all my commits and squashed them down to one * cargo upgrade * added migrtation for spend in accounting table. will run test-deposit next * trying to get request from polygon * first iteration /user/balance/:tx_hash works, needs to add accepted tokens next * creating the referral code seems to work * will now check if spending enough credits will lead to both parties receiving credits * rpcstats takes care of accounting for spend data * removed track spend from table * Revert "removed track spend from table" This reverts commit a50802d6ae75f786864c5ec42d0ceb2cb27124ed. * Revert "rpcstats takes care of accounting for spend data" This reverts commit 1cec728bf241e4cfd24351134637ed81c1a5a10b. * removed rpc request table entity * updated referral code to use ulid s * credits used are aggregated * added a bunch of fields to referrer * added database logic whenever an aggregate stats is added. will have to iterate over this a couple times i think. go to (1) detecting accepted stables next, (2) fix influxdb bug and (3) start to write test * removed track spend as this will occur in the database * will first work on "balance", then referral. these should really be treated as two separate PRs (although already convoluted) * balance logic initial commit * breaking WIP, changing the RPC call logic functions * will start testing next * got rid of warnings & lint * will proceed with subtracting / adding to balance * added decimal points, balance tracking seems to work * will beautify code a bit * removed deprecated dependency, and added topic + deposit contract to app.yaml * brownie test suite does not rely on local contract files it pulls all from polygonscan * will continue with referral * should perhaps (in a future revision) recordhow much the referees got for free. marking referrals seems to work rn * user is upgraded to premium if they deposit more than 10$. we dont accept more than $10M in a single tx * will start PR, referral seems to be fine so far, perhaps up to some numbers that still may need tweaking * will start PR * removed rogue comments, cleaned up payments a bit * changes before PR * apply stats * added unique constraint * some refactoring such that the user file is not too bloated * compiling * progress with subusers, creating a table entry seems to work * good response type is there as well now, will work on getters from primary user and secondary user next * subuser logic also seems fine now * downgrade logic * fixed bug influxdb does not support different types in same query (which makes sense) * WIP temporary commit * merging with PR * Delete daemon.rs there are multiple daemons now, so this was moved to `proxyd` * will remove request clone to &mut * multiple request handles for payment * making requests still seem fine * removed redundant commented out bits * added deposit endpoint, added deposit amount and deposit user, untested yet * small bug with downgrade tier id * will add authorization so balance can be received for users * balance history should be set now too * will check balance over time again * subususer can see rpc key balance if admin or owner * stats also seems to work fine now with historical balance * things seem to be building and working * removed clone from OpenRequestHandle * removed influxdb from workspace members * changed config files * reran sea-orm generate entities, added a foreign key, should be proper now * removed contract from commit * made deposit contract optional * added topic in polygon dev * changed deposit contract to deposit factory contract * added selfrelation on user_tier * added payment required * changed chain id to u64 * add wss in polygon llamarpc * removed origin and method from the table * added onchain transactions naming (and forgot to add a migration before) * changed foreign key to be the referrer (id), not the code itself * forgot to add id as the target foreign key * WIP adding cache to update role * fixed merge conflicts --------- Co-authored-by: Bryan Stitt <bryan@llamanodes.com> Co-authored-by: Bryan Stitt <bryan@stitthappens.com>
2023-05-12 19:45:15 +03:00
let balance = balance::Entity::find()
.filter(balance::Column::UserId.eq(user_model.id))
.one(db_replica.conn())
.await?
.expect("related balance")
.available_balance;
2022-11-01 22:12:57 +03:00
let user_tier_model =
user_tier::Entity::find_by_id(user_model.user_tier_id)
.one(db_replica.conn())
2022-11-01 22:12:57 +03:00
.await?
.expect("related user tier");
2022-10-27 00:39:26 +03:00
let allowed_ips: Option<Vec<IpNet>> =
2022-10-27 03:12:42 +03:00
if let Some(allowed_ips) = rpc_key_model.allowed_ips {
2022-10-27 00:39:26 +03:00
let x = allowed_ips
.split(',')
.map(|x| x.trim().parse::<IpNet>())
2022-10-27 00:39:26 +03:00
.collect::<Result<Vec<_>, _>>()?;
Some(x)
} else {
None
};
let allowed_origins: Option<Vec<Origin>> =
2022-10-27 03:12:42 +03:00
if let Some(allowed_origins) = rpc_key_model.allowed_origins {
2022-10-27 00:39:26 +03:00
// TODO: do this without collecting twice?
let x = allowed_origins
.split(',')
.map(|x| HeaderValue::from_str(x.trim()))
2022-10-27 00:39:26 +03:00
.collect::<Result<Vec<_>, _>>()?
.into_iter()
2022-10-27 00:39:26 +03:00
.map(|x| Origin::decode(&mut [x].iter()))
.collect::<Result<Vec<_>, _>>()?;
Some(x)
} else {
None
};
let allowed_referers: Option<Vec<Referer>> =
2022-10-27 03:12:42 +03:00
if let Some(allowed_referers) = rpc_key_model.allowed_referers {
2022-10-27 00:39:26 +03:00
let x = allowed_referers
.split(',')
.map(|x| {
x.trim()
.parse::<Referer>()
.or(Err(Web3ProxyError::InvalidReferer))
})
2022-10-27 00:39:26 +03:00
.collect::<Result<Vec<_>, _>>()?;
Some(x)
} else {
None
};
let allowed_user_agents: Option<Vec<UserAgent>> =
2022-10-27 03:12:42 +03:00
if let Some(allowed_user_agents) = rpc_key_model.allowed_user_agents {
2022-10-27 00:39:26 +03:00
let x: Result<Vec<_>, _> = allowed_user_agents
.split(',')
.map(|x| {
x.trim()
.parse::<UserAgent>()
.or(Err(Web3ProxyError::InvalidUserAgent))
})
2022-10-27 00:39:26 +03:00
.collect();
Some(x?)
} else {
None
};
2022-09-23 00:03:37 +03:00
let rpc_key_id =
Some(rpc_key_model.id.try_into().expect("db ids are never 0"));
Ok(AuthorizationChecks {
2022-10-27 03:12:42 +03:00
user_id: rpc_key_model.user_id,
2023-01-19 03:17:43 +03:00
rpc_secret_key: Some(rpc_secret_key),
rpc_secret_key_id: rpc_key_id,
allowed_ips,
allowed_origins,
allowed_referers,
allowed_user_agents,
tracking_level: rpc_key_model.log_level,
2022-10-27 03:12:42 +03:00
log_revert_chance: rpc_key_model.log_revert_chance,
2022-11-01 22:12:57 +03:00
max_concurrent_requests: user_tier_model.max_concurrent_requests,
max_requests_per_period: user_tier_model.max_requests_per_period,
2023-01-12 01:51:01 +03:00
private_txs: rpc_key_model.private_txs,
2023-03-03 04:39:50 +03:00
proxy_mode,
User Balance + Referral Logic (#44) * will implement balance topup endpoint * will quickly fix other PR reviews * merging from master * will finish up godmoe * will finish up login * added logic to top up balance (first iteration) * should implement additional columns soon (currency, amount, tx-hash), as well as a new table for spend * updated migrations, will account for spend next * get back to this later * will merge PR from stats-v2 * stats v2 rebased all my commits and squashed them down to one * cargo upgrade * added migrtation for spend in accounting table. will run test-deposit next * trying to get request from polygon * first iteration /user/balance/:tx_hash works, needs to add accepted tokens next * creating the referral code seems to work * will now check if spending enough credits will lead to both parties receiving credits * rpcstats takes care of accounting for spend data * removed track spend from table * Revert "removed track spend from table" This reverts commit a50802d6ae75f786864c5ec42d0ceb2cb27124ed. * Revert "rpcstats takes care of accounting for spend data" This reverts commit 1cec728bf241e4cfd24351134637ed81c1a5a10b. * removed rpc request table entity * updated referral code to use ulid s * credits used are aggregated * added a bunch of fields to referrer * added database logic whenever an aggregate stats is added. will have to iterate over this a couple times i think. go to (1) detecting accepted stables next, (2) fix influxdb bug and (3) start to write test * removed track spend as this will occur in the database * will first work on "balance", then referral. these should really be treated as two separate PRs (although already convoluted) * balance logic initial commit * breaking WIP, changing the RPC call logic functions * will start testing next * got rid of warnings & lint * will proceed with subtracting / adding to balance * added decimal points, balance tracking seems to work * will beautify code a bit * removed deprecated dependency, and added topic + deposit contract to app.yaml * brownie test suite does not rely on local contract files it pulls all from polygonscan * will continue with referral * should perhaps (in a future revision) recordhow much the referees got for free. marking referrals seems to work rn * user is upgraded to premium if they deposit more than 10$. we dont accept more than $10M in a single tx * will start PR, referral seems to be fine so far, perhaps up to some numbers that still may need tweaking * will start PR * removed rogue comments, cleaned up payments a bit * changes before PR * apply stats * added unique constraint * some refactoring such that the user file is not too bloated * compiling * progress with subusers, creating a table entry seems to work * good response type is there as well now, will work on getters from primary user and secondary user next * subuser logic also seems fine now * downgrade logic * fixed bug influxdb does not support different types in same query (which makes sense) * WIP temporary commit * merging with PR * Delete daemon.rs there are multiple daemons now, so this was moved to `proxyd` * will remove request clone to &mut * multiple request handles for payment * making requests still seem fine * removed redundant commented out bits * added deposit endpoint, added deposit amount and deposit user, untested yet * small bug with downgrade tier id * will add authorization so balance can be received for users * balance history should be set now too * will check balance over time again * subususer can see rpc key balance if admin or owner * stats also seems to work fine now with historical balance * things seem to be building and working * removed clone from OpenRequestHandle * removed influxdb from workspace members * changed config files * reran sea-orm generate entities, added a foreign key, should be proper now * removed contract from commit * made deposit contract optional * added topic in polygon dev * changed deposit contract to deposit factory contract * added selfrelation on user_tier * added payment required * changed chain id to u64 * add wss in polygon llamarpc * removed origin and method from the table * added onchain transactions naming (and forgot to add a migration before) * changed foreign key to be the referrer (id), not the code itself * forgot to add id as the target foreign key * WIP adding cache to update role * fixed merge conflicts --------- Co-authored-by: Bryan Stitt <bryan@llamanodes.com> Co-authored-by: Bryan Stitt <bryan@stitthappens.com>
2023-05-12 19:45:15 +03:00
balance: Some(balance),
2022-09-23 00:03:37 +03:00
})
}
None => Ok(AuthorizationChecks::default()),
2022-09-23 00:03:37 +03:00
}
})
.await;
authorization_checks.map_err(Web3ProxyError::Arc)
2022-09-23 00:03:37 +03:00
}
/// Authorized the ip/origin/referer/useragent and rate limit and concurrency
pub async fn rate_limit_by_rpc_key(
2022-11-01 21:54:39 +03:00
&self,
ip: IpAddr,
origin: Option<Origin>,
2023-03-03 04:39:50 +03:00
proxy_mode: ProxyMode,
referer: Option<Referer>,
2022-11-01 21:54:39 +03:00
rpc_key: RpcSecretKey,
user_agent: Option<UserAgent>,
) -> Web3ProxyResult<RateLimitResult> {
2023-03-03 04:39:50 +03:00
let authorization_checks = self.authorization_checks(proxy_mode, rpc_key).await?;
2022-09-23 00:03:37 +03:00
// if no rpc_key_id matching the given rpc was found, then we can't rate limit by key
2023-01-19 03:17:43 +03:00
if authorization_checks.rpc_secret_key_id.is_none() {
2022-09-23 00:03:37 +03:00
return Ok(RateLimitResult::UnknownKey);
}
// TODO: rpc_key should have an option to rate limit by ip instead of by key
// only allow this rpc_key to run a limited amount of concurrent requests
// TODO: rate limit should be BEFORE the semaphore!
let semaphore = self
.registered_user_semaphore(&authorization_checks)
.await?;
let authorization = Authorization::try_new(
authorization_checks,
self.db_conn(),
ip,
origin,
referer,
user_agent,
2022-12-12 07:39:54 +03:00
AuthorizationType::Frontend,
)?;
2022-09-27 05:01:45 +03:00
let user_max_requests_per_period = match authorization.checks.max_requests_per_period {
2022-09-28 06:35:55 +03:00
None => {
return Ok(RateLimitResult::Allowed(authorization, semaphore));
2022-09-27 05:01:45 +03:00
}
2022-09-23 00:03:37 +03:00
Some(x) => x,
};
// user key is valid. now check rate limits
if let Some(rate_limiter) = &self.frontend_registered_user_rate_limiter {
2022-09-23 00:03:37 +03:00
match rate_limiter
.throttle(
authorization.checks.user_id,
Some(user_max_requests_per_period),
1,
)
2022-09-23 00:03:37 +03:00
.await
{
2022-09-27 05:01:45 +03:00
Ok(DeferredRateLimitResult::Allowed) => {
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-27 05:01:45 +03:00
}
2022-09-23 00:03:37 +03:00
Ok(DeferredRateLimitResult::RetryAt(retry_at)) => {
// TODO: set headers so they know when they can retry
// TODO: debug or trace?
// this is too verbose, but a stat might be good
// TODO: keys are secrets! use the id instead
2022-10-10 07:15:07 +03:00
// TODO: emit a stat
2022-11-12 11:24:32 +03:00
// // trace!(?rpc_key, "rate limit exceeded until {:?}", retry_at);
Ok(RateLimitResult::RateLimited(authorization, Some(retry_at)))
2022-09-23 00:03:37 +03:00
}
Ok(DeferredRateLimitResult::RetryNever) => {
// TODO: keys are secret. don't log them!
2022-11-12 11:24:32 +03:00
// // trace!(?rpc_key, "rate limit is 0");
2022-10-10 07:15:07 +03:00
// TODO: emit a stat
Ok(RateLimitResult::RateLimited(authorization, None))
2022-09-23 00:03:37 +03:00
}
Err(err) => {
// internal error, not rate limit being hit
// TODO: i really want axum to do this for us in a single place.
2022-11-12 11:24:32 +03:00
error!("rate limiter is unhappy. allowing ip. err={:?}", err);
2022-09-27 05:01:45 +03:00
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
}
} else {
// TODO: if no redis, rate limit with just a local cache?
Ok(RateLimitResult::Allowed(authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
}
}
2023-01-19 03:17:43 +03:00
impl Authorization {
pub async fn check_again(
&self,
app: &Arc<Web3ProxyApp>,
) -> Web3ProxyResult<(Arc<Self>, Option<OwnedSemaphorePermit>)> {
2023-01-19 03:17:43 +03:00
// TODO: we could probably do this without clones. but this is easy
let (a, s) = if let Some(rpc_secret_key) = self.checks.rpc_secret_key {
key_is_authorized(
app,
rpc_secret_key,
self.ip,
self.origin.clone(),
2023-03-03 04:39:50 +03:00
self.checks.proxy_mode,
2023-01-19 03:17:43 +03:00
self.referer.clone(),
self.user_agent.clone(),
)
.await?
} else {
2023-03-03 04:39:50 +03:00
ip_is_authorized(app, self.ip, self.origin.clone(), self.checks.proxy_mode).await?
2023-01-19 03:17:43 +03:00
};
let a = Arc::new(a);
Ok((a, s))
}
}