add: new async backend

This new backend interact with the bui-monitor introduced in fff83dd5
A few things were refactored to take advantage of the queries
parallelization.

burpui/api/client.py

@@ -823,25 +823,9 @@ class ClientReport(Resource):
                 self.abort(500, str(exp))
         else:
             try:
-                client = bui.client.get_client(name, agent=server)
+                json = bui.client.get_backup_logs(-1, name, agent=server)
             except BUIserverException as exp:
                 self.abort(500, str(exp))
-            err = []
-            for back in client:
-                try:
-                    json.append(
-                        bui.client.get_backup_logs(
-                            back['number'],
-                            name,
-                            agent=server
-                        )
-                    )
-                except BUIserverException as exp:
-                    temp = [NOTIF_ERROR, str(exp)]
-                    if temp not in err:
-                        err.append(temp)
-            if err:
-                self.abort(500, err)
         return json
 
     @api.disabled_on_demo()

burpui/engines/monitor.py

@@ -12,8 +12,10 @@ import trio
 import json
 import struct
 import logging
+import datetime
 
 from itertools import count
+from async_generator import asynccontextmanager
 from logging.handlers import RotatingFileHandler
 
 from ..exceptions import BUIserverException
@@ -42,6 +44,11 @@ BUI_DEFAULTS = {
 class MonitorPool:
     logger = logging.getLogger('burp-ui')  # type: logging.Logger
 
+    # cache status results
+    _status_cache = {}
+    _last_status_cleanup = datetime.datetime.now()
+    _time_to_cache = datetime.timedelta(seconds=5)
+
     def __init__(self, conf=None, level=0, logfile=None, debug=False):
         self.debug = debug
         level = level or 0
@@ -105,6 +112,12 @@ class MonitorPool:
         ctx.load_cert_chain(self.sslcert, self.sslkey)
         return ctx
 
+    def _cleanup_cache(self):
+        now = datetime.datetime.now()
+        if now - self._last_status_cleanup > self._time_to_cache:
+            self._status_cache.clear()
+            self._last_status_cleanup = now
+
     async def receive_all(self, stream: trio.StapledStream, length=1024, bsize=None):
         buf = b''
         bsize = bsize if bsize is not None else 1024
@@ -126,6 +139,18 @@ class MonitorPool:
             received += len(newbuf)
         return buf
 
+    @asynccontextmanager
+    async def get_mon(self, ident) -> Monitor:
+        self.logger.info(f'{ident} - Waiting for a monitor...')
+        t1 = trio.current_time()
+        mon = await self.monitor_pool.get()  # type: Monitor
+        t2 = trio.current_time()
+        t = t2 - t1
+        self.logger.info(f'{ident} - Waited {t:.3f}s')
+        yield mon
+        self.logger.info(f'{ident} - Releasing monitor')
+        await self.monitor_pool.put(mon)
+
     async def handle(self, server_stream: trio.StapledStream):
         ident = next(CONNECTION_COUNTER)
         self.logger.info(f'{ident} - handle_request: started')
@@ -145,21 +170,27 @@ class MonitorPool:
             await server_stream.send_all(b'KO')
             return
         try:
-            if req.get('func') == 'monitor_version':
+            func = req.get('func')
+            if func == 'monitor_version':
                 response = json.dumps(__version__)
+            elif func in ['client_version', 'server_version']:
+                async with self.get_mon(ident) as mon:
+                    response = json.dumps(getattr(mon, func, ''))
             else:
                 query = req['query']
-                self.logger.info(f'{ident} - Waiting for a monitor...')
+                cache = req.get('cache', True)
-                t1 = trio.current_time()
-                mon = await self.monitor_pool.get()  # type: Monitor
-                t2 = trio.current_time()
-                t = t2 - t1
-                self.logger.info(f'{ident} - Waited {t:.3f}s')
 
-                response = mon.status(query, timeout=self.timeout, cache=req.get('cache', True))
+                self._cleanup_cache()
-                response = json.dumps(response)
+                # return cached results
-                self.logger.info(f'{ident} - Releasing monitor')
+                if cache and query in self._status_cache:
-                await self.monitor_pool.put(mon)
+                    response = self._status_cache[query]
+                else:
+                    async with self.get_mon(ident) as mon:
+                        response = mon.status(query, timeout=self.timeout, cache=False)
+                    response = json.dumps(response)
+
+                    if cache:
+                        self._status_cache[query] = response
             self.logger.debug(f'{ident} - Sending: {response}')
             await server_stream.send_all(b'OK')
         except BUIserverException as exc:

burpui/misc/backend/async.py

@@ -0,0 +1,988 @@
+# -*- coding: utf8 -*-
+"""
+.. module:: burpui.misc.backend.async
+    :platform: Unix
+    :synopsis: Burp-UI async backend module.
+
+.. moduleauthor:: Ziirish <hi+burpui@ziirish.me>
+
+"""
+import re
+import os
+import time
+import json
+import ssl
+import trio
+import struct
+import logging
+
+from .burp2 import Burp as Burp2
+from .burp1 import Burp as Burp1
+from .interface import BUIbackend
+from .burp.utils import Monitor
+from ..parser.burp2 import Parser
+from ...utils import human_readable as _hr, utc_to_local
+from ...exceptions import BUIserverException
+from ..._compat import to_unicode, to_bytes
+
+BURP_MINIMAL_VERSION = 'burp-2.0.18'
+BURP_LIST_BATCH = '2.0.48'
+BURP_STATUS_FORMAT_V2 = '2.1.10'
+BURP_REVERSE_COUNTERS = '2.1.6'
+
+BUI_DEFAULTS = {
+    'Async': {
+        'host': '::1',
+        'port': 11111,
+        'ssl': True,
+        'password': 'password123456',
+        'timeout': 15,
+    },
+}
+
+
+class Async:
+    logger = logging.getLogger('burp-ui')  # type: logging.Logger
+
+    def __init__(self, conf):
+        """Async client
+
+        :param conf: Configuration to use
+        :type conf: :class:`burpui.config.BUIConfig`
+        """
+
+        self.host = conf.safe_get('host', section='Async', defaults=BUI_DEFAULTS)
+        self.port = conf.safe_get('port', 'integer', section='Async', defaults=BUI_DEFAULTS)
+        self.ssl = conf.safe_get('ssl', 'boolean', section='Async', defaults=BUI_DEFAULTS)
+        self.password = conf.safe_get('password', section='Async', defaults=BUI_DEFAULTS)
+        self.timeout = conf.safe_get('timeout', 'integer', section='Async', defaults=BUI_DEFAULTS)
+
+        self.logger.info(f'Monitor {self.host}:{self.port} - ssl: {self.ssl}')
+
+        self.connected = False
+
+    async def conn(self):
+        if self.ssl:
+            ctx = ssl.SSLContext()
+            ctx.verify_mode = ssl.CERT_NONE
+            ctx.check_hostname = False
+            ctx.load_default_certs()
+            self.client_stream = await trio.open_ssl_over_tcp_stream(self.host, self.port, ssl_context=ctx)
+        else:
+            self.client_stream = await trio.open_tcp_stream(self.host, self.port)
+
+        self.logger.debug('Connected')
+        self.connected = True
+        return self.client_stream
+
+    async def _do_process(self, data):
+        res = '[]'
+        length = struct.pack('!Q', len(data))
+        await self.client_stream.send_all(length)
+        self.logger.debug(f'Sending: {data!r}')
+        await self.client_stream.send_all(to_bytes(data))
+        tmp = await self.client_stream.receive_some(2)
+        tmp = to_unicode(tmp)
+        if tmp == 'ER':
+            lengthbuf = await self.client_stream.receive_some(8)
+            length, = struct.unpack('!Q', lengthbuf)
+            err = await self.receive_all(length)
+            err = to_unicode(err)
+            raise BUIserverException(err)
+        if tmp != 'OK':
+            self.logger.debug('Ooops, unsuccessful!')
+            return res
+        self.logger.debug("Data sent successfully")
+        lengthbuf = await self.client_stream.receive_some(8)
+        length, = struct.unpack('!Q', lengthbuf)
+        res = await self.receive_all(length)
+        res = to_unicode(res)
+        self.logger.debug(f'Received: {res!r}')
+        return res
+
+    async def _process(self, data):
+        if not self.connected:
+            await self.conn()
+            async with self.client_stream:
+                result = await self._do_process(data)
+            self.connected = False
+        else:
+            result = await self._do_process(data)
+        return result
+
+    async def status(self, query, timeout=None, cache=True):
+
+        request = {
+            'query': query,
+            'timeout': timeout,
+            'cache': cache,
+            'password': self.password,
+        }
+        request = json.dumps(request)
+        result = await self._process(request)
+        return json.loads(result)
+
+    async def request(self, func, *args, **kwargs):
+        req = {
+            'func': func,
+            'args': args,
+            'kwargs': kwargs,
+            'password': self.password,
+        }
+        req = json.dumps(req)
+        result = await self._process(req)
+        return json.loads(result)
+
+    async def receive_all(self, length=1024, bsize=None):
+        buf = b''
+        bsize = bsize if bsize is not None else 1024
+        bsize = min(bsize, length)
+        received = 0
+        tries = 0
+        while received < length:
+            newbuf = await self.client_stream.receive_some(bsize)
+            if not newbuf:
+                # 3 successive read failure => raise exception
+                if tries > 3:
+                    raise Exception('Unable to read full response')
+                tries += 1
+                trio.sleep(0.1)
+                continue
+            # reset counter
+            tries = 0
+            buf += newbuf
+            received += len(newbuf)
+        return buf
+
+
+# Some functions are the same as in Burp1 backend
+class Burp(Burp2):
+    """The :class:`burpui.misc.backend.async.Burp` class provides a consistent
+    backend for ``burp-2`` servers through the bui-monitor pool. It is also able to
+    perform some operations asynchronously to speedup the whole API.
+
+    It extends the :class:`burpui.misc.backend.burp2.Burp` class because a few
+    functions can be reused. The rest is just overridden.
+
+    :param server: ``Burp-UI`` server instance in order to access logger
+                   and/or some global settings
+    :type server: :class:`burpui.engines.server.BUIServer`
+
+    :param conf: Configuration to use
+    :type conf: :class:`burpui.config.BUIConfig`
+    """
+
+    # backend version
+    _vers = 2
+    # cache to store the guessed OS
+    _os_cache = {}
+
+    _client_version = None
+    _server_version = None
+
+    def __init__(self, server=None, conf=None):
+        """
+        :param server: ``Burp-UI`` server instance in order to access logger
+                       and/or some global settings
+        :type server: :class:`burpui.engines.server.BUIServer`
+
+        :param conf: Configuration to use
+        :type conf: :class:`burpui.config.BUIConfig`
+        """
+
+        BUIbackend.__init__(self, server, conf)
+
+        self.parser = Parser(self)
+        self.conf = conf
+
+        # self.batch_list_supported = self.monitor.batch_list_supported
+        self.batch_list_supported = True
+
+        self.logger.info('burp conf cli: {}'.format(self.burpconfcli))
+        self.logger.info('burp conf srv: {}'.format(self.burpconfsrv))
+        self.logger.info('command timeout: {}'.format(self.timeout))
+        self.logger.info('tmpdir: {}'.format(self.tmpdir))
+        self.logger.info('zip64: {}'.format(self.zip64))
+        self.logger.info('includes: {}'.format(self.includes))
+        self.logger.info('enforce: {}'.format(self.enforce))
+        self.logger.info('revoke: {}'.format(self.revoke))
+
+    @property
+    def client_version(self):
+        if self._client_version is None:
+            self._client_version = trio.run(self._async_request, 'client_version')
+        return self._client_version
+
+    @property
+    def server_version(self):
+        if self._server_version is None:
+            self._server_version = trio.run(self._async_request, 'server_version')
+        return self._server_version
+
+    async def _async_status(self, query='c:\n', timeout=None, cache=True):
+        async_client = Async(self.conf)
+        return await async_client.status(query, timeout, cache)
+
+    async def _async_request(self, func, *args, **kwargs):
+        async_client = Async(self.conf)
+        return await async_client.request(func, *args, **kwargs)
+
+    def status(self, query='c:\n', timeout=None, cache=True, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.status`"""
+        return trio.run(self._async_status, query, timeout, cache)
+
+    async def _async_get_backup_logs(self, number, client, forward=False, store=None, limit=None):
+        async def _do_stuff():
+            nonlocal client
+            nonlocal number
+            nonlocal forward
+            ret = {}
+            query = await self._async_status('c:{0}:b:{1}\n'.format(client, number))
+            if not query:
+                return ret
+            try:
+                logs = query['clients'][0]['backups'][0]['logs']['list']
+            except KeyError:
+                self.logger.warning('No logs found')
+                return ret
+            if 'backup_stats' in logs:
+                ret = await self._async_parse_backup_stats(number, client, forward)
+
+            ret['encrypted'] = False
+            if 'files_enc' in ret and ret['files_enc']['total'] > 0:
+                ret['encrypted'] = True
+            return ret
+
+        if limit is not None:
+            async with limit:
+                res = await _do_stuff()
+        else:
+            res = await _do_stuff()
+
+        if store is not None:
+            # await store.put(res)
+            store.append(res)
+        else:
+            return res
+
+    async def _async_get_all_backup_logs(self, client, forward=False):
+        ret = []
+        backups = await self._async_get_client(client)
+        # queue = trio.Queue(len(backups))
+        queue = []
+        limit = trio.CapacityLimiter(5)
+        async with trio.open_nursery() as nursery:
+            for back in backups:
+                nursery.start_soon(self._async_get_backup_logs, back['number'], client, forward, queue, limit)
+
+        # while not queue.empty():
+        #     tmp = await queue.get()
+        #     ret.append(tmp)
+
+        # ret = sorted(ret, key=lambda x: x['number'])
+        ret = sorted(queue, key=lambda x: x['number'])
+        return ret
+
+    def get_backup_logs(self, number, client, forward=False, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_backup_logs`
+        """
+        ret = {}
+        if not client or not number:
+            return ret if number != -1 else []
+
+        if number == -1:
+            ret = trio.run(self._async_get_all_backup_logs, client, forward)
+        else:
+            ret = trio.run(self._async_get_backup_logs, number, client, forward)
+        return ret
+
+    def _guess_backup_protocol(self, number, client):
+        """The :func:`burpui.misc.backend.burp2.Burp._guess_backup_protocol`
+        function helps you determine if the backup is protocol 2 or 1
+
+        :param number: Backup number to work on
+        :type number: int
+
+        :param client: Client name to work on
+        :type client: str
+
+        :returns: 1 or 2
+        """
+        query = self.status('c:{0}:b:{1}:l:backup\n'.format(client, number))
+        try:
+            log = query['clients'][0]['backups'][0]['logs']['backup']
+            for line in log:
+                if re.search(r'Protocol: 2$', line):
+                    return 2
+        except KeyError:
+            # Assume protocol 1 in all cases unless explicitly found Protocol 2
+            return 1
+        return 1
+
+    async def _async_parse_backup_stats(self, number, client, forward=False, agent=None):
+        """The :func:`burpui.misc.backend.burp2.Burp._async_parse_backup_stats`
+        function is used to parse the burp logs.
+
+        :param number: Backup number to work on
+        :type number: int
+
+        :param client: Client name to work on
+        :type client: str
+
+        :param forward: Is the client name needed in later process
+        :type forward: bool
+
+        :param agent: What server to ask (only in multi-agent mode)
+        :type agent: str
+
+        :returns: Dict containing the backup log
+        """
+        ret = {}
+        backup = {'os': await self._async_guess_os(client), 'number': int(number)}
+        if forward:
+            backup['name'] = client
+        translate = {
+            'time_start': 'start',
+            'time_end': 'end',
+            'time_taken': 'duration',
+            'bytes': 'totsize',
+            'bytes_received': 'received',
+            'bytes_estimated': 'estimated_bytes',
+            'files': 'files',
+            'files_encrypted': 'files_enc',
+            'directories': 'dir',
+            'soft_links': 'softlink',
+            'hard_links': 'hardlink',
+            'meta_data': 'meta',
+            'meta_data_encrypted': 'meta_enc',
+            'special_files': 'special',
+            'efs_files': 'efs',
+            'vss_headers': 'vssheader',
+            'vss_headers_encrypted': 'vssheader_enc',
+            'vss_footers': 'vssfooter',
+            'vss_footers_encrypted': 'vssfooter_enc',
+            'total': 'total',
+            'grand_total': 'total',
+        }
+        counts = {
+            'new': 'count',
+            'changed': 'changed',
+            'unchanged': 'same',
+            'deleted': 'deleted',
+            'total': 'scanned',
+            'scanned': 'scanned',
+        }
+        single = [
+            'time_start',
+            'time_end',
+            'time_taken',
+            'bytes_received',
+            'bytes_estimated',
+            'bytes'
+        ]
+        query = await self._async_status(
+            'c:{0}:b:{1}:l:backup_stats\n'.format(client, number)
+        )
+        if not query:
+            return ret
+        try:
+            back = query['clients'][0]['backups'][0]
+        except KeyError:
+            self.logger.warning('No backup found')
+            return ret
+        if 'backup_stats' not in back['logs']:
+            self.logger.warning('No stats found for backup')
+            return ret
+        stats = None
+        try:
+            stats = json.loads(''.join(back['logs']['backup_stats']))
+        except:
+            stats = back['logs']['backup_stats']
+        if not stats:
+            return ret
+        # server was upgraded but backup comes from an older version
+        if 'counters' not in stats:
+            return Burp1._parse_backup_stats(
+                self,
+                number,
+                client,
+                forward,
+                stats
+            )
+        counters = stats['counters']
+        for counter in counters:
+            name = counter['name']
+            if name in translate:
+                name = translate[name]
+            if counter['name'] in single:
+                backup[name] = counter['count']
+            else:
+                backup[name] = {}
+                for (key, val) in counts.items():
+                    if val in counter:
+                        backup[name][key] = counter[val]
+                    else:
+                        backup[name][key] = 0
+        if 'start' in backup and 'end' in backup:
+            backup['duration'] = backup['end'] - backup['start']
+            # convert utc timestamp to local
+            # example: 1468850307 -> 1468857507
+            backup['start'] = utc_to_local(backup['start'])
+            backup['end'] = utc_to_local(backup['end'])
+
+        # Needed for graphs
+        if 'received' not in backup:
+            backup['received'] = 1
+
+        return backup
+
+    # TODO: support old clients
+    # NOTE: this should now be partly done since we fallback to the Burp1 code
+    # def _parse_backup_log(self, fh, number, client=None, agent=None):
+    #    """
+    #    parse_backup_log parses the log.gz of a given backup and returns a
+    #    dict containing different stats used to render the charts in the
+    #    reporting view
+    #    """
+    #    return {}
+
+    # def get_clients_report(self, clients, agent=None):
+
+    def get_counters(self, name=None, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.get_counters`"""
+        ret = {}
+        query = self.status('c:{0}\n'.format(name), cache=False)
+        # check the status returned something
+        if not query:
+            return ret
+
+        try:
+            client = query['clients'][0]
+        except KeyError:
+            self.logger.warning('Client not found')
+            return ret
+
+        # check the client is currently backing-up
+        if 'run_status' not in client or client['run_status'] != 'running':
+            return ret
+
+        backup = None
+        phases = ['working', 'finishing']
+        try:
+            for child in client['children']:
+                if 'action' in child and child['action'] == 'backup':
+                    backup = child
+                    break
+        except KeyError:
+            for back in client['backups']:
+                if 'flags' in back and any([x in back['flags'] for x in phases]):
+                    backup = back
+                    break
+        # check we found a working backup
+        if not backup:
+            return ret
+
+        # list of single counters (type CNTR_SINGLE_FIELD in cntr.c)
+        single = [
+            'bytes_estimated',
+            'bytes',
+            'bytes_received',
+            'bytes_sent',
+            'time_start',
+            'time_end',
+            'warnings',
+            'errors'
+        ]
+
+        # translation table to be compatible with burp1
+        def translate(cntr):
+            translate_table = {'bytes_estimated': 'estimated_bytes'}
+            try:
+                return translate_table[cntr]
+            except KeyError:
+                return cntr
+
+        for counter in backup.get('counters', {}):
+            name = translate(counter['name'])
+            if counter['name'] not in single:
+                # Prior burp-2.1.6 some counters are reversed
+                # See https://github.com/grke/burp/commit/adeb3ad68477303991a393fa7cd36bc94ff6b429
+                if self.server_version and self.server_version < BURP_REVERSE_COUNTERS:
+                    ret[name] = [
+                        counter['count'],
+                        counter['same'],     # reversed
+                        counter['changed'],  # reversed
+                        counter['deleted'],
+                        counter['scanned']
+                    ]
+                else:
+                    ret[name] = [
+                        counter['count'],
+                        counter['changed'],
+                        counter['same'],
+                        counter['deleted'],
+                        counter['scanned']
+                    ]
+            else:
+                ret[name] = counter['count']
+
+        if 'phase' in backup:
+            ret['phase'] = backup['phase']
+        else:
+            for phase in phases:
+                if phase in backup.get('flags', []):
+                    ret['phase'] = phase
+                    break
+
+        if 'bytes' not in ret:
+            ret['bytes'] = 0
+        if set(['time_start', 'estimated_bytes', 'bytes']) <= set(ret.keys()):
+            try:
+                diff = time.time() - int(ret['time_start'])
+                byteswant = int(ret['estimated_bytes'])
+                bytesgot = int(ret['bytes'])
+                bytespersec = bytesgot / diff
+                bytesleft = byteswant - bytesgot
+                ret['speed'] = bytespersec
+                if bytespersec > 0:
+                    timeleft = int(bytesleft / bytespersec)
+                    ret['timeleft'] = timeleft
+                else:
+                    ret['timeleft'] = -1
+            except:
+                ret['timeleft'] = -1
+        try:
+            ret['percent'] = round(
+                float(ret['bytes']) / float(ret['estimated_bytes']) * 100
+            )
+        except:
+            # You know... division by 0
+            ret['percent'] = 0
+
+        return ret
+
+    def is_backup_running(self, name=None, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.is_backup_running`
+        """
+        if not name:
+            return False
+        try:
+            query = self.status('c:{0}\n'.format(name))
+        except BUIserverException:
+            return False
+        if not query:
+            return False
+        try:
+            return query['clients'][0]['run_status'] in ['running']
+        except KeyError:
+            self.logger.warning('Client not found')
+            return False
+        return False
+
+    def is_one_backup_running(self, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.is_one_backup_running`
+        """
+        ret = []
+        try:
+            clients = self.get_all_clients()
+        except BUIserverException:
+            return ret
+        for client in clients:
+            if client['state'] in ['running']:
+                ret.append(client['name'])
+        return ret
+
+    def _status_human_readable(self, status):
+        """The label has changed in burp2, we override it to be compatible with
+        burp1's format
+
+        :param status: The status returned by the burp2 server
+        :type status: str
+
+        :returns: burp1 status compatible
+        """
+        if not status:
+            return None
+        if status == 'c crashed':
+            return 'client crashed'
+        if status == 's crashed':
+            return 'server crashed'
+        return status
+
+    async def _async_get_last_backup(self, name):
+        """Return the last backup of a given client
+
+        :param name: Name of the client
+        :type name: str
+
+        :returns: The last backup
+        """
+        try:
+            clients = await self._async_status('c:{}'.format(name))
+            client = clients['clients'][0]
+            return client['backups'][0]
+        except (KeyError, BUIserverException):
+            return None
+
+    def _get_last_backup(self, name):
+        """Return the last backup of a given client
+
+        :param name: Name of the client
+        :type name: str
+
+        :returns: The last backup
+        """
+        return trio.run(self._async_get_last_backup, name)
+
+    async def _async_guess_os(self, name):
+        """Return the OS of the given client based on the magic *os* label
+
+        :param name: Name of the client
+        :type name: str
+
+        :returns: The guessed OS of the client
+
+        ::
+
+            grep label /etc/burp/clientconfdir/toto
+            label = os: Darwin OS
+        """
+        ret = 'Unknown'
+        if name in self._os_cache:
+            return self._os_cache[name]
+
+        labels = await self._async_get_client_labels(name)
+        OSES = []
+
+        for label in labels:
+            if re.match('os:', label, re.IGNORECASE):
+                _os = label.split(':', 1)[1].strip()
+                if _os not in OSES:
+                    OSES.append(_os)
+
+        if OSES:
+            ret = OSES[-1]
+        else:
+            # more aggressive check
+            last = await self._async_get_last_backup(name)
+            if last:
+                try:
+                    tree = await self._async_get_tree(name, last['number'])
+
+                    if tree[0]['name'] != '/':
+                        ret = 'Windows'
+                    else:
+                        ret = 'Unix/Linux'
+                except (IndexError, KeyError, BUIserverException):
+                    pass
+
+        self._os_cache[name] = ret
+        return ret
+
+    def _guess_os(self, name):
+        """Return the OS of the given client based on the magic *os* label
+
+        :param name: Name of the client
+        :type name: str
+
+        :returns: The guessed OS of the client
+
+        ::
+
+            grep label /etc/burp/clientconfdir/toto
+            label = os: Darwin OS
+        """
+        return trio.run(self._async_guess_os, name)
+
+    def get_all_clients(self, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_all_clients`
+        """
+        ret = []
+        query = self.status()
+        if not query or 'clients' not in query:
+            return ret
+        clients = query['clients']
+        for client in clients:
+            cli = {}
+            cli['name'] = client['name']
+            cli['state'] = self._status_human_readable(client['run_status'])
+            infos = client['backups']
+            if cli['state'] in ['running']:
+                cli['last'] = 'now'
+            elif not infos:
+                cli['last'] = 'never'
+            else:
+                infos = infos[0]
+                cli['last'] = infos['timestamp']
+            ret.append(cli)
+        return ret
+
+    def get_client_status(self, name=None, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.get_client_status`"""
+        ret = {}
+        if not name:
+            return ret
+        query = self.status('c:{0}\n'.format(name))
+        if not query:
+            return ret
+        try:
+            client = query['clients'][0]
+        except (KeyError, IndexError):
+            self.logger.warning('Client not found')
+            return ret
+        ret['state'] = self._status_human_readable(client['run_status'])
+        infos = client['backups']
+        if ret['state'] in ['running']:
+            try:
+                ret['phase'] = client['phase']
+            except KeyError:
+                for child in client.get('children', []):
+                    if 'action' in child and child['action'] == 'backup':
+                        ret['phase'] = child['phase']
+                        break
+            counters = self.get_counters(name)
+            if 'percent' in counters:
+                ret['percent'] = counters['percent']
+            else:
+                ret['percent'] = 0
+            ret['last'] = 'now'
+        elif not infos:
+            ret['last'] = 'never'
+        else:
+            infos = infos[0]
+            ret['last'] = infos['timestamp']
+        return ret
+
+    async def _async_get_client(self, name=None):
+        return await self._async_get_client_filtered(name)
+
+    def get_client(self, name=None, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.get_client`"""
+        return trio.run(self._async_get_client, name)
+
+    async def _async_get_client_filtered(self, name=None, limit=-1, page=None, start=None, end=None):
+        ret = []
+        if not name:
+            return ret
+        query = await self._async_status('c:{0}\n'.format(name))
+        if not query:
+            return ret
+        try:
+            backups = query['clients'][0]['backups']
+        except (KeyError, IndexError):
+            self.logger.warning('Client not found')
+            return ret
+
+        async def __parse_log(backup, client, back, ret, limiter):
+            async with limiter:
+                append = True
+                log = await self._async_get_backup_logs(backup['number'], client)
+                try:
+                    back['encrypted'] = log['encrypted']
+                    try:
+                        back['received'] = log['received']
+                    except KeyError:
+                        back['received'] = 0
+                    try:
+                        back['size'] = log['totsize']
+                    except KeyError:
+                        back['size'] = 0
+                    back['end'] = log['end']
+                    # override date since the timestamp is odd
+                    back['date'] = log['start']
+                except Exception:
+                    self.logger.warning('Unable to parse logs')
+                    append = False
+
+                if append:
+                    # await ret.put(back)
+                    ret.append(back)
+
+        # queue = trio.Queue(len(backups))
+        queue = []
+        limiter = trio.CapacityLimiter(5)
+
+        async with trio.open_nursery() as nursery:
+            for idx, backup in enumerate(backups):
+                back = {}
+                # skip the first elements if we are in a page
+                if page and page > 1 and limit > 0:
+                    if idx < (page - 1) * limit:
+                        continue
+
+                # skip running backups since data will be inconsistent
+                if 'flags' in backup and 'working' in backup['flags']:
+                    continue
+                back['number'] = backup['number']
+                if 'flags' in backup and 'deletable' in backup['flags']:
+                    back['deletable'] = True
+                else:
+                    back['deletable'] = False
+                back['date'] = backup['timestamp']
+                # skip backups before "start"
+                if start and backup['timestamp'] < start:
+                    continue
+                # skip backups after "end"
+                if end and backup['timestamp'] > end:
+                    continue
+
+                nursery.start_soon(__parse_log, backup, name, back, queue, limiter)
+
+                # stop after "limit" elements
+                if page and page > 1 and limit > 0:
+                    if idx >= page * limit:
+                        break
+                elif limit > 0 and idx >= limit:
+                    break
+
+        # while not queue.empty():
+        #     tmp = await queue.get()
+        #     ret.append(tmp)
+
+        # Here we need to reverse the array so the backups are sorted by num
+        # ASC
+        # ret = sorted(ret, key=lambda x: x['number'])
+        ret = sorted(queue, key=lambda x: x['number'])
+        return ret
+
+    def get_client_filtered(self, name=None, limit=-1, page=None, start=None, end=None, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.get_client_filtered`"""
+        return trio.run(self._async_get_client_filtered, name, limit, page, start, end)
+
+    def is_backup_deletable(self, name=None, backup=None, agent=None):
+        """Check if a given backup is deletable"""
+        if not name or not backup:
+            return False
+        query = self.status('c:{0}:b:{1}\n'.format(name, backup))
+        if not query:
+            return False
+        try:
+            flags = query['clients'][0]['backups'][0]['flags']
+            return 'deletable' in flags
+        except KeyError:
+            return False
+
+    async def _async_get_tree(self, name=None, backup=None, root=None, level=-1):
+        ret = []
+        if not name or not backup:
+            return ret
+        if not root:
+            top = ''
+        else:
+            top = to_unicode(root)
+
+        # we know this operation may take a while so we arbitrary increase the
+        # read timeout
+        timeout = None
+        if top == '*':
+            timeout = max(self.timeout, 300)
+
+        query = await self._async_status(
+            'c:{0}:b:{1}:p:{2}\n'.format(name, backup, top),
+            timeout
+        )
+        if not query:
+            return ret
+        try:
+            backup = query['clients'][0]['backups'][0]
+        except KeyError:
+            return ret
+        for entry in backup['browse']['entries']:
+            data = {}
+            base = None
+            dirn = None
+            if top == '*':
+                base = os.path.basename(entry['name'])
+                dirn = os.path.dirname(entry['name'])
+            if entry['name'] == '.':
+                continue
+            else:
+                data['name'] = base or entry['name']
+            data['mode'] = self._human_st_mode(entry['mode'])
+            if re.match('^(d|l)', data['mode']):
+                data['type'] = 'd'
+                data['folder'] = True
+            else:
+                data['type'] = 'f'
+                data['folder'] = False
+            data['inodes'] = entry['nlink']
+            data['uid'] = entry['uid']
+            data['gid'] = entry['gid']
+            data['parent'] = dirn or top
+            data['size'] = '{0:.1eM}'.format(_hr(entry['size']))
+            data['date'] = entry['mtime']
+            data['fullname'] = os.path.join(top, entry['name']) if top != '*' \
+                else entry['name']
+            data['level'] = level
+            data['children'] = []
+            ret.append(data)
+        return ret
+
+    def get_tree(self, name=None, backup=None, root=None, level=-1, agent=None):
+        """See :func:`burpui.misc.backend.interface.BUIbackend.get_tree`"""
+        return trio.run(self._async_get_tree, name, backup, root, level)
+
+    def get_client_version(self, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_client_version`
+        """
+        return self.client_version
+
+    def get_server_version(self, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_server_version`
+        """
+        if not self.server_version:
+            self.status()
+        return self.server_version
+
+    async def _async_get_client_labels(self, client=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_client_labels`
+        """
+        ret = []
+        if not client:
+            return ret
+        # micro optimization since the status results are cached in memory for a
+        # couple seconds, using the same global query and iterating over it
+        # will be more efficient than filtering burp-side
+        query = await self._async_status('c:\n')
+        if not query:
+            return ret
+        try:
+            for cli in query['clients']:
+                if cli['name'] == client:
+                    return cli['labels']
+        except KeyError:
+            return ret
+
+    def get_client_labels(self, client=None, agent=None):
+        """See
+        :func:`burpui.misc.backend.interface.BUIbackend.get_client_labels`
+        """
+        return trio.run(self._async_get_client_labels, client)
+
+    # Same as in Burp1 backend
+    # def restore_files(
+    #     self,
+    #     name=None,
+    #     backup=None,
+    #     files=None,
+    #     strip=None,
+    #     archive='zip',
+    #     password=None,
+    #     agent=None):
+
+    # def read_conf_cli(self, agent=None):
+
+    # def read_conf_srv(self, agent=None):
+
+    # def store_conf_cli(self, data, agent=None):
+
+    # def store_conf_srv(self, data, agent=None):
+
+    # def get_parser_attr(self, attr=None, agent=None):

burpui/misc/backend/burp/utils.py

@@ -250,7 +250,7 @@ class Monitor(object):
         try:
             timeout = timeout or self.timeout
             query = sanitize_string(query.rstrip())
-            self.logger.info(f"{self.ident} - query: '{query}'")
+            self.logger.info(f"{self.ident} - query: '{query}' (cache: {cache})")
             query = '{0}\n'.format(query)
 
             self._cleanup_cache()
@@ -261,7 +261,7 @@ class Monitor(object):
             if not self._proc_is_alive():
                 self._spawn_burp()
 
-            _, write, _ = select([], [self.proc.stdin], [], self.timeout)
+            _, write, _ = select([], [self.proc.stdin], [], timeout)
             if self.proc.stdin not in write:
                 raise TimeoutError('Write operation timed out')
             self.proc.stdin.write(to_bytes(query))
@@ -271,7 +271,7 @@ class Monitor(object):
                 self.logger.debug('Nothing interesting to return')
                 return None
 
-            self.logger.debug(f'=> {jso}')
+            self.logger.debug(f'{self.ident} => {jso}')
 
             if cache:
                 self._status_cache[query] = jso

contrib/systemd/bui-monitor.service

@@ -0,0 +1,13 @@
+[Unit]
+Description=Burp-UI monitor service
+After=network.target
+
+[Service]
+User=burpui
+Group=burpui
+RuntimeDirectory=bui-monitor
+RuntimeDirectoryMode=0770
+ExecStart=/usr/bin/bui-monitor -c /etc/burp/bui-monitor.cfg
+
+[Install]
+WantedBy=multi-user.target

docs/advanced_usage.rst

@@ -7,7 +7,7 @@ Advanced usage
 `Burp`_ from the stable to the latest versions. `Burp`_ exists in two major
 versions: 1.x.x and 2.x.x.
 
-Both `Versions`_ are supported by `Burp-UI`_ thanks to its modular design.
+Both versions are supported by `Burp-UI`_ thanks to its modular design.
 The consequence is you have various options in the configuration file to suite
 everybody needs.
 
@@ -31,8 +31,13 @@ The `burpui.cfg`_ configuration file contains a ``[Global]`` section as follow:
 ::
 
     [Global]
-    # burp backend to load either one of 'burp1', 'burp2', or 'multi'.
+    # burp backend to load either one of 'burp1', 'burp2', 'async' or 'multi'.
-    # If you choose 'multi', you will have to declare at lease one 'Agent' section
+    # If you choose 'multi', you will have to declare at lease one 'Agent' section.
+    # If you choose 'async', you need to configure the [Async] section.
+    # If you choose either 'burp1' or 'burp2', you need to configure the [Burp]
+    # section.
+    # The [Burp] section is also used with the 'async' backend for the restoration
+    # process.
     # You can also use whatever custom backend you like if it is located in the
     # 'plugins' directory and if it implements the right interface.
     backend = burp2
@@ -253,6 +258,7 @@ Backends
 - `Burp1`_
 - `Burp2`_
 - `Multi`_
+- `Async`_
 
 These backends allow you to either connect to a `Burp`_ server version 1.x.x or
 2.x.x.
@@ -355,6 +361,46 @@ Once this backend is enabled, you have to create **one** ``[Agent]`` section
 To configure your agents, please refer to the `bui-agent`_ page.
 
 
+Async
+^^^^^
+
+The *async* backend allows you to connect to the *bui-monitor* pool. It can be
+enabled by setting the *backend* option to *async* in the ``[Global]`` section
+of your `burpui.cfg`_ file:
+
+::
+
+    [Global]
+    backend = async
+
+
+This backend allows you to access `Burp`_ servers through the `bui-monitor`_
+pool.
+The architecture is available on the bui-monitor
+`page <buimonitor.html#architecture>`__.
+
+
+Once this backend is enabled, you have to configure the ``[Async]`` section.
+
+::
+
+    # async backend specific options
+    [Async]
+    # address of the monitor pool
+    host = ::1
+    # port of the monitor pool
+    port = 11111
+    # how many time to wait for the monitor pool to answer (in seconds)
+    timeout = 15
+    # monitor pool password
+    password = password123456
+    # enable SSL
+    ssl = true
+
+
+To configure your monitor pool, please refer to the `bui-monitor`_ page.
+
+
 Options
 ^^^^^^^
 

docs/buimonitor.rst

@@ -30,39 +30,39 @@ The architecture is described bellow:
 
 ::
 
-	+---------------------+
+    +---------------------+
-	|                     |
+    |                     |
-	|     celery          |
+    |     celery          |
-	|                     |
+    |                     |
-	+---------------------+
+    +---------------------+
-	| +-----------------+ |                                 +----------------------+
+    | +-----------------+ |                                 +----------------------+
-	| |                 | |                                 |                      |
+    | |                 | |                                 |                      |
-	| |  worker 1       +----------------+------------------>     bui-monitor      |
+    | |  worker 1       +----------------+------------------>     bui-monitor      |
-	| |                 | |              |                  |                      |
+    | |                 | |              |                  |                      |
-	| +-----------------+ |              |                  +----------------------+
+    | +-----------------+ |              |                  +----------------------+
-	| +-----------------+ |              |                  | +------------------+ |
+    | +-----------------+ |              |                  | +------------------+ |
-	| |                 | |              |                  | |                  | |
+    | |                 | |              |                  | |                  | |
-	| |  worker n       +----------------+                  | | burp -a m   (1)  | |
+    | |  worker n       +----------------+                  | | burp -a m   (1)  | |
-	| |                 | |              |                  | |                  | |
+    | |                 | |              |                  | |                  | |
-	| +-----------------+ |              |                  | +------------------+ |
+    | +-----------------+ |              |                  | +------------------+ |
-	+---------------------+              |                  | +------------------+ |
+    +---------------------+              |                  | +------------------+ |
-										 |                  | |                  | |
+                                         |                  | |                  | |
-	+---------------------+              |                  | | burp -a m   (2)  | |
+    +---------------------+              |                  | | burp -a m   (2)  | |
-	|                     |              |                  | |                  | |
+    |                     |              |                  | |                  | |
-	|     burp-ui         |              |                  | +------------------+ |
+    |     burp-ui         |              |                  | +------------------+ |
-	|                     |              |                  | +------------------+ |
+    |                     |              |                  | +------------------+ |
-	+---------------------+              |                  | |                  | |
+    +---------------------+              |                  | |                  | |
-	| +-----------------+ |              |                  | | burp -a m   (n)  | |
+    | +-----------------+ |              |                  | | burp -a m   (n)  | |
-	| |                 | |              |                  | |                  | |
+    | |                 | |              |                  | |                  | |
-	| |  worker 1       +----------------+                  | +------------------+ |
+    | |  worker 1       +----------------+                  | +------------------+ |
-	| |                 | |              |                  +----------------------+
+    | |                 | |              |                  +----------------------+
-	| +-----------------+ |              |
+    | +-----------------+ |              |
-	| +-----------------+ |              |
+    | +-----------------+ |              |
-	| |                 | |              |
+    | |                 | |              |
-	| |  worker n       +----------------+
+    | |  worker n       +----------------+
-	| |                 | |
+    | |                 | |
-	| +-----------------+ |
+    | +-----------------+ |
-	+---------------------+
+    +---------------------+
 
 
 Requirements
@@ -75,7 +75,7 @@ You can launch it with the ``bui-monitor`` command.
 Configuration
 -------------
 
-There is a specific `buimanager.cfg`_ configuration file with a ``[Global]``
+There is a specific `buimonitor.cfg`_ configuration file with a ``[Global]``
 section as below:
 
 ::

requirements.txt

@@ -14,3 +14,4 @@ pyOpenSSL>=17.5.0
 configobj==5.0.6
 pyasn1>=0.2.3
 cffi>=1.10.0
+async_generator

share/burpui/etc/buiagent.sample.cfg

@@ -1,6 +1,4 @@
 # Burp-UI agent configuration file
-# @version@ - 0.4.2
-# @release@ - stable
 [Global]
 # On which port is the application listening
 port = 10000
@@ -15,12 +13,11 @@ sslcert = /var/lib/burp/ssl/server/ssl_cert-server.pem
 # ssl key
 sslkey = /var/lib/burp/ssl/server/ssl_cert-server.key
 # burp backend to load either 'burp1' or 'burp2'.
-# If you choose 'multi', you will have to declare at lease one 'Agent' section
 # You can also use whatever custom backend you like if it is located in the
 # 'plugins' directory and if it implements the right interface.
 backend = burp2
 # agent password
-password = password
+password = azerty
 
 [Security]
 ## This section contains some security options. Make sure you understand the

share/burpui/etc/burpui.sample.cfg

@@ -1,9 +1,12 @@
 # Burp-UI configuration file
-# @version@ - 0.6.0
-# @release@ - stable
 [Global]
-# burp backend to load either one of 'burp1', 'burp2', or 'multi'.
+# burp backend to load either one of 'burp1', 'burp2', 'async' or 'multi'.
-# If you choose 'multi', you will have to declare at lease one 'Agent' section
+# If you choose 'multi', you will have to declare at lease one 'Agent' section.
+# If you choose 'async', you need to configure the [Async] section.
+# If you choose either 'burp1' or 'burp2', you need to configure the [Burp]
+# section.
+# The [Burp] section is also used with the 'async' backend for the restoration
+# process.
 # You can also use whatever custom backend you like if it is located in the
 # 'plugins' directory and if it implements the right interface.
 backend = burp2
@@ -174,6 +177,19 @@ noserverrestore = false
 ## how many time to wait for the monitor to answer (in seconds)
 #timeout = 15
 
+## async backend specific options
+#[Async]
+## address of the monitor pool
+#host = ::1
+## port of the monitor pool
+#port = 11111
+## how many time to wait for the monitor pool to answer (in seconds)
+#timeout = 15
+## monitor pool password
+#password = password123456
+## enable SSL
+#ssl = true
+
 ## ldapauth specific options
 #[LDAP]
 ## Backend priority. Higher is first
@@ -307,7 +323,7 @@ noserverrestore = false
 ## bui-agent
 #[Agent:agent1]
 ## bui-agent address
-#host = 192.168.1.1
+#host = ::1
 ## bui-agent port
 #port = 10000
 ## bui-agent password