<commit_before>#!/usr/bin/env python3
# -*-coding:UTF-8 -*

import os
import sys
import redis
import datetime

sys.path.append(os.path.join(os.environ['AIL_BIN'], 'packages/'))
import Date

sys.path.append(os.path.join(os.environ['AIL_BIN'], 'lib/'))
import ConfigLoader
import item_basic

from pytaxonomies import Taxonomies
from pymispgalaxies import Galaxies, Clusters

config_loader = ConfigLoader.ConfigLoader()
r_serv_tags = config_loader.get_redis_conn("ARDB_Tags")
r_serv_metadata = config_loader.get_redis_conn("ARDB_Metadata")
config_loader = None

def build_unsafe_tags():
    unsafe_tags = set()
    ## CE content
    unsafe_tags.add('dark-web:topic="pornography-child-exploitation"')
    # add copine-scale tags
    taxonomies = Taxonomies()
    copine_scale = taxonomies.get('copine-scale')
    if copine_scale:
        for tag in copine_scale.machinetags():
            unsafe_tags.add(tag)
    return unsafe_tags

# set of unsafe tags
unsafe_tags = build_unsafe_tags()

def is_tags_safe(ltags):
    '''
    Check if a list of tags contain an unsafe tag (CE, ...)

    :param ltags: list of tags
    :type ltags: list
    :return: is a tag in the unsafe set
    :rtype: boolean
    '''
    return unsafe_tags.isdisjoint(ltags)

#### Taxonomies - Galaxies ####

def get_taxonomie_from_tag(tag):
    return tag.split(':')[0]

def get_galaxy_from_tag(tag):
    galaxy = tag.split(':')[1]
    galaxy = galaxy.split('=')[0]
    return galaxy

def get_active_taxonomies():
    return r_serv_tags.smembers('active_taxonomies')

def get_active_galaxies():
    return r_serv_tags.smembers('active_galaxies')

def get_all_taxonomies_tags(): # # TODO: add + REMOVE + Update
    return r_serv_tags.smembers('active_taxonomies_tags')

def get_all_galaxies_tags(): # # TODO: add + REMOVE + Update
    return r_serv_tags.smembers('active_galaxies_tags')

def is_taxonomie_tag_enabled(taxonomie, tag):
    if tag in r_serv_tags.smembers('active_tag_' + taxonomie):
        return True
    else:
        return False

def is_galaxy_tag_enabled(galaxy, tag):
    if tag in r_serv_tags.smembers('active_tag_galaxies_' + galaxy):
        return True
    else:
        return False

def enable_taxonomy(taxonomie, enable_tags=True):
    '''
    Enable a taxonomy. (UI)

    :param taxonomie: MISP taxonomy
    :type taxonomie: str
    :param enable_tags: crawled domain
    :type enable_tags: boolean
    '''
    taxonomies = Taxonomies()
    if enable_tags:
        taxonomie_info = taxonomies.get(taxonomie)
        if taxonomie_info:
            # activate taxonomie
            r_serv_tags.sadd('active_taxonomies', taxonomie)
            # activate taxonomie tags
            for tag in taxonomie_info.machinetags():
                r_serv_tags.sadd('active_tag_{}'.format(taxonomie), tag)
                #r_serv_tags.sadd('active_taxonomies_tags', tag)
        else:
            print('Error: {}, please update pytaxonomies'.format(taxonomie))

# Check if tags are enabled in AIL
def is_valid_tags_taxonomies_galaxy(list_tags, list_tags_galaxy):
    if list_tags:
        active_taxonomies = get_active_taxonomies()

        for tag in list_tags:
            taxonomie = get_taxonomie_from_tag(tag)
            if taxonomie not in active_taxonomies:
                return False
            if not is_taxonomie_tag_enabled(taxonomie, tag):
                return False

    if list_tags_galaxy:
        active_galaxies = get_active_galaxies()

        for tag in list_tags_galaxy:
            galaxy = get_galaxy_from_tag(tag)
            if galaxy not in active_galaxies:
                return False
            if not is_galaxy_tag_enabled(galaxy, tag):
                return False
    return True

####  ####

def is_tag_in_all_tag(tag):
    if r_serv_tags.sismember('list_tags', tag):
        return True
    else:
        return False

def get_min_tag(tag):
    tag = tag.split('=')
    if len(tag) > 1:
        if tag[1] != '':
            tag = tag[1][1:-1]
        # no value
        else:
            tag = tag[0][1:-1]
    # custom tags
    else:
        tag = tag[0]
    return tag

def get_obj_tags_minimal(item_id):
    return [ {"tag": tag, "min_tag": get_min_tag(tag)} for tag in get_obj_tag(item_id) ]

def unpack_str_tags_list(str_tags_list):
    str_tags_list = str_tags_list.replace('"','\"')
    if str_tags_list:
        return str_tags_list.split(',')
    else:
        return []

# used by modal
def get_modal_add_tags(item_id, object_type='item'):
    '''
    Modal: add tags to domain or Paste
    '''
    return {"active_taxonomies": get_active_taxonomies(), "active_galaxies": get_active_galaxies(),
            "object_id": item_id, "object_type": object_type}

######## NEW VERSION ########
def get_tag_first_seen(tag, r_int=False):
    '''
    Get tag first seen (current: item only)
    '''
    res = r_serv_tags.hget('tag_metadata:{}'.format(tag), 'first_seen')
    if r_int:
        if res is None:
            return 99999999
        else:
            return int(res)
    return res

def get_tag_last_seen(tag, r_int=False):
    '''
    Get tag last seen (current: item only)
    '''
    res = r_serv_tags.hget('tag_metadata:{}'.format(tag), 'last_seen')
    if r_int:
        if res is None:
            return 0
        else:
            return int(res)
    return res

def get_tag_metadata(tag, r_int=False):
    '''
    Get tag metadata (current: item only)
    '''
    tag_metadata = {"tag": tag}
    tag_metadata['first_seen'] = get_tag_first_seen(tag, r_int=r_int)
    tag_metadata['last_seen'] = get_tag_last_seen(tag, r_int=r_int)
    return tag_metadata

def get_tags_min_last_seen(l_tags, r_int=False):
    '''
    Get max last seen from a list of tags (current: item only)
    '''
    min_last_seen = 99999999
    for tag in l_tags:
        last_seen = get_tag_last_seen(tag, r_int=True)
        if last_seen < min_last_seen:
                min_last_seen = last_seen
    if r_int:
        return min_last_seen
    else:
        return str(min_last_seen)

def is_obj_tagged(object_id, tag):
    '''
    Check if a object is tagged

    :param object_id: object id
    :type domain: str
    :param tag: object type
    :type domain: str

    :return: is object tagged
    :rtype: boolean
    '''
    return r_serv_metadata.sismember('tag:{}'.format(object_id), tag)

def get_all_tags():
    return list(r_serv_tags.smembers('list_tags'))

def get_all_obj_tags(object_type):
    return list(r_serv_tags.smembers('list_tags:{}'.format(object_type)))

def get_obj_tag(object_id):
    '''
    Retun all the tags of a given object.
    :param object_id: (item_id, domain, ...)
    '''
    res = r_serv_metadata.smembers('tag:{}'.format(object_id))
    if res:
        return list(res)
    else:
        return []

def update_tag_first_seen(tag, tag_first_seen, tag_last_seen):
    if tag_first_seen == tag_last_seen:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_first_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'first_seen', tag_first_seen)
        # no tag in db
        else:
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'first_seen')
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'last_seen')
    else:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_first_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'first_seen', tag_first_seen)
        else:
            tag_first_seen = Date.date_add_day(tag_first_seen)
            update_tag_first_seen(tag, tag_first_seen, tag_last_seen)

def update_tag_last_seen(tag, tag_first_seen, tag_last_seen):
    if tag_first_seen == tag_last_seen:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_last_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'last_seen', tag_last_seen)
        # no tag in db
        else:
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'first_seen')
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'last_seen')
    else:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_last_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'last_seen', tag_last_seen)
        else:
            # # TODO: # FIXME: 
            #tag_last_seen = Date.date_substract_day(str(tag_last_seen))
            #update_tag_last_seen(tag, tag_first_seen, tag_last_seen)
            pass<commit_msg>fix: [tags] invalid tags<commit_after>#!/usr/bin/env python3
# -*-coding:UTF-8 -*

import os
import sys
import redis
import datetime

sys.path.append(os.path.join(os.environ['AIL_BIN'], 'packages/'))
import Date

sys.path.append(os.path.join(os.environ['AIL_BIN'], 'lib/'))
import ConfigLoader
import item_basic

from pytaxonomies import Taxonomies
from pymispgalaxies import Galaxies, Clusters

config_loader = ConfigLoader.ConfigLoader()
r_serv_tags = config_loader.get_redis_conn("ARDB_Tags")
r_serv_metadata = config_loader.get_redis_conn("ARDB_Metadata")
config_loader = None

def build_unsafe_tags():
    unsafe_tags = set()
    ## CE content
    unsafe_tags.add('dark-web:topic="pornography-child-exploitation"')
    # add copine-scale tags
    taxonomies = Taxonomies()
    copine_scale = taxonomies.get('copine-scale')
    if copine_scale:
        for tag in copine_scale.machinetags():
            unsafe_tags.add(tag)
    return unsafe_tags

# set of unsafe tags
unsafe_tags = build_unsafe_tags()

def is_tags_safe(ltags):
    '''
    Check if a list of tags contain an unsafe tag (CE, ...)

    :param ltags: list of tags
    :type ltags: list
    :return: is a tag in the unsafe set
    :rtype: boolean
    '''
    return unsafe_tags.isdisjoint(ltags)

#### Taxonomies - Galaxies ####

def get_taxonomie_from_tag(tag):
    try:
        return tag.split(':')[0]
    except IndexError:
        return None

def get_galaxy_from_tag(tag):
    try:
        galaxy = tag.split(':')[1]
        galaxy = galaxy.split('=')[0]
        return galaxy
    except IndexError:
        return None

def get_active_taxonomies():
    return r_serv_tags.smembers('active_taxonomies')

def get_active_galaxies():
    return r_serv_tags.smembers('active_galaxies')

def get_all_taxonomies_tags(): # # TODO: add + REMOVE + Update
    return r_serv_tags.smembers('active_taxonomies_tags')

def get_all_galaxies_tags(): # # TODO: add + REMOVE + Update
    return r_serv_tags.smembers('active_galaxies_tags')

def is_taxonomie_tag_enabled(taxonomie, tag):
    if tag in r_serv_tags.smembers('active_tag_' + taxonomie):
        return True
    else:
        return False

def is_galaxy_tag_enabled(galaxy, tag):
    if tag in r_serv_tags.smembers('active_tag_galaxies_' + galaxy):
        return True
    else:
        return False

def enable_taxonomy(taxonomie, enable_tags=True):
    '''
    Enable a taxonomy. (UI)

    :param taxonomie: MISP taxonomy
    :type taxonomie: str
    :param enable_tags: crawled domain
    :type enable_tags: boolean
    '''
    taxonomies = Taxonomies()
    if enable_tags:
        taxonomie_info = taxonomies.get(taxonomie)
        if taxonomie_info:
            # activate taxonomie
            r_serv_tags.sadd('active_taxonomies', taxonomie)
            # activate taxonomie tags
            for tag in taxonomie_info.machinetags():
                r_serv_tags.sadd('active_tag_{}'.format(taxonomie), tag)
                #r_serv_tags.sadd('active_taxonomies_tags', tag)
        else:
            print('Error: {}, please update pytaxonomies'.format(taxonomie))

# Check if tags are enabled in AIL
def is_valid_tags_taxonomies_galaxy(list_tags, list_tags_galaxy):
    if list_tags:
        active_taxonomies = get_active_taxonomies()

        for tag in list_tags:
            taxonomie = get_taxonomie_from_tag(tag)
            if taxonomie is None:
                return False
            if taxonomie not in active_taxonomies:
                return False
            if not is_taxonomie_tag_enabled(taxonomie, tag):
                return False

    if list_tags_galaxy:
        active_galaxies = get_active_galaxies()

        for tag in list_tags_galaxy:
            galaxy = get_galaxy_from_tag(tag)
            if galaxy is None:
                return False
            if galaxy not in active_galaxies:
                return False
            if not is_galaxy_tag_enabled(galaxy, tag):
                return False
    return True

####  ####

def is_tag_in_all_tag(tag):
    if r_serv_tags.sismember('list_tags', tag):
        return True
    else:
        return False

def get_min_tag(tag):
    tag = tag.split('=')
    if len(tag) > 1:
        if tag[1] != '':
            tag = tag[1][1:-1]
        # no value
        else:
            tag = tag[0][1:-1]
    # custom tags
    else:
        tag = tag[0]
    return tag

def get_obj_tags_minimal(item_id):
    return [ {"tag": tag, "min_tag": get_min_tag(tag)} for tag in get_obj_tag(item_id) ]

def unpack_str_tags_list(str_tags_list):
    str_tags_list = str_tags_list.replace('"','\"')
    if str_tags_list:
        return str_tags_list.split(',')
    else:
        return []

# used by modal
def get_modal_add_tags(item_id, object_type='item'):
    '''
    Modal: add tags to domain or Paste
    '''
    return {"active_taxonomies": get_active_taxonomies(), "active_galaxies": get_active_galaxies(),
            "object_id": item_id, "object_type": object_type}

######## NEW VERSION ########
def get_tag_first_seen(tag, r_int=False):
    '''
    Get tag first seen (current: item only)
    '''
    res = r_serv_tags.hget('tag_metadata:{}'.format(tag), 'first_seen')
    if r_int:
        if res is None:
            return 99999999
        else:
            return int(res)
    return res

def get_tag_last_seen(tag, r_int=False):
    '''
    Get tag last seen (current: item only)
    '''
    res = r_serv_tags.hget('tag_metadata:{}'.format(tag), 'last_seen')
    if r_int:
        if res is None:
            return 0
        else:
            return int(res)
    return res

def get_tag_metadata(tag, r_int=False):
    '''
    Get tag metadata (current: item only)
    '''
    tag_metadata = {"tag": tag}
    tag_metadata['first_seen'] = get_tag_first_seen(tag, r_int=r_int)
    tag_metadata['last_seen'] = get_tag_last_seen(tag, r_int=r_int)
    return tag_metadata

def get_tags_min_last_seen(l_tags, r_int=False):
    '''
    Get max last seen from a list of tags (current: item only)
    '''
    min_last_seen = 99999999
    for tag in l_tags:
        last_seen = get_tag_last_seen(tag, r_int=True)
        if last_seen < min_last_seen:
                min_last_seen = last_seen
    if r_int:
        return min_last_seen
    else:
        return str(min_last_seen)

def is_obj_tagged(object_id, tag):
    '''
    Check if a object is tagged

    :param object_id: object id
    :type domain: str
    :param tag: object type
    :type domain: str

    :return: is object tagged
    :rtype: boolean
    '''
    return r_serv_metadata.sismember('tag:{}'.format(object_id), tag)

def get_all_tags():
    return list(r_serv_tags.smembers('list_tags'))

def get_all_obj_tags(object_type):
    return list(r_serv_tags.smembers('list_tags:{}'.format(object_type)))

def get_obj_tag(object_id):
    '''
    Retun all the tags of a given object.
    :param object_id: (item_id, domain, ...)
    '''
    res = r_serv_metadata.smembers('tag:{}'.format(object_id))
    if res:
        return list(res)
    else:
        return []

def update_tag_first_seen(tag, tag_first_seen, tag_last_seen):
    if tag_first_seen == tag_last_seen:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_first_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'first_seen', tag_first_seen)
        # no tag in db
        else:
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'first_seen')
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'last_seen')
    else:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_first_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'first_seen', tag_first_seen)
        else:
            tag_first_seen = Date.date_add_day(tag_first_seen)
            update_tag_first_seen(tag, tag_first_seen, tag_last_seen)

def update_tag_last_seen(tag, tag_first_seen, tag_last_seen):
    if tag_first_seen == tag_last_seen:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_last_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'last_seen', tag_last_seen)
        # no tag in db
        else:
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'first_seen')
            r_serv_tags.hdel('tag_metadata:{}'.format(tag), 'last_seen')
    else:
        if r_serv_tags.scard('{}:{}'.format(tag, tag_last_seen)) > 0:
            r_serv_tags.hset('tag_metadata:{}'.format(tag), 'last_seen', tag_last_seen)
        else:
            # # TODO: # FIXME:
            #tag_last_seen = Date.date_substract_day(str(tag_last_seen))
            #update_tag_last_seen(tag, tag_first_seen, tag_last_seen)
            pass
{'fraction_non_alphanumeric': 0.11433457161389232, 'fraction_numerical': 0.004127601863317412, 'mean_word_length': 2.946939725389807, 'pattern_counts': {'":': 14, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3698581', 'n_tokens_mistral': 5926, 'n_tokens_neox': 5838, 'n_words': 1216}