Two-Factor & Multi-Factor Authentication -- Complete Implementation Guide

import pyotp
import hashlib
import base64
import os
from datetime import datetime

class TOTPManager:
    def __init__(self, issuer="DodaTech"):
        self.issuer = issuer

    def generate_secret(self):
        """Generate a new TOTP secret."""
        secret = pyotp.random_base32()
        return secret

    def get_provisioning_uri(self, secret, email):
        """Get the otpauth:// URI for QR code generation."""
        totp = pyotp.TOTP(secret)
        uri = totp.provisioning_uri(email, issuer_name=self.issuer)
        return uri

    def verify_code(self, secret, code):
        """Verify a TOTP code within a 90-second window."""
        totp = pyotp.TOTP(secret)
        return totp.verify(code, valid_window=1)

    def get_current_code(self, secret):
        """Get the current valid TOTP code (for testing)."""
        totp = pyotp.TOTP(secret)
        return totp.now()

# Usage
manager = TOTPManager()
secret = manager.generate_secret()
print(f"Secret: {secret}")
# Expected: JBSWY3DPEHPK3PXP

uri = manager.get_provisioning_uri(secret, "user@dodatech.com")
print(f"URI: {uri}")
# Expected: otpauth://totp/DodaTech:user@dodatech.com?secret=JBSWY3DPEHPK3PXP&issuer=DodaTech

# Simulate verification
code = manager.get_current_code(secret)
result = manager.verify_code(secret, code)
print(f"Verification result: {result}")
# Expected: True

wrong_code = "000000"
result = manager.verify_code(secret, wrong_code)
print(f"Wrong code result: {result}")
# Expected: False

// WebAuthn registration with platform authenticator
async function registerPasskey(username, userId) {
  const challenge = new Uint8Array(32);
  crypto.getRandomValues(challenge);

  const publicKeyCredential = await navigator.credentials.create({
    publicKey: {
      challenge,
      rp: {
        name: "DodaTech",
        id: window.location.hostname,
      },
      user: {
        id: new TextEncoder().encode(userId),
        name: username,
        displayName: username,
      },
      pubKeyCredParams: [
        { type: "public-key", alg: -7 },
        { type: "public-key", alg: -257 },
      ],
      authenticatorSelection: {
        authenticatorAttachment: "platform",
        userVerification: "required",
        residentKey: "required",
      },
    },
  });

  return {
    id: publicKeyCredential.id,
    rawId: Array.from(new Uint8Array(publicKeyCredential.rawId)),
    type: publicKeyCredential.type,
    clientDataJSON: Array.from(
      new Uint8Array(publicKeyCredential.response.clientDataJSON)
    ),
    attestationObject: Array.from(
      new Uint8Array(publicKeyCredential.response.attestationObject)
    ),
  };
}

// WebAuthn authentication
async function authenticateWithPasskey() {
  const challenge = new Uint8Array(32);
  crypto.getRandomValues(challenge);

  const assertion = await navigator.credentials.get({
    publicKey: {
      challenge,
      timeout: 60000,
      userVerification: "required",
    },
  });

  return {
    id: assertion.id,
    rawId: Array.from(new Uint8Array(assertion.rawId)),
    type: assertion.type,
    clientDataJSON: Array.from(
      new Uint8Array(assertion.response.clientDataJSON)
    ),
    authenticatorData: Array.from(
      new Uint8Array(assertion.response.authenticatorData)
    ),
    signature: Array.from(new Uint8Array(assertion.response.signature)),
    userHandle: assertion.response.userHandle
      ? Array.from(new Uint8Array(assertion.response.userHandle))
      : null,
  };
}

// Usage
registerPasskey("admin@dodatech.com", "admin-001")
  .then((cred) => console.log("Passkey registered:", cred.id))
  .catch((err) => console.error("Registration failed:", err.message));

# Step-up authentication for sensitive actions
from functools import wraps
from flask import session, request, jsonify

STEP_UP_ACTIONS = {
    "password_change": "MFA required for password changes",
    "api_key_generate": "MFA required for API key generation",
    "payment": "MFA required for payments over $1000",
}

def require_step_up(action):
    """Decorator that requires MFA for sensitive actions."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user_id = session.get("user_id")
            if not user_id:
                return jsonify({"error": "Not authenticated"}), 401

            # Check if user has already authenticated via MFA in this session
            mfa_verified = session.get(f"mfa_verified_{action}", False)
            mfa_timestamp = session.get(f"mfa_timestamp_{action}", 0)

            from time import time
            # MFA is valid for 15 minutes per action
            if mfa_verified and (time() - mfa_timestamp) < 900:
                return f(*args, **kwargs)

            # Require MFA verification
            return jsonify({
                "error": "step_up_required",
                "action": action,
                "message": STEP_UP_ACTIONS.get(action, "MFA required"),
                "session_token": session.get("step_up_token"),
            }), 403

        return decorated_function
    return decorator

# Apply step-up to sensitive routes
@app.route("/api/settings/password", methods=["POST"])
@require_step_up("password_change")
def change_password():
    # Password change logic here
    return jsonify({"status": "password_updated"})

@app.route("/api/verify-step-up", methods=["POST"])
def verify_step_up():
    """Verify MFA for step-up authentication."""
    data = request.get_json()
    code = data.get("code")
    action = data.get("action")

    # Verify TOTP code
    # ... verification logic ...

    from time import time
    session[f"mfa_verified_{action}"] = True
    session[f"mfa_timestamp_{action}"] = time()
    return jsonify({"status": "verified"})

import hashlib
import os
import json

class RecoveryCodeManager:
    def __init__(self, code_count=10, code_length=12):
        self.code_count = code_count
        self.code_length = code_length

    def generate_codes(self):
        """Generate recovery codes and their hashed versions."""
        codes = []
        hashed_codes = []

        for _ in range(self.code_count):
            code = os.urandom(self.code_length).hex()[:self.code_length]
            hashed = hashlib.sha256(code.encode()).hexdigest()
            codes.append(code)
            hashed_codes.append(hashed)

        return codes, hashed_codes

    def verify_code(self, code, hashed_codes, used_codes):
        """Verify a recovery code and mark it as used."""
        code_hash = hashlib.sha256(code.encode()).hexdigest()

        if code_hash in used_codes:
            return False, "Code already used"

        if code_hash in hashed_codes:
            return True, code_hash

        return False, "Invalid code"

# Usage
manager = RecoveryCodeManager()
codes, hashed = manager.generate_codes()

print("Save these recovery codes:")
for code in codes:
    print(f"  {code}")
# Expected: 12-character hex strings (printed once during enrollment)

used_codes = set()
valid, result = manager.verify_code(codes[0], hashed, used_codes)
print(f"First code valid: {valid}")
# Expected: True

used_codes.add(result)
valid, result = manager.verify_code(codes[0], hashed, used_codes)
print(f"Reuse attempt: {valid}")
# Expected: False (code already used)