chore: align

Author: ovr

packages/cubejs-api-gateway/src/SubscriptionServer.ts

@@ -53,7 +53,7 @@ export class SubscriptionServer {
       }
 
       if (message.authorization) {
-        authContext = { isSubscription: true };
+        authContext = { isSubscription: true, protocol: 'ws' };
         await this.apiGateway.checkAuthFn(authContext, message.authorization);
         const acceptanceResult = await this.contextAcceptor(authContext);
         if (!acceptanceResult.accepted) {

packages/cubejs-api-gateway/test/auth.test.ts

@@ -30,11 +30,9 @@ class ApiGatewayOpenAPI extends ApiGateway {
   public async shutdownSQLServer(): Promise<void> {
     try {
       await this.sqlServer.shutdown('fast');
-    } catch (error) {
-      console.log(`Error while shutting down server: ${error}`);
+    } finally {
+      this.isRunning = null;
     }
-
-    this.isRunning = null;
   }
 }
 
@@ -110,10 +108,15 @@ describe('test authorization with native gateway', () => {
   });
 
   afterAll(async () => {
-    await apiGateway.shutdownSQLServer();
+    try {
+      await apiGateway.shutdownSQLServer();
+    } catch (error) {
+      // TODO: Figure out, why ApiGatewayServer cannot shutdown!?
+      console.log(`Error while shutting down server: ${error}`);
+    }
   });
 
-  it('default authorization', async () => {
+  it('default authorization - success', async () => {
     const token = generateAuthToken({ uid: 5, });
 
     await request(app)
@@ -128,6 +131,35 @@ describe('test authorization with native gateway', () => {
 
     await apiGateway.shutdownSQLServer();
   });
+
+  it('default authorization - wrong secret', async () => {
+    const badToken = generateAuthToken({ uid: 5, }, {}, 'bad');
+
+    await request(app)
+      .get('/cubejs-api/v2/stream')
+      .set('Authorization', `${badToken}`)
+      .expect(403);
+
+    // No bad logs
+    expect(loggerMock.mock.calls.length).toEqual(0);
+    // We should not call js handler, request should go into rust code
+    expect(handlerMock.mock.calls.length).toEqual(0);
+
+    await apiGateway.shutdownSQLServer();
+  });
+
+  it('default authorization - missing auth header', async () => {
+    await request(app)
+      .get('/cubejs-api/v2/stream')
+      .expect(403);
+
+    // No bad logs
+    expect(loggerMock.mock.calls.length).toEqual(0);
+    // We should not call js handler, request should go into rust code
+    expect(handlerMock.mock.calls.length).toEqual(0);
+
+    await apiGateway.shutdownSQLServer();
+  });
 });
 
 describe('test authorization', () => {

packages/cubejs-backend-native/js/index.ts

@@ -43,8 +43,10 @@ export interface ContextToApiScopesPayload {
 
 export type ContextToApiScopesResponse = string[];
 
+export interface CheckAuthPayloadRequestMeta extends BaseMeta {}
+
 export interface CheckAuthPayload {
-  request: Request<undefined>,
+  request: Request<CheckAuthPayloadRequestMeta>,
   token: string,
 }
 

packages/cubejs-backend-native/src/auth.rs

@@ -121,8 +121,7 @@ impl SqlAuthService for NodeBridgeAuthService {
 
 #[derive(Debug, Serialize)]
 struct CheckAuthTransportRequest {
-    request: TransportRequest,
-    req: GatewayCheckAuthRequest,
+    request: GatewayCheckAuthRequest,
     token: String,
 }
 
@@ -154,19 +153,13 @@ type ContextToApiScopesTransportResponse = Vec<String>;
 impl GatewayAuthService for NodeBridgeAuthService {
     async fn authenticate(
         &self,
-        req: GatewayCheckAuthRequest,
+        request: GatewayCheckAuthRequest,
         token: String,
     ) -> Result<GatewayAuthenticateResponse, CubeError> {
         trace!("[auth] Request ->");
 
-        let request_id = Uuid::new_v4().to_string();
-
         let extra = serde_json::to_string(&CheckAuthTransportRequest {
-            request: TransportRequest {
-                id: format!("{}-span-1", request_id),
-                meta: None,
-            },
-            req,
+            request,
             token: token.clone(),
         })?;
         let response: CheckAuthTransportResponse = call_js_with_channel_as_callback(

packages/cubejs-backend-native/src/gateway/auth_middleware.rs

@@ -2,6 +2,7 @@ use crate::gateway::http_error::HttpError;
 use crate::gateway::state::ApiGatewayStateRef;
 use crate::gateway::{GatewayAuthContextRef, GatewayAuthService, GatewayCheckAuthRequest};
 use axum::extract::State;
+use axum::http::HeaderValue;
 use axum::response::IntoResponse;
 
 #[derive(Debug, Clone)]
@@ -15,6 +16,18 @@ impl AuthExtension {
     }
 }
 
+fn parse_token(header_value: &HeaderValue) -> Result<&str, HttpError> {
+    let trimmed = header_value.to_str()?.trim();
+
+    if let Some(stripped) = trimmed.strip_prefix("Bearer ") {
+        Ok(stripped)
+    } else if let Some(stripped) = trimmed.strip_prefix("bearer ") {
+        Ok(stripped)
+    } else {
+        Ok(trimmed)
+    }
+}
+
 pub async fn gateway_auth_middleware(
     State(state): State<ApiGatewayStateRef>,
     mut req: axum::extract::Request,
@@ -26,6 +39,8 @@ pub async fn gateway_auth_middleware(
         ));
     };
 
+    let bearer_token = parse_token(token_header_value)?;
+
     let auth = state
         .injector_ref()
         .get_service_typed::<dyn GatewayAuthService>()
@@ -35,7 +50,7 @@ pub async fn gateway_auth_middleware(
         GatewayCheckAuthRequest {
             protocol: "http".to_string(),
         },
-        token_header_value.to_str()?.to_string(),
+        bearer_token.to_string(),
     );
 
     let auth_response = auth_fut

packages/cubejs-backend-native/src/gateway/auth_service.rs

@@ -31,7 +31,7 @@ pub struct GatewayContextToApiScopesResponse {
 pub trait GatewayAuthService: Send + Sync + Debug {
     async fn authenticate(
         &self,
-        request: GatewayCheckAuthRequest,
+        req: GatewayCheckAuthRequest,
         token: String,
     ) -> Result<GatewayAuthenticateResponse, CubeError>;
 

packages/cubejs-backend-native/src/gateway/handlers/stream.rs

@@ -23,7 +23,7 @@ pub async fn stream_handler_v2(
     Ok((
         StatusCode::NOT_IMPLEMENTED,
         Json(HandlerResponse {
-            message: "Not implemented".to_string(),
+            message: "/v2/stream is not implemented".to_string(),
         }),
     ))
 }