from fastapi import Request from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from nacl.signing import VerifyKey from nacl.exceptions import BadSignatureError import hashlib import secrets import base58 import requests import json import httpx from geopy.geocoders import Nominatim from geopy.exc import GeocoderServiceError geolocator = Nominatim(user_agent="velocity") def GEOCODER(coords): try: lat = coords.get("latitude") lon = coords.get("longitude") if lat is None or lon is None: return { "status": "error", "message": "Missing latitude or longitude.", "data": None } location = geolocator.reverse(f"{lat}, {lon}", exactly_one=True) if location and "address" in location.raw: country_name = location.raw["address"].get("country") country_code = location.raw["address"].get("country_code") return { "status": "success", "message": "Location resolved successfully.", "data": { "country": country_name, "code": country_code.upper() if country_code else None } } return { "status": "not_found", "message": "Location could not be resolved.", "data": None } except GeocoderServiceError as e: return { "status": "error", "message": f"Geocoding service error: {str(e)}", "data": None } except Exception as e: return { "status": "error", "message": f"Unexpected error: {str(e)}", "data": None } def SignatureVerification(X_401_Addr,X_401_Nonce,X_401_Sign,challange): if X_401_Addr and X_401_Nonce and X_401_Sign: try: signature_bytes = base58.b58decode(X_401_Sign) verify_key = VerifyKey(base58.b58decode(X_401_Addr)) payload_bytes = challange.encode("utf-8") verify_key.verify(payload_bytes, signature_bytes) return True except BadSignatureError: return False def SecretNonceGenerator(): random_bytes = secrets.token_bytes(32) return hashlib.sha256(random_bytes).hexdigest() def TokenCheck(walletPublicKey): url = f"https://mainnet.helius-rpc.com/?api-key=4e833ada-d32c-48c5-b020-c11b2253f25b" payload = { "jsonrpc": "2.0", "id": "1", "method": "getTokenAccountsByOwner", "params": [ walletPublicKey, {"mint":"3d4XyPWkUJzruF5c2qc1QfpLgsaNaDLMtTya1bWBpump" }, {"encoding": "jsonParsed"} ] } headers = {"Content-Type": "application/json"} try: response = requests.post(url, json=payload, headers=headers) response.raise_for_status() data = response.json() except requests.exceptions.RequestException as e: return {"status": False, "message": f"API Request Failed: {e}"} token_accounts = data.get("result", {}).get("value") if not token_accounts: return False try: account_info = token_accounts[0]["account"]["data"]["parsed"]["info"] token_amount = account_info["tokenAmount"] ui_amount = float(token_amount.get("uiAmount")) except (TypeError, KeyError, IndexError): return {"status": False, "message": "Could not parse token account data"} if ui_amount >= 100000.0: print(ui_amount) return True else: print(ui_amount) return False class x401Kit(BaseHTTPMiddleware): def __init__(self, app, protected_paths: list, required_mint: str, mint_amount:int , helius_api_key: str, geo_code:bool ,geo_code_locs:list,secret_domain="VELOCITY"): super().__init__(app) self.protected_paths = protected_paths self.required_mint = required_mint self.mint_amount=mint_amount self.helius_api_key = helius_api_key self.secret_domain = secret_domain self.geo_code=geo_code self.geo_code_locs=geo_code_locs async def dispatch(self, request: Request, call_next): if request.method == "OPTIONS": return await call_next(request) if not any(request.url.path.startswith(p) for p in self.protected_paths): return await call_next(request) NONCE=SecretNonceGenerator() X_401_Nonce=request.headers.get("X-401-Nonce") X_401_Sign=request.headers.get("X-401-Signature") X_401_Addr=request.headers.get("X-401-Addr") lat=request.headers.get("X-Lat") long=request.headers.get("X-Long") coords={ "latitude":lat, "longitude":long } REQUIRED_SERVICE=None if not X_401_Addr and not X_401_Nonce and not X_401_Sign : payload401={ "X-401-Status":"Authrequired", "x-401-Mechanism":"SOLANA", "X-401-Nonce":NONCE, "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true", "Access-Control-Expose-Headers": "x-401-Nonce, x-401-Mechanism, x-401-Status" } return JSONResponse(content={ "message":"401 Auth Required", "information":"Non persistant stateless auth" },headers=payload401,status_code=401) challange=f"CHALLENGE::{X_401_Nonce}::{request.url.path}::{self.secret_domain}" signverify=SignatureVerification(X_401_Addr,X_401_Nonce,X_401_Sign,challange) tokenverify=TokenCheck(X_401_Addr) print(tokenverify) if signverify == True and tokenverify == True: if self.geo_code==True: country=GEOCODER(coords) print(country) if country["data"] is None: return JSONResponse( content={"status": "locerror", "message": f"Auth error"}, headers= { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true" }, status_code=500 ) if country["data"]["code"] in self.geo_code_locs: return JSONResponse( content={"status": "locdeny", "message":f"access denied for {country['data']['country']}"}, headers= { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true" }, status_code=401 ) response = await call_next(request) if response.headers.get("content-type") == "application/json": body_bytes = b"" async for chunk in response.body_iterator: body_bytes += chunk try: data = json.loads(body_bytes.decode()) except json.JSONDecodeError: return response data["address"] = X_401_Addr response_headers = dict(response.headers) response_headers.pop("content-length", None) print(data) return JSONResponse( content=data, status_code=response.status_code, headers=response_headers ) return response elif tokenverify==False: return JSONResponse( content={"status": "error", "message": "Missing required token"}, status_code=500, headers={ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true" } ) elif signverify==False: print("failed") return JSONResponse( content={"status": "error", "message": "bad signature"}, status_code=500, headers={ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true" } ) else: return JSONResponse( content={"status": "error", "message": "Authentication failed"}, status_code=500, headers={ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true" } )