notifier/services/auth.py

import logging

from aiohttp import ClientSession
from strawberry.extensions import Extension

from orm.author import Author
from services.db import local_session
from settings import AUTH_URL


logger = logging.getLogger('\t[services.auth]\t')
logger.setLevel(logging.DEBUG)


async def check_auth(req) -> str | None:
    token = req.headers.get('Authorization')
    user_id = ''
    if token:
        query_name = 'validate_jwt_token'
        operation = 'ValidateToken'
        headers = {
            'Content-Type': 'application/json',
        }

        variables = {
            'params': {
                'token_type': 'access_token',
                'token': token,
            }
        }

        gql = {
            'query': f'query {operation}($params: ValidateJWTTokenInput!) {{ {query_name}(params: $params) {{ is_valid claims }} }}',
            'variables': variables,
            'operationName': operation,
        }
        try:
            # Asynchronous HTTP request to the authentication server
            async with ClientSession() as session:
                async with session.post(
                    AUTH_URL, json=gql, headers=headers
                ) as response:
                    print(
                        f'[services.auth] HTTP Response {response.status} {await response.text()}'
                    )
                    if response.status == 200:
                        data = await response.json()
                        errors = data.get('errors')
                        if errors:
                            print(f'errors: {errors}')
                        else:
                            user_id = (
                                data.get('data', {})
                                .get(query_name, {})
                                .get('claims', {})
                                .get('sub')
                            )
                            if user_id:
                                print(f'[services.auth] got user_id: {user_id}')
                                return user_id
        except Exception as e:
            import traceback

            traceback.print_exc()
            # Handling and logging exceptions during authentication check
            print(f'[services.auth] Error {e}')

    return None


class LoginRequiredMiddleware(Extension):
    async def on_request_start(self):
        context = self.execution_context.context
        req = context.get('request')
        user_id = await check_auth(req)
        if user_id:
            context['user_id'] = user_id.strip()
            with local_session() as session:
                author = session.query(Author).filter(Author.user == user_id).first()
                if author:
                    context['author_id'] = author.id
        context['user_id'] = user_id or None

        self.execution_context.context = context