crafty-4/app/classes/web/api_handler.py

import logging
import re

from app.classes.web.base_handler import BaseHandler

logger = logging.getLogger(__name__)
bearer_pattern = re.compile(r'^Bearer', flags=re.IGNORECASE)

class ApiHandler(BaseHandler):

    def return_response(self, status: int, data: dict):
        # Define a standardized response
        self.set_status(status)
        self.write(data)

    def access_denied(self, user, reason=''):
        if reason:
            reason = ' because ' + reason
        logger.info("User %s from IP %s was denied access to the API route " + self.request.path + reason, user, self.get_remote_ip())
        self.finish(self.return_response(403, {
            'error':'ACCESS_DENIED',
            'info':'You were denied access to the requested resource'
        }))

    def authenticate_user(self) -> bool:
        try:
            logger.debug("Searching for specified token")

            api_token = self.get_argument('token', '')
            if api_token is None and self.request.headers.get('Authorization'):
                api_token = bearer_pattern.sub('', self.request.headers.get('Authorization'))
            elif api_token is None:
                api_token = self.get_cookie('token')
            user_data = self.controller.users.get_user_by_api_token(api_token)

            logger.debug("Checking results")
            if user_data:
                # Login successful! Check perms
                logger.info(f"User {user_data['username']} has authenticated to API")
                # TODO: Role check

                return True # This is to set the "authenticated"
            else:
                logging.debug("Auth unsuccessful")
                self.access_denied("unknown", "the user provided an invalid token")
                return False
        except Exception as e:
            logger.warning("An error occured while authenticating an API user: %s", e)
            self.finish(self.return_response(403, {
                'error':'ACCESS_DENIED',
                'info':'An error occured while authenticating the user'
            }))
            return False


class ServersStats(ApiHandler):
    def get(self):
        """Get details about all servers"""
        authenticated = self.authenticate_user()
        if not authenticated:
            return

        # Get server stats
        # TODO Check perms
        self.finish(self.write({"servers": self.controller.stats.get_servers_stats()}))


class NodeStats(ApiHandler):
    def get(self):
        """Get stats for particular node"""
        authenticated = self.authenticate_user()
        if not authenticated:
            return

        # Get node stats
        node_stats = self.controller.stats.get_node_stats()
        node_stats.pop("servers")
        self.finish(self.write(node_stats))