"""Validate planned interactions against a manifest.""" from __future__ import annotations from typing import Optional from lxml import html from ..errors import InteractionValidationError from ..models import Manifest, ResourceRule, ValidationResult from ..utils.specificity import Specificity, SpecificityCalculator def validate_interaction(manifest: Manifest, *, verb: str, element) -> ValidationResult: """Return an allow/deny decision for the supplied interaction.""" if verb not in manifest.allowed_verbs(): return ValidationResult( allowed=False, matched_rule=None, reasons=[f"Unknown verb '{verb}'"], ) if not hasattr(element, "getroottree"): return ValidationResult( allowed=False, matched_rule=None, reasons=["Provided element is not an lxml HTML element"], ) try: root = element.getroottree().getroot() except Exception as exc: return ValidationResult( allowed=False, matched_rule=None, reasons=[f"Could not obtain root from element: {exc}"], ) matching_rule = _find_matching_rule(manifest, verb, root, element) if matching_rule is None: return ValidationResult( allowed=False, matched_rule=None, reasons=["No matching resource rule"], ) result = ValidationResult( allowed=matching_rule.allowed, matched_rule=matching_rule.model_dump(mode="json"), ) if matching_rule.modifiers: modifiers = matching_rule.modifiers if modifiers.burst is not None: result.require("burst", modifiers.burst) if modifiers.rate_limit is not None: result.require( "rate_limit", { "max_requests": modifiers.rate_limit.max_requests, "window_seconds": modifiers.rate_limit.window_seconds, }, ) if modifiers.human_in_the_loop is not None: result.require("human_in_the_loop", modifiers.human_in_the_loop) if modifiers.time_window is not None: result.require("time_window", modifiers.time_window) if not matching_rule.allowed: result.add_reason("Matching rule forbids this interaction") return result def ensure_interaction_allowed( manifest: Manifest, *, verb: str, element, ) -> ValidationResult: """Validate and raise InteractionValidationError if the interaction is denied.""" result = validate_interaction(manifest, verb=verb, element=element) if not result.allowed: raise InteractionValidationError( "; ".join(result.reasons) if result.reasons else "Interaction denied by manifest" ) return result def _find_matching_rule( manifest: Manifest, verb: str, root: html.HtmlElement, target ) -> Optional[ResourceRule]: verb_specific_rules = [] wildcard_rules = [] for index, rule in enumerate(manifest.resource_rules): if not _selector_matches(rule.selector, root, target): continue entry = (rule, _specificity(rule.selector), index) if rule.verb == verb: verb_specific_rules.append(entry) elif rule.verb == "all": wildcard_rules.append(entry) chosen_bucket = verb_specific_rules or wildcard_rules if not chosen_bucket: return None # CSS specificity: higher tuple wins; if equal, later rules override. chosen_bucket.sort(key=lambda item: (item[1].to_list(), item[2])) return chosen_bucket[-1][0] def _selector_matches(rule_selector: str, root: html.HtmlElement, target) -> bool: selector = rule_selector.strip() if selector == "*": return True try: matched_elements = root.cssselect(selector) except Exception: return False return target in matched_elements def _specificity(selector: str) -> Specificity: selector = selector.strip() if not selector or selector == "*": return Specificity() try: return SpecificityCalculator.calculate(selector) except Exception: # Fall back to lowest specificity on parse errors. return Specificity()