refactor: use auth middleware

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

crates/service/src/error.rs

@@ -22,14 +22,8 @@ pub enum IndexerServiceError {
     ReceiptError(#[from] tap_core::Error),
     #[error("No attestation signer found for allocation `{0}`")]
     NoSignerForAllocation(Address),
-    #[error("Invalid request body: {0}")]
-    InvalidRequest(anyhow::Error),
     #[error("Error while processing the request: {0}")]
     ProcessingError(SubgraphServiceError),
-    #[error("No valid receipt or free query auth token provided")]
-    Unauthorized,
-    #[error("Invalid free query auth token")]
-    InvalidFreeQueryAuthToken,
     #[error("Failed to sign attestation")]
     FailedToSignAttestation,
 
@@ -47,15 +41,9 @@ impl IntoResponse for IndexerServiceError {
         }
 
         let status = match self {
-            Unauthorized => StatusCode::UNAUTHORIZED,
-
             NoSignerForAllocation(_) | FailedToSignAttestation => StatusCode::INTERNAL_SERVER_ERROR,
 
-            ReceiptError(_)
-            | InvalidRequest(_)
-            | InvalidFreeQueryAuthToken
-            | EscrowAccount(_)
-            | ProcessingError(_) => StatusCode::BAD_REQUEST,
+            ReceiptError(_) | EscrowAccount(_) | ProcessingError(_) => StatusCode::BAD_REQUEST,
             ReceiptNotFound => StatusCode::PAYMENT_REQUIRED,
         };
         tracing::error!(%self, "An IndexerServiceError occoured.");

crates/service/src/middleware.rs

@@ -1,7 +1,7 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-mod auth;
+pub mod auth;
 mod inject_allocation;
 mod inject_deployment;
 mod inject_labels;
@@ -10,8 +10,9 @@ mod inject_sender;
 mod prometheus_metrics;
 
 pub use inject_allocation::{allocation_middleware, Allocation, AllocationState};
+pub use inject_context::context_middleware;
 pub use inject_deployment::deployment_middleware;
 pub use inject_labels::labels_middleware;
 pub use inject_receipt::receipt_middleware;
-pub use inject_sender::{sender_middleware, Sender, SenderState};
+pub use inject_sender::{sender_middleware, SenderState};
 pub use prometheus_metrics::PrometheusMetricsMiddlewareLayer;

crates/service/src/routes/request_handler.rs

@@ -3,87 +3,26 @@
 
 use std::sync::Arc;
 
-use crate::{
-    error::IndexerServiceError,
-    metrics::FAILED_RECEIPT,
-    middleware::{Allocation, Sender},
-    tap::AgoraQuery,
-};
+use crate::{error::IndexerServiceError, middleware::Allocation};
 use axum::{
     extract::{Path, State},
-    http::HeaderMap,
     response::IntoResponse,
     Extension,
 };
-use axum_extra::TypedHeader;
 use reqwest::StatusCode;
-use serde_json::value::RawValue;
-use tap_core::receipt::Context;
 use thegraph_core::DeploymentId;
 use tracing::trace;
 
-use crate::service::{AttestationOutput, IndexerServiceResponse, IndexerServiceState, TapReceipt};
+use crate::service::{AttestationOutput, IndexerServiceResponse, IndexerServiceState};
 
 pub async fn request_handler(
     Path(manifest_id): Path<DeploymentId>,
-    TypedHeader(receipt): TypedHeader<TapReceipt>,
-    Extension(Sender(sender)): Extension<Sender>,
     Extension(Allocation(allocation_id)): Extension<Allocation>,
     State(state): State<Arc<IndexerServiceState>>,
-    headers: HeaderMap,
     req: String,
 ) -> Result<impl IntoResponse, IndexerServiceError> {
     trace!("Handling request for deployment `{manifest_id}`");
 
-    let request: QueryBody =
-        serde_json::from_str(&req).map_err(|e| IndexerServiceError::InvalidRequest(e.into()))?;
-
-    if let Some(receipt) = receipt.into_signed_receipt() {
-        let variables = request
-            .variables
-            .as_ref()
-            .map(ToString::to_string)
-            .unwrap_or_default();
-        let mut ctx = Context::new();
-        ctx.insert(AgoraQuery {
-            deployment_id: manifest_id,
-            query: request.query.clone(),
-            variables,
-        });
-
-        // Verify the receipt and store it in the database
-        state
-            .tap_manager
-            .verify_and_store_receipt(&ctx, receipt)
-            .await
-            .inspect_err(|_| {
-                FAILED_RECEIPT
-                    .with_label_values(&[
-                        &manifest_id.to_string(),
-                        &allocation_id.to_string(),
-                        &sender.to_string(),
-                    ])
-                    .inc()
-            })
-            .map_err(IndexerServiceError::ReceiptError)?;
-    } else {
-        match headers
-            .get("authorization")
-            .and_then(|v| v.to_str().ok())
-            .and_then(|s| s.strip_prefix("Bearer "))
-            .map(|s| s.to_string())
-        {
-            None => return Err(IndexerServiceError::Unauthorized),
-            Some(ref token) => {
-                if Some(token) != state.config.service.free_query_auth_token.as_ref() {
-                    return Err(IndexerServiceError::InvalidFreeQueryAuthToken);
-                }
-            }
-        }
-
-        trace!(?manifest_id, "New free query");
-    }
-
     // Check if we have an attestation signer for the allocation the receipt was created for
     let signer = state
         .attestation_signers
@@ -94,7 +33,7 @@ pub async fn request_handler(
 
     let response = state
         .service_impl
-        .process_request(manifest_id, request)
+        .process_request(manifest_id, &req)
         .await
         .map_err(IndexerServiceError::ProcessingError)?;
 
@@ -112,9 +51,3 @@ pub async fn request_handler(
 
     Ok((StatusCode::OK, response))
 }
-
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
-pub struct QueryBody {
-    pub query: String,
-    pub variables: Option<Box<RawValue>>,
-}

crates/service/src/service.rs

@@ -98,7 +98,7 @@ impl SubgraphService {
     pub async fn process_request<Request: DeserializeOwned + Send + std::fmt::Debug + Serialize>(
         &self,
         deployment: DeploymentId,
-        request: Request,
+        request: &Request,
     ) -> Result<SubgraphServiceResponse, SubgraphServiceError> {
         let deployment_url = self
             .state
@@ -110,7 +110,7 @@ impl SubgraphService {
             .state
             .graph_node_client
             .post(deployment_url)
-            .json(&request)
+            .json(request)
             .send()
             .await
             .map_err(SubgraphServiceError::QueryForwardingError)?;

crates/service/src/service/indexer_service.rs

@@ -2,9 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use anyhow;
-use axum::extract::MatchedPath;
-use axum::extract::Request as ExtractRequest;
 use axum::{
+    extract::{MatchedPath, Request as ExtractRequest},
     http::{Method, Request},
     middleware::{from_fn, from_fn_with_state},
     response::IntoResponse,
@@ -26,20 +25,24 @@ use std::{
 };
 use tap_core::{manager::Manager, receipt::checks::CheckList, tap_eip712_domain};
 use thegraph_core::{Address, Attestation};
-use tokio::net::TcpListener;
-use tokio::signal;
-use tokio::sync::watch::Receiver;
+use tokio::{net::TcpListener, signal, sync::watch::Receiver};
 use tower::ServiceBuilder;
 use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
-use tower_http::validate_request::ValidateRequestHeaderLayer;
-use tower_http::{cors, cors::CorsLayer, normalize_path::NormalizePath, trace::TraceLayer};
-use tracing::warn;
-use tracing::{error, info, info_span};
+use tower_http::{
+    auth::AsyncRequireAuthorizationLayer,
+    cors::{self, CorsLayer},
+    normalize_path::NormalizePath,
+    trace::TraceLayer,
+    validate_request::ValidateRequestHeaderLayer,
+};
+use tracing::{error, info, info_span, warn};
 
 use crate::{
-    metrics::{HANDLER_FAILURE, HANDLER_HISTOGRAM},
+    metrics::{FAILED_RECEIPT, HANDLER_FAILURE, HANDLER_HISTOGRAM},
     middleware::{
-        allocation_middleware, deployment_middleware, labels_middleware, receipt_middleware,
+        allocation_middleware,
+        auth::{self, Bearer, OrExt},
+        context_middleware, deployment_middleware, labels_middleware, receipt_middleware,
         sender_middleware, AllocationState, PrometheusMetricsMiddlewareLayer, SenderState,
     },
     routes::{health, request_handler, static_subgraph_request_handler},
@@ -96,7 +99,6 @@ pub struct IndexerServiceOptions {
 pub struct IndexerServiceState {
     pub config: Config,
     pub attestation_signers: Receiver<HashMap<Address, AttestationSigner>>,
-    pub tap_manager: Manager<IndexerTapContext>,
     pub service_impl: SubgraphService,
 }
 
@@ -266,16 +268,15 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
     )
     .await;
 
-    let tap_manager = Manager::new(
+    let tap_manager = Box::leak(Box::new(Manager::new(
         domain_separator.clone(),
         indexer_context,
         CheckList::new(checks),
-    );
+    )));
 
     let state = Arc::new(IndexerServiceState {
         config: options.config.clone(),
         attestation_signers,
-        tap_manager,
         service_impl: options.service_impl,
     });
 
@@ -361,6 +362,22 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
 
     misc_routes = misc_routes.with_state(state.clone());
 
+    let mut request_handler_route = post(request_handler);
+
+    // inject auth
+    let failed_receipt_metric = Box::leak(Box::new(FAILED_RECEIPT.clone()));
+    let tap_auth = auth::tap_receipt_authorize(tap_manager, failed_receipt_metric);
+
+    if let Some(free_auth_token) = &options.config.service.serve_auth_token {
+        let free_query = Bearer::new(free_auth_token);
+        let result = free_query.or(tap_auth);
+        let auth_layer = AsyncRequireAuthorizationLayer::new(result);
+        request_handler_route = request_handler_route.layer(auth_layer);
+    } else {
+        let auth_layer = AsyncRequireAuthorizationLayer::new(tap_auth);
+        request_handler_route = request_handler_route.layer(auth_layer);
+    }
+
     let deployment_to_allocation = deployment_to_allocation(allocations);
     let allocation_state = AllocationState {
         deployment_to_allocation,
@@ -385,15 +402,19 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
         .layer(PrometheusMetricsMiddlewareLayer::new(
             HANDLER_HISTOGRAM.clone(),
             HANDLER_FAILURE.clone(),
-        ));
+        ))
+        // tap context
+        .layer(from_fn(context_middleware));
+
+    request_handler_route = request_handler_route.layer(service_builder);
 
     let data_routes = Router::new()
         .route(
             PathBuf::from(&options.config.service.url_prefix)
                 .join(format!("{}/id/:id", options.url_namespace))
                 .to_str()
                 .expect("Failed to set up `/{url_namespace}/id/:id` route"),
-            post(request_handler).route_layer(service_builder),
+            request_handler_route,
         )
         .with_state(state.clone());