"""360Shield Guard — FastAPI Drop this file in your project root. Add one line to main.py. Done. from guard_fastapi import shield shield(app, key="YOUR_AGENT_KEY", domain="example.com") The key MUST be an AGENT key (ada_agent_*), NOT a master key (ada_live_*). Generate one per domain from your dashboard at https://360shield.net/dashboard ("Guard & API" tab) — the dashboard auto-creates a key for each verified domain. Open source — https://github.com/asimetry/360shield-guard Version: 1.0.1 | License: MIT | Zero dependencies (stdlib only) """ import os, sys, json, time, hashlib, hmac, threading, collections from datetime import datetime, timezone from urllib.request import Request, urlopen from urllib.error import URLError __version__ = '1.0.3' _API = 'https://360shield.net/v1/agent' _HB_INTERVAL = 30 _FLOOD_LIMIT = 50 _FLOOD_WINDOW = 300 _BUFFER_SIZE = 500 _buf = collections.deque(maxlen=_BUFFER_SIZE) _flood = {} _blocked = set() _executed_cmd_ids = [] _maintenance = False _lock = threading.Lock() _agent_id = None _booted = False def _ip(request): for h in ('cf-connecting-ip', 'x-real-ip', 'x-forwarded-for'): v = request.headers.get(h) if v: return v.split(',')[0].strip() if request.client: return request.client.host return '' def _is_bot(ua): ua = ua.lower() bots = ('bot', 'crawler', 'spider', 'scraper', 'curl', 'wget', 'python-requests', 'go-http', 'headless', 'phantom', 'selenium', 'puppeteer', 'postman', 'httpie') return any(b in ua for b in bots) or not ua def _api_call(key, endpoint, body, sign=False): headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {key}', 'X-Guard-Version': __version__, 'User-Agent': f'360Shield-Guard/{__version__}', } if sign: headers['X-Guard-Signature'] = hmac.new( key.encode(), body, hashlib.sha256).hexdigest() req = Request(_API + endpoint, data=body, headers=headers, method='POST') try: with urlopen(req, timeout=10) as r: return json.loads(r.read()) except URLError as e: print(f'[360Shield] API error: {e}', file=sys.stderr) return None def _boot(key, domain): global _booted if _booted: return _booted = True def _run(): global _agent_id time.sleep(2) try: body = json.dumps({'domain': domain}).encode() resp = _api_call(key, '/register', body) if resp and resp.get('agent_id'): _agent_id = resp['agent_id'] print(f'[360Shield] Guard registered: {_agent_id}') except Exception as e: print(f'[360Shield] register failed: {e}', file=sys.stderr) while True: try: _heartbeat(key, domain) except Exception as e: print(f'[360Shield] heartbeat error: {e}', file=sys.stderr) time.sleep(_HB_INTERVAL) threading.Thread(target=_run, daemon=True).start() def _heartbeat(key, domain): global _executed_cmd_ids, _maintenance reqs = list(_buf) now = int(time.time()) window = [r for r in reqs if now - r['time'] < 300] total = len(window) sampled = window[-500:] if len(window) > 500 else window recent_requests = [{ 'ts': float(r.get('time', 0)), 'ip': r.get('ip', ''), 'method': r.get('method', 'GET'), 'path': r.get('path', '/'), 'status': r.get('status') or 200, 'is_bot': bool(r.get('is_bot')), 'ua': r.get('ua', ''), } for r in sampled] sent_acks = list(_executed_cmd_ids) _executed_cmd_ids = [] payload = { 'agent_id': _agent_id or 'pending', 'domain': domain, 'guard_version': __version__, 'timestamp': datetime.now(timezone.utc).isoformat(), 'traffic': { 'total_requests': total, 'bot_requests': sum(1 for r in window if r.get('is_bot')), 'unique_ips': len(set(r['ip'] for r in window)), 'rpm': round(total / 5, 1), }, 'recent_requests': recent_requests, 'security': {'platform': 'fastapi'}, 'guard_state': { 'blocked_count': len(_blocked), 'buffer_size': len(_buf), 'maintenance_mode': _maintenance, 'bot_patterns': 25, }, 'acked_command_ids': sent_acks, } body = json.dumps(payload, default=str).encode() resp = _api_call(key, '/heartbeat', body, sign=True) if resp: for cmd in resp.get('commands', []): cid = cmd.get('id') action = cmd.get('action', '') try: if action == 'block_ip' and cmd.get('ip'): _blocked.add(cmd['ip']) elif action == 'unblock_ip': _blocked.discard(cmd.get('ip', '')) elif action == 'reset_blocked': _blocked.clear() elif action == 'set_maintenance': _maintenance = bool(cmd.get('enabled', False)) elif action == 'force_scan': pass if cid is not None: _executed_cmd_ids.append(cid) except Exception: pass else: _executed_cmd_ids = sent_acks + _executed_cmd_ids def shield(app, key=None, domain=None): """One-line Guard activation for FastAPI. from guard_fastapi import shield shield(app, key="ada_live_xxx", domain="example.com") """ from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse key = key or os.environ.get('SHIELD_AGENT_KEY', '') domain = domain or os.environ.get('SHIELD_DOMAIN', '') if not key or not domain: print('[360Shield] WARNING: SHIELD_AGENT_KEY and SHIELD_DOMAIN required. Guard NOT active.', file=sys.stderr) return _boot(key, domain) class _ShieldMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): path = request.url.path if path.startswith('/static/') or path == '/favicon.ico': return await call_next(request) ip = _ip(request) if ip in _blocked: return JSONResponse({'error': 'blocked'}, status_code=403) now = int(time.time()) with _lock: ts = _flood.get(ip, []) ts = [t for t in ts if now - t < _FLOOD_WINDOW] ts.append(now) _flood[ip] = ts[-200:] if len(ts) >= _FLOOD_LIMIT: return JSONResponse({'error': 'rate_limited', 'retry_after': 60}, status_code=429) ua = request.headers.get('user-agent', '') entry = { 'time': now, 'ip': ip, 'method': request.method, 'path': path[:80], 'ua': ua[:120], 'is_bot': _is_bot(ua), 'status': None, 'ms': None, } _buf.append(entry) t0 = time.time() response = await call_next(request) entry['status'] = response.status_code entry['ms'] = int((time.time() - t0) * 1000) response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'DENY' response.headers['X-XSS-Protection'] = '1; mode=block' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' return response app.add_middleware(_ShieldMiddleware)