From 23b353c0558c384a0a48d58c1071786ab4f60999 Mon Sep 17 00:00:00 2001 From: 0x-Apollyon <149272988+0x-Apollyon@users.noreply.github.com> Date: Tue, 30 Dec 2025 08:55:16 +0530 Subject: [PATCH 1/2] main commit --- .gitignore | 3 +- nova/__init__.py | 2 + nova/core/__init__.py | 2 + nova/core/matcher.py | 76 +++++++++++++++++++++++++++++++---- nova/core/parser.py | 78 +++++++++++++++++++++++++++++------- nova/core/rules.py | 15 +++++++ nova/evaluators/base.py | 4 ++ nova/evaluators/condition.py | 46 +++++++++++++++++---- nova/evaluators/fuzzy.py | 59 +++++++++++++++++++++++++++ nova/novarun.py | 15 ++++++- 10 files changed, 267 insertions(+), 33 deletions(-) create mode 100644 nova/evaluators/fuzzy.py diff --git a/.gitignore b/.gitignore index ad08a9d..1d21095 100644 --- a/.gitignore +++ b/.gitignore @@ -7,7 +7,8 @@ wheels/ *.egg-info # Virtual environments .venv +venv/ optimized_matcher.py optimized_test.py # Pytest cache -tests/ \ No newline at end of file +tests/ diff --git a/nova/__init__.py b/nova/__init__.py index 87e67a8..ed8d13d 100644 --- a/nova/__init__.py +++ b/nova/__init__.py @@ -27,6 +27,7 @@ KeywordPattern, SemanticPattern, LLMPattern, + FuzzyPattern, NovaRule ) from nova.core.matcher import NovaMatcher @@ -37,6 +38,7 @@ __all__ = [ 'KeywordPattern', + 'FuzzyPattern', 'SemanticPattern', 'LLMPattern', 'NovaRule', diff --git a/nova/core/__init__.py b/nova/core/__init__.py index 16d526f..6b0080e 100644 --- a/nova/core/__init__.py +++ b/nova/core/__init__.py @@ -9,6 +9,7 @@ from nova.core.rules import ( KeywordPattern, + FuzzyPattern, SemanticPattern, LLMPattern, NovaRule @@ -19,6 +20,7 @@ __all__ = [ 'KeywordPattern', + 'FuzzyPattern', 'SemanticPattern', 'LLMPattern', 'NovaRule', diff --git a/nova/core/matcher.py b/nova/core/matcher.py index 2ce22ea..f4f49ea 100644 --- a/nova/core/matcher.py +++ b/nova/core/matcher.py @@ -10,7 +10,7 @@ from typing import Dict, List, Tuple, Optional, Any, Set, Callable import re -from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern +from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern, FuzzyPattern from nova.evaluators.keywords import DefaultKeywordEvaluator from nova.evaluators.condition import evaluate_condition from nova.utils.logger import get_logger @@ -22,6 +22,7 @@ DefaultSemanticEvaluator = None OpenAIEvaluator = None LLMEvaluator = None +DefaultFuzzyEvaluator = None try: from nova.evaluators.semantics import DefaultSemanticEvaluator @@ -33,6 +34,11 @@ except ImportError: pass +try: + from nova.evaluators.fuzzy import DefaultFuzzyEvaluator +except ImportError: + pass + class NovaMatcher: """ @@ -44,6 +50,7 @@ class NovaMatcher: def __init__(self, rule: NovaRule, keyword_evaluator: Optional[DefaultKeywordEvaluator] = None, + fuzzy_evaluator: Optional[Any] = None, semantic_evaluator: Optional[Any] = None, # DefaultSemanticEvaluator might not be available llm_evaluator: Optional[Any] = None, # LLMEvaluator might not be available create_llm_evaluator: bool = True): @@ -77,7 +84,14 @@ def __init__(self, needs_llm = True elif rule and 'llm.' in rule.condition.lower(): needs_llm = True - + + needs_fuzzy = False + # Check if rule has fuzzy section (handle attribute safely) + if rule and rule.fuzzy: + needs_fuzzy = True + elif rule and 'fuzzy' in rule.condition.lower(): + needs_fuzzy = True + # Only initialize semantic evaluator if needed if needs_semantic: if semantic_evaluator: @@ -107,6 +121,18 @@ def __init__(self, if needs_llm: logger.warning("Rule requires LLM evaluation but no evaluator provided and creation disabled.") # Remove the verbose message for rules that don't need LLM evaluation + + if fuzzy_evaluator: + self.fuzzy_evaluator = fuzzy_evaluator + elif needs_fuzzy: + if DefaultFuzzyEvaluator is not None: + self.fuzzy_evaluator = DefaultFuzzyEvaluator() + else: + self.fuzzy_evaluator = None + logger.warning("Rule requires fuzzy evaluation but rapidfuzz not available.") + else: + self.fuzzy_evaluator = None + # Pre-compile keyword patterns for performance if rule: @@ -144,23 +170,24 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: """ needed_patterns = { 'keywords': set(), + 'fuzzy' : set(), 'semantics': set(), 'llm': set(), 'section_wildcards': set() } # Check for section wildcards - for section in ['keywords', 'semantics', 'llm']: + for section in ['keywords', 'semantics', 'llm' , 'fuzzy']: if f"{section}.*" in condition: needed_patterns['section_wildcards'].add(section) # Check for "any of" section wildcards - for section in ['keywords', 'semantics', 'llm']: + for section in ['keywords', 'semantics', 'llm' , 'fuzzy']: if f"any of {section}.*" in condition: needed_patterns['section_wildcards'].add(section) # Check for direct variable references with section prefixes - for section in ['keywords', 'semantics', 'llm']: + for section in ['keywords', 'semantics', 'llm' , 'fuzzy']: # Exact references: "section.$var" pattern = rf'{section}\.\$([a-zA-Z0-9_]+)(?!\*)' for match in re.finditer(pattern, condition): @@ -188,6 +215,8 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: needed_patterns['semantics'].add(var_name) elif var_name in self.rule.llms: needed_patterns['llm'].add(var_name) + elif var_name in self.rule.fuzzy: + needed_patterns['fuzzy'].add(var_name) # Check for "any of" wildcards any_of_pattern = r'any\s+of\s+\(\$([a-zA-Z0-9_]+)\*\)' @@ -197,6 +226,7 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: # Add all matching variables from all sections for section, patterns in [ ('keywords', self.rule.keywords), + ('fuzzy' , self.rule.fuzzy), ('semantics', self.rule.semantics), ('llm', self.rule.llms) ]: @@ -222,6 +252,7 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: # Track all evaluation results for debugging all_keyword_matches = {} + all_fuzzy_matches = {} all_semantic_matches = {} all_semantic_scores = {} all_llm_matches = {} @@ -232,6 +263,7 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: # Initialize filtered dictionaries to hold results needed for condition evaluation keyword_matches = {} + fuzzy_matches = {} semantic_matches = {} llm_matches = {} @@ -258,6 +290,11 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: temperature = pattern.threshold lazy_evaluations[('llm', key)] = lambda p=pattern, t=temperature: self.llm_evaluator.evaluate_prompt(p.pattern, prompt, temperature=t) + if self.fuzzy_evaluator: + for key , pattern in self.rule.fuzzy.items(): + if key in needed_patterns['fuzzy'] or 'fuzzy' in needed_patterns['section_wildcards']: + lazy_evaluations[('fuzzy', key)] = lambda p=pattern, k=key: self.fuzzy_evaluator.evaluate(p, prompt) + # First evaluate only the patterns that are directly referenced in the condition # Create a list of sections and variables to evaluate based on what evaluators are available patterns_to_evaluate = [('keywords', name) for name in needed_patterns['keywords']] @@ -267,6 +304,9 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: if self.llm_evaluator: patterns_to_evaluate += [('llm', name) for name in needed_patterns['llm']] + + if self.fuzzy_evaluator: + patterns_to_evaluate += [('fuzzy', name) for name in needed_patterns['fuzzy']] # Evaluate each pattern for section, var_name in patterns_to_evaluate: @@ -287,6 +327,10 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: all_llm_matches[var_name] = matched all_llm_scores[var_name] = confidence llm_matches[var_name] = matched + elif section == 'fuzzy': + matched = lazy_evaluations[eval_key]() + all_fuzzy_matches[var_name] = matched + fuzzy_matches[var_name] = matched except Exception as e: logger.error(f"Error evaluating {section}.{var_name}: {str(e)}") # Default to False on error @@ -301,6 +345,9 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: all_llm_matches[var_name] = False all_llm_scores[var_name] = 0.0 llm_matches[var_name] = False + elif section == 'fuzzy': + all_fuzzy_matches[var_name] = False + fuzzy_matches[var_name] = False # Process section wildcards if needed for section in needed_patterns['section_wildcards']: @@ -338,6 +385,14 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: except Exception: all_llm_matches[key] = False all_llm_scores[key] = 0.0 + elif section == 'fuzzy' and self.fuzzy_evaluator and not all_fuzzy_matches: + for key, pattern in self.rule.fuzzy.items(): + if key not in all_fuzzy_matches: + try: + result = self.fuzzy_evaluator.evaluate(pattern, prompt) + all_fuzzy_matches[key] = result + except Exception: + all_fuzzy_matches[key] = False # Evaluate condition if provided has_match = False @@ -351,13 +406,16 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: semantic_matches.update(all_semantic_matches) if 'llm' in needed_patterns['section_wildcards'] and self.llm_evaluator: llm_matches.update(all_llm_matches) + if 'fuzzy' in needed_patterns['section_wildcards'] and self.fuzzy_evaluator: + fuzzy_matches.update(all_fuzzy_matches) # Use the condition evaluator with filtered match types condition_result = evaluate_condition( condition, - keyword_matches, + keyword_matches, semantic_matches, - llm_matches + llm_matches, + fuzzy_matches ) has_match = condition_result else: @@ -372,6 +430,7 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: 'matching_keywords': {k: v for k, v in keyword_matches.items() if v}, 'matching_semantics': {k: v for k, v in semantic_matches.items() if v}, 'matching_llm': {k: v for k, v in llm_matches.items() if v}, + 'matching_fuzzy': {k: v for k, v in fuzzy_matches.items() if v}, 'semantic_scores': all_semantic_scores, 'llm_scores': all_llm_scores, 'debug': { @@ -379,7 +438,8 @@ def check_prompt(self, prompt: str) -> Dict[str, Any]: 'condition_result': condition_result, 'all_keyword_matches': all_keyword_matches, 'all_semantic_matches': all_semantic_matches, - 'all_llm_matches': all_llm_matches + 'all_llm_matches': all_llm_matches, + 'all_fuzzy_matches': all_fuzzy_matches } } diff --git a/nova/core/parser.py b/nova/core/parser.py index 75180fa..a16bb9f 100644 --- a/nova/core/parser.py +++ b/nova/core/parser.py @@ -9,7 +9,7 @@ import re from typing import Dict, List, Optional, Set, Any -from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern +from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern , FuzzyPattern from nova.utils.logger import get_logger # Get logger for this module @@ -18,10 +18,10 @@ # Precompile regex patterns for better performance RULE_NAME_PATTERN = re.compile(r'rule\s+(\w+)(?:\s*{)?') RULE_START_PATTERN = re.compile(r'rule\s+\w+\s*{?') -SECTION_WILDCARD_PATTERN = re.compile(r'(keywords|semantics|llm)\.\*') -VAR_PREFIX_PATTERN = re.compile(r'(keywords|semantics|llm)\.\$([a-zA-Z0-9_]+)\*') +SECTION_WILDCARD_PATTERN = re.compile(r'(keywords|semantics|llm|fuzzy)\.\*') +VAR_PREFIX_PATTERN = re.compile(r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)\*') ANY_OF_WILDCARD_PATTERN = re.compile(r'any\s+of\s+\(\$([a-zA-Z0-9_]+)\*\)') -DIRECT_VAR_PATTERN = re.compile(r'(keywords|semantics|llm)\.\$([a-zA-Z0-9_]+)') +DIRECT_VAR_PATTERN = re.compile(r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)') STANDALONE_VAR_PATTERN = re.compile(r'(? NovaRule: @@ -68,7 +69,8 @@ def parse(self, content: str) -> NovaRule: self.variable_names = { 'keywords': set(), 'semantics': set(), - 'llm': set() + 'llm': set(), + 'fuzzy': set() } # Check rule declaration and extract name @@ -118,6 +120,8 @@ def _parse_section(self, section: str, content: List[str]): self.rule.meta = self._parse_meta_section(content) elif section == "keywords": self.rule.keywords = self._parse_keywords_section(content) + elif section == "fuzzy": + self.rule.fuzzy = self._parse_fuzzy_section(content) elif section == "semantics": self.rule.semantics = self._parse_semantics_section(content) elif section == "llm": @@ -151,6 +155,51 @@ def _parse_meta_section(self, content: List[str]) -> Dict[str, str]: result[key] = value return result + + def _parse_fuzzy_section(self , content: List[str]) -> Dict[str, FuzzyPattern]: + """Parse fuzzy patterns from the fuzzy section.""" + result = {} + seen_variables = set() + + for line_num, line in enumerate(content): + # Skip empty lines and comments + if not line or line.strip().startswith('//') or line.strip().startswith('#'): + continue + + # Syntax: $var = "pattern" (threshold) + if '=' not in line: + raise NovaParserError( + f"Invalid fuzzy pattern at line {line_num}: '{line}'. Format: '$var = \"pattern\" (threshold)'") + + key, value_part = map(str.strip, line.split('=', 1)) + + # Validation + if not key.startswith('$'): + raise NovaParserError(f"Invalid fuzzy variable '{key}'. Must start with $") + + if key in seen_variables: + raise NovaParserError(f"Duplicate variable '{key}' in fuzzy section.") + + seen_variables.add(key) + self.variable_names['fuzzy'].add(key) + + # Extract pattern and threshold + # Expected value_part: "admin" (80) + match = re.search(r'"([^"]+)"\s*\((\d+)\)', value_part) + if not match: + raise NovaParserError( + f"Invalid fuzzy value '{value_part}'. format: \"text\" (80)") + + pattern_text = match.group(1) + threshold = int(match.group(2)) + + if not (0 <= threshold <= 100): + raise NovaParserError(f"Fuzzy threshold {threshold} must be between 0 and 100") + + # Create object + result[key] = FuzzyPattern(pattern=pattern_text, threshold=threshold) + + return result def _parse_semantics_section(self, content: List[str]) -> Dict[str, SemanticPattern]: """Parse semantic patterns from the semantics section.""" @@ -400,7 +449,7 @@ def _parse_condition_section(self, content: List[str]) -> str: f"Invalid nested quantifiers at position {pos}: '{match.group(0)}'. Quantifiers cannot be nested. Context: '...{context}...'") # Validate section references - check for misspelled section names - valid_sections = ['keywords', 'semantics', 'llm'] + valid_sections = ['keywords', 'semantics', 'llm' , 'fuzzy'] # Find all potential section references that aren't in our valid list for match in SECTION_REF_PATTERN.finditer(condition): @@ -527,7 +576,7 @@ def _validate_rule_structure(self): """ try: # Check that at least one pattern type is specified - if not self.rule.keywords and not self.rule.semantics and not self.rule.llms: + if not self.rule.keywords and not self.rule.semantics and not self.rule.llms and not self.rule.fuzzy: raise NovaParserError( f"Rule '{self.rule.name}' must specify at least one of: keywords, semantics, llm") @@ -592,12 +641,12 @@ def _validate_rule_structure(self): syntax_check_condition = cleaned_condition # Replace section wildcards first since they're easier to detect - section_wildcards = ["keywords.*", "semantics.*", "llm.*"] + section_wildcards = ["keywords.*", "semantics.*", "llm.*" , "fuzzy.*"] for wildcard in section_wildcards: syntax_check_condition = syntax_check_condition.replace(wildcard, "True") # Replace "any of section.*" patterns - these cause syntax errors - syntax_check_condition = re.sub(r'any\s+of\s+(keywords|semantics|llm)\.\*', 'True', syntax_check_condition) + syntax_check_condition = re.sub(r'any\s+of\s+(keywords|semantics|llm|fuzzy)\.\*', 'True', syntax_check_condition) # Replace variable references with True syntax_check_condition = re.sub(r'[a-zA-Z0-9_]+\.\$[a-zA-Z0-9_]+\*?', 'True', syntax_check_condition) @@ -690,7 +739,8 @@ def _check_unused_variables(self): section_wildcards = { 'keywords': False, 'semantics': False, - 'llm': False + 'llm': False, + 'fuzzy': False } for section in section_wildcards: @@ -700,7 +750,7 @@ def _check_unused_variables(self): used_variables.update(self.variable_names[section]) # Check for section-specific prefix wildcards (e.g., keywords.$bypass*) - for section in ['keywords', 'semantics', 'llm']: + for section in ['keywords', 'semantics', 'llm' , 'fuzzy']: pattern = rf'{section}\.\$([a-zA-Z0-9_]+)\*' for match in re.finditer(pattern, condition): prefix = match.group(1) @@ -720,7 +770,7 @@ def _check_unused_variables(self): used_variables.add(var) # Check for direct variable references - var_pattern = r'(keywords|semantics|llm)\.\$([a-zA-Z0-9_]+)(?!\*)' + var_pattern = r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)(?!\*)' for match in re.finditer(var_pattern, condition): section = match.group(1) var_name = f"${match.group(2)}" @@ -743,7 +793,7 @@ def _validate_direct_variables(self): # Simple check for explicitly referenced variables (only for non-wildcard vars) # Patterns like 'semantics.$var' and 'llm.$var' - for section in ['semantics', 'llm']: + for section in ['semantics', 'llm' , 'fuzzy']: pattern = rf'{section}\.\$(\w+)' for match in re.finditer(pattern, condition): var_name = f"${match.group(1)}" diff --git a/nova/core/rules.py b/nova/core/rules.py index 8f8ea10..f4bf5af 100644 --- a/nova/core/rules.py +++ b/nova/core/rules.py @@ -26,6 +26,20 @@ class KeywordPattern: case_sensitive: bool = False # Default to case-insensitive +@dataclass +class FuzzyPattern: + """ + Pattern for fuzzy matching with support for regex and case sensitivity. + + Attributes: + pattern: The string or regex pattern to match + case_sensitive: Whether the match should be case-sensitive + """ + pattern: str + threshold: float + case_sensitive: bool = False # Default to case-insensitive + + @dataclass class SemanticPattern: """ @@ -70,4 +84,5 @@ class NovaRule: keywords: Dict[str, KeywordPattern] = field(default_factory=dict) semantics: Dict[str, SemanticPattern] = field(default_factory=dict) llms: Dict[str, LLMPattern] = field(default_factory=dict) + fuzzy: Dict[str , FuzzyPattern] = field(default_factory=dict) condition: str = "" \ No newline at end of file diff --git a/nova/evaluators/base.py b/nova/evaluators/base.py index 504c2b2..a903f42 100644 --- a/nova/evaluators/base.py +++ b/nova/evaluators/base.py @@ -33,6 +33,10 @@ class KeywordEvaluator(BaseEvaluator): """Base class for keyword pattern evaluators.""" pass +class FuzzyEvaluator(BaseEvaluator): + """Base class for fuzzy pattern evaluators.""" + pass + class SemanticEvaluator(BaseEvaluator): """Base class for semantic pattern evaluators.""" diff --git a/nova/evaluators/condition.py b/nova/evaluators/condition.py index d7aede6..4a3ceaf 100644 --- a/nova/evaluators/condition.py +++ b/nova/evaluators/condition.py @@ -11,7 +11,8 @@ import re def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], - semantic_matches: Dict[str, bool], llm_matches: Dict[str, bool] = None) -> bool: + semantic_matches: Dict[str, bool], llm_matches: Dict[str, bool] = None , + fuzzy_matches: Dict[str, bool] = None) -> bool: """ Evaluate a condition expression against pattern match results. Handles wildcards correctly with improved parsing for complex expressions. @@ -32,6 +33,10 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], # Initialize llm_matches if not provided if llm_matches is None: llm_matches = {} + + # Initialise fuzzy_matches if not provided + if fuzzy_matches is None: + fuzzy_matches = {} # Make a copy of the original condition for debugging original_condition = condition @@ -70,10 +75,14 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], if "any of llm.*" in eval_condition: any_llm = any(llm_matches.values()) eval_condition = eval_condition.replace("any of llm.*", "True" if any_llm else "False") + + if "any of fuzzy.*" in eval_condition: + any_fuzzy = any(fuzzy_matches.values()) + eval_condition = eval_condition.replace("any of fuzzy.*", "True" if any_fuzzy else "False") # Handle section-specific prefix wildcards # Pattern matches "section.$prefix*" - pattern = r'(keywords|semantics|llm)\.\$([a-zA-Z0-9_]+)\*' + pattern = r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)\*' for match in re.finditer(pattern, eval_condition): section = match.group(1).lower() prefix = match.group(2) @@ -83,7 +92,8 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], matches_dict = { 'keywords': keyword_matches, 'semantics': semantic_matches, - 'llm': llm_matches + 'llm': llm_matches, + 'fuzzy': fuzzy_matches, }.get(section, {}) # Check if any variable with this prefix matches @@ -108,6 +118,10 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], if "llm.*" in eval_condition: any_llm = any(llm_matches.values()) eval_condition = eval_condition.replace("llm.*", "True" if any_llm else "False") + + if "fuzzy.*" in eval_condition: + any_fuzzy = any(fuzzy_matches.values()) + eval_condition = eval_condition.replace("fuzzy.*", "True" if any_fuzzy else "False") # Handle "any of" with wildcards - pattern matches: "any of ($prefix*)" any_of_pattern = r'any\s+of\s+\(\$([a-zA-Z0-9_]+)\*\)' @@ -117,7 +131,7 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], # Check if any variable with this prefix matches in any section matches = False - for var_dict in [keyword_matches, semantic_matches, llm_matches]: + for var_dict in [keyword_matches, semantic_matches, llm_matches , fuzzy_matches]: for var, value in var_dict.items(): if var[1:].startswith(prefix) and value: matches = True @@ -129,7 +143,7 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], eval_condition = eval_condition.replace(original, "True" if matches else "False") # Handle "N of" pattern - replace with actual boolean result - n_of_pattern = r'(\d+)\s+of\s+(keywords|semantics|llm)' + n_of_pattern = r'(\d+)\s+of\s+(keywords|semantics|llm|fuzzy)' for match in re.finditer(n_of_pattern, eval_condition): original = match.group(0) n = int(match.group(1)) @@ -145,6 +159,9 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], elif category == "llm": match_count = sum(llm_matches.values()) result = match_count >= n + elif category == "fuzzy": + match_count = sum(fuzzy_matches.values()) + result = match_count >= n else: result = False @@ -155,7 +172,7 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], # Handle different formats: "section.$var", "$var" # First, handle fully qualified variables (section.$var) - section_var_pattern = r'(keywords|semantics|llm)\.\$([a-zA-Z0-9_]+)(?!\*)' + section_var_pattern = r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)(?!\*)' for match in re.finditer(section_var_pattern, eval_condition): section = match.group(1) var_name = "$" + match.group(2) @@ -169,6 +186,8 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], match_value = semantic_matches[var_name] elif section == "llm" and var_name in llm_matches: match_value = llm_matches[var_name] + elif section == "fuzzy" and var_name in fuzzy_matches: + match_value = fuzzy_matches[var_name] # Replace in evaluation condition eval_condition = eval_condition.replace(original, "True" if match_value else "False") @@ -187,6 +206,8 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], match_value = semantic_matches[var_name] elif var_name in llm_matches: match_value = llm_matches[var_name] + elif var_name in fuzzy_matches: + match_value = fuzzy_matches[var_name] # Replace in evaluation condition eval_condition = eval_condition.replace(original, "True" if match_value else "False") @@ -232,7 +253,7 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], # Special case handling for common patterns that might fail in eval # If the condition is just a single section.$ reference and there's a match - if re.match(r'^(keywords|semantics|llm)\.\$[a-zA-Z0-9_]+$', original_condition): + if re.match(r'^(keywords|semantics|llm|fuzzy)\.\$[a-zA-Z0-9_]+$', original_condition): try: section, var = original_condition.split('.') if section == "keywords" and var in keyword_matches: @@ -241,6 +262,8 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], return semantic_matches[var] elif section == "llm" and var in llm_matches: return llm_matches[var] + elif section == "fuzzy" and var in fuzzy_matches: + return fuzzy_matches[var] except Exception: # If any error occurs in this special case handling, continue to the next one pass @@ -262,6 +285,9 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], elif part.startswith("llm.$"): var = part.replace("llm.", "") results.append(llm_matches.get(var, False)) + elif part.startswith("fuzzy.$"): + var = part.replace("fuzzy.", "") + results.append(fuzzy_matches.get(var, False)) elif part.startswith("$"): if part in keyword_matches: results.append(keyword_matches[part]) @@ -269,6 +295,8 @@ def evaluate_condition(condition: str, keyword_matches: Dict[str, bool], results.append(semantic_matches[part]) elif part in llm_matches: results.append(llm_matches[part]) + elif part in fuzzy_matches: + results.append(fuzzy_matches[part]) else: results.append(False) @@ -322,12 +350,14 @@ def check_prompt_safe(prompt, matcher_obj): "matching_keywords": {}, "matching_semantics": {}, "matching_llm": {}, + "matching_fuzzy" : {}, "debug": { "condition": matcher_obj.rule.condition, "condition_result": False, "all_keyword_matches": {}, "all_semantic_matches": {}, - "all_llm_matches": {} + "all_llm_matches": {}, + "all_fuzzy_matches": {} } } diff --git a/nova/evaluators/fuzzy.py b/nova/evaluators/fuzzy.py new file mode 100644 index 0000000..7cd8e09 --- /dev/null +++ b/nova/evaluators/fuzzy.py @@ -0,0 +1,59 @@ +import re +from typing import Dict, Union +from nova.core.rules import FuzzyPattern +from nova.evaluators.base import FuzzyEvaluator +from nova.utils.logger import get_logger +try: + from rapidfuzz import fuzz + RAPIDFUZZ_AVAILABLE = True +except ImportError: + RAPIDFUZZ_AVAILABLE = False + fuzz = None + +# Get logger for this module +logger = get_logger("nova.evaluators.fuzzy") + + +class DefaultFuzzyEvaluator(FuzzyEvaluator): + """Default fuzzy pattern evaluator supporting case sensitivity.""" + + def __init__(self): + if not RAPIDFUZZ_AVAILABLE: + logger.error("RapidFuzz library not found. Fuzzy matching will fail. Install with: pip install rapidfuzz") + + + def evaluate(self, pattern: FuzzyPattern , text: str) -> bool: + """ + Check if a fuzzy pattern matches the text. + + Args: + pattern: The FuzzyPattern to match + text: The text to evaluate + + Returns: + Boolean indicating whether the pattern matches + """ + + if not RAPIDFUZZ_AVAILABLE or fuzz is None: + return False + + if not text or not pattern.pattern: + return False + + target = pattern.pattern + content = text + + #Handle Case Sensitivity + if not pattern.case_sensitive: + target = target.lower() + content = content.lower() + + #Calculate Score + # partial_ratio is best for "needle in haystack" + # It detects "admin" inside "I am administrator" with score 100 + score = fuzz.partial_ratio(target, content) + + is_match = (score >= pattern.threshold) + return is_match + + \ No newline at end of file diff --git a/nova/novarun.py b/nova/novarun.py index 12d7d8a..4f0517e 100644 --- a/nova/novarun.py +++ b/nova/novarun.py @@ -362,12 +362,12 @@ def print_result(result: Dict[str, Any], rule_path: str, prompt: str, verbose: b if result['matched']: print(f"\n{Fore.CYAN}Matching Patterns:") - if result['matching_keywords']: + if 'matching_keywords' in result and result['matching_keywords']: print(f" {Fore.GREEN}Keywords:") for key in result['matching_keywords']: print(f" {Fore.WHITE}• {key}") - if result['matching_semantics']: + if 'matching_semantics' in result and result['matching_semantics']: print(f" {Fore.GREEN}Semantics:") for key in result['matching_semantics']: print(f" {Fore.WHITE}• {key}") @@ -376,6 +376,11 @@ def print_result(result: Dict[str, Any], rule_path: str, prompt: str, verbose: b print(f" {Fore.GREEN}LLM:") for key in result['matching_llm']: print(f" {Fore.WHITE}• {key}") + + if 'matching_fuzzy' in result and result['matching_fuzzy']: + print(f" {Fore.GREEN}Fuzzy Matches:") + for key in result['matching_fuzzy']: + print(f" {Fore.WHITE}• {key}") # Print debug information if verbose mode is enabled if verbose: @@ -399,6 +404,12 @@ def print_result(result: Dict[str, Any], rule_path: str, prompt: str, verbose: b for key, value in debug['all_keyword_matches'].items(): match_color = Fore.GREEN if value else Fore.RED print(f" {Fore.CYAN}{key}: {match_color}{value}") + + if 'all_fuzzy_matches' in debug: + print(f"\n{Fore.MAGENTA}Fuzzy Matches (Detailed):") + for key, value in debug['all_fuzzy_matches'].items(): + match_color = Fore.GREEN if value else Fore.RED + print(f" {Fore.CYAN}{key}: {match_color}{value}") if 'all_semantic_matches' in debug: print(f"\n{Fore.MAGENTA}Semantic Matches:") From 3d5c77032e8c2ea7b3dab066c7b01e4771373f01 Mon Sep 17 00:00:00 2001 From: 0x-Apollyon <149272988+0x-Apollyon@users.noreply.github.com> Date: Thu, 5 Feb 2026 00:07:20 +0530 Subject: [PATCH 2/2] Fuzzy --- nova/core/matcher.py | 459 ++++++++++++++++++++++----------------- nova/core/parser.py | 269 +++++++++-------------- nova/core/rules.py | 2 + nova/evaluators/fuzzy.py | 4 +- 4 files changed, 358 insertions(+), 376 deletions(-) diff --git a/nova/core/matcher.py b/nova/core/matcher.py index e5597f0..18e98f3 100644 --- a/nova/core/matcher.py +++ b/nova/core/matcher.py @@ -10,50 +10,34 @@ from typing import Dict, List, Tuple, Optional, Any, Set, Callable import re -from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern +from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern, FuzzyPattern from nova.evaluators.keywords import DefaultKeywordEvaluator -from nova.evaluators.condition import ( - evaluate_condition, - can_semantics_change_outcome, - can_llm_change_outcome -) +from nova.evaluators.condition import evaluate_condition from nova.utils.logger import get_logger # Get logger for this module logger = get_logger("nova.matcher") -# Lazy-loaded module references - avoids expensive imports at module load time -_DefaultSemanticEvaluator = None -_OpenAIEvaluator = None -_LLMEvaluator = None -_imports_attempted = {'semantic': False, 'llm': False} +# Optional imports - these may not be available if extras not installed +DefaultSemanticEvaluator = None +OpenAIEvaluator = None +LLMEvaluator = None +DefaultFuzzyEvaluator = None +try: + from nova.evaluators.semantics import DefaultSemanticEvaluator +except ImportError: + pass -def _get_semantic_evaluator_class(): - """Lazy import of DefaultSemanticEvaluator - only loads when actually needed.""" - global _DefaultSemanticEvaluator, _imports_attempted - if not _imports_attempted['semantic']: - _imports_attempted['semantic'] = True - try: - from nova.evaluators.semantics import DefaultSemanticEvaluator - _DefaultSemanticEvaluator = DefaultSemanticEvaluator - except ImportError: - pass - return _DefaultSemanticEvaluator +try: + from nova.evaluators.llm import OpenAIEvaluator, LLMEvaluator +except ImportError: + pass - -def _get_llm_evaluator_classes(): - """Lazy import of LLM evaluators - only loads when actually needed.""" - global _OpenAIEvaluator, _LLMEvaluator, _imports_attempted - if not _imports_attempted['llm']: - _imports_attempted['llm'] = True - try: - from nova.evaluators.llm import OpenAIEvaluator, LLMEvaluator - _OpenAIEvaluator = OpenAIEvaluator - _LLMEvaluator = LLMEvaluator - except ImportError: - pass - return _OpenAIEvaluator, _LLMEvaluator +try: + from nova.evaluators.fuzzy import DefaultFuzzyEvaluator +except ImportError: + pass class NovaMatcher: @@ -66,6 +50,7 @@ class NovaMatcher: def __init__(self, rule: NovaRule, keyword_evaluator: Optional[DefaultKeywordEvaluator] = None, + fuzzy_evaluator: Optional[Any] = None, semantic_evaluator: Optional[Any] = None, # DefaultSemanticEvaluator might not be available llm_evaluator: Optional[Any] = None, # LLMEvaluator might not be available create_llm_evaluator: bool = True): @@ -99,47 +84,59 @@ def __init__(self, needs_llm = True elif rule and 'llm.' in rule.condition.lower(): needs_llm = True - + + needs_fuzzy = False + # Check if rule has fuzzy section (handle attribute safely) + if rule and rule.fuzzy: + needs_fuzzy = True + elif rule and 'fuzzy' in rule.condition.lower(): + needs_fuzzy = True + # Only initialize semantic evaluator if needed if needs_semantic: if semantic_evaluator: self.semantic_evaluator = semantic_evaluator + elif DefaultSemanticEvaluator is not None: + self.semantic_evaluator = DefaultSemanticEvaluator() else: - # Lazy import - only load the module when actually needed - SemanticEvalClass = _get_semantic_evaluator_class() - if SemanticEvalClass is not None: - self.semantic_evaluator = SemanticEvalClass() - else: - self.semantic_evaluator = None - logger.warning("Rule requires semantic evaluation but sentence-transformers not available. Install with: pip install nova-hunting") + self.semantic_evaluator = None + logger.warning("Rule requires semantic evaluation but sentence-transformers not available. Install with: pip install nova-hunting[llm]") else: self.semantic_evaluator = None - + # Handle LLM evaluator initialization if llm_evaluator: # Use provided evaluator regardless of need self.llm_evaluator = llm_evaluator elif needs_llm and create_llm_evaluator: - # Lazy import - only load the module when actually needed - OpenAIEval, _ = _get_llm_evaluator_classes() - if OpenAIEval is not None: - self.llm_evaluator = OpenAIEval() + # Create a new evaluator if needed and allowed to create + if OpenAIEvaluator is not None: + self.llm_evaluator = OpenAIEvaluator() else: self.llm_evaluator = None - logger.warning("Rule requires LLM evaluation but LLM dependencies not available. Install with: pip install nova-hunting") + logger.warning("Rule requires LLM evaluation but LLM dependencies not available. Install with: pip install nova-hunting[llm]") else: # No evaluator provided and either not needed or not allowed to create self.llm_evaluator = None if needs_llm: logger.warning("Rule requires LLM evaluation but no evaluator provided and creation disabled.") + # Remove the verbose message for rules that don't need LLM evaluation + + if fuzzy_evaluator: + self.fuzzy_evaluator = fuzzy_evaluator + elif needs_fuzzy: + if DefaultFuzzyEvaluator is not None: + self.fuzzy_evaluator = DefaultFuzzyEvaluator() + else: + self.fuzzy_evaluator = None + logger.warning("Rule requires fuzzy evaluation but rapidfuzz not available.") + else: + self.fuzzy_evaluator = None + # Pre-compile keyword patterns for performance if rule: self._precompile_patterns() - # Pre-compute condition analysis (static for rule's lifetime) - self._cached_needed_patterns = self._analyze_condition(rule.condition) - else: - self._cached_needed_patterns = None def _precompile_patterns(self): """Pre-compile regex patterns for better performance.""" @@ -154,17 +151,12 @@ def set_rule(self, rule: NovaRule): """ Update the matcher with a new rule. This is more efficient than creating a new matcher instance. - + Args: rule: The new NovaRule to match against """ self.rule = rule self._precompile_patterns() - # Update cached condition analysis for new rule - if rule: - self._cached_needed_patterns = self._analyze_condition(rule.condition) - else: - self._cached_needed_patterns = None def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: """ @@ -178,23 +170,24 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: """ needed_patterns = { 'keywords': set(), + 'fuzzy': set(), # Track which keywords need fuzzy evaluation 'semantics': set(), 'llm': set(), 'section_wildcards': set() } - # Check for section wildcards - for section in ['keywords', 'semantics', 'llm']: + # 1. Check for section wildcards (e.g., keywords.*, fuzzy.*) + for section in ['keywords', 'semantics', 'llm', 'fuzzy']: if f"{section}.*" in condition: needed_patterns['section_wildcards'].add(section) - # Check for "any of" section wildcards - for section in ['keywords', 'semantics', 'llm']: + # 2. Check for "any of" section wildcards + for section in ['keywords', 'semantics', 'llm', 'fuzzy']: if f"any of {section}.*" in condition: needed_patterns['section_wildcards'].add(section) - # Check for direct variable references with section prefixes - for section in ['keywords', 'semantics', 'llm']: + # 3. Check for direct variable references with section prefixes (e.g., fuzzy.$var) + for section in ['keywords', 'semantics', 'llm', 'fuzzy']: # Exact references: "section.$var" pattern = rf'{section}\.\$([a-zA-Z0-9_]+)(?!\*)' for match in re.finditer(pattern, condition): @@ -205,17 +198,26 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: wildcard_pattern = rf'{section}\.\$([a-zA-Z0-9_]+)\*' for match in re.finditer(wildcard_pattern, condition): prefix = match.group(1) + + # REDIRECT: If the condition asks for 'fuzzy.', we look for the data in 'keywords' + if section == 'fuzzy': + search_in = self.rule.keywords + else: + search_in = getattr(self.rule, section, {}) + # Add all matching variables to needed patterns - for var_name in getattr(self.rule, section, {}): + for var_name in search_in: if var_name[1:].startswith(prefix): # Remove $ from var name needed_patterns[section].add(var_name) - # Check for standalone variables ($var) + # 4. Check for standalone variables ($var) var_pattern = r'(? Dict[str, Set[str]]: elif var_name in self.rule.llms: needed_patterns['llm'].add(var_name) - # Check for "any of" wildcards + # 5. Check for "any of" wildcards (e.g., any of ($prefix*)) any_of_pattern = r'any\s+of\s+\(\$([a-zA-Z0-9_]+)\*\)' for match in re.finditer(any_of_pattern, condition): prefix = match.group(1) - # Add all matching variables from all sections + # We removed 'fuzzy' from this list because the data is in 'keywords' for section, patterns in [ ('keywords', self.rule.keywords), ('semantics', self.rule.semantics), @@ -243,168 +245,219 @@ def _analyze_condition(self, condition: str) -> Dict[str, Set[str]]: def check_prompt(self, prompt: str) -> Dict[str, Any]: """ Check if a prompt matches the rule. - - Uses short-circuit evaluation: evaluates patterns in order of cost - (keywords → semantics → LLM) and stops as soon as condition is satisfied. - + Args: prompt: The prompt text to check - + Returns: Dictionary containing match results and details """ - # Use cached condition analysis (pre-computed at initialization) + # Extract variables needed based on the condition condition = self.rule.condition - needed_patterns = self._cached_needed_patterns or self._analyze_condition(condition) - + needed_patterns = self._analyze_condition(condition) + # Track all evaluation results for debugging all_keyword_matches = {} + all_fuzzy_matches = {} all_semantic_matches = {} all_semantic_scores = {} all_llm_matches = {} all_llm_scores = {} - + + # Dictionary to hold evaluation functions for lazy evaluation + lazy_evaluations = {} + # Initialize filtered dictionaries to hold results needed for condition evaluation keyword_matches = {} + fuzzy_matches = {} semantic_matches = {} llm_matches = {} - - # Helper function to build final results - def build_results(has_match: bool, condition_result: Any) -> Dict[str, Any]: - return { - 'matched': has_match, - 'rule_name': self.rule.name, - 'meta': self.rule.meta, - 'matching_keywords': {k: v for k, v in keyword_matches.items() if v}, - 'matching_semantics': {k: v for k, v in semantic_matches.items() if v}, - 'matching_llm': {k: v for k, v in llm_matches.items() if v}, - 'semantic_scores': all_semantic_scores, - 'llm_scores': all_llm_scores, - 'debug': { - 'condition': condition, - 'condition_result': condition_result, - 'all_keyword_matches': all_keyword_matches, - 'all_semantic_matches': all_semantic_matches, - 'all_llm_matches': all_llm_matches - } - } - - # ------ STAGE 1: EVALUATE KEYWORDS (fastest) ------ - - # Evaluate keywords needed by condition or section wildcards + + # ------ LAZY PATTERN EVALUATION ------ + + # Set up lazy evaluation functions for keywords for key, pattern in self.rule.keywords.items(): + # Only include if explicitly needed by condition or section wildcard is used if key in needed_patterns['keywords'] or 'keywords' in needed_patterns['section_wildcards']: - try: - result = self.keyword_evaluator.evaluate(pattern, prompt, key) - all_keyword_matches[key] = result - keyword_matches[key] = result - except Exception as e: - logger.error(f"Error evaluating keywords.{key}: {str(e)}") - all_keyword_matches[key] = False - keyword_matches[key] = False - - # Update keyword_matches for wildcards - if 'keywords' in needed_patterns['section_wildcards']: - keyword_matches.update(all_keyword_matches) - - # SHORT-CIRCUIT CHECK: Try to evaluate condition with just keywords - # If condition is satisfied (e.g., OR condition with keyword match), skip expensive evaluations - if condition: - try: - early_result = evaluate_condition(condition, keyword_matches, {}, {}) - if early_result: - # Condition satisfied with just keywords - no need for semantic/LLM - return build_results(True, early_result) - except Exception: - # Can't determine yet, continue with more patterns - pass - - # ------ STAGE 2: EVALUATE SEMANTICS (medium cost) ------ - - # Check if semantics could possibly change the outcome before running expensive evaluation - skip_semantics = False - if condition and not can_semantics_change_outcome(condition, keyword_matches): - # Semantics can't change the outcome - skip this evaluation stage - skip_semantics = True - logger.debug(f"Skipping semantic evaluation for rule '{self.rule.name}' - can't change outcome") - - if self.semantic_evaluator and not skip_semantics: + lazy_evaluations[('keywords', key)] = lambda p=pattern, k=key: self.keyword_evaluator.evaluate(p, prompt, k) + + # Set up lazy evaluation functions for semantics (if evaluator exists) + if self.semantic_evaluator: for key, pattern in self.rule.semantics.items(): + # Only include if explicitly needed by condition or section wildcard is used if key in needed_patterns['semantics'] or 'semantics' in needed_patterns['section_wildcards']: - try: - matched, score = self.semantic_evaluator.evaluate(pattern, prompt) - all_semantic_matches[key] = matched - all_semantic_scores[key] = score - semantic_matches[key] = matched - except Exception as e: - logger.error(f"Error evaluating semantics.{key}: {str(e)}") - all_semantic_matches[key] = False - all_semantic_scores[key] = 0.0 - semantic_matches[key] = False - - # Update semantic_matches for wildcards - if 'semantics' in needed_patterns['section_wildcards']: - semantic_matches.update(all_semantic_matches) - - # SHORT-CIRCUIT CHECK: Try with keywords + semantics - if condition: - try: - early_result = evaluate_condition(condition, keyword_matches, semantic_matches, {}) - if early_result: - # Condition satisfied - no need for expensive LLM evaluation - return build_results(True, early_result) - except Exception: - # Can't determine yet, continue - pass - - # ------ STAGE 3: EVALUATE LLM PATTERNS (most expensive) ------ - - # Check if LLM could possibly change the outcome before running expensive API calls - skip_llm = False - if condition and not can_llm_change_outcome(condition, keyword_matches, semantic_matches): - # LLM can't change the outcome - skip this evaluation stage - skip_llm = True - logger.debug(f"Skipping LLM evaluation for rule '{self.rule.name}' - can't change outcome") - - if self.llm_evaluator and not skip_llm: + lazy_evaluations[('semantics', key)] = lambda p=pattern: self.semantic_evaluator.evaluate(p, prompt) + + # Set up lazy evaluation functions for LLMs (if evaluator exists) + if self.llm_evaluator: for key, pattern in self.rule.llms.items(): + # Only include if explicitly needed by condition or section wildcard is used if key in needed_patterns['llm'] or 'llm' in needed_patterns['section_wildcards']: - try: - # Note: pattern.threshold is used as LLM temperature (0.0=deterministic, 1.0=creative) - # The LLM itself determines if the pattern matches; threshold controls response variability - llm_temperature = pattern.threshold - matched, confidence, details = self.llm_evaluator.evaluate_prompt( - pattern.pattern, prompt, temperature=llm_temperature - ) - all_llm_matches[key] = matched - all_llm_scores[key] = confidence - llm_matches[key] = matched - except Exception as e: - logger.error(f"Error evaluating llm.{key}: {str(e)}") - all_llm_matches[key] = False - all_llm_scores[key] = 0.0 - llm_matches[key] = False - - # Update llm_matches for wildcards - if 'llm' in needed_patterns['section_wildcards']: - llm_matches.update(all_llm_matches) - - # ------ FINAL CONDITION EVALUATION ------ - + temperature = pattern.threshold + lazy_evaluations[('llm', key)] = lambda p=pattern, t=temperature: self.llm_evaluator.evaluate_prompt(p.pattern, prompt, temperature=t) + + if self.fuzzy_evaluator: + for var_name in needed_patterns['fuzzy']: + pattern_obj = self.rule.keywords.get(var_name) + if pattern_obj: + lazy_evaluations[('fuzzy', var_name)] = lambda p=pattern_obj, k=var_name:self.fuzzy_evaluator.evaluate(p, prompt, key=k) + + # First evaluate only the patterns that are directly referenced in the condition + # Create a list of sections and variables to evaluate based on what evaluators are available + patterns_to_evaluate = [('keywords', name) for name in needed_patterns['keywords']] + + if self.semantic_evaluator: + patterns_to_evaluate += [('semantics', name) for name in needed_patterns['semantics']] + + if self.llm_evaluator: + patterns_to_evaluate += [('llm', name) for name in needed_patterns['llm']] + + if self.fuzzy_evaluator: + patterns_to_evaluate += [('fuzzy', name) for name in needed_patterns['fuzzy']] + + # Evaluate each pattern + for section, var_name in patterns_to_evaluate: + eval_key = (section, var_name) + if eval_key in lazy_evaluations: + try: + if section == 'keywords': + result = lazy_evaluations[eval_key]() + all_keyword_matches[var_name] = result + keyword_matches[var_name] = result + elif section == 'semantics': + matched, score = lazy_evaluations[eval_key]() + all_semantic_matches[var_name] = matched + all_semantic_scores[var_name] = score + semantic_matches[var_name] = matched + elif section == 'llm': + matched, confidence, details = lazy_evaluations[eval_key]() + all_llm_matches[var_name] = matched + all_llm_scores[var_name] = confidence + llm_matches[var_name] = matched + elif section == 'fuzzy': + matched = lazy_evaluations[eval_key]() + all_fuzzy_matches[var_name] = matched + fuzzy_matches[var_name] = matched + except Exception as e: + logger.error(f"Error evaluating {section}.{var_name}: {str(e)}") + # Default to False on error + if section == 'keywords': + all_keyword_matches[var_name] = False + keyword_matches[var_name] = False + elif section == 'semantics': + all_semantic_matches[var_name] = False + all_semantic_scores[var_name] = 0.0 + semantic_matches[var_name] = False + elif section == 'llm': + all_llm_matches[var_name] = False + all_llm_scores[var_name] = 0.0 + llm_matches[var_name] = False + elif section == 'fuzzy': + all_fuzzy_matches[var_name] = False + fuzzy_matches[var_name] = False + + # Process section wildcards if needed + for section in needed_patterns['section_wildcards']: + if section == 'keywords' and not all_keyword_matches: + # Only evaluate all keywords if needed and not already evaluated + for key, pattern in self.rule.keywords.items(): + if key not in all_keyword_matches: + try: + result = self.keyword_evaluator.evaluate(pattern, prompt, key) + all_keyword_matches[key] = result + except Exception: + all_keyword_matches[key] = False + + elif section == 'semantics' and self.semantic_evaluator and not all_semantic_matches: + # Only evaluate all semantics if needed and not already evaluated + for key, pattern in self.rule.semantics.items(): + if key not in all_semantic_matches: + try: + matched, score = self.semantic_evaluator.evaluate(pattern, prompt) + all_semantic_matches[key] = matched + all_semantic_scores[key] = score + except Exception: + all_semantic_matches[key] = False + all_semantic_scores[key] = 0.0 + + elif section == 'llm' and self.llm_evaluator and not all_llm_matches: + # Only evaluate all LLM patterns if needed and not already evaluated + for key, pattern in self.rule.llms.items(): + if key not in all_llm_matches: + try: + temperature = pattern.threshold + matched, confidence, details = self.llm_evaluator.evaluate_prompt(pattern.pattern, prompt, temperature=temperature) + all_llm_matches[key] = matched + all_llm_scores[key] = confidence + except Exception: + all_llm_matches[key] = False + all_llm_scores[key] = 0.0 + elif section == 'fuzzy' and self.fuzzy_evaluator: + # Proactively scan all keywords marked as fuzzy for the fuzzy.* wildcard + for key, pattern in self.rule.keywords.items(): + # Only process keywords the user marked with the 'fuzzy' keyword + if getattr(pattern, 'is_fuzzy', False): + if key not in all_fuzzy_matches: + try: + result = self.fuzzy_evaluator.evaluate(pattern, prompt, key=key) + all_fuzzy_matches[key] = result + # We must put the result in fuzzy_matches for the condition evaluator + fuzzy_matches[key] = result + except Exception: + all_fuzzy_matches[key] = False + fuzzy_matches[key] = False + else: + # If already evaluated by a direct reference earlier, + # ensure it's copied into the condition dict + fuzzy_matches[key] = all_fuzzy_matches[key] + + # Evaluate condition if provided has_match = False condition_result = None - + if condition: - # Final evaluation with all pattern results + # Make sure eval_condition gets all variables that might be needed by wildcards + if 'keywords' in needed_patterns['section_wildcards']: + keyword_matches.update(all_keyword_matches) + if 'semantics' in needed_patterns['section_wildcards'] and self.semantic_evaluator: + semantic_matches.update(all_semantic_matches) + if 'llm' in needed_patterns['section_wildcards'] and self.llm_evaluator: + llm_matches.update(all_llm_matches) + if 'fuzzy' in needed_patterns['section_wildcards'] and self.fuzzy_evaluator: + fuzzy_matches.update(all_fuzzy_matches) + + # Use the condition evaluator with filtered match types condition_result = evaluate_condition( - condition, + condition, keyword_matches, - semantic_matches, - llm_matches + semantic_matches, + llm_matches, + fuzzy_matches ) has_match = condition_result else: # Fall back to original behavior if no condition is specified has_match = any(keyword_matches.values()) or any(semantic_matches.values()) or any(llm_matches.values()) - - return build_results(has_match, condition_result) \ No newline at end of file + + # Build results with matching variables only + results = { + 'matched': has_match, + 'rule_name': self.rule.name, + 'meta': self.rule.meta, + 'matching_keywords': {k: v for k, v in keyword_matches.items() if v}, + 'matching_semantics': {k: v for k, v in semantic_matches.items() if v}, + 'matching_llm': {k: v for k, v in llm_matches.items() if v}, + 'matching_fuzzy': {k: v for k, v in fuzzy_matches.items() if v}, + 'semantic_scores': all_semantic_scores, + 'llm_scores': all_llm_scores, + 'debug': { + 'condition': condition, + 'condition_result': condition_result, + 'all_keyword_matches': all_keyword_matches, + 'all_semantic_matches': all_semantic_matches, + 'all_llm_matches': all_llm_matches, + 'all_fuzzy_matches': all_fuzzy_matches + } + } + + return results \ No newline at end of file diff --git a/nova/core/parser.py b/nova/core/parser.py index a16bb9f..ec7e246 100644 --- a/nova/core/parser.py +++ b/nova/core/parser.py @@ -9,7 +9,7 @@ import re from typing import Dict, List, Optional, Set, Any -from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern , FuzzyPattern +from nova.core.rules import NovaRule, KeywordPattern, SemanticPattern, LLMPattern from nova.utils.logger import get_logger # Get logger for this module @@ -46,8 +46,7 @@ def __init__(self): self.variable_names = { 'keywords': set(), 'semantics': set(), - 'llm': set(), - 'fuzzy': set() + 'llm': set() } def parse(self, content: str) -> NovaRule: @@ -69,8 +68,7 @@ def parse(self, content: str) -> NovaRule: self.variable_names = { 'keywords': set(), 'semantics': set(), - 'llm': set(), - 'fuzzy': set() + 'llm': set() } # Check rule declaration and extract name @@ -120,8 +118,6 @@ def _parse_section(self, section: str, content: List[str]): self.rule.meta = self._parse_meta_section(content) elif section == "keywords": self.rule.keywords = self._parse_keywords_section(content) - elif section == "fuzzy": - self.rule.fuzzy = self._parse_fuzzy_section(content) elif section == "semantics": self.rule.semantics = self._parse_semantics_section(content) elif section == "llm": @@ -156,50 +152,6 @@ def _parse_meta_section(self, content: List[str]) -> Dict[str, str]: return result - def _parse_fuzzy_section(self , content: List[str]) -> Dict[str, FuzzyPattern]: - """Parse fuzzy patterns from the fuzzy section.""" - result = {} - seen_variables = set() - - for line_num, line in enumerate(content): - # Skip empty lines and comments - if not line or line.strip().startswith('//') or line.strip().startswith('#'): - continue - - # Syntax: $var = "pattern" (threshold) - if '=' not in line: - raise NovaParserError( - f"Invalid fuzzy pattern at line {line_num}: '{line}'. Format: '$var = \"pattern\" (threshold)'") - - key, value_part = map(str.strip, line.split('=', 1)) - - # Validation - if not key.startswith('$'): - raise NovaParserError(f"Invalid fuzzy variable '{key}'. Must start with $") - - if key in seen_variables: - raise NovaParserError(f"Duplicate variable '{key}' in fuzzy section.") - - seen_variables.add(key) - self.variable_names['fuzzy'].add(key) - - # Extract pattern and threshold - # Expected value_part: "admin" (80) - match = re.search(r'"([^"]+)"\s*\((\d+)\)', value_part) - if not match: - raise NovaParserError( - f"Invalid fuzzy value '{value_part}'. format: \"text\" (80)") - - pattern_text = match.group(1) - threshold = int(match.group(2)) - - if not (0 <= threshold <= 100): - raise NovaParserError(f"Fuzzy threshold {threshold} must be between 0 and 100") - - # Create object - result[key] = FuzzyPattern(pattern=pattern_text, threshold=threshold) - - return result def _parse_semantics_section(self, content: List[str]) -> Dict[str, SemanticPattern]: """Parse semantic patterns from the semantics section.""" @@ -345,74 +297,73 @@ def _parse_keywords_section(self, content: List[str]) -> Dict[str, KeywordPatter seen_variables = set() for line_num, line in enumerate(content): - # Skip empty lines and comments if not line or line.strip().startswith('#'): continue - # Split the line into variable and pattern try: key, value = map(str.strip, line.split('=', 1)) except ValueError: - # Line doesn't contain an equals sign - raise NovaParserError( - f"Invalid keyword pattern at line {line_num}: '{line}'. Patterns must be in the format '$var = \"pattern\"'") + raise NovaParserError(f"Invalid keyword pattern at line {line_num}: '{line}'") - # Check for "$var" format if not key.startswith('$'): - raise NovaParserError( - f"Invalid keyword variable at line {line_num}: '{key}'. Variable names must start with $") - - # Check for duplicate variable names - if key in seen_variables: - raise NovaParserError( - f"Duplicate variable '{key}' at line {line_num}. Variable names must be unique within each section.") + raise NovaParserError(f"Invalid variable name '{key}' at line {line_num}") - # Track that we've seen this variable seen_variables.add(key) - - # Track variable name for reference self.variable_names['keywords'].add(key) value = value.strip() - + + # --- ADD THESE INITIALIZATIONS HERE --- + is_fuzzy = False + threshold = 80 + case_sensitive = False + # -------------------------------------- + + # 1. Check for Fuzzy Marker at the end of the line + # Matches: "pattern" (85) fuzzy OR "pattern" fuzzy + fuzzy_match = re.search(r'\((?P\d+)\)\s+fuzzy$|(?Pfuzzy)$', value, re.IGNORECASE) + if fuzzy_match: + is_fuzzy = True + if fuzzy_match.group('threshold'): + threshold = int(fuzzy_match.group('threshold')) + + # Remove the fuzzy part from the value string so it doesn't break regex/string logic + value = re.sub(r'\s*\(\d+\)\s+fuzzy$|\s+fuzzy$', '', value, flags=re.IGNORECASE).strip() + + # 2. Identify if it is a Regex is_regex = value.startswith('/') and value.rstrip('i').endswith('/') - case_sensitive = False # Default to case-insensitive for all patterns if is_regex: - # Regex pattern handling if value.endswith('i'): - value = value[1:-2] # Remove /.../ and i flag + value = value[1:-2] else: - value = value[1:-1] # Remove /.../ only - - # Check if there's a case:true modifier + value = value[1:-1] + if "case:true" in value: - parts = value.split("case:true", 1) - value = parts[0].strip() + value = value.replace("case:true", "").strip() case_sensitive = True - # Validate the regex pattern try: re.compile(value) except re.error as e: - raise NovaParserError( - f"Invalid regex pattern at line {line_num}: '{value}'. Regex error: {str(e)}") + raise NovaParserError(f"Invalid regex: {e}") else: - # Regular string pattern handling - if not (value.startswith('"') and value.endswith('"')) and not (value.startswith("'") and value.endswith("'")): - raise NovaParserError( - f"Invalid keyword pattern at line {line_num}: '{value}'. Patterns must be in quotes or as regex.") - - # Extract pattern without quotes - value = value[1:-1] + # 3. Handle standard quoted strings + if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")): + value = value[1:-1] - # Check for case sensitivity modifier if "case:true" in value: - parts = value.split("case:true", 1) - value = parts[0].strip() + value = value.replace("case:true", "").strip() case_sensitive = True - - result[key] = KeywordPattern(pattern=value, is_regex=is_regex, case_sensitive=case_sensitive) + + # Now all variables (is_fuzzy, threshold, etc.) are defined! + result[key] = KeywordPattern( + pattern=value, + is_regex=is_regex, + case_sensitive=case_sensitive, + is_fuzzy=is_fuzzy, + threshold=threshold + ) return result @@ -576,7 +527,7 @@ def _validate_rule_structure(self): """ try: # Check that at least one pattern type is specified - if not self.rule.keywords and not self.rule.semantics and not self.rule.llms and not self.rule.fuzzy: + if not self.rule.keywords and not self.rule.semantics and not self.rule.llms: raise NovaParserError( f"Rule '{self.rule.name}' must specify at least one of: keywords, semantics, llm") @@ -641,12 +592,12 @@ def _validate_rule_structure(self): syntax_check_condition = cleaned_condition # Replace section wildcards first since they're easier to detect - section_wildcards = ["keywords.*", "semantics.*", "llm.*" , "fuzzy.*"] + section_wildcards = ["keywords.*", "semantics.*", "llm.*"] for wildcard in section_wildcards: syntax_check_condition = syntax_check_condition.replace(wildcard, "True") # Replace "any of section.*" patterns - these cause syntax errors - syntax_check_condition = re.sub(r'any\s+of\s+(keywords|semantics|llm|fuzzy)\.\*', 'True', syntax_check_condition) + syntax_check_condition = re.sub(r'any\s+of\s+(keywords|semantics|llm)\.\*', 'True', syntax_check_condition) # Replace variable references with True syntax_check_condition = re.sub(r'[a-zA-Z0-9_]+\.\$[a-zA-Z0-9_]+\*?', 'True', syntax_check_condition) @@ -723,8 +674,8 @@ def _validate_rule_structure(self): def _check_unused_variables(self): """ - Check for variables defined in pattern sections but not used in the condition, - even when wildcards are present. + Check for variables defined in pattern sections but not used in the condition. + Updated to handle fuzzy-to-keyword mapping. """ # Get all defined variables all_variables = set() @@ -735,50 +686,41 @@ def _check_unused_variables(self): used_variables = set() condition = self.rule.condition - # Check for variables used via section wildcards - section_wildcards = { - 'keywords': False, - 'semantics': False, - 'llm': False, - 'fuzzy': False - } + # 1. Check for section wildcards (e.g., keywords.*, fuzzy.*) + section_wildcards = ['keywords', 'semantics', 'llm', 'fuzzy'] for section in section_wildcards: - if f"{section}.*" in condition: - section_wildcards[section] = True - # Mark all variables in this section as used - used_variables.update(self.variable_names[section]) - - # Check for section-specific prefix wildcards (e.g., keywords.$bypass*) - for section in ['keywords', 'semantics', 'llm' , 'fuzzy']: + if f"{section}.*" in condition or f"any of {section}.*" in condition: + if section == 'fuzzy': + # 'fuzzy.*' uses all keywords marked as fuzzy + for var_name, pattern in self.rule.keywords.items(): + if getattr(pattern, 'is_fuzzy', False): + used_variables.add(var_name) + elif section in self.variable_names: + used_variables.update(self.variable_names[section]) + + # 2. Check for prefix wildcards (e.g., keywords.$bypass*) + for section in ['keywords', 'semantics', 'llm', 'fuzzy']: pattern = rf'{section}\.\$([a-zA-Z0-9_]+)\*' for match in re.finditer(pattern, condition): prefix = match.group(1) - # Mark all variables with this prefix in this section as used - for var in self.variable_names[section]: - if var[1:].startswith(prefix): # Remove $ from var name - used_variables.add(var) - - # Check for "any of" wildcard patterns - any_of_pattern = r'any\s+of\s+\(\$([a-zA-Z0-9_]+)\*\)' - for match in re.finditer(any_of_pattern, condition): - prefix = match.group(1) - # Mark variables with this prefix from all sections as used - for section_vars in self.variable_names.values(): - for var in section_vars: - if var[1:].startswith(prefix): - used_variables.add(var) - - # Check for direct variable references + # Map fuzzy prefix to keywords section + target_section = 'keywords' if section == 'fuzzy' else section + if target_section in self.variable_names: + for var in self.variable_names[target_section]: + if var[1:].startswith(prefix): + used_variables.add(var) + + # 3. Check for direct variable references (e.g., fuzzy.$admin) + # We add 'fuzzy' to the capture group here var_pattern = r'(keywords|semantics|llm|fuzzy)\.\$([a-zA-Z0-9_]+)(?!\*)' for match in re.finditer(var_pattern, condition): - section = match.group(1) var_name = f"${match.group(2)}" used_variables.add(var_name) - # Check for standalone variables - var_pattern = r'(? bool: + def evaluate(self, pattern: KeywordPattern, text: str, key: str = None) -> bool: """ Check if a fuzzy pattern matches the text.