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

741 lines
28 KiB
Rust
Raw Normal View History

2022-10-18 00:47:58 +03:00
//! Utilities for authorization of logged in and anonymous users.
2022-09-23 00:03:37 +03:00
use super::errors::FrontendErrorResponse;
use crate::app::{AuthorizationChecks, Web3ProxyApp, APP_USER_AGENT};
2022-10-10 07:15:07 +03:00
use crate::jsonrpc::JsonRpcRequest;
2022-10-31 23:05:58 +03:00
use crate::user_token::UserBearerToken;
2022-09-23 00:03:37 +03:00
use anyhow::Context;
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;
2022-09-23 00:03:37 +03:00
use deferred_rate_limiter::DeferredRateLimitResult;
2022-11-01 21:54:39 +03:00
use entities::{rpc_key, user, user_tier};
use hashbrown::HashMap;
2022-10-27 00:39:26 +03:00
use http::HeaderValue;
use ipnet::IpNet;
2022-11-12 11:24:32 +03:00
use log::error;
2022-10-26 00:10:05 +03:00
use redis_rate_limiter::redis::AsyncCommands;
2022-09-24 06:59:21 +03:00
use redis_rate_limiter::RedisRateLimitResult;
2022-11-01 21:54:39 +03:00
use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter};
2022-09-24 08:53:45 +03:00
use std::fmt::Display;
2022-10-25 06:41:59 +03:00
use std::sync::atomic::{AtomicBool, AtomicU64};
2022-09-24 08:53:45 +03:00
use std::{net::IpAddr, str::FromStr, sync::Arc};
2022-09-28 06:35:55 +03:00
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
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,
}
/// TODO: include the authorization checks in this?
2022-10-27 00:39:26 +03:00
#[derive(Clone, Debug)]
pub struct Authorization {
pub checks: AuthorizationChecks,
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-10-10 07:15:07 +03:00
}
#[derive(Debug)]
2022-10-10 07:15:07 +03:00
pub struct RequestMetadata {
pub start_datetime: chrono::DateTime<Utc>,
pub start_instant: tokio::time::Instant,
// TODO: better name for this
pub period_seconds: u64,
2022-10-11 20:34:25 +03:00
pub request_bytes: u64,
// TODO: do we need atomics? seems like we should be able to pass a &mut around
2022-11-03 02:14:16 +03:00
// TODO: "archive" isn't really a boolean.
pub archive_request: AtomicBool,
/// if this is 0, there was a cache_hit
2022-10-25 06:41:59 +03:00
pub backend_requests: AtomicU64,
pub no_servers: AtomicU64,
pub error_response: AtomicBool,
pub response_bytes: AtomicU64,
pub response_millis: AtomicU64,
2022-10-10 07:15:07 +03:00
}
impl RequestMetadata {
pub fn new(period_seconds: u64, request: &JsonRpcRequest) -> anyhow::Result<Self> {
// TODO: how can we do this without turning it into a string first. this is going to slow us down!
let request_bytes = serde_json::to_string(request)
.context("finding request size")?
.len()
.try_into()?;
2022-10-10 07:15:07 +03:00
let new = Self {
start_instant: Instant::now(),
start_datetime: Utc::now(),
period_seconds,
2022-10-11 20:34:25 +03:00
request_bytes,
2022-11-03 02:14:16 +03:00
archive_request: false.into(),
backend_requests: 0.into(),
no_servers: 0.into(),
error_response: false.into(),
response_bytes: 0.into(),
response_millis: 0.into(),
};
Ok(new)
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 {
2022-09-24 08:53:45 +03:00
type Err = anyhow::Error;
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
2022-09-24 08:53:45 +03:00
Err(anyhow::anyhow!("UserKey was not a ULID or UUID"))
}
}
}
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 local(db_conn: Option<DatabaseConnection>) -> anyhow::Result<Self> {
let authorization_checks = AuthorizationChecks {
// any error logs on a local (internal) query are likely problems. log them all
log_revert_chance: 1.0,
// 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)
}
pub fn public(
allowed_origin_requests_per_period: &HashMap<String, u64>,
db_conn: Option<DatabaseConnection>,
ip: IpAddr,
origin: Option<Origin>,
referer: Option<Referer>,
user_agent: Option<UserAgent>,
) -> anyhow::Result<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();
// TODO: default or None?
let authorization_checks = AuthorizationChecks {
max_requests_per_period,
..Default::default()
};
Self::try_new(
authorization_checks,
db_conn,
ip,
origin,
referer,
user_agent,
)
}
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>,
) -> anyhow::Result<Self> {
// check ip
match &authorization_checks.allowed_ips {
None => {}
Some(allowed_ips) => {
if !allowed_ips.iter().any(|x| x.contains(&ip)) {
return Err(anyhow::anyhow!("IP is not allowed!"));
}
}
}
// check origin
match (&origin, &authorization_checks.allowed_origins) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(anyhow::anyhow!("Origin required")),
(Some(origin), Some(allowed_origins)) => {
2022-10-27 00:39:26 +03:00
if !allowed_origins.contains(origin) {
return Err(anyhow::anyhow!("IP is not allowed!"));
}
}
}
// check referer
match (&referer, &authorization_checks.allowed_referers) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(anyhow::anyhow!("Referer required")),
(Some(referer), Some(allowed_referers)) => {
if !allowed_referers.contains(referer) {
return Err(anyhow::anyhow!("Referer is not allowed!"));
}
}
}
// check user_agent
match (&user_agent, &authorization_checks.allowed_user_agents) {
(None, None) => {}
(Some(_), None) => {}
(None, Some(_)) => return Err(anyhow::anyhow!("User agent required")),
(Some(user_agent), Some(allowed_user_agents)) => {
if !allowed_user_agents.contains(user_agent) {
return Err(anyhow::anyhow!("User agent is not allowed!"));
}
}
}
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,
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
2022-09-24 06:59:21 +03:00
pub async fn login_is_authorized(
app: &Web3ProxyApp,
ip: IpAddr,
) -> Result<Authorization, FrontendErrorResponse> {
let authorization = match app.rate_limit_login(ip).await? {
RateLimitResult::Allowed(authorization, None) => authorization,
RateLimitResult::RateLimited(authorization, retry_at) => {
return Err(FrontendErrorResponse::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(
app: &Web3ProxyApp,
ip: IpAddr,
origin: Option<Origin>,
) -> Result<(Authorization, Option<OwnedSemaphorePermit>), FrontendErrorResponse> {
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
.rate_limit_by_ip(&app.config.allowed_origin_requests_per_period, ip, origin)
.await?
{
RateLimitResult::Allowed(authorization, semaphore) => (authorization, semaphore),
RateLimitResult::RateLimited(authorization, retry_at) => {
return Err(FrontendErrorResponse::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),
};
Ok((authorization, semaphore))
2022-09-23 00:03:37 +03:00
}
/// like app.rate_limit_by_rpc_key but converts to a FrontendErrorResponse;
2022-09-23 00:03:37 +03:00
pub async fn key_is_authorized(
app: &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>,
2022-09-23 00:03:37 +03:00
referer: Option<Referer>,
user_agent: Option<UserAgent>,
) -> Result<(Authorization, Option<OwnedSemaphorePermit>), FrontendErrorResponse> {
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
.rate_limit_by_rpc_key(ip, origin, referer, rpc_key, user_agent)
.await?
{
RateLimitResult::Allowed(authorization, semaphore) => (authorization, semaphore),
RateLimitResult::RateLimited(authorization, retry_at) => {
return Err(FrontendErrorResponse::RateLimited(authorization, retry_at));
2022-09-23 00:03:37 +03:00
}
RateLimitResult::UnknownKey => return Err(FrontendErrorResponse::UnknownKey),
};
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.
2022-10-25 07:01:41 +03:00
pub async fn ip_semaphore(&self, ip: IpAddr) -> anyhow::Result<Option<OwnedSemaphorePermit>> {
if let Some(max_concurrent_requests) = self.config.public_max_concurrent_requests {
let semaphore = self
.ip_semaphores
.get_with(ip, async move {
// 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 rpc_key_semaphore(
2022-09-28 06:35:55 +03:00
&self,
authorization_checks: &AuthorizationChecks,
2022-09-28 06:35:55 +03:00
) -> anyhow::Result<Option<OwnedSemaphorePermit>> {
if let Some(max_concurrent_requests) = authorization_checks.max_concurrent_requests {
let rpc_key_id = authorization_checks.rpc_key_id.context("no rpc_key_id")?;
2022-09-28 06:35:55 +03:00
let semaphore = self
2022-10-27 03:12:42 +03:00
.rpc_key_semaphores
.get_with(rpc_key_id, async move {
2022-10-25 07:31:18 +03:00
let s = Semaphore::new(max_concurrent_requests as usize);
2022-11-12 11:24:32 +03:00
// // trace!("new semaphore for rpc_key_id {}", rpc_key_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,
) -> anyhow::Result<(user::Model, OwnedSemaphorePermit)> {
// limit concurrent requests
let semaphore = self
.bearer_token_semaphores
.get_with(bearer.token().to_string(), async move {
let s = Semaphore::new(self.config.bearer_token_max_concurrent_requests as usize);
Arc::new(s)
})
.await;
let semaphore_permit = semaphore.acquire_owned().await?;
// get the user id for this bearer token
// TODO: move redis key building to a helper function
2022-10-31 23:05:58 +03:00
let bearer_cache_key = UserBearerToken::try_from(bearer)?.to_string();
2022-10-26 00:10:05 +03:00
// get the attached address from redis for the given auth_token.
let mut redis_conn = self.redis_conn().await?;
let user_id: u64 = redis_conn
.get::<_, Option<u64>>(bearer_cache_key)
.await
.context("fetching bearer cache key from redis")?
.context("unknown bearer token")?;
// turn user id into a user
let db_conn = self.db_conn().context("Getting database connection")?;
let user = user::Entity::find_by_id(user_id)
.one(&db_conn)
.await
.context("fetching user from db by id")?
.context("unknown user id")?;
Ok((user, semaphore_permit))
}
2022-09-24 06:59:21 +03:00
pub async fn rate_limit_login(&self, ip: IpAddr) -> anyhow::Result<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::public(
&self.config.allowed_origin_requests_per_period,
self.db_conn(),
ip,
None,
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>,
2022-10-21 23:59:05 +03:00
) -> anyhow::Result<RateLimitResult> {
// ip rate limits don't check referer or user agent
// the do check
let authorization = Authorization::public(
allowed_origin_requests_per_period,
self.db_conn.clone(),
ip,
origin,
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,
rpc_secret_key: RpcSecretKey,
) -> anyhow::Result<AuthorizationChecks> {
let authorization_checks: Result<_, Arc<anyhow::Error>> = self
2022-11-01 21:54:39 +03:00
.rpc_secret_key_cache
.try_get_with(rpc_secret_key.into(), async move {
2022-11-12 11:24:32 +03:00
// // trace!(?rpc_secret_key, "user cache miss");
2022-09-23 00:03:37 +03:00
2022-10-20 09:17:20 +03:00
let db_conn = self.db_conn().context("Getting database connection")?;
2022-09-23 00:03:37 +03:00
2022-11-01 21:54:39 +03:00
let rpc_secret_key: Uuid = rpc_secret_key.into();
2022-09-24 08:53:45 +03:00
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()
.filter(rpc_key::Column::SecretKey.eq(rpc_secret_key))
.filter(rpc_key::Column::Active.eq(true))
2022-10-20 09:17:20 +03:00
.one(&db_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_conn)
.await?
.expect("related user");
let user_tier_model =
user_tier::Entity::find_by_id(user_model.user_tier_id)
.one(&db_conn)
.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.parse::<IpNet>())
.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(HeaderValue::from_str)
.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.parse::<Referer>())
.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.parse::<UserAgent>())
.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,
rpc_key_id,
allowed_ips,
allowed_origins,
allowed_referers,
allowed_user_agents,
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,
2022-09-23 00:03:37 +03:00
})
}
None => Ok(AuthorizationChecks::default()),
2022-09-23 00:03:37 +03:00
}
})
.await;
2022-09-30 07:18:18 +03:00
// TODO: what's the best way to handle this arc? try_unwrap will not work
authorization_checks.map_err(|err| anyhow::anyhow!(err))
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>,
referer: Option<Referer>,
2022-11-01 21:54:39 +03:00
rpc_key: RpcSecretKey,
user_agent: Option<UserAgent>,
2022-11-01 21:54:39 +03:00
) -> anyhow::Result<RateLimitResult> {
let authorization_checks = self.authorization_checks(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
if authorization_checks.rpc_key_id.is_none() {
2022-09-23 00:03:37 +03:00
return Ok(RateLimitResult::UnknownKey);
}
// only allow this rpc_key to run a limited amount of concurrent requests
// TODO: rate limit should be BEFORE the semaphore!
let semaphore = self.rpc_key_semaphore(&authorization_checks).await?;
let authorization = Authorization::try_new(
authorization_checks,
self.db_conn(),
ip,
origin,
referer,
user_agent,
)?;
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_key_rate_limiter {
match rate_limiter
2022-10-27 03:12:42 +03:00
.throttle(rpc_key.into(), 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
}
}
}