FM / bakauth

/opt/imh-python/lib/python3.9/site-packages/bakauth/

/opt/imh-python/lib/python3.9/site-packages/bakauth

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
exc.py	3648	0644	edit dl rm
hints.py	1949	0644	edit dl rm
sess.py	3974	0644	edit dl rm
__init__.py	40347	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/bakauth/__init__.py (40347B)

"""Code for accessing the backup authority server api

.. data:: bakauth.AUTH_JSON:

"/opt/backups/etc/auth.json"
    *(str)*

.. data:: bakauth.BAKAUTH1:

"ash-sys-pro-bakauth1.imhadmin.net"
    *(str)*

.. data:: bakauth.BAKAUTH2:

"ash-sys-dev-bakauth2.imhadmin.net"
    *(str)*

.. data:: bakauth.BAKAUTH3:

"lax-sys-pro-bakauth3.imhadmin.net"
    *(str)*

.. data:: bakauth.SHARED_CLASSES:

{'imh_reseller', 'imh_shared', 'hub_shared'}
    *(set)*
"""

import sys
from typing import Union, Any
import functools
import logging
import random
import json
import time
import platform
import distro
import rads
from restic import Restic, ResticRepo
from .hints import (
    SharedFailoverLocks,
    AgentClientLookup,
    AgentCpuserLookup,
    VznodeBackupLookup,
    VznodeRestoreLookup,
    UserBuckets,
    RegDetails,
)
from .sess import Status, MdsState, post, DEFAULT_TIMEOUT, DEFAULT_RETRIES
from .exc import (
    BakAuthError,
    BakAuthDown,
    ClusterUnavailable,
    AMPDownError,
    WrongServerClass,
    BakAuthLoginFailed,
    BakAuthWrongLogin,
    VpsRestricted,
    LookupMissing,
    WrongSharedServer,
    NoAmpAccount,
    DedicatedMoved,
    InternalQuota,
    Unregistered,
)

__version__ = '1.5.5'

BAKAUTH1 = 'ash-sys-pro-bakauth1.imhadmin.net'  # prod main
BAKAUTH2 = 'ash-sys-dev-bakauth2.imhadmin.net'  # testing
BAKAUTH3 = 'lax-sys-pro-bakauth3.imhadmin.net'  # prod replicant

AUTH_JSON = '/opt/backups/etc/auth.json'

SHARED_CLASSES = {'imh_reseller', 'imh_shared', 'hub_shared'}

class BakAuth:
    """Handles backup authority requests"""

def __init__(self):
        try:
            with open(AUTH_JSON, encoding='utf-8') as handle:
                data = json.load(handle)
        except FileNotFoundError as exc:
            raise Unregistered(str(exc)) from exc
        self._post = functools.partial(
            post, auth=(data['apiuser'], data['authkey'])
        )

def _post_main(
        self,
        *,
        uri: str,
        timeout: int,
        retries: int,
        log_retries: bool = True,
        **data,
    ) -> tuple[Status, Any]:
        """Perform a post request that only the primary bakauth server can
        handle

Args:
            uri (str): HTTP request URI
            timeout (int): HTTP request timeout in seconds
            retries (int): HTTP request auto-retries after timeout
            log_retries (bool): whether to log on auto-retries. Defaults False
            **data: POST form data

Returns:
            tuple[Status, Any]: (``Status`` enum, data)
        """
        return self._post(
            bakauth_host=BAKAUTH1,
            uri=uri,
            timeout=timeout,
            retries=retries,
            log_retries=log_retries,
            data=data,
        )

def _post_pref(
        self,
        *,
        uri: str,
        timeout: int,
        retries: int,
        log_retries: bool = True,
        pref_main: bool = True,
        **data,
    ) -> tuple[Status, Any, str]:
        """Perform a post request that the primary bakauth server should handle,
        but will failover to a replicant bakauth server if needed

Returns:
            tuple[Status, Any, str]: (``Status`` enum, data, bakauth host used)
        """
        kwargs = {
            'uri': uri,
            'timeout': timeout,
            'retries': retries,
            'log_retries': log_retries,
            'data': data,
        }
        if pref_main:
            bakauth_hosts = [BAKAUTH1, BAKAUTH3]
        else:
            bakauth_hosts = [BAKAUTH3, BAKAUTH1]
        status, data = self._post(bakauth_host=bakauth_hosts[0], **kwargs)
        if status is Status.REQUEST_EXCEPTION:
            if log_retries:
                logging.warning(
                    '%s::%s: request exception - retrying using %s',
                    bakauth_hosts[0],
                    uri,
                    bakauth_hosts[1],
                )
            ret = self._post(bakauth_host=bakauth_hosts[1], **kwargs)
            return *ret, bakauth_hosts[1]
        return status, data, bakauth_hosts[0]

def _post_either(
        self,
        *,
        uri: str,
        timeout: int,
        retries: int,
        log_retries: bool = True,
        **data,
    ) -> tuple[Status, Any]:
        """Perform a post request that any production bakauth server can
        handle (round robin)

Returns:
            tuple[Status, Any]: (``Status`` enum, data)
        """
        prio = [BAKAUTH1, BAKAUTH3]
        random.shuffle(prio)
        kwargs = {
            'uri': uri,
            'timeout': timeout,
            'retries': retries,
            'log_retries': log_retries,
            'data': data,
        }
        status, data = self._post(bakauth_host=prio[0], **kwargs)
        if status is Status.REQUEST_EXCEPTION:
            if log_retries:
                logging.warning(
                    '%s:%s: request exception - retrying using %s',
                    prio[0],
                    uri,
                    prio[1],
                )
            return self._post(bakauth_host=prio[1], **kwargs)
        return status, data

def task_wait(
        self,
        task_id: str,
        *,
        wait_mins: int = 240,
        poll_secs: int = 3,
        bakauth_host: str = BAKAUTH1,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_TIMEOUT,
    ) -> None:
        """Wait for a celery task to finish on the Backup Authority server

Args:
            task_id (str): task identifier string
            wait_mins (int, optional): max minutes to wait. Defaults to 240.
            poll_secs (int, optional): secondss between checking the state of
                the task. Defaults to 3.
            timeout (int, optional): request timeout in secs
                (per poll request, not in total)
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: error checking the state of the task
        """
        state_msg = 'QUEUED'
        wait = wait_mins * 60
        start = time.time()
        while state_msg in ('QUEUED', 'STARTED') and time.time() - start < wait:
            time.sleep(poll_secs)
            status, state_msg = self._post(
                bakauth_host=bakauth_host,
                uri='/lookup/check_task',
                timeout=timeout,
                retries=retries,
                log_retries=True,
                data={'task_id': task_id},
            )
            if status is not Status.OKAY:
                raise BakAuthError(status=status, data=state_msg)

@staticmethod
    def register(
        *,
        svr_class: str,
        host: str,
        require_amp: bool = False,
    ) -> dict[str, str]:
        """Register a (non-internal) server in backup authority

Args:
            svr_class (str): server classification, one in:
                imh_vps, imh_reseller, imh_shared, hub_shared, or imh_ded
            host (str): short hostname which should match what AMP knows
            require_amp (bool): if True, instruct bakauth to reject registration
                if the supplied host does not match a known AMP account

Raises:
            BakAuthError: any error registering the server

Returns:
            dict[str, str]: contains keys "authkey" and "apiuser"
        """
        status, data = post(
            bakauth_host=BAKAUTH1,
            uri='/register',
            timeout=DEFAULT_TIMEOUT,
            retries=DEFAULT_RETRIES,
            log_retries=True,
            data={
                'host': host,
                'svr_class': svr_class,
                'require_amp': '1' if require_amp else '0',
            },
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        return data

def vzclient_backup(
        self,
        *,
        veids: dict[int, str],
        net: Union[str, None] = None,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> VznodeBackupLookup:
        """Requests information form bakauth needed to backup a vz node

Args:
            veids (dict[int, str]): If performing a vps backup run, provide the
                IDs of all vps found on this compute node, mapped to their
                FQDNs. If only backing up mds, send an empty dict
            net (str | None): set "lan" or "wan" to override which
                ceph network to use
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: error looking up data

Returns:
            VznodeBackupLookup: restic info needed to backup a vznode.
            Contains keys "endpoints", "node_keys", and "vps_keys"
        """
        status, data = self._post_main(
            uri='/vzclient/backup',
            veids=json.dumps(veids),
            net=net,
            timeout=timeout,
            retries=retries,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        for task_id in data['task_ids']:
            try:
                self.task_wait(task_id, timeout=timeout, retries=retries)
            except BakAuthError as exc:
                logging.warning('%s', exc)
        return {
            'endpoints': data['endpoints'],
            'node_keys': ResticRepo(**data['node_keys']),
            'vps_keys': {
                int(k): ResticRepo(**v) for k, v in data['vps_keys'].items()
            },
            'changed': {int(k): v for k, v in data['changed'].items()},
        }

def vzclient_restore(
        self,
        veid: Union[int, None],
        *,
        net: Union[str, None] = None,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> VznodeRestoreLookup:
        """Requests information from bakauth needed to restore data for a vps
        from a vznode

Args:
            veid (int | None): ID for the vps you're trying to restore.
                Explicitly set to None to fetch info for the node itself.
            net (str | None): set "lan" or "wan" to override which
                ceph network to use
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthDown: if bakauth could not be reached
            BakAuthLoginFailed: API user/password is wrong
            LookupMissing: if a requested vps's keys do not exist
            VpsRestricted: if a requested vps is internal
            BakAuthError: catch-all for any other api error

Returns:
            VznodeRestoreLookup: restic info needed to restore a vps from a
            vznode. The dict has keys "this_endpoint", "key_info",
            and "all_endpoints"
        """
        if veid is None:
            kwargs = {'node': '1'}
        else:
            kwargs = {'veid': veid}
        status, data = self._post_either(
            uri='/vzclient/restore',
            net=net,
            timeout=timeout,
            retries=retries,
            **kwargs,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        data['key_info'] = ResticRepo(**data['key_info'])
        return data

def get_shared_failover_locks(
        self,
        user: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> SharedFailoverLocks:
        """Check which failover backups should not be rotated"""
        status, data, _ = self._post_pref(
            uri='/failover/get_shared_locks',
            timeout=timeout,
            retries=retries,
            pref_main=False,
            user=user,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        return data

def monitoring_vz_update(
        self,
        *,
        version: str,
        mds_state: MdsState,
        vps_crit: int,
        vps_warn: int,
        vps_sizes: dict[int, int],
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> None:
        """Post backup monitoring status for a vznode

Args:
            version (str): RPM version
            mds_state (MdsState): MDS status enum
            vps_crit (int): number of vps old enough to be a nrpe critical
            vps_warn (int): number of vps behind schedule but not critical
            vps_sizes (dict[int, int]): veids mapped to VPS sizes in MiB
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: error updating monitoring status
        """
        py_version = sys.version_info
        status, data = self._post_main(
            uri='/monitoring/vz_update',
            version=version,
            mds_state=mds_state.value,
            vps_crit=vps_crit,
            vps_warn=vps_warn,
            vps_sizes=json.dumps(vps_sizes),
            os_info=json.dumps(distro.info()),
            py_info=f'{py_version[0]}.{py_version[1]}',
            timeout=timeout,
            retries=retries,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def monitoring_update(
        self,
        *,
        version: str,
        running: bool,
        ded_moved: bool,
        errors: list[str],
        num_old: int,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> None:
        """Updates bakauth with client server status

Args:
            version (str): backup client RPM version
            running (bool): whether the backup runner daemon is running
            ded_moved (bool): True if this is a dedi and its bucket was moved
                to a new server
            errors (list[str]): error messages to display in monitoring dash
            num_old (int): number of old tasks in queue
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: error updating monitoring status
        """
        py_version = sys.version_info
        status, data = self._post_main(
            uri='/monitoring/update',
            timeout=timeout,
            retries=retries,
            version=version,
            running=1 if running else 0,
            ded_moved=1 if ded_moved else 0,
            errors=json.dumps(errors),
            num_old=num_old,
            os_info=json.dumps(distro.info()),
            py_info=f'{py_version[0]}.{py_version[1]}',
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def note_auto_ticket(
        self,
        *,
        task: str,
        plugin: str,
        user: str,
        ipaddr: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> None:
        """Add an account note when a ticket is generated

Args:
            task (str): backup/restore task name
            plugin (str): plugin task name (e.g. "cPanel Backup Manager")
            user (str): username
            ipaddr (str): IP address of user
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: error posting note
        """
        status, data = self._post_main(
            uri='/note/auto_ticket',
            timeout=timeout,
            retries=retries,
            task=task,
            plugin=plugin,
            user=user,
            ipaddr=ipaddr,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def post_user_sizes(
        self,
        user_sizes: dict[str, dict[str, int]],
        *,
        notify: list[str],
        reset: list[str],
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> None:
        """Post a user's account size for AMP to access

Args:
            user_sizes (dict[str, dict[str, int]]): user account sizes (in MiB).
                each dict should have keys "total_mb" and "usage_mb"
            notify (list[str]): users over quota
            reset (list[str]): users under quota (to reset their notify times)
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error updating usage
        """
        status, data = self._post_main(
            uri='/usage/set/shared',
            timeout=timeout,
            retries=retries,
            users=json.dumps(user_sizes),
            notify=json.dumps(notify),
            reset=json.dumps(reset),
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def post_vded_size(
        self,
        usage: int,
        total: int,
        *,
        notify: bool = False,
        reset: bool = False,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> None:
        """Post a v/ded server's size for AMP to access

Args:
            usage (int): disk usage of selected backups (in MiB)
            total (int): disk usage of all account data (in MiB)
            notify (bool, optional): notify as over quota. Defaults to False.
            reset (bool, optional): reset notify counter. Defaults to False.
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error updating usage
        """
        status, data = self._post_main(
            uri='/usage/set/vded',
            timeout=timeout,
            retries=retries,
            usage=usage,
            total=total,
            notify='1' if notify else '0',
            reset='1' if reset else '0',
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def agent_get_user_bucket(
        self,
        user: str,
        key: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> AgentCpuserLookup:
        """Used by support tooling to request a shared user's bucket details
        from a server other than where the bucket belongs

Args:
            user (str): username
            key (str): agent key from cpjump
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting bucket details

Returns:
            AgentCpuserLookup: restic auth info and wans available
        """
        status, data = self._post_main(
            uri='/agent/get_bucket/cpuser',
            timeout=timeout,
            retries=retries,
            cpuser=user,
            key=key,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        wans = data.pop('wans')
        return {'wans': wans, 'repo': ResticRepo(**data)}

def agent_get_server_bucket(
        self,
        host: str,
        key: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> AgentClientLookup:
        """Used by support tooling to request v/ded bucket details for a
        server with an active registration, from a server other than where
        the bucket belongs

Args:
            host (str): short hostname
            key (str): agent key from cpjump
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting bucket details

Returns:
            AgentClientLookup: restic auth info and wans available
        """
        status, data = self._post_main(
            uri='/agent/get_bucket/client',
            timeout=timeout,
            retries=retries,
            host=host,
            key=key,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        wans = data.pop('wans')
        svr_class = data.pop('svr_class')
        return {
            'wans': wans,
            'svr_class': svr_class,
            'repo': ResticRepo(**data),
        }

def agent_get_stashed_bucket(
        self,
        bucket: str,
        key: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> AgentClientLookup:
        """Used by support tooling (baksync) to request v/ded bucket details
        for a bucket which is marked for deletion

Args:
            bucket (str): full bucket name from cpjump
            key (str): agent key from cpjump
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting bucket details

Returns:
            AgentClientLookup: restic auth info and wans available
        """
        status, data = self._post_main(
            uri='/agent/get_bucket/client',
            timeout=timeout,
            retries=retries,
            stashed_bucket=bucket,
            key=key,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        wans = data.pop('wans')
        svr_class = data.pop('svr_class')
        return {
            'wans': wans,
            'svr_class': svr_class,
            'repo': ResticRepo(**data),
        }

def get_user_buckets(
        self,
        users: list[str],
        *,
        wait_mins: int,
        nocache: bool = False,
        suspends: Union[dict, None] = None,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> UserBuckets:
        """Get quota and bucket details for multiple shared users

Args:
            users (list[str]): users to get bucket data for
            wait_mins (int): max number of minutes to wait for bucket creation
                (per cluster)
            nocache (bool): Instruct bakauth to skip cache when
                looking up quota information. Defaults to False.
            suspends (dict | None): only used by the scheduler
                cron. This is a dict of suspended users and when/why.
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting bucket details

Returns:
            UserBuckets: dict where "quotas_gb" contains a dict
            of usernames to quotas in GiB. "repos" contains a dict of usernames
            to ResticRepo objects. "missing" is specific to the replicant,
            ``BAKAUTH3``. If ``BAKAUTH1`` did not reply, we use
            ``BAKAUTH3``, which is unable to setup new users. "copy_users" are
            users which get cross-coast backups
        """
        if not isinstance(users, list):
            raise TypeError
        kwargs = {'users': json.dumps(users), 'nocache': 1 if nocache else 0}
        if suspends is not None:
            kwargs['suspends'] = json.dumps(suspends)
        status, data, _ = self._post_pref(
            uri='/buckets/users',
            timeout=timeout,
            retries=retries,
            **kwargs,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        if wait_mins:
            for task_id in data['task_ids']:
                try:
                    self.task_wait(task_id, wait_mins=wait_mins)
                except BakAuthError as exc:
                    logging.warning(
                        'error when waiting for ceph user setup: %s', exc
                    )
        quotas = {k: v.pop('quota') for k, v in data['users'].items()}
        return {
            'copy_users': data['copy_users'],
            'quotas_gb': quotas,
            'repos': {k: ResticRepo(**v) for k, v in data['users'].items()},
            'missing': data['missing'],
        }

def get_user_bucket(
        self,
        user: str,
        *,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> tuple[int, ResticRepo]:
        """Like get_user_buckets but for only one user and wait_mins=0.

Args:
            user (str): username
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            WrongSharedServer: The requested user does not belong here
            BakAuthDown: Backup Authority server was unreachable
            BakAuthError: any other error

Returns:
            tuple[int, ResticRepo]: quota in GiB and ResticRepo object
        """
        if not isinstance(user, str):
            raise TypeError(f'get_user_bucket({user=})')
        # BakAuthError may raise here
        lookup = self.get_user_buckets(
            [user], nocache=True, wait_mins=0, timeout=timeout, retries=retries
        )
        if user not in lookup['repos']:
            raise WrongSharedServer(
                status=Status.ERROR,
                data=f'{user} is not assigned to this server',
            )
        if user in lookup['missing']:
            # This means we tried bakauth1 and failed, then successfully
            # reached bakauth2, which replied saying the bucket was
            # missing, which is something only bakauth1 can fix.
            raise BakAuthDown(
                status=Status.REQUEST_EXCEPTION,
                data=f'Could not connect to {BAKAUTH1}',
            )
        return lookup['quotas_gb'][user], lookup['repos'][user]

def get_user_bucket_v2(
        self,
        user: str,
        *,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> tuple[int, ResticRepo, bool]:
        """Like get_user_buckets but for only one user and wait_mins=0.

Args:
            user (str): username
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            WrongSharedServer: The requested user does not belong here
            BakAuthDown: Backup Authority server was unreachable
            BakAuthError: any other error

Returns:
            tuple[int, ResticRepo, bool]: quota in GiB, ResticRepo object, and
                whether this user gets cross-coast backups.
        """
        if not isinstance(user, str):
            raise TypeError(f'get_user_bucket({user=})')
        # BakAuthError may raise here
        lookup = self.get_user_buckets(
            [user], nocache=True, wait_mins=0, timeout=timeout, retries=retries
        )
        if user not in lookup['repos']:
            raise WrongSharedServer(
                status=Status.ERROR,
                data=f'{user} is not assigned to this server',
            )
        if user in lookup['missing']:
            # This means we tried bakauth1 and failed, then successfully
            # reached bakauth2, which replied saying the bucket was
            # missing, which is something only bakauth1 can fix.
            raise BakAuthDown(
                status=Status.REQUEST_EXCEPTION,
                data=f'Could not connect to {BAKAUTH1}',
            )
        geo = user in lookup['copy_users']
        return lookup['quotas_gb'][user], lookup['repos'][user], geo

def get_failover_limit(
        self,
        *,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> int:
        """Get the shared failover account size limit in GiB

Args:
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting the size limit

Returns:
            int: shared failover account size limit in GiB
        """
        status, data = self._post_either(
            uri='/buckets/failover_limit',
            timeout=timeout,
            retries=retries,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        return data['failover_gib']

def get_vded_quota(
        self,
        *,
        nocache=False,
        timeout=DEFAULT_TIMEOUT,
        retries=DEFAULT_RETRIES,
    ) -> int:
        """Get this v/ded server's quota as an int in GiB

Args:
            nocache (bool): Instruct bakauth to skip cache when
                looking up quota information. Defaults to False.
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthError: any error requesting the server's quota

Returns:
            int: this v/ded server's quota as an int in GiB
        """
        status, data = self._post_either(
            uri='/buckets/vded_quota',
            timeout=timeout,
            retries=retries,
            ver=1,
            nocache=int(nocache),
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        return data

def get_vded_quota_v2(
        self,
        *,
        nocache=False,
        timeout=DEFAULT_TIMEOUT,
        retries=DEFAULT_RETRIES,
    ) -> tuple[int, bool]:
        """Get this v/ded server's quota as an int in GiB

Raises:
            BakAuthError: any error requesting the server's quota

Returns:
            int: this v/ded server's quota as an int in GiB
            bool: whether this server gets cross-coast backups
        """
        status, data = self._post_either(
            uri='/buckets/vded_quota',
            timeout=timeout,
            retries=retries,
            ver=2,
            nocache=int(nocache),
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        return data['quota'], data['copy']

def get_reg_details(
        self,
        *,
        net: Union[str, None] = None,
        startup: bool = False,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
    ) -> RegDetails:
        """Fetch registration details needed for backup-runner to start

Args:
            startup (bool): If True, bakauth performs extra checks on buckets.
                This should only be set to True on initial backup-runner
                startup. Defaults false.
            net(str, optional): "wan" or "lan" to request that ceph endpoint
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout

Raises:
            BakAuthLoginFailed: auth.json is invalid
            BakAuthWrongLogin: this server has the wrong server's auth.json
            BakAuthError: any other error getting the server's registration info

Returns:
            RegDetails: contains "svr_class", "client_host",
            "endpoint", and "repo"
        """
        data = {}
        if net:
            data['net'] = net
        if startup:
            data['startup'] = '1'
        status, data = self._post_either(
            uri='/buckets/reg_details',
            timeout=timeout,
            retries=retries,
            **data,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        if data['svr_class'] in SHARED_CLASSES:
            if data['client_host'] != platform.node().split('.')[0]:
                raise BakAuthWrongLogin(
                    status=Status.ERROR,
                    data="Shared server registration incorrect! "
                    f"Registered as {data['client_host']!r}",
                )

return {
            'name': data['name'],
            'svr_class': data['svr_class'],
            'client_host': data['client_host'],
            'endpoint': data['endpoint'],
            'location': data['location'],
            'wans': data['wans'],
            'copy': data['copy'],
            'repo': ResticRepo(
                bucket=data['bucket'],
                restic_pass=data['restic_pass'],
                access_key=data['access_key'],
                secret_key=data['secret_key'],
            ),
        }

def ping(
        self,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
        log_retries: bool = False,
    ) -> None:
        """Check if backup authority is reachable and raise an exception if not

Args:
            timeout (int): request timeout in seconds
            retries (int): number of times to retry each server on timeout
            log_retries (bool): whether to log on auto-retries. Defaults False

Raises:
            BakAuthError: backup authority is not reachable
        """
        status, data = self._post_either(
            uri='/monitoring/ping',
            timeout=timeout,
            retries=retries,
            log_retries=log_retries,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)

def check_bucket_index(
        self,
        *,
        user: str,
        bucket: str,
        timeout: int = DEFAULT_TIMEOUT,
        retries: int = DEFAULT_RETRIES,
        geo: Union[bool, None] = None,
    ) -> None:
        """Tell ceph to repair a bucket's index

Args:
            user (str): username that owns the bucket (root if v/ded)
            bucket (str): full bucket name
            timeout (int): request timeout in seconds
            retries (int): number of times to retry the request on timeout
            geo (bool | None): if True, check opposite coast; if False, check
                same-coast. If None (default), check both if present.

Raises:
            BakAuthError: error either requesting the repair, or checking on it
        """
        if geo is None:
            kwargs = {}
        else:
            kwargs = {'geo': int(geo)}
        # queue the repair
        status, data, auth_host = self._post_pref(
            uri='/buckets/queue_repair',
            timeout=timeout,
            retries=retries,
            pref_main=False,
            user=user,
            bucket=bucket,
            **kwargs,
        )
        if status is not Status.OKAY:
            raise BakAuthError(status=status, data=data)
        # wait until it's finished
        # use wait_mins=16 because celery crits after 15 mins
        # task_wait can also raise BakAuthError
        self.task_wait(data, wait_mins=16, poll_secs=10, bakauth_host=auth_host)

def get_restic(
        self, user: str = 'root', *, geo: bool = False, **kwargs
    ) -> Restic:
        """Get a Restic instance for a specific user on a backups 3.x client,
        assuming the current server is setup with it

Args:
            user (str): If on a shared server, this is the username to get the
                Restic instance for. For reseller children, supply their
                reseller's name. If on vps/dedicated, always lookup "root"
            geo (bool): If True, use the secondary, geographically separated
                cluster.
            **kwargs: other keyword arguments to send as-is to the
                Restic() constructor

Raises:
            ValueError: Incorrect user argument supplied
            WrongSharedServer: The requested user does not belong here
            BakAuthDown: Backup Authority server was unreachable
            BakAuthError: Any other API error

Returns:
            Restic: restic instance
        """
        reg = self.get_reg_details()
        if geo:
            if not all(
                (
                    reg['copy']['endpoint'],
                    reg['copy']['name'],
                    reg['copy']['location'],
                )
            ):
                raise ClusterUnavailable(
                    status=Status.ERROR, data="This cluster is not available"
                )
            endpoint = reg['copy']['endpoint']
            cluster = reg['copy']['name']
        else:
            endpoint = reg['endpoint']
            cluster = reg['name']
        is_shared = reg['svr_class'] in SHARED_CLASSES
        if not is_shared and user != 'root':
            raise ValueError(f'{user=}; should be root on {reg["svr_class"]}')
        if user == 'root':
            return Restic(
                endpoint=endpoint, cluster=cluster, repo=reg['repo'], **kwargs
            )
        _, repo = self.get_user_bucket(user)
        return Restic(endpoint=endpoint, cluster=cluster, repo=repo, **kwargs)

def all_restics(
        self, users: Union[list[str], None] = None, geo: bool = False, **kwargs
    ) -> dict[str, Restic]:
        """If users are not supplied, get all restic instances for this shared
        server, including root. Otherwise, get that set of users

Args:
            users (list[str] | None): List of users to obtain restic
                instances for. This function will not resolve child account
                names to resellers, so be sure to only request main cpanel users
            geo (bool): If True, use the secondary, geographically separated
                cluster.
            **kwargs: other keyword arguments to send as-is to the
                Restic() constructor

Raises:
            BakAuthDown: Backup Authority server was unreachable
            WrongServerClass: The current server is not a shared server
            BakAuthError: Any other API error

Returns:
            dict[str, Restic]: usernames mapped to Restic instances. If users
                was supplied, not all usernames requested may be in the result,
                if bakauth did not recognize some as belonging to this server
        """
        if users is None:
            users = rads.main_cpusers()
        repos = self.get_user_buckets(users=users, wait_mins=0)['repos']
        reg = self.get_reg_details()
        if geo:
            if not all(
                reg['copy']['endpoint'],
                reg['copy']['name'],
                reg['copy']['location'],
            ):
                raise ClusterUnavailable(
                    status=Status.ERROR, data="This cluster is not available"
                )
            endpoint = reg['copy']['endpoint']
            cluster = reg['copy']['name']
        else:
            endpoint = reg['endpoint']
            cluster = reg['name']
        ret = {
            'root': Restic(
                endpoint=endpoint, cluster=cluster, repo=reg['repo'], **kwargs
            )
        }
        for user, repo in repos.items():
            ret[user] = Restic(
                endpoint=endpoint, cluster=cluster, repo=repo, **kwargs
            )
        return ret

cmd: