epicyon/crawlers.py

__filename__ = "crawlers.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.5.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "Core"

import os
import time
from utils import save_json
from utils import user_agent_domain
from utils import remove_eol
from blocking import get_mil_domains_list
from blocking import update_blocked_cache
from blocking import is_blocked_domain

default_user_agent_blocks = [
    'fedilist', 'ncsc scan'
]


def update_known_crawlers(ua_str: str,
                          base_dir: str, known_crawlers: {},
                          last_known_crawler: int):
    """Updates a dictionary of known crawlers accessing nodeinfo
    or the masto API
    """
    if not ua_str:
        return None

    curr_time = int(time.time())
    if known_crawlers.get(ua_str):
        known_crawlers[ua_str]['hits'] += 1
        known_crawlers[ua_str]['lastseen'] = curr_time
    else:
        known_crawlers[ua_str] = {
            "lastseen": curr_time,
            "hits": 1
        }

    if curr_time - last_known_crawler >= 30:
        # remove any old observations
        remove_crawlers = []
        for uagent, item in known_crawlers.items():
            if curr_time - item['lastseen'] >= 60 * 60 * 24 * 30:
                remove_crawlers.append(uagent)
        for uagent in remove_crawlers:
            del known_crawlers[uagent]
        # save the list of crawlers
        save_json(known_crawlers,
                  base_dir + '/accounts/knownCrawlers.json')
    return curr_time


def load_known_web_bots(base_dir: str) -> []:
    """Returns a list of known web bots
    """
    known_bots_filename = base_dir + '/accounts/knownBots.txt'
    if not os.path.isfile(known_bots_filename):
        return []
    crawlers_str = None
    try:
        with open(known_bots_filename, 'r', encoding='utf-8') as fp_crawlers:
            crawlers_str = fp_crawlers.read()
    except OSError:
        print('EX: unable to load web bots from ' +
              known_bots_filename)
    if not crawlers_str:
        return []
    known_bots = []
    crawlers_list = crawlers_str.split('\n')
    for crawler in crawlers_list:
        if not crawler:
            continue
        crawler = remove_eol(crawler).strip()
        if not crawler:
            continue
        if crawler not in known_bots:
            known_bots.append(crawler)
    return known_bots


def _save_known_web_bots(base_dir: str, known_bots: []) -> bool:
    """Saves a list of known web bots
    """
    known_bots_filename = base_dir + '/accounts/knownBots.txt'
    known_bots_str = ''
    for crawler in known_bots:
        known_bots_str += crawler.strip() + '\n'
    try:
        with open(known_bots_filename, 'w+', encoding='utf-8') as fp_crawlers:
            fp_crawlers.write(known_bots_str)
    except OSError:
        print("EX: unable to save known web bots to " +
              known_bots_filename)
        return False
    return True


def blocked_user_agent(calling_domain: str, agent_str: str,
                       news_instance: bool, debug: bool,
                       user_agents_blocked: [],
                       blocked_cache_last_updated,
                       base_dir: str,
                       blocked_cache: [],
                       block_federated: [],
                       blocked_cache_update_secs: int,
                       crawlers_allowed: [],
                       known_bots: [], path: str,
                       block_military: {}):
    """Should a GET or POST be blocked based upon its user agent?
    """
    if not agent_str:
        return True, blocked_cache_last_updated

    agent_str_lower = agent_str.lower()
    for ua_block in default_user_agent_blocks:
        if ua_block in agent_str_lower:
            print('BLOCK: Blocked User agent 1: ' + ua_block)
            return True, blocked_cache_last_updated

    agent_domain = None

    if agent_str:
        # is this a web crawler? If so then block it by default
        # unless this is a news instance or if it is in the allowed list
        bot_strings = ('bot/', 'bot-', '/bot', '/robot', 'gptbot',
                       '-ai/', ' ai/', '-ai ', ' ai ', 'spider/',
                       'externalhit/', 'chatgpt')
        contains_bot_string = False
        for bot_str in bot_strings:
            if bot_str in agent_str_lower:
                if '://bot' not in agent_str_lower and \
                   '://robot' not in agent_str_lower and \
                   'pixelfedbot/' not in agent_str_lower:
                    contains_bot_string = True
                    break
        if contains_bot_string:
            if agent_str_lower not in known_bots:
                known_bots.append(agent_str_lower)
                known_bots.sort()
                _save_known_web_bots(base_dir, known_bots)
            # if this is a news instance then we want it
            # to be indexed by search engines
            if news_instance:
                return False, blocked_cache_last_updated
            # is this crawler allowed?
            for crawler in crawlers_allowed:
                if crawler.lower() in agent_str_lower:
                    return False, blocked_cache_last_updated
            print('BLOCK: Blocked Crawler: ' + agent_str)
            return True, blocked_cache_last_updated
        # get domain name from User-Agent
        agent_domain = user_agent_domain(agent_str, debug)
    else:
        # no User-Agent header is present
        return True, blocked_cache_last_updated

    # is the User-Agent type blocked? eg. "Mastodon"
    if user_agents_blocked:
        blocked_ua = False
        for agent_name in user_agents_blocked:
            if agent_name in agent_str:
                blocked_ua = True
                break
        if blocked_ua:
            return True, blocked_cache_last_updated

    if not agent_domain:
        return False, blocked_cache_last_updated

    # is the User-Agent domain blocked
    blocked_ua = False
    if not agent_domain.startswith(calling_domain):
        blocked_cache_last_updated = \
            update_blocked_cache(base_dir, blocked_cache,
                                 blocked_cache_last_updated,
                                 blocked_cache_update_secs)

        blocked_ua = \
            is_blocked_domain(base_dir, agent_domain,
                              blocked_cache, block_federated)
        if blocked_ua:
            print('BLOCK: Blocked User agent 2: ' + agent_domain)

    # optionally block military domains on a per account basis
    if not blocked_ua and block_military:
        if '/users/' in path:
            # which accounts is this?
            nickname = path.split('/users/')[1]
            if '/' in nickname:
                nickname = nickname.split('/')[0]
            # does this account block military domains?
            if block_military.get(nickname):
                mil_domains = get_mil_domains_list()
                for domain_str in mil_domains:
                    if '.' not in domain_str:
                        tld = domain_str
                        if agent_domain.endswith('.' + tld):
                            blocked_ua = True
                            print('BLOCK: Blocked military tld user agent: ' +
                                  agent_domain)
                            break
                    else:
                        if agent_domain.endswith(domain_str):
                            blocked_ua = True
                            print('BLOCK: Blocked military user agent: ' +
                                  agent_domain)
                            break

    return blocked_ua, blocked_cache_last_updated