HEX
Server: LiteSpeed
System: Linux spg20.cloudpowerdns.com 5.14.0-611.35.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Wed Feb 25 03:46:09 EST 2026 x86_64
User: carrerup (3153)
PHP: 8.2.30
Disabled: NONE
Upload Files
File: //opt/bitninja-python-dojo/embedded/lib/python3.9/site-packages/plyara/utils.py
#!/usr/bin/env python3
# Copyright 2014 Christian Buia
# Copyright 2020 plyara Maintainers
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""plyara utility functions.

This module contains various utility functions for working with plyara output.
"""
import hashlib
import logging
import re

from plyara.core import Parser

# Initialize the logger
logger = logging.getLogger(__name__)


def is_valid_rule_name(entry):
    """Check to see if entry is a valid rule name.

    Args:
        entry: String containing rule name.

    Returns:
        bool
    """
    # Check if entry is blank
    if not entry:
        return False

    # Check length
    if len(entry) > 128:
        return False

    # Ensure doesn't start with a digit
    if entry[0].isdigit():
        return False

    # Accept only alphanumeric and underscores
    if not re.match(r'[a-zA-Z_][a-zA-Z_0-9]*$', entry):
        return False

    # Verify not in keywords
    if entry in Parser.KEYWORDS:
        return False

    return True


def is_valid_rule_tag(entry):
    """Check to see if entry is a valid rule tag.

    Args:
        entry: String containing tag.

    Returns:
        bool
    """
    # Same lexical conventions as name
    return is_valid_rule_name(entry)


def detect_imports(rule):
    """Take a parsed yararule and provide a list of required imports based on condition.

    Args:
        rule: Dict output from a parsed rule.

    Returns:
        list: Imports that are required.
    """
    detected_imports = list()
    condition_terms = rule['condition_terms']

    for imp in Parser.IMPORT_OPTIONS:
        imp_module = '{}.'.format(imp)

        for term in condition_terms:
            if term.startswith(imp_module):
                detected_imports.append(imp)
                break

    return detected_imports


def detect_dependencies(rule):
    """Take a parsed yararule and provide a list of external rule dependencies.

    Args:
        rule: Dict output from a parsed rule.

    Returns:
        list: External rule dependencies.
    """
    dependencies = list()
    string_iteration_variables = list()
    condition_terms = rule['condition_terms']

    # Number of terms for index iteration and reference
    term_count = len(condition_terms)

    for index in range(0, term_count):
        # Grab term by index
        term = condition_terms[index]

        if is_valid_rule_name(term) and (term not in Parser.IMPORT_OPTIONS):
            # Grab reference to previous term for logic checks
            if index > 0:
                previous_term = condition_terms[index - 1]
            else:
                previous_term = None

            # Grab reference to next term for logic checks
            if index < (term_count - 1):
                next_term = condition_terms[index + 1]
            else:
                next_term = None

            # Extend term indexes beyond wrapping parentheses for logic checks
            if previous_term == '(' and next_term == ')':
                if (index - 2) >= 0:
                    previous_term = condition_terms[index - 2]
                else:
                    previous_term = None

                if (index + 2) < term_count:
                    next_term = condition_terms[index + 2]
                else:
                    next_term = None

            # Check if reference is a variable for string iteration
            if term in string_iteration_variables:
                continue

            if previous_term in ('any', 'all', ) and next_term == 'in':
                string_iteration_variables.append(term)
                continue

            # Check for external string variable dependency
            if next_term in ('matches', 'contains',) or previous_term in ('matches', 'contains',):
                continue

            # Check for external integer variable dependency
            if next_term in Parser.COMPARISON_OPERATORS or previous_term in Parser.COMPARISON_OPERATORS:
                continue

            # Check for external boolean dependency may not be possible without stripping out valid rule references

            # Checks for likely rule reference
            if previous_term is None and next_term is None:
                dependencies.append(term)
            elif previous_term in ('and', 'or', ) or next_term in ('and', 'or', ):
                dependencies.append(term)

    return dependencies


def generate_logic_hash(rule):
    """Calculate hash value of rule strings and condition.

    Args:
        rule: Dict output from a parsed rule.

    Returns:
        str: Hexdigest SHA-256.
    """
    import warnings
    warnings.warn(
        'Utility generate_logic_hash() may be deprecated, see generate_hash()',
        PendingDeprecationWarning
    )
    strings = rule.get('strings', list())
    conditions = rule['condition_terms']

    string_values = list()
    condition_mapping = list()
    string_mapping = {'anonymous': list(), 'named': dict()}

    for entry in strings:
        name = entry['name']
        modifiers = entry.get('modifiers', list())

        # Handle string modifiers
        if modifiers:
            value = '{}<MODIFIED>{}'.format(entry['value'], ' & '.join(sorted(modifiers)))
        else:
            value = entry['value']

        if name == '$':
            # Track anonymous strings
            string_mapping['anonymous'].append(value)
        else:
            # Track named strings
            string_mapping['named'][name] = value

        # Track all string values
        string_values.append(value)

    # Sort all string values
    sorted_string_values = sorted(string_values)

    for condition in conditions:
        # All string references (sort for consistency)
        if condition == 'them' or condition == '$*':
            condition_mapping.append('<STRINGVALUE>{}'.format(' | '.join(sorted_string_values)))

        elif condition.startswith('$') and condition != '$':
            # Exact Match
            if condition in string_mapping['named']:
                condition_mapping.append('<STRINGVALUE>{}'.format(string_mapping['named'][condition]))
            # Wildcard Match
            elif '*' in condition:
                wildcard_strings = list()
                condition = condition.replace('$', r'\$').replace('*', '.*')
                pattern = re.compile(condition)

                for name, value in string_mapping['named'].items():
                    if pattern.match(name):
                        wildcard_strings.append(value)

                wildcard_strings.sort()
                condition_mapping.append('<STRINGVALUE>{}'.format(' | '.join(wildcard_strings)))
            else:
                logger.error('[!] Unhandled String Condition {}'.format(condition))

        # Count Match
        elif condition.startswith('#') and condition != '#':
            condition = condition.replace('#', '$')

            if condition in string_mapping['named']:
                condition_mapping.append('<COUNTOFSTRING>{}'.format(string_mapping['named'][condition]))
            else:
                logger.error('[!] Unhandled String Count Condition {}'.format(condition))

        else:
            condition_mapping.append(condition)

    logic_hash = hashlib.sha256(''.join(condition_mapping).encode()).hexdigest()
    return logic_hash


def generate_hash(rule, secure_hash=None):
    """Calculate a secure hash of the logic in the rule strings and condition.

    If the resultant hashes are identical for two YARA rules, the rules will match on identical content.
    The reverse it not true, so two rules that match the same content may not generate the same hash.
    For example, if a rule only contains one string, the logic for 'any of' and 'all of' generate different hashes,
    but the rules contain the same logic.

    Args:
        rule: Dict output from a parsed rule.
        secure_hash: Alternate hash function, defaults to SHA-256

    Returns:
        str: hexdigest
    """
    condition_string_prefaces = ('$', '!', '#', '@')
    if secure_hash is None:
        hf = hashlib.sha256()
    else:
        hf = secure_hash()

    strings = rule.get('strings', list())
    conditions = rule['condition_terms']

    string_values = list()
    condition_mapping = list()
    string_mapping = {'anonymous': list(), 'named': dict()}

    for entry in strings:
        name = entry['name']
        modifiers = entry.get('modifiers', list())

        if entry['type'] == 'byte':
            value = re.sub(r'[^-a-fA-F?0-9\[\]{}]+', '', entry['value'])
        elif entry['type'] == 'text':
            value = '{}'.format(entry['value'])
        else:
            value = entry['value']

        # Handle string modifiers
        if modifiers:
            value += '<MODIFIED>{}'.format(' & '.join(sorted(modifiers)))

        if name == '$':
            # Track anonymous strings
            string_mapping['anonymous'].append(value)
        else:
            # Track named strings
            string_mapping['named'][name] = value

        # Track all string values
        string_values.append(value)

    # Sort all string values
    string_values.sort()

    for condition in conditions:
        # All string references (sort for consistency)
        if condition == 'them' or condition == '$*':
            all_values = '<STRINGVALUE>{}'.format(' | '.join(string_values))
            if condition == 'them':
                condition_mapping.extend(['(', all_values, ')'])
            else:
                condition_mapping.append(all_values)

        elif condition.startswith('$') and condition != '$':
            # Exact Match
            if condition in string_mapping['named']:
                condition_mapping.append('<STRINGVALUE>{}'.format(string_mapping['named'][condition]))
            # Wildcard Match
            elif '*' in condition:
                wildcard_strings = list()
                condition = condition.replace('$', r'\$').replace('*', '.*')
                pattern = re.compile(condition)

                for name, value in string_mapping['named'].items():
                    if pattern.match(name):
                        wildcard_strings.append(value)

                wildcard_strings.sort()
                condition_mapping.append('<STRINGVALUE>{}'.format(' | '.join(wildcard_strings)))
            else:
                logger.error('[!] Unhandled String Condition "{}" in "{}"'.format(condition, ' '.join(conditions)))

        # Count Match
        elif condition[:1] in condition_string_prefaces and condition not in ('#', '!='):
            symbol = condition[:1]
            condition = '${}'.format(condition[1:])
            if symbol == '#':
                symbol_type = 'COUNTOFSTRING'
            elif symbol == '@':
                symbol_type = 'POSITIONOFSTRING'
            elif symbol == '!':
                symbol_type = 'LENGTHOFSTRING'
            elif symbol == condition == '$':
                symbol_type = 'ANONYMOUSSTRING'
            else:
                symbol_type = 'UNKNOWN'

            if condition in string_mapping['named']:
                condition_mapping.append('<{}>{}'.format(symbol_type, string_mapping['named'][condition]))
            else:
                condition_mapping.append('<{}>{}'.format(symbol_type, condition))
                logger.error('[!] Unhandled {} Condition "{}" in "{}"'.format(
                    symbol_type, symbol, ' '.join(conditions))
                )

        else:
            condition_mapping.append(condition)
    hf.update(''.join(condition_mapping).encode())
    hexdigest = hf.hexdigest()

    return hexdigest


def rebuild_yara_rule(rule, condition_indents=False):
    """Take a parsed yararule and rebuild it into a usable one.

    Args:
        rule: Dict output from a parsed rule.
        condition_indents: Use nested indentation for condition

    Returns:
        str: Formatted text string of YARA rule.
    """
    rule_format = "{imports}{scopes}rule {rulename}{tags}\n{{{meta}{strings}{condition}\n}}\n"

    rule_name = rule['rule_name']

    # Rule Imports
    if rule.get('imports'):
        unpacked_imports = ['import "{}"\n'.format(entry) for entry in rule['imports']]
        rule_imports = '{}\n'.format(''.join(unpacked_imports))
    else:
        rule_imports = str()

    # Rule Scopes
    if rule.get('scopes'):
        rule_scopes = '{} '.format(' '.join(rule['scopes']))
    else:
        rule_scopes = str()

    # Rule Tags
    if rule.get('tags'):
        rule_tags = ' : {}'.format(' '.join(rule['tags']))
    else:
        rule_tags = str()

    # Rule Metadata
    if rule.get('metadata'):
        unpacked_meta = []
        kv_list = [(k, ) + (v, ) for dic in rule['metadata'] for k, v in dic.items()]

        # Check for and handle correctly quoting string metadata
        for k, v in kv_list:
            if isinstance(v, bool):
                v = str(v).lower()
            elif isinstance(v, int):
                v = str(v)
            else:
                v = '"{}"'.format(v)
            unpacked_meta.append('\n\t\t{key} = {value}'.format(key=k, value=v))
        rule_meta = '\n\tmeta:{}\n'.format(''.join(unpacked_meta))
    else:
        rule_meta = str()

    # Rule Strings
    if rule.get('strings'):

        string_container = list()

        for rule_string in rule['strings']:
            if 'modifiers' in rule_string:
                string_modifiers = [x for x in rule_string['modifiers'] if isinstance(x, str)]

                if rule_string['type'] == 'text':
                    string_format = '\n\t\t{} = "{}" {}'
                else:
                    string_format = '\n\t\t{} = {} {}'
                fstring = string_format.format(rule_string['name'], rule_string['value'], ' '.join(string_modifiers))

            else:
                if rule_string['type'] == 'text':
                    string_format = '\n\t\t{} = "{}"'
                else:
                    string_format = '\n\t\t{} = {}'
                fstring = string_format.format(rule_string['name'], rule_string['value'])

            string_container.append(fstring)

        rule_strings = '\n\tstrings:{}\n'.format(''.join(string_container))
    else:
        rule_strings = str()

    if rule.get('condition_terms'):
        # Format condition with appropriate whitespace between keywords
        cond = list()
        indents = '\n\t\t'
        for term in rule['condition_terms']:

            if condition_indents:
                if term == '(':
                    indents = indents + '\t'
                if term == ')' and len(indents) > 3:
                    indents = indents[:-1]

            if not cond:

                if term in Parser.FUNCTION_KEYWORDS:
                    cond.append(term)

                elif term in Parser.KEYWORDS:
                    cond.append(term)
                    cond.append(' ')

                else:
                    cond.append(term)

            else:

                if cond[-1][-1] in (' ', '\t') and term in Parser.FUNCTION_KEYWORDS:
                    cond.append(term)

                elif cond[-1][-1] not in (' ', '\t') and term in Parser.FUNCTION_KEYWORDS:
                    cond.append(' ')
                    cond.append(term)

                elif cond[-1][-1] in (' ', '\t') and term in Parser.KEYWORDS:
                    cond.append(term)
                    cond.append(' ')
                    if condition_indents and term in ('and', 'or'):
                        cond.append(indents)

                elif cond[-1][-1] not in (' ', '\t') and term in Parser.KEYWORDS:
                    cond.append(' ')
                    cond.append(term)
                    cond.append(' ')
                    if condition_indents and term in ('and', 'or'):
                        cond.append(indents)

                elif cond[-1][-1] in (' ', '\t') and term == ':':
                    cond.append(term)
                    cond.append(' ')

                elif cond[-1][-1] not in (' ', '\t') and term == ':':
                    cond.append(' ')
                    cond.append(term)
                    cond.append(' ')

                else:
                    cond.append(term)

        fcondition = ''.join(cond).rstrip(' ')
        rule_condition = '\n\tcondition:{}{}'.format('\n\t\t', fcondition)
    else:
        rule_condition = str()

    formatted_rule = rule_format.format(imports=rule_imports,
                                        rulename=rule_name,
                                        tags=rule_tags,
                                        meta=rule_meta,
                                        scopes=rule_scopes,
                                        strings=rule_strings,
                                        condition=rule_condition)

    return formatted_rule