Protection for Two-Factor Authentication: How We Did It

import pyotp
import secrets
import hashlib
import time
from typing import Optional, List, Dict, Tuple
from dataclasses import dataclass, field
from enum import Enum


class RateLimitStatus(Enum):
    ALLOWED = "allowed"
    COOLDOWN = "cooldown"
    LOCKED = "locked"


@dataclass
class RateLimiter:
    """Sliding-window rate limiter for 2FA attempts.

    Tracks individual attempt timestamps and enforces a maximum
    number of tries within a configurable time window. After
    exceeding the limit, the account is locked for a configurable
    lockout duration.
    """
    max_attempts: int = 5
    window_seconds: int = 300
    lockout_seconds: int = 900
    _attempts: List[float] = field(default_factory=list, repr=False)
    _locked_until: Optional[float] = None

    def check(self) -> RateLimitStatus:
        """Return the current rate-limit status without recording."""
        if self._locked_until is not None and time.time() < self._locked_until:
            return RateLimitStatus.LOCKED
        cutoff = time.time() - self.window_seconds
        self._attempts = [t for t in self._attempts if t > cutoff]
        if len(self._attempts) >= self.max_attempts:
            self._locked_until = time.time() + self.lockout_seconds
            return RateLimitStatus.LOCKED
        return RateLimitStatus.ALLOWED

    def record(self) -> None:
        """Record a single attempt."""
        self._attempts.append(time.time())

    def reset(self) -> None:
        """Clear all state after successful verification."""
        self._attempts.clear()
        self._locked_until = None

    @property
    def seconds_remaining(self) -> float:
        if self._locked_until is None:
            return 0.0
        return max(0.0, self._locked_until - time.time())


@dataclass
class TwoFactorManager:
    """Manages TOTP secrets, hashed backup codes, and rate limiting.

    Secrets are generated with 160 bits of entropy (base32-encoded
    by pyotp). Backup codes are 8 hex characters each, hashed with
    SHA-256 before storage. The manager never persists plaintext
    secrets or codes — that responsibility belongs to the caller.
    """
    totp_secret: str
    _backup_hashes: List[str] = field(default_factory=list, repr=False)
    _rate_limiter: RateLimiter = field(default_factory=RateLimiter, repr=False)

    @classmethod
    def create(cls) -> Tuple["TwoFactorManager", List[str]]:
        """Create a new 2FA enrollment.

        Returns the manager and the plaintext backup codes. The caller
        MUST display these codes to the user immediately and then
        discard the plaintext list.
        """
        secret = pyotp.random_base32()
        plaintext_codes = [secrets.token_hex(4).upper() for _ in range(10)]
        hashed_codes = [
            hashlib.sha256(code.encode()).hexdigest()
            for code in plaintext_codes
        ]
        manager = cls(totp_secret=secret, _backup_hashes=hashed_codes)
        return manager, plaintext_codes

    def verify_totp(self, code: str) -> Dict:
        """Verify a TOTP code with rate-limit enforcement."""
        status = self._rate_limiter.check()
        if status != RateLimitStatus.ALLOWED:
            return {
                "success": False,
                "error": "rate_limited",
                "detail": f"Account locked. Try again in {int(self._rate_limiter.seconds_remaining)}s."
            }

        self._rate_limiter.record()
        totp = pyotp.TOTP(self.totp_secret)
        is_valid = totp.verify(code, valid_window=1)

        if is_valid:
            self._rate_limiter.reset()
            return {"success": True, "method": "totp"}

        return {
            "success": False,
            "error": "invalid_code",
            "attempts_remaining": max(0, self._rate_limiter.max_attempts - len(self._rate_limiter._attempts))
        }

    def verify_backup_code(self, code: str) -> Dict:
        """Verify and consume a single backup code."""
        status = self._rate_limiter.check()
        if status != RateLimitStatus.ALLOWED:
            return {
                "success": False,
                "error": "rate_limited",
                "detail": f"Account locked. Try again in {int(self._rate_limiter.seconds_remaining)}s."
            }

        self._rate_limiter.record()
        target_hash = hashlib.sha256(code.encode()).hexdigest()

        if target_hash in self._backup_hashes:
            self._backup_hashes.remove(target_hash)
            self._rate_limiter.reset()
            return {
                "success": True,
                "method": "backup_code",
                "remaining": len(self._backup_hashes)
            }

        return {
            "success": False,
            "error": "invalid_code",
            "attempts_remaining": max(0, self._rate_limiter.max_attempts - len(self._rate_limiter._attempts))
        }

    def regenerate_backup_codes(self, count: int = 10) -> List[str]:
        """Rotate all backup codes. Invalidates every existing code."""
        new_codes = [secrets.token_hex(4).upper() for _ in range(count)]
        self._backup_hashes = [
            hashlib.sha256(c.encode()).hexdigest() for c in new_codes
        ]
        return new_codes

    @property
    def backup_codes_remaining(self) -> int:
        return len(self._backup_hashes)

from django.db import models
from django.contrib.auth.models import AbstractUser
from enum import Enum


class MFAMethod(Enum):
    NONE = "none"
    SMS = "sms"
    TOTP = "totp"
    WEBAUTHN = "webauthn"


class User(AbstractUser):
    mfa_method = models.CharField(
        max_length=20,
        choices=[(m.value, m.name) for m in MFAMethod],
        default=MFAMethod.NONE.value
    )
    totp_secret = models.CharField(max_length=32, blank=True, null=True)
    webauthn_credential_id = models.BinaryField(blank=True, null=True)
    backup_code_hash = models.JSONField(default=list)
    failed_2fa_attempts = models.IntegerField(default=0)
    mfa_locked_until = models.DateTimeField(blank=True, null=True)

import base64
import json
import hashlib
from typing import Optional, Dict, Any
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone


def _b64url_decode(data: str) -> bytes:
    """Decode a URL-safe base64 string to bytes."""
    padding = 4 - len(data) % 4
    if padding != 4:
        data += "=" * padding
    return base64.urlsafe_b64decode(data)


def _b64url_encode(data: bytes) -> str:
    """Encode bytes to a URL-safe base64 string without padding."""
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


@dataclass
class WebAuthnRegistration:
    """Handles WebAuthn credential registration (attestation).

    This flow creates a new passkey. The server generates a challenge
    and a set of acceptable credential IDs, sends them to the browser,
    and the browser prompts the user to register a new authenticator.
    """
    challenge: bytes = field(default_factory=lambda: secrets_token(32))
    rp_id: str = "example.com"
    rp_name: str = "Example Corp"
    user_id: bytes = field(default_factory=lambda: secrets_token(16))
    user_name: str = ""
    user_display_name: str = ""
    timeout: int = 60000

    def to_public_key_credential_creation_options(self) -> Dict[str, Any]:
        """Return the options dict that gets serialized to JSON for the browser."""
        return {
            "challenge": _b64url_encode(self.challenge),
            "rp": {
                "name": self.rp_name,
                "id": self.rp_id,
            },
            "user": {
                "id": _b64url_encode(self.user_id),
                "name": self.user_name,
                "displayName": self.user_display_name,
            },
            "pubKeyCredParams": [
                {"type": "public-key", "alg": -7},   # ES256
                {"type": "public-key", "alg": -8},   # EdDSA
                {"type": "public-key", "alg": -257}, # RS256
            ],
            "timeout": self.timeout,
            "attestation": "direct",
            "authenticatorSelectionCriteria": [
                {
                    "authenticatorAttachment": "platform",
                    "residentKey": "discouraged",
                },
                {
                    "authenticatorAttachment": "cross-platform",
                    "residentKey": "preferred",
                },
            ],
        }


@dataclass
class WebAuthnAuthentication:
    """Handles WebAuthn authentication (assertion).

    The server sends a challenge scoped to the user's registered
    credential IDs. The browser signs the challenge with the stored
    private key and returns the assertion. This is immune to phishing
    because the browser refuses to sign for a mismatching origin.
    """
    challenge: bytes = field(default_factory=lambda: secrets_token(32))
    rp_id: str = "example.com"
    allowed_credential_ids: list = field(default_factory=list)
    user_handle: Optional[bytes] = None
    timeout: int = 60000

    def to_public_key_credential_request_options(self) -> Dict[str, Any]:
        """Return the options dict for the browser's getAssertion call."""
        return {
            "challenge": _b64url_encode(self.challenge),
            "rpId": self.rp_id,
            "allowCredentials": [
                {
                    "type": "public-key",
                    "id": _b64url_encode(cred_id),
                    "transports": ["usb", "nfc", "ble", "internal"],
                }
                for cred_id in self.allowed_credential_ids
            ],
            "userVerification": "required",
            "timeout": self.timeout,
        }


def verify_authentication_response(
    credential_id: bytes,
    authenticator_data: bytes,
    client_data_json: bytes,
    signature: bytes,
    user_handle: bytes,
    expected_challenge: bytes,
    expected_origin: str,
    credential_public_key: bytes,
) -> Dict[str, Any]:
    """Verify a WebAuthn authentication assertion on the server side.

    This function implements the core verification steps defined in
    the WebAuthn spec (https://www.w3.org/TR/webauthn-2/#sctn-verifying-assertion).
    Returns a dict with success status and detailed error info on failure.

    Args:
        credential_id: The ID of the credential being used.
        authenticator_data: The authenticator data bytes from the assertion.
        client_data_json: The JSON-serialized client data.
        signature: The cryptographic signature over clientDataJSON.
        user_handle: The user handle (if present).
        expected_challenge: The challenge originally sent to the browser.
        expected_origin: The origin the assertion must match (e.g. 'https://example.com').
        credential_public_key: The stored public key for this credential.

    Returns:
        A dict with 'success' (bool), 'error' (str or None), and 'sign_count' (int).
    """
    import hashlib

    try:
        # Step 1: Verify clientDataJSON structure and challenge.
        client_data = json.loads(client_data_json.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        return {"success": False, "error": f"Invalid clientDataJSON: {exc}"}

    if client_data.get("type") != "webauthn.get":
        return {"success": False, "error": f"Wrong type: expected 'webauthn.get', got '{client_data.get('type')}'"}

    # The challenge in clientData is base64url-encoded by the browser.
    returned_challenge = _b64url_decode(client_data.get("challenge", ""))
    if returned_challenge != expected_challenge:
        return {"success": False, "error": "Challenge mismatch — possible replay or phishing"}

    # Step 2: Verify the origin matches the expected relying party origin.
    if client_data.get("origin") != expected_origin:
        return {"success": False, "error": f"Origin mismatch: expected {expected_origin}, got {client_data.get('origin')}"}

    # Step 3: Verify the Token Binding status (if present).
    token_binding = client_data.get("tokenBinding", {})
    if token_binding.get("status") == "present":
        # In production, validate the TokenBinding ID against the TLS channel.
        pass  # Simplified for this example.

    # Step 4: Compute the clientDataHash.
    client_data_hash = hashlib.sha256(client_data_json).digest()

    # Step 5: Parse authenticatorData and verify the signature.
    # authenticatorData = rpIdHash (32) + flags (1) + signCount (4) + extensions (var)
    if len(authenticator_data) < 37:
        return {"success": False, "error": "Authenticator data too short"}

    rp_id_hash_received = authenticator_data[:32]
    expected_rp_id_hash = hashlib.sha256(expected_origin.encode()).digest()
    if rp_id_hash_received != expected_rp_id_hash:
        return {"success": False, "error": "RP ID hash mismatch"}

    flags = authenticator_data[32]
    sign_count = int.from_bytes(authenticator_data[33:37], "big")

    # Verify user presence flag (bit 0) is set.
    if not (flags & 0x01):
        return {"success": False, "error": "User presence flag not set"}

    # Verify user verification flag (bit 2) is set (required for this flow).
    if not (flags & 0x04):
        return {"success": False, "error": "User verification flag not set"}

    # Step 6: Verify the cryptographic signature.
    # This requires the credential's public key and ECDSA/PSS verification.
    # In production, use a library like `cose` or `cryptography`.
    verification_input = authenticator_data + client_data_hash
    try:
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import ec, utils, ed25519
        from cryptography.hazmat.backends import default_backend

        public_key = serialization.load_der_public_key(
            credential_public_key, backend=default_backend()
        )
        if isinstance(public_key, (ec.EllipticCurvePublicKey,)):
            public_key.verify(
                signature,
                verification_input,
                ec.ECDSA(hashes.SHA256()),
            )
        elif isinstance(public_key, ed25519.Ed25519PublicKey):
            public_key.verify(signature, verification_input)
        else:
            # RSA
            public_key.verify(
                signature,
                verification_input,
                padding.PKCS1v15(),
                hashes.SHA256(),
            )
    except Exception as exc:
        return {"success": False, "error": f"Signature verification failed: {exc}"}

    # Step 7: Check signCount for token-cloning detection.
    if sign_count > 0 and sign_count <= _get_stored_sign_count(credential_id):
        return {"success": False, "error": "signCount not increasing — possible token clone"}

    _store_sign_count(credential_id, sign_count)
    return {"success": True, "sign_count": sign_count}


def secrets_token(n: int) -> bytes:
    """Generate n cryptographically random bytes."""
    import secrets as _secrets
    return _secrets.token_bytes(n)


def _get_stored_sign_count(credential_id: bytes) -> int:
    """Placeholder: retrieve the stored signCount from the database."""
    return 0


def _store_sign_count(credential_id: bytes, count: int) -> None:
    """Placeholder: persist the signCount to the database."""
    pass

import hashlib
import hmac
import os
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass, field


@dataclass
class DeviceFingerprint:
    """Represents a device fingerprint derived from TLS and HTTP signals."""
    tls_client_hello_hash: str
    user_agent_hash: str
    ip_subnet: str  # /24 for IPv4, /48 for IPv6
    screen_resolution: Optional[str] = None
    timezone: Optional[str] = None

    def compute_binding_token(self, server_secret: bytes) -> str:
        """HMAC-SHA256 of the fingerprint fields bound to a server secret."""
        raw = "|".join([
            self.tls_client_hello_hash,
            self.user_agent_hash,
            self.ip_subnet,
            self.screen_resolution or "",
            self.timezone or "",
        ]).encode()
        return hmac.new(server_secret, raw, hashlib.sha256).hexdigest()


@dataclass
class SessionStore:
    """In-memory session store. Replace with Redis or your DB in production."""
    _sessions: Dict[str, Dict[str, Any]] = field(
        default_factory=dict, repr=False
    )
    _max_age_seconds: int = 3600
    _binding_secret: bytes = field(default_factory=lambda: os.urandom(32))

    def create_session(
        self,
        user_id: int,
        fingerprint: DeviceFingerprint,
        mfa_verified: bool = True,
    ) -> str:
        """Create a new session token bound to the device fingerprint."""
        token = os.urandom(32).hex()
        binding = fingerprint.compute_binding_token(self._binding_secret)
        self._sessions[token] = {
            "user_id": user_id,
            "created_at": time.time(),
            "last_seen": time.time(),
            "binding": binding,
            "mfa_verified": mfa_verified,
            "fingerprint_snapshot": {
                "tls_hash": fingerprint.tls_client_hello_hash,
                "ua_hash": fingerprint.user_agent_hash,
                "ip_subnet": fingerprint.ip_subnet,
            },
        }
        return token

    def validate_session(
        self, token: str, fingerprint: DeviceFingerprint
    ) -> Dict[str, Any]:
        """Validate a session token and check device binding."""
        session = self._sessions.get(token)
        if session is None:
            return {"valid": False, "error": "session_not_found"}

        if time.time() - session["last_seen"] > self._max_age_seconds:
            del self._sessions[token]
            return {"valid": False, "error": "session_expired"}

        # Verify device binding.
        current_binding = fingerprint.compute_binding_token(self._binding_secret)
        if not hmac.compare_digest(current_binding, session["binding"]):
            return {
                "valid": False,
                "error": "device_mismatch",
                "action": "re-authenticate",
            }

        # Update last-seen timestamp.
        session["last_seen"] = time.time()
        return {
            "valid": True,
            "user_id": session["user_id"],
            "mfa_verified": session["mfa_verified"],
        }

    def revoke_session(self, token: str) -> bool:
        """Explicitly revoke a session token."""
        if token in self._sessions:
            del self._sessions[token]
            return True
        return False

    def revoke_all_user_sessions(self, user_id: int) -> int:
        """Revoke all sessions for a given user (useful on password change)."""
        tokens_to_remove = [
            t for t, s in self._sessions.items()
            if s["user_id"] == user_id
        ]
        for t in tokens_to_remove:
            del self._sessions[t]
        return len(tokens_to_remove)