ghstedpixel/app.py
0
1import re2import httpx3from config import SAFE_BROWSING_API4 5def scrub_conversational_bloat(query: str) -> str:6 """Strips conversational noise words and store hooks to maximize index matching accuracy"""7 if query.startswith(("http://", "https://")): return query8 noise_pattern = r'\b(buy|cheap|online|sale|authentic|store|deals?|discount|get|best|price|lowest)\b'9 scrubbed = re.sub(noise_pattern, '', query.lower())10 clean_string = " ".join(scrubbed.split())11 if clean_string != query.lower():12 print(f"Core normalization: Cleaned token string layout from '{query}' -> '{clean_string}'")13 return clean_string if clean_string else query14 15async def check_url_safe(url: str, client: httpx.AsyncClient) -> bool:16 """Vets outbound destination URLs against malware and deceptive targets using Google Safe Browsing APIs"""17 if not SAFE_BROWSING_API:18 print("⚠️ WARNING: SAFE_BROWSING_API key missing. Bypassing live link validation safety verification.")19 return True20 api_url = f"https://safebrowsing.googleapis.com/v4/threatMatches:find?key={SAFE_BROWSING_API}"21 payload = {22 "client": {"clientId": "tarz-ai-search-agent", "clientVersion": "1.1.0"},23 "threatInfo": {24 "threatTypes": ["MALWARE", "SOCIAL_ENGINEERING", "UNWANTED_SOFTWARE", "POTENTIALLY_HARMFUL_APPLICATION"],25 "platformTypes": ["ANY_PLATFORM"],26 "threatEntryTypes": ["URL"],27 "threatEntries": [{"url": url}]28 }29 }30 try:31 response = await client.post(api_url, json=payload, timeout=5.0)32 if response.status_code == 200:33 res_data = response.json()34 if "matches" in res_data and len(res_data["matches"]) > 0:35 print(f"❌ [Security Flag] Blocked malicious link threat match vector: {url}")36 return False37 return True38 else:39 print(f"⚠️ [Safe Browsing Error] API returned state {response.status_code}. Defaulting shut for containment.")40 return False41 except Exception as e:42 print(f"❌ [Safe Browsing Exception] Connection failed: {e}. Isolation fallback activated.")43 return False