File size: 2,106 Bytes
27d37fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | # src/hitl/validators.py
import re
from typing import List, Tuple, Set
class RiskValidator:
def __init__(self, blocked_keywords: List[str], sensitive_topics: List[str]):
self.blocked_keywords = [kw.lower() for kw in blocked_keywords]
self.sensitive_topics = [st.lower() for st in sensitive_topics]
# Regex to detect emails and IPv4 addresses.
# Each IPv4 octet is constrained to 0-255 and the whole match must be
# bounded by non-dot/digit characters, so version strings like
# "3.11.4.2" or "1.0.0.0" (and longer dotted numbers) don't false-positive.
self.email_regex = re.compile(r'[\w\.-]+@[\w\.-]+\.\w+')
_octet = r'(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)'
self.ip_regex = re.compile(
r'(?<![\d.])' + r'\.'.join([_octet] * 4) + r'(?![\d.])'
)
def analyze(self, question: str, answer: str) -> Tuple[str, List[str]]:
"""
Runs rules-based scans across proposed edits.
Returns:
Tuple[str, List[str]]: (Assigned Risk Level, List of flags raised)
"""
flags = []
combined_text = (question + " " + answer).lower()
# 1. Check for blocked/malicious system prompt injection keywords
for word in self.blocked_keywords:
if word in combined_text:
flags.append(f"BLOCKED_KEYWORD: '{word}'")
# 2. Scan for flagged security vectors
for topic in self.sensitive_topics:
if topic in combined_text:
flags.append(f"SENSITIVE_TOPIC_RISK: '{topic}'")
# 3. Detect PII Leakage (Emails, IPs)
if self.email_regex.search(combined_text):
flags.append("PII_DETECTED: EMAIL")
if self.ip_regex.search(combined_text):
flags.append("PII_DETECTED: IP_ADDRESS")
# Determine structural severity
if any("BLOCKED_KEYWORD" in f for f in flags):
risk_level = "HIGH"
elif len(flags) > 0:
risk_level = "MEDIUM"
else:
risk_level = "LOW"
return risk_level, flags
|