1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / bakauth /
Filename : sess.py
back
"""Handles automatic retries for Backup Authority API calls"""

from typing import Union, Any
import logging
import enum
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests

DEFAULT_TIMEOUT = 20
DEFAULT_RETRIES = 2

logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)


class Status(enum.Enum):
    """Backup Authority Error Codes"""

    REQUEST_EXCEPTION = -1  # error contacting backup authority
    OKAY = 0  # normal response
    ERROR = 1  # general error
    ARGS_MISSING = 2  # incorrect api usage, missing arguments
    UNHANDLED_EXCEPTION = 3  # exception raised on bakauth server (a bug)
    SQLALCHEMY_ERROR = 4  # SQL error raised. Possibly a MySQL timeout
    AMP_DOWN = 5  # PowerPanel API is down
    WRONG_SERVER_CLASS = 6  # Wrong API endpoint for registered server class
    AUTH_FAILED = 7  # API login failed
    INTERNAL_NO_QUOTA = 8  # requested quota on an internal server
    PRODUCTION_ONLY = 9  # function does not work on testing bakauth server
    VDED_AMP_MISSING = 10  # could not locate amp account for a v/ded server
    LOOKUP_MISSING = 11  # requested data not found in bakauth database
    AUTH_MIGRATED = 12  # ceph bucket was moved to a new server (dedi)
    NO_REPLICANT = 13  # function does not work on the bakauth replicant
    INTERNAL_VPS = 14  # access denied to S3 creds for an internal VPS (vznode)


class MdsState(enum.Enum):
    """Possible states for vznode MDS backups"""

    OKAY = 0
    WARN = 1
    CRIT = 2


class LoggedRetry(Retry):
    """Logs on HTTP retries"""

    def increment(
        self,
        method=None,
        url=None,
        response=None,
        error=None,
        _pool=None,
        _stacktrace=None,
    ):
        if error is not None:
            logging.debug('bakauth::%s: %s', url, error)
        return super().increment(
            method, url, response, error, _pool, _stacktrace
        )


class BakAuthSession(requests.Session):
    """Custom requests.Session for bakauth requests"""

    def __init__(self, retries: int, log_retries: bool):
        super().__init__()
        retry_class = LoggedRetry if log_retries else Retry
        self.mount(
            'https://',
            HTTPAdapter(
                max_retries=retry_class(
                    total=retries,
                    read=retries,
                    connect=retries,
                    status=retries,
                    status_forcelist=[500],
                    backoff_factor=1.0,
                )
            ),
        )


def post(
    *,
    bakauth_host: str,
    uri: str,
    data: dict,
    timeout: int,
    retries: int,
    auth: Union[tuple, None] = None,
    log_retries: bool = True,
) -> tuple[Status, Any]:
    """Performs a Backup Authority API request

    Args:
        bakauth_host (str): bakauth server fqdn
        uri (str): endpoint uri to use
        data (dict): POST form data
        timeout (int): seconds before raising BakAuthDown
        retries (int): number of automatic retries
        auth (tuple | None): HTTP auth info

    Returns:
        (Status, data): Status response (enum) and returned data
    """
    try:
        with BakAuthSession(retries, log_retries) as session:
            raw = session.post(
                url=f"https://{bakauth_host}:4002/{uri.lstrip('/')}",
                auth=auth,
                timeout=timeout,
                data=data,
            )
    except requests.exceptions.RequestException as exc:
        try:
            url = exc.request.url
        except AttributeError:
            return Status.REQUEST_EXCEPTION, str(exc)
        return Status.REQUEST_EXCEPTION, f"{url}: {exc}"
    try:
        ret = raw.json()
    except ValueError:
        return Status.REQUEST_EXCEPTION, f'Invalid JSON from auth server: {raw}'
    try:
        status = Status(ret['status'])
    except ValueError:
        status = Status.ERROR
    return status, ret['data']