Web Auth · Multi-2FA¶

from enum import Enum
from uuid import UUID, uuid4
from datetime import datetime
from pydantic import BaseModel, EmailStr


class UserRole(str, Enum):
    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"
    GUEST = "guest"  # TTL 48h


class ThemePreference(str, Enum):
    LIGHT = "light"
    DARK = "dark"
    SYSTEM = "system"


class PrivacyMode(str, Enum):
    STANDARD = "standard"
    REDUCED = "reduced"      # niente testo conversazioni in log
    ANONYMOUS = "anonymous"  # solo metadati tecnici


class UserProfile(BaseModel):
    id: UUID = uuid4()
    email: EmailStr
    display_name: str
    avatar_url: str | None = None
    role: UserRole = UserRole.MEMBER
    theme: ThemePreference = ThemePreference.SYSTEM
    language: str = "it"
    tts_voice: str | None = None
    privacy_mode: PrivacyMode = PrivacyMode.STANDARD
    mfa_enabled: bool = False
    created_at: datetime
    last_login_at: datetime | None = None
    is_active: bool = True

from fastapi import Depends, HTTPException

ROLE_HIERARCHY = {UserRole.OWNER: 4, UserRole.ADMIN: 3, UserRole.MEMBER: 2, UserRole.GUEST: 1}


def require_role(min_role: UserRole):
    def dep(current_user: UserProfile = Depends(get_current_user)):
        if ROLE_HIERARCHY[current_user.role] < ROLE_HIERARCHY[min_role]:
            raise HTTPException(403, "Permessi insufficienti")
        return current_user
    return dep


# Usage:
# @router.delete("/users/{uid}")
# async def delete_user(uid: UUID, _=Depends(require_role(UserRole.ADMIN))):

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

# Parametri OWASP 2025: memory=128MB, time=3, parallelism=4
_ph = PasswordHasher(time_cost=3, memory_cost=131072, parallelism=4, hash_len=32, salt_len=16)


def hash_password(plain: str) -> str:
    return _ph.hash(plain)


def verify_password(plain: str, hashed: str) -> bool:
    try:
        return _ph.verify(hashed, plain)
    except VerifyMismatchError:
        return False


PASSWORD_POLICY = {
    "min_length": 12,
    "require_uppercase": True,
    "require_digit": True,
    "require_special": True,
    "pwned_check": True,  # API HaveIBeenPwned k-anonymity
}

SESSION_COOKIE_SETTINGS = {
    "key": "jarvis_session",
    "httponly": True,
    "secure": True,
    "samesite": "strict",
    "max_age": 86400 * 14,
    "path": "/",
}

import pyotp
import qrcode
import io
import base64
import secrets


def generate_totp_secret() -> str:
    return pyotp.random_base32()


def get_totp_uri(secret: str, email: str, issuer: str = "open-jarvis") -> str:
    return pyotp.TOTP(secret).provisioning_uri(name=email, issuer_name=issuer)


def generate_qr_base64(uri: str) -> str:
    img = qrcode.make(uri)
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode()


def verify_totp(secret: str, code: str) -> bool:
    return pyotp.TOTP(secret).verify(code, valid_window=1)  # ±30s NTP drift


def generate_backup_codes(count: int = 10) -> list[str]:
    return [secrets.token_hex(5).upper() for _ in range(count)]

import webauthn
from webauthn.helpers.structs import PublicKeyCredentialDescriptor, AuthenticatorTransport

RP_ID = "jarvis.local"
RP_NAME = "open-jarvis"
ORIGIN = "https://jarvis.local"


def begin_registration(user_id: bytes, username: str, existing: list):
    return webauthn.generate_registration_options(
        rp_id=RP_ID,
        rp_name=RP_NAME,
        user_id=user_id,
        user_name=username,
        exclude_credentials=[
            PublicKeyCredentialDescriptor(id=c.credential_id) for c in existing
        ],
    )


def complete_registration(credential, challenge: bytes, expected_origin: str):
    return webauthn.verify_registration_response(
        credential=credential,
        expected_challenge=challenge,
        expected_rp_id=RP_ID,
        expected_origin=expected_origin,
    )


def begin_authentication(credentials: list):
    return webauthn.generate_authentication_options(
        rp_id=RP_ID,
        allow_credentials=[
            PublicKeyCredentialDescriptor(
                id=c.credential_id, transports=[AuthenticatorTransport.INTERNAL]
            )
            for c in credentials
        ],
    )


def complete_authentication(credential, challenge: bytes, stored_cred):
    return webauthn.verify_authentication_response(
        credential=credential,
        expected_challenge=challenge,
        expected_rp_id=RP_ID,
        expected_origin=ORIGIN,
        credential_public_key=stored_cred.public_key,
        credential_current_sign_count=stored_cred.sign_count,
    )

from datetime import datetime, timedelta, timezone
from fastapi import Request, HTTPException

STEP_UP_WINDOW = timedelta(minutes=5)


def require_step_up(request: Request) -> None:
    session = request.state.session
    verified_at = session.get("step_up_verified_at")
    if not verified_at:
        raise HTTPException(403, "step_up_required")
    if datetime.now(timezone.utc) - verified_at > STEP_UP_WINDOW:
        raise HTTPException(403, "step_up_expired")

from fastapi import APIRouter, Response, Request, Depends, HTTPException
from pydantic import BaseModel, EmailStr
from uuid import UUID

router = APIRouter(prefix="/auth", tags=["auth"])


class LoginRequest(BaseModel):
    email: EmailStr
    password: str


class TOTPVerifyRequest(BaseModel):
    session_token: str
    code: str


@router.post("/register", status_code=201)
async def register(body: dict):
    validate_password_policy(body["password"])
    hashed = hash_password(body["password"])
    user = await create_user(body["email"], hashed, body["display_name"])
    await send_verification_email(user.email)
    return {"message": "Verifica la tua email."}


@router.post("/login")
async def login(body: LoginRequest, response: Response):
    user = await get_user_by_email(body.email)
    if not user or not verify_password(body.password, user.password_hash):
        await record_failed_attempt(body.email)
        raise HTTPException(401, "Credenziali non valide.")
    if user.mfa_enabled:
        pending_token = create_pending_session(user.id)  # 5 min
        return {"mfa_required": True, "session_token": pending_token,
                "mfa_methods": user.enrolled_mfa_methods}
    set_session_cookie(response, create_session(user.id))
    return {"message": "Login completato."}


@router.post("/2fa/totp/verify")
async def verify_totp_endpoint(body: TOTPVerifyRequest, response: Response):
    user = await get_user_from_pending(body.session_token)
    if not verify_totp(user.totp_secret, body.code):
        raise HTTPException(401, "Codice TOTP non valido.")
    set_session_cookie(response, create_session(user.id))
    return {"message": "2FA completato."}


@router.post("/2fa/email/send")
async def send_email_otp(body: dict):
    user = await get_user_from_pending(body["session_token"])
    otp = generate_email_otp(user.id)
    await send_otp_email(user.email, otp)
    return {"message": "OTP inviato."}


@router.post("/2fa/email/verify")
async def verify_email_otp(body: dict, response: Response):
    user = await get_user_from_pending(body["session_token"])
    if not check_email_otp(user.id, body["code"]):
        raise HTTPException(401, "OTP non valido o scaduto.")
    set_session_cookie(response, create_session(user.id))
    return {"message": "2FA via email completato."}


@router.post("/passkey/register/begin")
async def passkey_register_begin(request: Request,
                                  current_user=Depends(get_current_user)):
    existing = await get_user_credentials(current_user.id)
    options = begin_registration(current_user.id.bytes, current_user.email, existing)
    request.session["webauthn_challenge"] = options.challenge
    return options


@router.post("/passkey/login")
async def passkey_login(request: Request, credential: dict, response: Response):
    user_handle = extract_user_handle(credential)
    user = await get_user_by_id(UUID(bytes=user_handle))
    stored = await get_credential(credential["id"])
    challenge = request.session.pop("webauthn_challenge", None)
    result = complete_authentication(credential, challenge, stored)
    await update_sign_count(stored.id, result.new_sign_count)
    set_session_cookie(response, create_session(user.id))
    return {"message": "Login con passkey completato."}

from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timezone


class AuditEvent(str, Enum):
    LOGIN_SUCCESS = "login_success"
    LOGIN_FAILED = "login_failed"
    MFA_ENROLLED = "mfa_enrolled"
    MFA_VERIFIED = "mfa_verified"
    PASSWORD_CHANGED = "password_changed"
    PASSKEY_REGISTERED = "passkey_registered"
    ACCOUNT_LOCKED = "account_locked"
    DATA_EXPORT_REQUESTED = "data_export_requested"
    ACCOUNT_DELETED = "account_deleted"


@dataclass(frozen=True)
class AuditRecord:
    user_id: UUID
    event: AuditEvent
    ip_address: str
    user_agent: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    metadata: dict = field(default_factory=dict)


async def record_audit_event(record: AuditRecord) -> None:
    await db.execute(
        "INSERT INTO audit_log VALUES ($1,$2,$3,$4,$5,$6)",
        record.user_id, record.event.value, record.ip_address,
        record.user_agent, record.timestamp, record.metadata
    )