glances_restful_api.py « outputs « glances - glances - Mirror of https://github.com/nicolargo/glances

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2023 Nicolas Hennion <nicolas@nicolargo.com>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""RestFull API interface class."""

import os
import sys
import tempfile
from io import open
import webbrowser
import socket
from urllib.parse import urljoin
# Replace typing_extensions by typing when Python 3.8 support will be dropped
# from typing import Annotated
from typing_extensions import Annotated

from glances import __version__
from glances.password import GlancesPassword
from glances.timer import Timer
from glances.logger import logger

# FastAPI import
try:
    from fastapi import FastAPI, Depends, HTTPException, status, APIRouter, Request
    from fastapi.security import HTTPBasic, HTTPBasicCredentials
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.middleware.gzip import GZipMiddleware
    from fastapi.responses import HTMLResponse, ORJSONResponse
    from fastapi.templating import Jinja2Templates
    from fastapi.staticfiles import StaticFiles
except ImportError:
    logger.critical('FastAPI import error. Glances cannot start in web server mode.')
    sys.exit(2)

try:
    import uvicorn
except ImportError:
    logger.critical('Uvicorn import error. Glances cannot start in web server mode.')
    sys.exit(2)

security = HTTPBasic()


class GlancesRestfulApi(object):
    """This class manages the Restful API server."""

    API_VERSION = '4'

    def __init__(self, config=None, args=None):
        # Init config
        self.config = config

        # Init args
        self.args = args

        # Init stats
        # Will be updated within Bottle route
        self.stats = None

        # cached_time is the minimum time interval between stats updates
        # i.e. HTTP/RESTful calls will not retrieve updated info until the time
        # since last update is passed (will retrieve old cached info instead)
        self.timer = Timer(0)

        # Load configuration file
        self.load_config(config)

        # Set the bind URL
        self.bind_url = urljoin('http://{}:{}/'.format(self.args.bind_address,
                                                       self.args.port),
                                self.url_prefix)

        # FastAPI Init
        if self.args.password:
            self._app = FastAPI(dependencies=[Depends(self.authentication)])
            self._password = GlancesPassword(username=args.username,
                                             config=config)

        else:
            self._app = FastAPI()
            self._password = None

        # Change the default root path
        if self.url_prefix != '/':
            self._app.include_router(APIRouter(prefix=self.url_prefix))

        # Set path for WebUI
        self.STATIC_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'static/public')
        self.TEMPLATE_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'static/templates')
        self._templates = Jinja2Templates(directory=self.TEMPLATE_PATH)

        # FastAPI Enable CORS
        # https://fastapi.tiangolo.com/tutorial/cors/
        self._app.add_middleware(
            CORSMiddleware,
            # allow_origins=["*"],
            allow_origins=[
                self.bind_url
            ],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

        # FastAPI Enable GZIP compression
        # https://fastapi.tiangolo.com/advanced/middleware/
        self._app.add_middleware(GZipMiddleware,
                                 minimum_size=1000)

        # FastAPI Define routes
        self._app.include_router(self._router())

    def load_config(self, config):
        """Load the outputs section of the configuration file."""
        # Limit the number of processes to display in the WebUI
        self.url_prefix = '/'
        if config is not None and config.has_section('outputs'):
            n = config.get_value('outputs', 'max_processes_display', default=None)
            logger.debug('Number of processes to display in the WebUI: {}'.format(n))
            self.url_prefix = config.get_value('outputs', 'url_prefix', default='/')
            logger.debug('URL prefix: {}'.format(self.url_prefix))

    def __update__(self):
        # Never update more than 1 time per cached_time
        if self.timer.finished():
            self.stats.update()
            self.timer = Timer(self.args.cached_time)

    def app(self):
        return self._app()

    def authentication(self, creds: Annotated[HTTPBasicCredentials, Depends(security)]):
        """Check if a username/password combination is valid."""
        if creds.username == self.args.username:
            # check_password and get_hash are (lru) cached to optimize the requests
            if self._password.check_password(self.args.password,
                                             self._password.get_hash(creds.password)):
                return creds.username

        # If the username/password combination is invalid, return an HTTP 401
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )

    def _router(self):
        """Define a custom router for Glances path."""
        router = APIRouter()

        # REST API
        router.add_api_route('/api/%s/status' % self.API_VERSION,
                             status_code=status.HTTP_200_OK,
                             response_class=ORJSONResponse,
                             endpoint=self._api_status)

        router.add_api_route('/api/%s/config' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_config)
        router.add_api_route('/api/%s/config/{section}' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_config_section)
        router.add_api_route('/api/%s/config/{section}/{item}' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_config_section_item)

        router.add_api_route('/api/%s/args' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_args)
        router.add_api_route('/api/%s/args/{item}' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_args_item)

        router.add_api_route('/api/%s/pluginslist' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_plugins)
        router.add_api_route('/api/%s/all' % self.API_VERSION,
                             response_class=ORJSONResponse,
                             endpoint=self._api_all)
        router.add_api_route('/api/%s/all/limits' % self.API_VERSION