Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, HTTPException, Header, Depends | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| import random | |
| import string | |
| from twilio.rest import Client | |
| import time | |
| app = FastAPI() | |
| # --- SECURITY CONFIGURATION --- | |
| # Ye wo secret password hai jo aap apne client ko denge (ya khud use karenge) | |
| # Iske bina koi API use nahi kar payega. | |
| MY_SECRET_API_KEY = "malik-special-key-12345" | |
| # --- TWILIO CONFIGURATION --- | |
| TWILIO_ACCOUNT_SID = "ACa674f655da36f0e6854aa206df3870e2" # Apna sahi wala dalen | |
| TWILIO_AUTH_TOKEN = "your_auth_token_here" # Apna sahi wala dalen | |
| TWILIO_SENDER_NUMBER = "+19204826070" | |
| # Twilio Client | |
| try: | |
| client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN) | |
| except: | |
| pass | |
| # Storage & Rate Limiting | |
| otp_storage = {} | |
| rate_limit_storage = {} # Ye record rakhega ke kisne kab SMS bheja | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| class GenerateOTPRequest(BaseModel): | |
| phone_number: str | |
| class VerifyOTPRequest(BaseModel): | |
| phone_number: str | |
| otp: str | |
| # --- SECURITY FUNCTION --- | |
| async def verify_api_key(x_api_key: str = Header(None)): | |
| if x_api_key != MY_SECRET_API_KEY: | |
| raise HTTPException(status_code=403, detail="β Access Denied: Invalid API Key") | |
| return x_api_key | |
| def read_root(): | |
| return {"status": "Secure API Running"} | |
| def generate_otp(request: GenerateOTPRequest, api_key: str = Depends(verify_api_key)): | |
| """ | |
| OTP generate karne se pehle API Key check karega aur Rate Limit lagayega. | |
| """ | |
| phone = request.phone_number.strip().replace(" ", "").replace("-", "").replace(".", "") | |
| if phone.startswith("03"): phone = "+92" + phone[1:] | |
| # --- RATE LIMITING (Bot Protection) --- | |
| current_time = time.time() | |
| last_sent = rate_limit_storage.get(phone, 0) | |
| # Agar 60 seconds se pehle dobara request ayi to block kar do | |
| if current_time - last_sent < 60: | |
| raise HTTPException(status_code=429, detail=f"β³ Please wait {int(60 - (current_time - last_sent))} seconds.") | |
| # Generate OTP | |
| otp_code = ''.join(random.choices(string.digits, k=6)) | |
| # Save Data | |
| otp_storage[phone] = otp_code | |
| rate_limit_storage[phone] = current_time # Time update karo | |
| # Send SMS | |
| try: | |
| # Code yahan same rahega (Twilio wala) | |
| # client.messages.create(...) | |
| # (Testing ke liye print kar raha hoon taake aapka paisa na kate) | |
| print(f"Sending OTP {otp_code} to {phone}") | |
| return {"status": "success", "message": "SMS sent successfully!"} | |
| except Exception as e: | |
| return {"status": "error", "message": str(e)} | |
| def verify_otp(request: VerifyOTPRequest, api_key: str = Depends(verify_api_key)): | |
| phone = request.phone_number.strip().replace(" ", "").replace("-", "").replace(".", "") | |
| if phone.startswith("03"): phone = "+92" + phone[1:] | |
| if phone in otp_storage and otp_storage[phone] == request.otp: | |
| del otp_storage[phone] | |
| return {"status": "success", "message": "β Verified!"} | |
| raise HTTPException(status_code=400, detail="β Invalid OTP") |