chg: [crawler] add interactive browser crawler capture. user can open a live browser to crawl content

Author: Terrtia

bin/crawlers/Crawler.py

@@ -392,6 +392,7 @@ def compute(self, capture):  # TODO ADD FUNCTION TO MANUALLY IMPORT ???
         if crawlers.is_domain_correlation_cache(self.original_domain.id):
             crawlers.save_domain_correlation_cache(self.original_domain.was_up(), domain)
 
+        crawlers.release_interactive_session_by_capture(capture.uuid, status='completed')
         task.remove()
         self.root_item = None
 

bin/lib/crawlers.py

@@ -2318,6 +2318,310 @@ def can_launch_forum_crawler_account():
     return get_nb_running_forum_crawler_accounts() < get_forum_crawler_max_accounts()
 
 
+#### INTERACTIVE CRAWLER SESSIONS ####
+
+INTERACTIVE_SESSION_TTL = 3600
+INTERACTIVE_SESSION_META_TTL = 3600
+INTERACTIVE_ACTIVE_STATES = {'starting', 'ready', 'finishing'}
+INTERACTIVE_FINAL_STATES = {'completed', 'expired', 'error', 'closed'}
+
+def get_max_interactive_crawler():
+    nb = r_cache.hget('crawler:lacus', 'max_interactive_crawler')
+    if not nb:
+        nb = r_db.hget('crawler:lacus', 'max_interactive_crawler')
+        if not nb:
+            nb = 1
+            save_max_interactive_crawler(nb)
+        else:
+            r_cache.hset('crawler:lacus', 'max_interactive_crawler', int(nb))
+    return int(nb)
+
+def save_max_interactive_crawler(nb):
+    r_db.hset('crawler:lacus', 'max_interactive_crawler', int(nb))
+    r_cache.hset('crawler:lacus', 'max_interactive_crawler', int(nb))
+
+def api_set_max_interactive_crawler(data):
+    nb = data.get('nb', 1)
+    try:
+        nb = int(nb)
+        if nb < 0:
+            nb = 0
+    except (TypeError, ValueError):
+        return {'error': 'Invalid number of interactive crawler sessions'}, 400
+    save_max_interactive_crawler(nb)
+    return nb, 200
+
+def _cleanup_interactive_task_capture(task_uuid=None, capture_uuid=None):
+    if capture_uuid:
+        capture = CrawlerCapture(capture_uuid)
+        if capture.exists():
+            capture.delete()
+    if task_uuid:
+        task = CrawlerTask(task_uuid)
+        if task.exists():
+            task.delete()
+
+def cleanup_stale_interactive_sessions(now=None):
+    if now is None:
+        now = int(time.time())
+    for session_uuid, launch_time in r_cache.zrange('crawler:interactive:sessions', 0, -1, withscores=True):
+        session = InteractiveCrawlerSession(session_uuid)
+        if not session.exists():
+            r_cache.srem('crawler:interactive:active', session_uuid)
+            r_cache.zrem('crawler:interactive:sessions', session_uuid)
+            continue
+        status = session.get_status()
+        if status in INTERACTIVE_FINAL_STATES:
+            continue
+        if now - int(launch_time) > INTERACTIVE_SESSION_TTL:
+            session.expire()
+
+def get_nb_active_interactive_sessions():
+    cleanup_stale_interactive_sessions()
+    return r_cache.scard('crawler:interactive:active')
+
+def get_interactive_usage():
+    return {'active': get_nb_active_interactive_sessions(), 'max': get_max_interactive_crawler()}
+
+def get_interactive_session_by_capture(capture_uuid):
+    session_uuid = r_cache.hget('crawler:interactive:captures', capture_uuid)
+    if session_uuid:
+        return InteractiveCrawlerSession(session_uuid)
+    for candidate in r_cache.zrange('crawler:interactive:sessions', 0, -1):
+        session = InteractiveCrawlerSession(candidate)
+        if session.get_capture_uuid() == capture_uuid:
+            return session
+    return None
+
+def release_interactive_session_by_capture(capture_uuid, status='completed'):
+    session = get_interactive_session_by_capture(capture_uuid)
+    if session and session.exists():
+        session.release(status=status)
+
+def get_active_interactive_sessions():
+    cleanup_stale_interactive_sessions()
+    sessions = []
+    for session_uuid in r_cache.smembers('crawler:interactive:active'):
+        session = InteractiveCrawlerSession(session_uuid)
+        if session.exists():
+            sessions.append(session.get_meta())
+    return sorted(sessions, key=lambda m: m.get('launch_time', 0))
+
+def get_user_active_interactive_session(user_id):
+    cleanup_stale_interactive_sessions()
+    session_uuid = r_cache.hget('crawler:interactive:users', user_id)
+    if session_uuid:
+        session = InteractiveCrawlerSession(session_uuid)
+        if session.is_active():
+            return session
+        r_cache.hdel('crawler:interactive:users', user_id)
+    return None
+
+def reserve_interactive_session(user_id, url, task_uuid=None):
+    cleanup_stale_interactive_sessions()
+    if get_user_active_interactive_session(user_id):
+        return None, {'error': 'User already has an active interactive session'}, 409
+    max_sessions = get_max_interactive_crawler()
+    if max_sessions <= 0:
+        return None, {'error': 'Interactive crawler sessions are disabled'}, 403
+    session_uuid = gen_uuid()
+    launch_time = int(time.time())
+    if r_cache.hget('crawler:interactive:users', user_id):
+        return None, {'error': 'User already has an active interactive session'}, 409
+    if r_cache.scard('crawler:interactive:active') >= max_sessions:
+        return None, {'error': 'No interactive crawler slots available'}, 429
+    r_cache.hset(f'crawler:interactive:session:{session_uuid}', mapping={'user': user_id, 'url': url, 'status': 'starting', 'launch_time': launch_time})
+    if task_uuid:
+        r_cache.hset(f'crawler:interactive:session:{session_uuid}', 'task_uuid', task_uuid)
+    r_cache.expire(f'crawler:interactive:session:{session_uuid}', INTERACTIVE_SESSION_META_TTL)
+    r_cache.hset('crawler:interactive:users', user_id, session_uuid)
+    r_cache.sadd('crawler:interactive:active', session_uuid)
+    r_cache.zadd('crawler:interactive:sessions', {session_uuid: launch_time})
+    return InteractiveCrawlerSession(session_uuid), None, 200
+
+
+def _remote_headed_response_to_meta(response):
+    if response is None:
+        return {}
+    if isinstance(response, dict):
+        return response
+    meta = {}
+    for field in ('uuid', 'status', 'raw_status', 'finish_requested', 'view_url', 'created_at', 'expires_at', 'error'):
+        if hasattr(response, field):
+            meta[field] = getattr(response, field)
+    return meta
+
+def refresh_interactive_session_status(session):
+    capture_uuid = session.get_capture_uuid()
+    if not capture_uuid:
+        return session.get_meta()
+    try:
+        lacus = get_lacus()
+        remote = _remote_headed_response_to_meta(lacus.get_remote_headed_session(capture_uuid))
+        if remote.get('status'):
+            session.set('remote_status', remote['status'])
+        if remote.get('raw_status') is not None:
+            session.set('remote_raw_status', remote['raw_status'])
+        if remote.get('finish_requested') is not None:
+            session.set('finish_requested', str(remote['finish_requested']))
+        if remote.get('view_url'):
+            session.set('remote_url', remote['view_url'])
+            if session.get_status() == 'starting':
+                session.set('status', 'ready')
+        if remote.get('expires_at'):
+            session.set('expires_at', remote['expires_at'])
+        if remote.get('error'):
+            session.set('error', remote['error'])
+            session.release(status='error')
+        capture_status = lacus.get_capture_status(capture_uuid)
+        session.set('capture_status', int(capture_status))
+    except Exception as e:
+        session.set('last_status_error', str(e))
+    return session.get_meta()
+
+class InteractiveCrawlerSession:
+    def __init__(self, session_uuid):
+        self.uuid = session_uuid
+
+    def exists(self):
+        return r_cache.exists(f'crawler:interactive:session:{self.uuid}')
+
+    def get(self, field):
+        return r_cache.hget(f'crawler:interactive:session:{self.uuid}', field)
+
+    def set(self, field, value):
+        return r_cache.hset(f'crawler:interactive:session:{self.uuid}', field, value)
+
+    def get_user(self):
+        return self.get('user')
+
+    def get_status(self):
+        return self.get('status') or 'unknown'
+
+    def is_active(self):
+        return self.exists() and self.get_status() in INTERACTIVE_ACTIVE_STATES
+
+    def get_capture_uuid(self):
+        return self.get('capture_uuid')
+
+    def get_task_uuid(self):
+        return self.get('task_uuid')
+
+    def get_meta(self):
+        meta = r_cache.hgetall(f'crawler:interactive:session:{self.uuid}')
+        meta['uuid'] = self.uuid
+        try:
+            meta['launch_time'] = int(meta.get('launch_time', 0))
+        except (TypeError, ValueError):
+            meta['launch_time'] = 0
+        return meta
+
+    def release(self, status='completed'):
+        user = self.get_user()
+        task_uuid = self.get_task_uuid()
+        capture_uuid = self.get_capture_uuid()
+        self.set('status', status)
+        self.set('end_time', int(time.time()))
+        r_cache.expire(f'crawler:interactive:session:{self.uuid}', INTERACTIVE_SESSION_META_TTL)
+        r_cache.srem('crawler:interactive:active', self.uuid)
+        r_cache.zrem('crawler:interactive:sessions', self.uuid)
+        if capture_uuid:
+            r_cache.hdel('crawler:interactive:captures', capture_uuid)
+        if user:
+            r_cache.hdel('crawler:interactive:users', user)
+        if status in {'error', 'expired', 'closed'}:
+            _cleanup_interactive_task_capture(task_uuid=task_uuid, capture_uuid=capture_uuid)
+
+    def expire(self):
+        self.release(status='expired')
+
+def api_start_interactive_capture(data, user_org, user_id):
+    task, resp = api_parse_task_dict_basic(data, user_id)
+    if resp != 200:
+        return task, resp
+    if task.get('urls'):
+        return {'error': 'Interactive capture accepts only one URL'}, 400
+    task['depth_limit'] = 0
+    filter_local_ips_error = api_validate_global_urls(url=task.get('url'))
+    if filter_local_ips_error:
+        return filter_local_ips_error
+    session, error, code = reserve_interactive_session(user_id, task['url'])
+    if error:
+        return error, code
+    try:
+        task_uuid = create_task(task['url'], depth=0, har=task['har'], screenshot=task['screenshot'], proxy=task['proxy'], tags=task['tags'], parent='interactive', priority=90, external=True)
+        if not task_uuid:
+            session.release(status='error')
+            return {'error': 'Aborted by Crawler'}, 400
+        session.set('task_uuid', task_uuid)
+        capture_uuid = session.uuid
+        lacus = get_lacus()
+        returned_uuid = lacus.enqueue(url=task['url'], depth=0, proxy=task['proxy'], with_favicon=True, force=True, uuid=capture_uuid, remote_headfull=True, general_timeout_in_sec=90)
+        capture_uuid = returned_uuid or capture_uuid
+        session.set('capture_uuid', capture_uuid)
+        r_cache.hset('crawler:interactive:captures', capture_uuid, session.uuid)
+        create_capture(capture_uuid, task_uuid)
+        CrawlerTask(task_uuid).start()
+        refresh_interactive_session_status(session)
+        return session.get_meta(), 200
+    except Exception as e:
+        session.set('error', str(e))
+        session.release(status='error')
+        return {'error': 'Unable to start interactive capture', 'details': str(e)}, 502
+
+def api_get_interactive_session(session_uuid, user_id, is_admin=False):
+    session = InteractiveCrawlerSession(session_uuid)
+    if not session.exists():
+        return {'error': 'Unknown interactive session'}, 404
+    if not is_admin and session.get_user() != user_id:
+        return {'error': 'Forbidden'}, 403
+    return refresh_interactive_session_status(session), 200
+
+def api_finish_interactive_session(session_uuid, user_id):
+    session = InteractiveCrawlerSession(session_uuid)
+    if not session.exists():
+        return {'error': 'Unknown interactive session'}, 404
+    if session.get_user() != user_id:
+        return {'error': 'Forbidden'}, 403
+    session.set('status', 'finishing')
+    capture_uuid = session.get_capture_uuid()
+    task_uuid = session.get_task_uuid()
+    try:
+        lacus = get_lacus()
+        remote = _remote_headed_response_to_meta(lacus.finish_remote_headed_session(capture_uuid))
+        if remote.get('status'):
+            session.set('remote_status', remote['status'])
+        if remote.get('finish_requested') is not None:
+            session.set('finish_requested', str(remote['finish_requested']))
+        if remote.get('view_url'):
+            session.set('remote_url', remote['view_url'])
+    except Exception as e:
+        session.set('error', str(e))
+    if capture_uuid and task_uuid:
+        refresh_interactive_session_status(session)
+    return session.get_meta(), 200
+
+def api_admin_close_interactive_session(session_uuid):
+    session = InteractiveCrawlerSession(session_uuid)
+    if not session.exists():
+        return {'error': 'Unknown interactive session'}, 404
+    capture_uuid = session.get_capture_uuid()
+    if capture_uuid:
+        try:
+            lacus = get_lacus()
+            remote = _remote_headed_response_to_meta(lacus.finish_remote_headed_session(capture_uuid))
+            if remote.get('status'):
+                session.set('remote_status', remote['status'])
+            if remote.get('finish_requested') is not None:
+                session.set('finish_requested', str(remote['finish_requested']))
+            if remote.get('view_url'):
+                session.set('remote_url', remote['view_url'])
+        except Exception as e:
+            session.set('error', str(e))
+    session.release(status='closed')
+    return session.get_meta(), 200
+
+
 #### CRAWLER CAPTURE ####
 
 def get_nb_crawler_captures():