SMS Multi-Factor Authentication — Implementation and Security Considerations

import random, time, os
from flask import Flask, request, jsonify
from twilio.rest import Client

app = Flask(__name__)

# Twilio configuration
TWILIO_ACCOUNT_SID = os.environ.get("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.environ.get("TWILIO_AUTH_TOKEN")
TWILIO_PHONE_NUMBER = os.environ.get("TWILIO_PHONE_NUMBER")
twilio_client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN) if TWILIO_ACCOUNT_SID else None

# In-memory code store (use Redis in production)
sms_codes = {}

def generate_sms_code(length=6):
    """Generate a numeric SMS verification code."""
    return ''.join(str(random.randint(0, 9)) for _ in range(length))

def send_sms(to_phone, message):
    """Send SMS via Twilio."""
    if not twilio_client:
        print(f"[MOCK SMS] To: {to_phone}, Message: {message}")
        return True

    try:
        twilio_client.messages.create(
            body=message,
            from_=TWILIO_PHONE_NUMBER,
            to=to_phone
        )
        return True
    except Exception as e:
        print(f"Failed to send SMS: {e}")
        return False

@app.route("/api/mfa/sms/send", methods=["POST"])
def send_mfa_code():
    """Generate and send SMS MFA code."""
    phone = request.json.get("phone", "")
    user_id = request.json.get("user_id", "")

    # Validate phone number format
    if not phone.startswith("+") or len(phone) < 10:
        return jsonify({"error": "Invalid phone number"}), 400

    # Check rate limit
    rate_key = f"sms_rate:{phone}"
    if is_rate_limited(rate_key, 3, 300):  # 3 SMS per 5 minutes
        return jsonify({
            "error": "rate_limited",
            "message": "Too many SMS requests. Try again later."
        }), 429

    code = generate_sms_code()
    expiry = time.time() + 300  # 5 minutes

    # Store code
    sms_codes[phone] = {
        "code": code,
        "expiry": expiry,
        "user_id": user_id,
        "attempts": 0,
        "used": False
    }

    message = f"Your Durga Antivirus verification code: {code}. Valid for 5 minutes."
    sent = send_sms(phone, message)

    if not sent:
        return jsonify({"error": "Failed to send SMS"}), 500

    return jsonify({
        "message": "Verification code sent",
        "expires_in": 300,
        "phone_masked": phone[-4:].rjust(len(phone) - 3, "*")
    })

@app.route("/api/mfa/sms/verify", methods=["POST"])
def verify_sms_code():
    """Verify an SMS MFA code."""
    phone = request.json.get("phone", "")
    code = request.json.get("code", "")

    stored = sms_codes.get(phone)
    if not stored:
        return jsonify({"error": "No code requested"}), 400

    if stored["used"]:
        return jsonify({"error": "Code already used"}), 400

    if time.time() > stored["expiry"]:
        del sms_codes[phone]
        return jsonify({"error": "Code expired"}), 400

    # Rate limit verification attempts
    stored["attempts"] += 1
    if stored["attempts"] > 5:
        del sms_codes[phone]
        return jsonify({
            "error": "Too many attempts. Request a new code."
        }), 429

    if stored["code"] != code:
        return jsonify({"error": "Invalid code", "attempts_remaining": 5 - stored["attempts"]}), 401

    # Code verified — mark as used
    stored["used"] = True

    # Issue MFA-verified token
    access_token = jwt.encode({
        "sub": stored["user_id"],
        "mfa_method": "sms",
        "mfa_verified": True,
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }, SECRET, algorithm="HS256")

    return jsonify({
        "verified": True,
        "access_token": access_token,
        "method": "sms"
    })

class SIMSwapDetector:
    """Detect potential SIM swap attacks."""

    def __init__(self):
        # Track phone number changes per user
        self.phone_history = {}

    def check_sim_swap(self, user_id, phone):
        """Check if phone number was recently changed."""
        last_phone = self.phone_history.get(user_id)

        if last_phone and last_phone != phone:
            # Phone number changed — could be SIM swap
            time_since_change = self.get_time_since_last_auth(user_id)
            if time_since_change < 3600:  # Within last hour
                return {
                    "risk": "high",
                    "action": "require_additional_verification",
                    "message": "Phone number recently changed. Additional verification required."
                }

        self.phone_history[user_id] = phone
        return {"risk": "low", "action": "proceed"}

    def get_time_since_last_auth(self, user_id):
        """Get time since user's last successful authentication."""
        last_auth = last_auth_times.get(user_id)
        if not last_auth:
            return float('inf')
        return time.time() - last_auth

# Integration with login flow
@app.route("/api/auth/login", methods=["POST"])
def login_with_sms_mfa():
    username = request.json.get("username")
    password = request.json.get("password")

    if not validate_password(username, password):
        return jsonify({"error": "Invalid credentials"}), 401

    user = get_user(username)
    if user.get("mfa_enabled") and user.get("mfa_method") == "sms":
        sim_check = SIMSwapDetector().check_sim_swap(username, user["phone"])
        if sim_check["risk"] == "high":
            # Require email verification in addition to SMS
            send_email_verification(username)

        return jsonify({
            "mfa_required": True,
            "mfa_method": "sms",
            "phone_masked": user["phone"][-4:].rjust(len(user["phone"]) - 3, "*")
        }), 403

    return issue_token(username)