Logs middleware

.github/workflows/test.yml

@@ -2,6 +2,10 @@ name: test
 
 on: push
 
+concurrency:
+  group: ${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   lint:
     runs-on: ubuntu-latest

examples/app.py

@@ -5,6 +5,7 @@
 from fastapi import FastAPI
 
 from fast_agave.middlewares import FastAgaveErrorHandler
+from fast_agave.middlewares.loggers import OpenSearchLog
 from .resources import app as resources
 from .middlewares import AuthedMiddleware
 from .tasks.task_example import dummy_task
@@ -13,7 +14,9 @@
 app = FastAPI(title='example')
 app.include_router(resources)
 
+
 app.add_middleware(AuthedMiddleware)
+app.add_middleware(OpenSearchLog)
 app.add_middleware(FastAgaveErrorHandler)
 
 

examples/models/cards.py

@@ -1,4 +1,4 @@
-from mongoengine import DateTimeField, StringField
+from mongoengine import DateTimeField, StringField, IntField
 from mongoengine_plus.aio import AsyncDocument
 from mongoengine_plus.models import BaseModel
 
@@ -8,5 +8,9 @@
 class Card(BaseModel, AsyncDocument):
     id = StringField(primary_key=True, default=uuid_field('CA'))
     number = StringField(required=True)
+    cvv = StringField(default='111')
+    exp_year = IntField(default=2040)
     user_id = StringField(required=True)
+    status = StringField(default='active')
     created_at = DateTimeField()
+    deactivated_at = DateTimeField()

examples/resources/base.py

@@ -8,6 +8,11 @@
 app = RestApiBlueprint()
 
 
+@app.get('/hello')
+def hello() -> str:
+    return 'hello!'
+
+
 @app.get('/healthy_auth')
 def health_auth_check() -> Dict:
     return dict(greeting="I'm authenticated and healthy !!!")

examples/resources/cards.py

@@ -1,27 +1,42 @@
+import datetime as dt
 from typing import Dict
+from fastapi import Request
 
 from fastapi.responses import JSONResponse as Response
 from fast_agave.filters import generic_query
 
 from ..models import Card as CardModel
-from ..validators import CardQuery
+from ..validators import CardQuery, CardUpdateRequest
 from .base import app
 
 
 @app.resource('/cards')
 class Card:
     model = CardModel
     query_validator = CardQuery
+    update_validator = CardUpdateRequest
     get_query_filter = generic_query
 
     @staticmethod
     async def retrieve(card: CardModel) -> Response:
         data = card.to_dict()
-        data['number'] = '*' * 16
+        data['last_four_digits'] = card.number[-4:]
         return Response(content=data)
 
     @staticmethod
     async def query(response: Dict):
         for item in response['items']:
-            item['number'] = '*' * 16
+            item['last_four_digits'] = item['number'][-4:]
         return response
+
+    @staticmethod
+    async def update(card: CardModel, request: CardUpdateRequest) -> Response:
+        card.status = request.status
+        await card.async_save()
+        return Response(content=card.to_dict(), status_code=200)
+
+    @staticmethod
+    async def delete(card: CardModel, _: Request) -> Response:
+        card.deactivated_at = dt.datetime.utcnow()
+        await card.async_save()
+        return Response(content=card.to_dict(), status_code=200)

examples/validators.py

@@ -81,6 +81,10 @@ class CardQuery(QueryParams):
     number: Optional[str] = None
 
 
+class CardUpdateRequest(BaseModel):
+    status: str
+
+
 class FileUploadValidator(BaseModel):
     file: bytes
     file_name: str

fast_agave/middlewares/__init__.py

@@ -1,3 +1,4 @@
 from .error_handlers import FastAgaveErrorHandler
+from .loggers import OpenSearchLog
 
-__all__ = ['FastAgaveErrorHandler']
+__all__ = ['FastAgaveErrorHandler', 'OpenSearchLog']

fast_agave/middlewares/loggers.py

@@ -0,0 +1,167 @@
+import asyncio
+import base64
+import binascii
+import json
+import os
+import re
+from dataclasses import asdict
+from typing import Any, Optional
+
+from aiohttp import ClientSession
+from cuenca_validations.errors import CuencaError
+from cuenca_validations.typing import DictStrAny
+from fastapi import Request, Response
+from starlette.datastructures import Headers
+from starlette.middleware.base import (
+    BaseHTTPMiddleware,
+    RequestResponseEndpoint,
+)
+
+from fast_agave.exc import FastAgaveError
+
+AUTHED_REQUIRED_HEADERS = {
+    'x-cuenca-token',
+    'x-cuenca-logintoken',
+    'x-cuenca-loginid',
+    'x-cuenca-sessionid',
+}
+
+SENSITIVE_DATA_FIELDS = {
+    'number',
+    'cvv2',
+    'cvv',
+    'icvv',
+    'exp_month',
+    'exp_year',
+    'pin',
+    'pin_block',
+    'pin_block_switch',
+}
+
+NON_VALID_SCOPE_VALUES = [
+    'headers',
+    'app',
+    'extensions',
+    'fastapi_astack',
+    'app',
+    'route_handler',
+    'router',
+    'endpoint',
+    'raw_path',
+]
+
+
+def basic_auth_decode(basic: str) -> str:
+    basic = re.sub(r'^Basic ', '', basic, flags=re.IGNORECASE)
+    try:
+        decoded = base64.b64decode(basic).decode('ascii')
+        ak, _ = decoded.split(':')
+    except (binascii.Error, UnicodeDecodeError, ValueError):
+        # If `basic` value is not valid then it is not a sensitive data
+        return basic
+    return ak
+
+
+def mask_sensitive_headers(headers: Headers) -> DictStrAny:
+    masked_dict = dict()
+    for key, val in headers.items():
+        if key == 'authorization':
+            masked_dict[key] = basic_auth_decode(val)
+        elif key in AUTHED_REQUIRED_HEADERS:
+            masked_dict[key] = val[0:5] + '*' * 5 if val else ''
+        else:
+            masked_dict[key] = val
+    return masked_dict
+
+
+def mask_sensitive_data(data: Any) -> Any:
+    if type(data) is dict:
+        for key, val in data.items():
+            if type(val) is list:
+                data[key] = [mask_sensitive_data(v) for v in val]
+            elif key == 'number':
+                data[key] = '*' * 12 + val[-4:]
+            elif key in SENSITIVE_DATA_FIELDS and type(val) is int:
+                data[key] = '***'
+            elif key in SENSITIVE_DATA_FIELDS:
+                data[key] = '*' * len(key)
+
+    return data
+
+
+class OpenSearchLog(BaseHTTPMiddleware):
+    async def dispatch(
+        self, request: Request, call_next: RequestResponseEndpoint
+    ) -> Response:
+        if not os.environ.get('LOGS_SERVER_URL', ''):
+            return await call_next(request)
+
+        request_data = dict(request.scope)
+
+        request_headers = mask_sensitive_headers(request.headers)
+        request_data['client'] = f'{request.client.host}:{request.client.port}'
+        request_data['query_string'] = str(request.query_params)
+        request_data['server'] = ':'.join(
+            (str(v) for v in request.scope['server'])
+        )
+
+        for value in NON_VALID_SCOPE_VALUES:
+            request_data.pop(value, None)
+
+        response_body = {}
+
+        try:
+            response = await call_next(request)
+        except FastAgaveError as exc:
+            response_body = asdict(exc)
+            raise
+        except CuencaError as exc:
+            response_body = dict(status_code=exc.status_code, code=str(exc))
+            raise
+        else:
+            # El objeto response es de tipo `starlette.responses.StreamingResponse`
+            # por lo que sus datos vienen en binario, se deben obtener manualmente
+            # y crear un custom Request
+            # https://stackoverflow.com/questions/71882419/fastapi-how-to-get-the-response-body-in-middleware
+            binary_response_body = b''
+            async for chunk in response.body_iterator:  # type: ignore
+                binary_response_body += chunk
+
+            response_body = json.loads(binary_response_body.decode())
+            response = Response(
+                content=binary_response_body,
+                status_code=response.status_code,
+                headers=dict(response.headers),
+                media_type=response.media_type,
+            )
+
+            response_body = mask_sensitive_data(response_body)
+        finally:
+            asyncio.create_task(
+                self.send_log_to_open_search(
+                    request.app.title,
+                    request_data,
+                    request_headers,
+                    response_body,
+                )
+            )
+        return response
+
+    @classmethod
+    async def send_log_to_open_search(
+        cls,
+        app_name: str,
+        request_data: DictStrAny,
+        request_headers: DictStrAny,
+        response_body: Optional[DictStrAny] = None,
+    ) -> None:
+        log_server_url = os.environ.get('LOGS_SERVER_URL', '')
+        data = dict(
+            app=app_name,
+            request_data=request_data,
+            request_headers=request_headers,
+            response_body=response_body,
+        )
+        async with ClientSession() as session:
+            async with session.put(log_server_url, json=data):
+                pass

fast_agave/version.py

@@ -1 +1 @@
-__version__ = '0.6.0'
+__version__ = '0.7.0'

requirements.txt

@@ -1,4 +1,5 @@
 aiobotocore==2.1.0
+aiohttp==3.7.4.post0
 cuenca-validations==0.10.7
 fastapi==0.68.2
 mongoengine-plus==0.0.3

setup.cfg

@@ -2,7 +2,7 @@
 test=pytest
 
 [tool:pytest]
-addopts = -p no:warnings -v --cov-report term-missing --cov=fast_agave
+addopts = -p no:warnings -v --cov-report term-missing --cov-report=html --cov=fast_agave
 
 [flake8]
 inline-quotes = '

setup.py

@@ -24,6 +24,7 @@
     python_requires='>=3.8',
     install_requires=[
         'aiobotocore>=2.1.0,<2.2.0',
+        'aiohttp>=3.7.4.post0,<3.8.0',
         'cuenca-validations>=0.9.4,<1.0.0',
         'fastapi>=0.63.0,<0.69.0',
         'mongoengine-plus>=0.0.2,<1.0.0',

tests/blueprint/test_blueprint.py

@@ -87,7 +87,8 @@ def test_update_resource_with_invalid_params(client: TestClient) -> None:
 def test_retrieve_custom_method(client: TestClient, card: Card) -> None:
     resp = client.get(f'/cards/{card.id}')
     assert resp.status_code == 200
-    assert resp.json()['number'] == '*' * 16
+    assert resp.json()['number'] == card.number
+    assert resp.json()['last_four_digits'] == card.number[-4:]
 
 
 def test_update_resource_that_doesnt_exist(client: TestClient) -> None:
@@ -229,13 +230,13 @@ def test_query_custom_method(client: TestClient) -> None:
     json_body = resp.json()
     assert resp.status_code == 200
     assert len(json_body['items']) == 2
-    assert all(card['number'] == '*' * 16 for card in json_body['items'])
+    assert all('last_four_digits' in card for card in json_body['items'])
 
     resp = client.get(json_body['next_page_uri'])
     json_body = resp.json()
     assert resp.status_code == 200
     assert len(json_body['items']) == 2
-    assert all(card['number'] == '*' * 16 for card in json_body['items'])
+    assert all('last_four_digits' in card for card in json_body['items'])
 
 
 def test_cannot_query_resource(client: TestClient) -> None:

tests/helpers.py

@@ -0,0 +1,9 @@
+import base64
+from typing import Dict
+
+
+def auth_header(username: str, password: str = '') -> Dict:
+    creds = base64.b64encode(f'{username}:{password}'.encode('ascii')).decode(
+        'utf-8'
+    )
+    return {'Authorization': f'Basic {creds}'}