demeter
Autonomous Hydroponic Intelligencegit clone https://github.com/maydayv7/demeter
agent/guardrails/validation.py (8560B)
1"""
2🛡️ DEMETER GUARDRAILS - Responsible AI Safety Module
3Prevents prompt injection, enforces bounds, detects violations.
4"""
5
6import re
7import json
8from typing import Dict, List, Tuple
9
10# ============================================================================
11# HARD BOUNDS - Absolute safety limits for farm operations
12# ============================================================================
13
14HARD_BOUNDS = {
15 # ATMOSPHERIC TARGETS (OUTPUT from AtmosphericAgent)
16 "air_temp": {"min": 10, "max": 35, "unit": "°C", "desc": "Air Temperature Target"},
17 "humidity": {"min": 30, "max": 90, "unit": "%", "desc": "Humidity Target"},
18 "co2": {"min": 300, "max": 1500, "unit": "ppm", "desc": "CO₂ Target"},
19 "light_intensity": {"min": 0, "max": 100, "unit": "%", "desc": "Light Intensity Target"},
20
21 # WATER TARGETS (OUTPUT from WaterAgent)
22 # These are TARGET values, not dosages. Dosages are computed from these targets.
23 "ph": {"min": 4.0, "max": 7.5, "unit": "pH", "desc": "pH Target"},
24 "ec": {"min": 0.1, "max": 3.0, "unit": "dS/m", "desc": "EC Target"},
25 "water_temp": {"min": 12, "max": 28, "unit": "°C", "desc": "Water Temperature Target"},
26}
27
28# ============================================================================
29# PROMPT INJECTION DETECTION
30# ============================================================================
31
32INJECTION_PATTERNS = [
33 r"ignore.*instructions",
34 r"forget.*previous",
35 r"disregard.*prompt",
36 r"override.*system",
37 r"execute.*command",
38 r"ignore.*safety",
39 r"bypass.*constraints",
40 r"(import|exec|eval|__)", # Code execution attempts
41 r"(SELECT|DROP|DELETE|INSERT).*FROM", # SQL injection
42 r"(curl|wget|bash|sh)\s+", # Shell command attempts
43]
44
45FARM_UNRELATED_KEYWORDS = [
46 "bitcoin", "weather", "politics", "personal", "financial advice",
47 "medical", "legal", "hack", "jailbreak", "bypass", "crack",
48]
49
50# ============================================================================
51# VALIDATION FUNCTIONS
52# ============================================================================
53
54def sanitize_input(text: str) -> Tuple[str, List[str]]:
55 """
56 Sanitizes user input to prevent injection attacks.
57
58 Returns:
59 (cleaned_text, violations_list)
60 """
61 violations = []
62 cleaned = text.strip()
63
64 # Check for injection patterns
65 for pattern in INJECTION_PATTERNS:
66 if re.search(pattern, cleaned, re.IGNORECASE):
67 violations.append(f"⚠️ Detected potential injection pattern: {pattern}")
68
69 # Check for off-topic keywords
70 for keyword in FARM_UNRELATED_KEYWORDS:
71 if re.search(rf"\b{keyword}\b", cleaned, re.IGNORECASE):
72 violations.append(f"⚠️ Query contains off-topic keyword: '{keyword}'")
73
74 # Remove markdown code blocks (common injection vector)
75 cleaned = re.sub(r"```[\s\S]*?```", "", cleaned)
76 cleaned = re.sub(r"`.*?`", "", cleaned)
77
78 return cleaned, violations
79
80
81def validate_bounds(parameter_name: str, value: float) -> Tuple[bool, str]:
82 """
83 Validates a parameter against hard bounds.
84
85 Returns:
86 (is_valid, message)
87 """
88 if parameter_name not in HARD_BOUNDS:
89 return False, f"❌ Unknown parameter: {parameter_name}"
90
91 bounds = HARD_BOUNDS[parameter_name]
92
93 try:
94 val = float(value)
95 except (ValueError, TypeError):
96 return False, f"❌ Invalid value for {parameter_name}: {value} (must be numeric)"
97
98 if val < bounds["min"]:
99 return False, (
100 f"❌ {parameter_name} = {val}{bounds['unit']} is BELOW minimum "
101 f"({bounds['min']}{bounds['unit']})"
102 )
103
104 if val > bounds["max"]:
105 return False, (
106 f"❌ {parameter_name} = {val}{bounds['unit']} is ABOVE maximum "
107 f"({bounds['max']}{bounds['unit']})"
108 )
109
110 return True, f"✅ {parameter_name} = {val}{bounds['unit']} is valid"
111
112
113def clamp_to_bounds(parameter_name: str, value: float) -> float:
114 """Clamps a value to hard bounds (lossy but safe)."""
115 if parameter_name not in HARD_BOUNDS:
116 return value
117
118 bounds = HARD_BOUNDS[parameter_name]
119 return max(bounds["min"], min(bounds["max"], float(value)))
120
121
122def validate_plan(plan: Dict, agent_type: str = "both") -> Dict:
123 """
124 Comprehensive plan validation.
125
126 Returns:
127 {
128 "valid": bool,
129 "violations": [str],
130 "warnings": [str],
131 "bounded_plan": dict,
132 "severity": "SAFE" | "WARNING" | "CRITICAL"
133 }
134 """
135 violations = []
136 warnings = []
137 bounded_plan = plan.copy() if plan else {}
138 severity = "SAFE"
139
140 if not isinstance(plan, dict):
141 return {
142 "valid": False,
143 "violations": ["Plan must be a valid JSON object"],
144 "warnings": [],
145 "bounded_plan": {},
146 "severity": "CRITICAL"
147 }
148
149 for param, value in plan.items():
150 if param not in HARD_BOUNDS:
151 warnings.append(f"⚠️ Unknown parameter: {param}")
152 continue
153
154 try:
155 val = float(value)
156 is_valid, message = validate_bounds(param, val)
157
158 if not is_valid:
159 violations.append(message)
160 bounded_plan[param] = clamp_to_bounds(param, val)
161 severity = "CRITICAL"
162 else:
163 bounded_plan[param] = val
164
165 except (ValueError, TypeError) as e:
166 violations.append(f"❌ {param}: Invalid numeric value '{value}'")
167 severity = "CRITICAL"
168
169 return {
170 "valid": len(violations) == 0,
171 "violations": violations,
172 "warnings": warnings,
173 "bounded_plan": bounded_plan,
174 "severity": severity
175 }
176
177
178def detect_hard_violations(plan: Dict) -> Tuple[bool, List[str]]:
179 """
180 Checks if a plan has HARD violations (cannot be auto-fixed).
181
182 Returns:
183 (has_violations, violation_list)
184 """
185 violations = []
186
187 if not isinstance(plan, dict):
188 return True, ["Plan is not a valid dictionary"]
189
190 for param, value in plan.items():
191 if param not in HARD_BOUNDS:
192 continue
193
194 try:
195 val = float(value)
196 bounds = HARD_BOUNDS[param]
197
198 # If value is MORE THAN 10% outside bounds → HARD VIOLATION
199 overflow = max(0, val - bounds["max"]) / bounds["max"] if bounds["max"] > 0 else 0
200 underflow = max(0, bounds["min"] - val) / bounds["min"] if bounds["min"] > 0 else 0
201
202 if overflow > 0.1:
203 violations.append(
204 f"🚫 HARD VIOLATION: {param} = {val} exceeds max {bounds['max']} by {overflow*100:.1f}%"
205 )
206 if underflow > 0.1:
207 violations.append(
208 f"🚫 HARD VIOLATION: {param} = {val} below min {bounds['min']} by {underflow*100:.1f}%"
209 )
210
211 except (ValueError, TypeError):
212 violations.append(f"🚫 HARD VIOLATION: {param} has non-numeric value: {value}")
213
214 return len(violations) > 0, violations
215
216
217def create_validation_report(plan: Dict) -> str:
218 """Creates a human-readable validation report."""
219 validation = validate_plan(plan)
220 report = []
221
222 report.append(f"\n{'='*60}")
223 report.append(f"🛡️ VALIDATION REPORT - Severity: {validation['severity']}")
224 report.append(f"{'='*60}\n")
225
226 if validation["valid"]:
227 report.append("✅ PLAN PASSED ALL CHECKS\n")
228 else:
229 report.append(f"❌ VIOLATIONS ({len(validation['violations'])}):")
230 for v in validation["violations"]:
231 report.append(f" {v}")
232 report.append("")
233
234 if validation["warnings"]:
235 report.append(f"⚠️ WARNINGS ({len(validation['warnings'])}):")
236 for w in validation["warnings"]:
237 report.append(f" {w}")
238 report.append("")
239
240 if validation["bounded_plan"] and validation["bounded_plan"] != plan:
241 report.append("📊 AUTO-CLAMPED PARAMETERS:")
242 for param, value in validation["bounded_plan"].items():
243 original = plan.get(param)
244 if original != value:
245 report.append(f" {param}: {original} → {value} {HARD_BOUNDS.get(param, {}).get('unit', '')}")
246 report.append("")
247
248 report.append(f"{'='*60}\n")
249 return "\n".join(report)