<commit_before>import demistomock as demisto
from CommonServerPython import *
from CommonServerUserPython import *

''' IMPORTS '''
import os
import traceback
import requests
import zipfile
import io
from datetime import datetime as dt

# Disable insecure warnings
requests.packages.urllib3.disable_warnings()

''' GLOBALS/PARAMS '''

# Remove trailing slash to prevent wrong URL path to service
API_URL = demisto.params()['url'].rstrip('/')

# Should we use SSL
USE_SSL = not demisto.params().get('insecure', False)

# Remove proxy if not set to true in params
if not demisto.params().get('proxy'):
    os.environ.pop('HTTP_PROXY', None)
    os.environ.pop('HTTPS_PROXY', None)
    os.environ.pop('http_proxy', None)
    os.environ.pop('https_proxy', None)

THRESHOLD = int(demisto.params().get('threshold', 1))

COMPROMISED_IS_MALICIOUS = demisto.params().get('compromised_is_malicious', False)

# Headers to be sent in requests
HEADERS = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Accept': 'application/json'
}

''' HELPER FUNCTIONS '''


def http_request(method, command, data=None):
    url = f'{API_URL}/{command}/'
    demisto.info(f'{method} {url}')
    res = requests.request(method,
                           url,
                           verify=USE_SSL,
                           data=data,
                           headers=HEADERS)

    if res.status_code != 200:
        raise Exception(f'Error in API call {url} [{res.status_code}] - {res.reason}')

    return res


def reformat_date(date):
    try:
        return dt.strptime(date.rstrip(' UTC'), '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dT%H:%M:%S')
    except Exception:
        return 'Unknown'


def extract_zipped_buffer(buffer):
    with io.BytesIO() as bio:
        bio.write(buffer)
        with zipfile.ZipFile(bio) as z:
            return z.read(z.namelist()[0])


def query_url_information(url):
    return http_request('POST',
                        'url',
                        f'url={url}')


def query_host_information(host):
    return http_request('POST',
                        'host',
                        f'host={host}')


def query_payload_information(hash_type, hash):
    return http_request('POST',
                        'payload',
                        f'{hash_type}_hash={hash}')


def download_malware_sample(sha256):
    return http_request('GET',
                        f'download/{sha256}')


''' COMMANDS + REQUESTS FUNCTIONS '''


def test_module():
    """
    Performs basic get request to get item samples
    """
    http_request('POST', 'url')


def calculate_dbot_score(blacklists, threshold, compromised_is_malicious):
    blacklist_appearances = []
    for blacklist, status in blacklists.items():
        if blacklist == 'spamhaus_dbl':
            if status.endswith('domain') or (status.startswith('abused') and compromised_is_malicious):
                blacklist_appearances.append((blacklist, status))
        elif status == 'listed':
            blacklist_appearances.append((blacklist,))

    if len(blacklist_appearances) >= threshold:
        description = ''
        for appearance in blacklist_appearances:
            if len(appearance) == 1:
                description += f'Listed in {appearance[0]}. '
            elif len(appearance) == 2:
                description += f'Listed as {appearance[1]} in {appearance[0]}. '
            else:
                raise Exception('Unknown blacklist format in the response')

        return 3, description
    elif len(blacklist_appearances) > 0:
        return 2, None
    else:
        return 1, None


def url_command():
    url = demisto.args().get('url')
    try:
        url_information = query_url_information(url).json()

        ec = {
            'URL': {
                'Data': url
            },
            'DBotScore': {
                'Type': 'url',
                'Vendor': 'URLhaus',
                'Indicator': url
            }
        }

        if url_information['query_status'] == 'ok':
            # URLhaus output
            blacklist_information = []
            blacklists = url_information.get('blacklists', {})
            for bl_name, bl_status in blacklists.items():
                blacklist_information.append({'Name': bl_name,
                                              'Status': bl_status})<commit_msg>Made the code compatible with mypy<commit_after>import demistomock as demisto
from CommonServerPython import *
from CommonServerUserPython import *

''' IMPORTS '''
import os
import traceback
import requests
import zipfile
import io
from datetime import datetime as dt

# Disable insecure warnings
requests.packages.urllib3.disable_warnings()

''' GLOBALS/PARAMS '''

# Remove trailing slash to prevent wrong URL path to service
API_URL = demisto.params()['url'].rstrip('/')

# Should we use SSL
USE_SSL = not demisto.params().get('insecure', False)

# Remove proxy if not set to true in params
if not demisto.params().get('proxy'):
    os.environ.pop('HTTP_PROXY', None)
    os.environ.pop('HTTPS_PROXY', None)
    os.environ.pop('http_proxy', None)
    os.environ.pop('https_proxy', None)

THRESHOLD = int(demisto.params().get('threshold', 1))

COMPROMISED_IS_MALICIOUS = demisto.params().get('compromised_is_malicious', False)

# Headers to be sent in requests
HEADERS = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Accept': 'application/json'
}

''' HELPER FUNCTIONS '''


def http_request(method, command, data=None):
    url = f'{API_URL}/{command}/'
    demisto.info(f'{method} {url}')
    res = requests.request(method,
                           url,
                           verify=USE_SSL,
                           data=data,
                           headers=HEADERS)

    if res.status_code != 200:
        raise Exception(f'Error in API call {url} [{res.status_code}] - {res.reason}')

    return res


def reformat_date(date):
    try:
        return dt.strptime(date.rstrip(' UTC'), '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dT%H:%M:%S')
    except Exception:
        return 'Unknown'


def extract_zipped_buffer(buffer):
    with io.BytesIO() as bio:
        bio.write(buffer)
        with zipfile.ZipFile(bio) as z:
            return z.read(z.namelist()[0])


def query_url_information(url):
    return http_request('POST',
                        'url',
                        f'url={url}')


def query_host_information(host):
    return http_request('POST',
                        'host',
                        f'host={host}')


def query_payload_information(hash_type, hash):
    return http_request('POST',
                        'payload',
                        f'{hash_type}_hash={hash}')


def download_malware_sample(sha256):
    return http_request('GET',
                        f'download/{sha256}')


''' COMMANDS + REQUESTS FUNCTIONS '''


def test_module():
    """
    Performs basic get request to get item samples
    """
    http_request('POST', 'url')


def calculate_dbot_score(blacklists, threshold, compromised_is_malicious):
    blacklist_appearances = []
    for blacklist, status in blacklists.items():
        if blacklist == 'spamhaus_dbl':
            if status.endswith('domain') or (status.startswith('abused') and compromised_is_malicious):
                blacklist_appearances.append((blacklist, status))
        elif status == 'listed':
            blacklist_appearances.append((blacklist, None))

    if len(blacklist_appearances) >= threshold:
        description = ''
        for appearance in blacklist_appearances:
            if not appearance[1]:
                description += f'Listed in {appearance[0]}. '
            else:
                description += f'Listed as {appearance[1]} in {appearance[0]}. '

        return 3, description
    elif len(blacklist_appearances) > 0:
        return 2, None
    else:
        return 1, None


def url_command():
    url = demisto.args().get('url')
    try:
        url_information = query_url_information(url).json()

        ec = {
            'URL': {
                'Data': url
            },
            'DBotScore': {
                'Type': 'url',
                'Vendor': 'URLhaus',
                'Indicator': url
            }
        }

        if url_information['query_status'] == 'ok':
            # URLhaus output
            blacklist_information = []
            blacklists = url_information.get('blacklists', {})
            for bl_name, bl_status in blacklists.items():
                blacklist_information.append({'Name': bl_name,
                                              'Status': bl_status})
{'fraction_non_alphanumeric': 0.1231681786461968, 'fraction_numerical': 0.004768550825773436, 'mean_word_length': 2.2908534251817834, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3806168', 'n_tokens_mistral': 2595, 'n_tokens_neox': 2490, 'n_words': 569}