List call_requests check auth with dependency

engineapi/engineapi/contracts_actions.py

@@ -430,7 +430,7 @@ def create_request_calls(
     return len(call_specs)
 
 
-def get_call_requests(
+def get_call_request(
     db_session: Session,
     request_id: uuid.UUID,
 ) -> Tuple[CallRequest, RegisteredContract]:

engineapi/engineapi/middleware.py

@@ -6,8 +6,9 @@ from uuid import UUID
 
 from bugout.data import BugoutResource, BugoutResources, BugoutUser
 from bugout.exceptions import BugoutResponseException
-from fastapi import HTTPException, Request, Response
+from fastapi import Header, HTTPException, Request, Response
 from pydantic import AnyHttpUrl, parse_obj_as
+from starlette.datastructures import Headers
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.middleware.cors import CORSMiddleware
 from starlette.responses import Response
@@ -34,6 +35,89 @@ from .settings import bugout_client as bc
 logger = logging.getLogger(__name__)
 
 
+class InvalidAuthHeaderFormat(Exception):
+    """
+    Raised when authorization header not pass validation.
+    """
+
+
+class BugoutUnverifiedAuth(Exception):
+    """
+    Raised when attempted access by unverified Brood account.
+    """
+
+
+class BugoutAuthWrongApp(Exception):
+    """
+    Raised when user does not belong to this application.
+    """
+
+
+def parse_auth_header(auth_header: str) -> Tuple[str, str]:
+    """
+    Returns: auth_format and user_token passed in authorization header.
+    """
+    auth_list = auth_header.split()
+    if len(auth_list) != 2:
+        raise InvalidAuthHeaderFormat("Wrong authorization header")
+
+    return auth_list[0], auth_list[1]
+
+
+def bugout_auth(token: str) -> BugoutUser:
+    """
+    Extended bugout.get_user with additional checks.
+    """
+    user: BugoutUser = bc.get_user(token)
+    if not user.verified:
+        raise BugoutUnverifiedAuth("Only verified accounts can have access")
+    if str(user.application_id) != str(MOONSTREAM_APPLICATION_ID):
+        raise BugoutAuthWrongApp("User does not belong to this application")
+
+    return user
+
+
+def user_for_auth_header(
+    authorization: str = Header(None),
+) -> Optional[BugoutUser]:
+    """
+    Fetch Bugout user if authorization token provided.
+    """
+    user: Optional[BugoutUser] = None
+    if authorization is not None:
+        user_token: str = ""
+        try:
+            _, user_token = parse_auth_header(auth_header=authorization)
+        except InvalidAuthHeaderFormat:
+            raise EngineHTTPException(
+                status_code=403, detail="Wrong authorization header"
+            )
+        except Exception as e:
+            logger.error(f"Error processing Brood response: {str(e)}")
+            raise EngineHTTPException(status_code=500, detail="Internal server error")
+
+        if user_token != "":
+            try:
+                user: BugoutUser = bugout_auth(token=user_token)
+            except BugoutUnverifiedAuth:
+                logger.info(f"Attempted access by unverified Brood account: {user.id}")
+                raise EngineHTTPException(
+                    status_code=403,
+                    detail="Only verified accounts can have access",
+                )
+            except BugoutAuthWrongApp:
+                raise EngineHTTPException(
+                    status_code=403, detail="User does not belong to this application"
+                )
+            except BugoutResponseException as e:
+                raise HTTPException(status_code=e.status_code, detail=e.detail)
+            except Exception as e:
+                logger.error(f"Error processing Brood response: {str(e)}")
+                raise HTTPException(status_code=500, detail="Internal server error")
+
+    return user
+
+
 class BroodAuthMiddleware(BaseHTTPMiddleware):
     """
     Checks the authorization header on the request. If it represents a verified Brood user,
@@ -59,25 +143,30 @@ class BroodAuthMiddleware(BaseHTTPMiddleware):
         if path in self.whitelist.keys() and self.whitelist[path] == method:
             return await call_next(request)
 
-        authorization_header = request.headers.get("authorization")
+        authorization = request.headers.get("authorization")
-        if authorization_header is None:
+        if authorization is None:
             return Response(
-                status_code=403, content="No authorization header passed with request"
+                status_code=403,
+                content="No authorization header passed with request",
             )
-        user_token_list = authorization_header.split()
-        if len(user_token_list) != 2:
-            return Response(status_code=403, content="Wrong authorization header")
-        user_token: str = user_token_list[-1]
 
         try:
-            user: BugoutUser = bc.get_user(user_token)
+            _, user_token = parse_auth_header(auth_header=authorization)
-            if not user.verified:
+        except InvalidAuthHeaderFormat:
+            return Response(status_code=403, content="Wrong authorization header")
+        except Exception as e:
+            logger.error(f"Error processing Brood response: {str(e)}")
+            return Response(status_code=500, content="Internal server error")
+
+        try:
+            user: BugoutUser = bugout_auth(token=user_token)
+        except BugoutUnverifiedAuth:
             logger.info(f"Attempted access by unverified Brood account: {user.id}")
             return Response(
                 status_code=403,
                 content="Only verified accounts can have access",
             )
-            if str(user.application_id) != str(MOONSTREAM_APPLICATION_ID):
+        except BugoutAuthWrongApp:
             return Response(
                 status_code=403, content="User does not belong to this application"
             )

engineapi/engineapi/routes/metatx.py

@@ -9,8 +9,7 @@ import logging
 from typing import Dict, List, Optional
 from uuid import UUID
 
-from bugout.data import BugoutResource, BugoutResources, BugoutUser
+from bugout.data import BugoutUser
-from bugout.exceptions import BugoutResponseException
 from fastapi import Body, Depends, FastAPI, Path, Query, Request
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import Session
@@ -20,9 +19,9 @@ from ..middleware import (
     BroodAuthMiddleware,
     BugoutCORSMiddleware,
     EngineHTTPException,
+    user_for_auth_header,
 )
-from ..settings import DOCS_TARGET_PATH, MOONSTREAM_APPLICATION_ID
+from ..settings import DOCS_TARGET_PATH
-from ..settings import bugout_client as bc
 from ..version import VERSION
 
 logger = logging.getLogger(__name__)
@@ -285,9 +284,12 @@ async def call_request_types_route(
     return call_request_types
 
 
-@app.get("/requests", tags=["requests"], response_model=List[data.CallRequestResponse])
+@app.get(
+    "/requests",
+    tags=["requests"],
+    response_model=List[data.CallRequestResponse],
+)
 async def list_requests_route(
-    request: Request,
     contract_id: Optional[UUID] = Query(None),
     contract_address: Optional[str] = Query(None),
     caller: str = Query(...),
@@ -295,6 +297,7 @@ async def list_requests_route(
     offset: Optional[int] = Query(None),
     show_expired: bool = Query(False),
     show_before_live_at: bool = Query(False),
+    user: Optional[BugoutUser] = Depends(user_for_auth_header),
     db_session: Session = Depends(db.yield_db_read_only_session),
 ) -> List[data.CallRequestResponse]:
     """
@@ -302,33 +305,6 @@ async def list_requests_route(
 
     At least one of `contract_id` or `contract_address` must be provided as query parameters.
     """
-    authorization_header = request.headers.get("authorization")
-    user: Optional[BugoutUser] = None
-    if authorization_header is not None:
-        try:
-            auth_list = authorization_header.split()
-            if len(auth_list) != 2:
-                return EngineHTTPException(
-                    status_code=403, content="Wrong authorization header"
-                )
-
-            user = bc.get_user(auth_list[-1])
-            if not user.verified:
-                logger.info(f"Attempted access by unverified Brood account: {user.id}")
-                return EngineHTTPException(
-                    status_code=403,
-                    content="Only verified accounts can have access",
-                )
-            if str(user.application_id) != str(MOONSTREAM_APPLICATION_ID):
-                return EngineHTTPException(
-                    status_code=403, content="User does not belong to this application"
-                )
-        except BugoutResponseException as e:
-            return EngineHTTPException(status_code=e.status_code, content=e.detail)
-        except Exception as e:
-            logger.error(f"Error processing Brood response: {str(e)}")
-            return EngineHTTPException(status_code=500, content="Internal server error")
-
     try:
         requests = contracts_actions.list_call_requests(
             db_session=db_session,
@@ -364,7 +340,7 @@ async def get_request(
     At least one of `contract_id` or `contract_address` must be provided as query parameters.
     """
     try:
-        request = contracts_actions.get_call_requests(
+        request = contracts_actions.get_call_request(
             db_session=db_session,
             request_id=request_id,
         )