Autenticación de Usuarios en FastHTML: OAuth, Login y Controles de Administrador

# Simplified Beforeware flow
login_redir = RedirectResponse('/login', status_code=303)

def before(req, sess):
    auth = req.scope['auth'] = sess.get('user_id', None)
    if not auth:
        return login_redir

bware = Beforeware(before, skip=[r'/favicon\.ico', r'/static/.*', r'.*\.css',
                                 '/', '/login', '/register', '/auth/github/callback'])

CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    username TEXT UNIQUE,
    email TEXT UNIQUE,
    password_hash TEXT,
    salt TEXT,
    github_id TEXT UNIQUE,
    is_admin BOOLEAN DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_login TIMESTAMP
)

CREATE TABLE IF NOT EXISTS title_history (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER,
    topic TEXT NOT NULL,
    platform TEXT NOT NULL,
    style TEXT NOT NULL,
    number_of_titles INTEGER NOT NULL,
    titles TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id)
)

your-app/
├── auth/
│   ├── auth_manager.py      # Session helpers, login_user(), is_admin()
│   └── email_auth.py        # Email/password validation and authentication
├── components/
│   ├── header.py            # Auth-aware navigation
│   ├── footer.py            # Footer component
│   └── page_layout.py       # Page wrapper with header/footer
├── db/
│   ├── database.py          # Database class, schema init, password hashing
│   ├── user_dao.py          # User CRUD operations
│   └── history_dao.py       # History CRUD operations
├── pages/
│   ├── home.py              # Landing page
│   ├── login.py             # Login form (email + GitHub OAuth)
│   ├── register.py          # Registration form
│   ├── title_generator.py   # Protected: title generation form
│   ├── history.py           # Protected: user history dashboard
│   └── admin.py             # Admin-only: user mgmt, global history
├── tools/
│   └── title_generator.py   # LLM title generation logic
├── main.py                  # Routes, OAuth subclass, Beforeware, app init
├── config.py                # Environment config
├── .env                     # Secrets (not committed)
└── requirements.txt         # python-fasthtml>=0.14,<0.15, python-dotenv, etc.

import os
import hashlib
import hmac
import secrets
from contextlib import contextmanager
import sqlite3
import config

class Database:
    def __init__(self, db_path=None):
        self.db_path = db_path or config.DB_PATH
        if not self.db_path:
            self.db_path = "tools.db"
        self._initialize_db()

    def _initialize_db(self):
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE,
                email TEXT UNIQUE,
                password_hash TEXT,
                salt TEXT,
                github_id TEXT UNIQUE,
                is_admin BOOLEAN DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_login TIMESTAMP
            )
            ''')
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS title_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER,
                topic TEXT NOT NULL,
                platform TEXT NOT NULL,
                style TEXT NOT NULL,
                number_of_titles INTEGER NOT NULL,
                titles TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
            ''')
            # Auto-create admin if configured
            cursor.execute("SELECT COUNT(*) FROM users WHERE is_admin = 1")
            if cursor.fetchone()[0] == 0 and config.ADMIN_EMAIL and config.ADMIN_PASSWORD:
                salt = secrets.token_hex(16)
                password_hash = self._hash_password(config.ADMIN_PASSWORD, salt)
                cursor.execute('''
                INSERT INTO users (username, email, password_hash, salt, is_admin)
                VALUES (?, ?, ?, ?, 1)
                ''', ('admin', config.ADMIN_EMAIL, password_hash, salt))
            conn.commit()

    @staticmethod
    def _hash_password(password, salt):
        """PBKDF2-HMAC-SHA256 with 600,000 iterations (OWASP minimum)."""
        return hashlib.pbkdf2_hmac(
            'sha256',
            password.encode('utf-8'),
            salt.encode('utf-8'),
            600_000  # was 100_000 in the original article
        ).hex()

    @contextmanager
    def get_connection(self):
        db_dir = os.path.dirname(self.db_path)
        if db_dir:
            os.makedirs(db_dir, exist_ok=True)
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
        finally:
            conn.close()

db = Database()

from typing import Optional, Dict, Any, List
import hmac
import secrets
from datetime import datetime
from .database import db

class UserDAO:
    @staticmethod
    def create_user(username: str, email: str, password: Optional[str] = None,
                    github_id: Optional[str] = None) -> int:
        try:
            with db.get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "SELECT id FROM users WHERE email = ? OR username = ? "
                    "OR (github_id = ? AND github_id IS NOT NULL)",
                    (email, username, github_id)
                )
                if cursor.fetchone():
                    return 0

                password_hash = None
                salt = None
                if password:
                    salt = secrets.token_hex(16)
                    password_hash = db._hash_password(password, salt)

                cursor.execute('''
                INSERT INTO users (username, email, password_hash, salt, github_id, is_admin)
                VALUES (?, ?, ?, ?, ?, 0)
                ''', (username, email, password_hash, salt, github_id))
                conn.commit()
                return cursor.lastrowid
        except Exception as e:
            print(f"Error creating user: {e}")
            return 0

    @staticmethod
    def authenticate_email(email: str, password: str) -> Optional[Dict[str, Any]]:
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM users WHERE email = ?", (email,)
            )
            user = cursor.fetchone()
            if not user:
                return None
            if not user['password_hash'] or not user['salt']:
                return None

            # Constant-time comparison (OWASP recommendation)
            password_hash = db._hash_password(password, user['salt'])
            if not hmac.compare_digest(password_hash, user['password_hash']):
                return None

            # Update last_login
            cursor.execute(
                "UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?",
                (user['id'],)
            )
            conn.commit()
            return dict(user)

    @staticmethod
    def find_or_create_github_user(github_id: str, username: str,
                                    email: str) -> Optional[Dict[str, Any]]:
        try:
            with db.get_connection() as conn:
                cursor = conn.cursor()
                # Try to find by github_id first
                cursor.execute("SELECT * FROM users WHERE github_id = ?", (github_id,))
                user = cursor.fetchone()
                if user:
                    cursor.execute(
                        "UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?",
                        (user['id'],)
                    )
                    conn.commit()
                    return dict(user)

                # Create new GitHub user
                cursor.execute('''
                INSERT INTO users (username, email, github_id, is_admin)
                VALUES (?, ?, ?, 0)
                ''', (username, email, github_id))
                conn.commit()
                cursor.execute("SELECT * FROM users WHERE id = ?", (cursor.lastrowid,))
                return dict(cursor.fetchone())
        except Exception as e:
            print(f"Error with GitHub user: {e}")
            return None

    @staticmethod
    def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]:
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
            user = cursor.fetchone()
            return dict(user) if user else None

    @staticmethod
    def is_user_admin(user_id: int) -> bool:
        """Check admin status from DB (not session) for live demotion."""
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT is_admin FROM users WHERE id = ?", (user_id,))
            user = cursor.fetchone()
            return bool(user and user['is_admin'])

    @staticmethod
    def get_all_users() -> List[Dict[str, Any]]:
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT id, username, email, github_id, is_admin, "
                          "created_at, last_login FROM users ORDER BY created_at DESC")
            return [dict(row) for row in cursor.fetchall()]

    @staticmethod
    def set_admin(user_id: int, is_admin: bool) -> bool:
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE users SET is_admin = ? WHERE id = ?",
                          (int(is_admin), user_id))
            conn.commit()
            return cursor.rowcount > 0

    @staticmethod
    def delete_user(user_id: int) -> bool:
        try:
            with db.get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("DELETE FROM title_history WHERE user_id = ?", (user_id,))
                cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
                conn.commit()
                return cursor.rowcount > 0
        except Exception as e:
            print(f"Error deleting user: {e}")
            return False

import os
from dotenv import load_dotenv

load_dotenv()

# Database
DB_PATH = os.getenv("DB_PATH", "tools.db")

# Security -- no placeholder default! Fail fast if unset.
SECRET_KEY = os.getenv("SECRET_KEY")
if not SECRET_KEY:
    raise ValueError("SECRET_KEY environment variable is required. "
                     "Generate one with: python -c 'import secrets; print(secrets.token_hex(32))'")

# Session
SESSION_EXPIRY = int(os.getenv("SESSION_EXPIRY", 604800))  # 7 days in seconds

# GitHub OAuth
GITHUB_CLIENT_ID = os.getenv("GITHUB_CLIENT_ID", "")
GITHUB_CLIENT_SECRET = os.getenv("GITHUB_CLIENT_SECRET", "")

# Admin bootstrap
ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "")
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "")

# LLM (inherited from Part 5)
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "openai/gpt-4o-mini")

# App
DEBUG = os.getenv("DEBUG", "false").lower() == "true"

from fasthtml.common import *
from fasthtml.oauth import GitHubAppClient, OAuth, redir_url
import secrets
import hmac
import config
from auth.auth_manager import AuthManager
from auth.email_auth import EmailAuth
from db.user_dao import UserDAO
from db.history_dao import HistoryDAO

login_redir = RedirectResponse('/login', status_code=303)

def before(req, sess):
    """Beforeware: check session auth, redirect to login if missing."""
    auth = req.scope['auth'] = sess.get('user_id', None)
    if not auth:
        return login_redir

bware = Beforeware(before, skip=[
    r'/favicon\.ico', r'/static/.*', r'.*\.css',
    '/', '/login', '/register',
    '/auth/github/callback',
    '/error',
])

app = FastHTML(
    secret_key=config.SECRET_KEY,
    before=bware,
    session_cookie="session",
    max_age=config.SESSION_EXPIRY,
    sess_path="/",
    same_site="lax",
    sess_https_only=not config.DEBUG,
)

# Route helper: get user_id from req.scope (set by Beforeware)
def get_auth(req): return req.scope['auth']

# In main.py, after app initialization

client = GitHubAppClient(
    config.GITHUB_CLIENT_ID,
    config.GITHUB_CLIENT_SECRET,
    scope="user:email"
)

class Auth(OAuth):
    def get_auth(self, info, ident, session, state):
        """
        Called after GitHub redirects back.

        Args:
            info: dict with 'login', 'email' (may be None), etc.
            ident: the GitHub user ID (from client.id_key == 'id')
            session: the Starlette session
            state: the CSRF state parameter
        """
        # GitHub may return email: null for non-public emails
        email = info.get('email') or f"{info['login']}-{ident}@github.user"

        user = UserDAO.find_or_create_github_user(
            str(ident), info['login'], email
        )
        if not user:
            return None  # redirects back to /login

        AuthManager.login_user(session, user)
        # Generate CSRF token on login
        session['csrf'] = secrets.token_hex(32)
        return RedirectResponse('/', status_code=303)

oauth = Auth(app, client, redir_path='/auth/github/callback', login_path='/login')

# pages/login.py
from fasthtml.common import *

def login_page(error=None, success=None, session=None):
    return Div(
        # Error/success messages...
        H2("Sign In", cls="text-2xl font-bold mb-6"),

        # GitHub OAuth button
        A("Sign in with GitHub",
          href="/login",  # OAuth's login_path handles the redirect
          cls="block w-full text-center bg-gray-800 text-white py-2 px-4 rounded mb-4 hover:bg-gray-700"),

        Div("or", cls="text-center text-gray-500 my-4"),

        # Email/password form
        Form(
            # CSRF token in every state-changing form
            Input(type="hidden", id="csrf", value=session.get('csrf', '')),
            Input(type="email", name="email", placeholder="Email", required=True,
                  cls="w-full p-2 border rounded mb-3"),
            Input(type="password", name="password", placeholder="Password",
                  required=True, cls="w-full p-2 border rounded mb-3"),
            Button("Sign In with Email", type="submit",
                   cls="w-full bg-blue-600 text-white py-2 px-4 rounded hover:bg-blue-700"),
            method="post", action="/auth/email/login"
        ),

        # Register link
        P(A("Create an account", href="/register", cls="text-blue-600"),
          cls="mt-4 text-center"),
        cls="max-w-md mx-auto mt-10 p-6 bg-white rounded shadow"
    )

import re
from typing import Optional, Dict, Any, Tuple

class EmailAuth:
    @staticmethod
    def validate_registration(username: str, email: str,
                               password: str) -> Tuple[bool, str]:
        """Validate registration input. Returns (is_valid, error_message)."""
        if not username or len(username) < 3:
            return False, "Username must be at least 3 characters"
        if not email or not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', email):
            return False, "Invalid email address"
        if not password or len(password) < 8:
            return False, "Password must be at least 8 characters"
        if not re.search(r'[A-Z]', password) or not re.search(r'[0-9]', password):
            return False, "Password must contain at least one uppercase letter and one number"
        return True, ""

    @staticmethod
    def authenticate(email: str, password: str) -> Optional[Dict[str, Any]]:
        """Authenticate user with email/password."""
        from db.user_dao import UserDAO
        return UserDAO.authenticate_email(email, password)

@app.post('/auth/email/register')
def email_register(req, username: str, email: str, password: str,
                   confirm_password: str, csrf: str = ""):
    session = req.session

    # CSRF validation
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        return RedirectResponse('/register', status_code=303)

    # Validate input
    valid, error = EmailAuth.validate_registration(username, email, password)
    if not valid:
        return register_page(error=error, session=session)

    if password != confirm_password:
        return register_page(error="Passwords do not match", session=session)

    # Create user
    from db.user_dao import UserDAO
    user_id = UserDAO.create_user(username=username, email=email, password=password)
    if not user_id:
        return register_page(error="Email or username already exists", session=session)

    return RedirectResponse('/login?registered=1', status_code=303)

@app.post('/auth/email/login')
def email_login(req, email: str, password: str, csrf: str = ""):
    session = req.session

    # CSRF validation
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        return login_page(error="Invalid request", session=session)

    user = EmailAuth.authenticate(email, password)
    if not user:
        return login_page(error="Invalid email or password", session=session)

    AuthManager.login_user(session, user)
    session['csrf'] = secrets.token_hex(32)
    return RedirectResponse('/', status_code=303)

# Helper to inject CSRF hidden field into forms
def csrf_field(session):
    return Input(type="hidden", id="csrf", name="csrf",
                 value=session.get('csrf', ''))

# In any POST form:
Form(
    csrf_field(session),
    Input(type="text", name="topic", ...),
    Button("Generate Titles", type="submit"),
    method="post", action="/generate"
)

# In any POST handler:
@app.post('/generate')
def generate_titles(req, topic: str, csrf: str = "", ...):
    session = req.session
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        return Div("Invalid request", cls="text-red-600")
    # ... rest of handler

# In main.py -- admin route helper
def require_admin(req):
    """Check admin status from DB. Returns user_id or raises."""
    user_id = get_auth(req)
    if not UserDAO.is_user_admin(user_id):
        raise HTTPException(403, "Admin access required")
    return user_id

# Admin dashboard
@app.get('/admin')
def admin_dashboard(req):
    user_id = require_admin(req)
    users = UserDAO.get_all_users()
    return page_layout("Admin Dashboard",
                      admin_dashboard_content(users),
                      current_page="/admin", session=req.session)

@app.post('/admin/make-admin')
def make_admin(req, user_id: int, csrf: str = ""):
    session = req.session
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        raise HTTPException(401, "Invalid CSRF token")
    admin_id = require_admin(req)
    UserDAO.set_admin(user_id, True)
    return RedirectResponse('/admin', status_code=303)

@app.post('/admin/delete-user')
def delete_user(req, user_id: int, csrf: str = ""):
    session = req.session
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        raise HTTPException(401, "Invalid CSRF token")
    admin_id = require_admin(req)
    if user_id == admin_id:
        return RedirectResponse('/admin?error=cannot_delete_self', status_code=303)
    UserDAO.delete_user(user_id)
    return RedirectResponse('/admin', status_code=303)

from fasthtml.common import *
from db.user_dao import UserDAO

def header(current_page="/", auth=None):
    """Auth-aware navigation header."""
    nav_items = [A("Home", href="/", cls="hover:text-blue-600")]

    if auth:
        nav_items.extend([
            A("Title Generator", href="/title-generator",
              cls="hover:text-blue-600"),
            A("My History", href="/history",
              cls="hover:text-blue-600"),
        ])
        # Admin link only for admins (DB check, not session)
        if UserDAO.is_user_admin(auth):
            nav_items.append(
                A("Admin", href="/admin", cls="hover:text-red-600 font-semibold")
            )
        nav_items.append(
            A("Logout", href="/logout", cls="hover:text-red-600")
        )
    else:
        nav_items.extend([
            A("Login", href="/login", cls="hover:text-blue-600"),
            A("Register", href="/register", cls="hover:text-blue-600"),
        ])

    return Nav(
        Div(
            A("AI Title Generator", href="/", cls="text-xl font-bold"),
            Div(*nav_items, cls="flex gap-4"),
            cls="container mx-auto px-4 py-3 flex justify-between items-center"
        ),
        cls="bg-white shadow"
    )

from fasthtml.common import *
from .header import header
from .footer import footer

def page_layout(title, content, current_page="/", auth=None):
    return Html(
        Head(
            Title(title),
            Meta(charset="UTF-8"),
            Meta(name="viewport", content="width=device-width, initial-scale=1.0"),
            Script(src="https://cdn.tailwindcss.com"),
        ),
        Body(
            Div(
                header(current_page, auth),
                Main(
                    Div(content, cls="container mx-auto px-4 py-8"),
                    cls="flex-grow"
                ),
                footer(),
                cls="flex flex-col min-h-screen"
            )
        )
    )

# Title generator -- protected by Beforeware
@app.get('/title-generator')
def title_generator_page(req):
    auth = get_auth(req)
    return page_layout("Title Generator",
                      title_generator_form(),
                      current_page="/title-generator", auth=auth)

@app.post('/generate')
def generate_titles(req, topic: str, platform: str, style: str,
                    number: int, csrf: str = ""):
    session = req.session
    auth = get_auth(req)

    # CSRF check
    if not hmac.compare_digest(session.get('csrf', ''), csrf):
        return title_generator_form(error="Invalid request")

    # Generate titles (LLM call from Part 5)
    from tools.title_generator import generate
    titles = generate(topic, platform, style, number)

    # Save to history with user_id
    HistoryDAO.save_generation(auth, topic, platform, style, number, titles)

    return title_generator_form(titles=titles, session=session, auth=auth)

# User history -- shows only the logged-in user's records
@app.get('/history')
def history_page(req):
    auth = get_auth(req)
    records = HistoryDAO.get_user_history(auth)
    return page_layout("My History",
                      user_history_content(records),
                      current_page="/history", auth=auth)

# Admin history -- shows all users (admin check via DB)
@app.get('/admin/history')
def admin_history(req):
    auth = require_admin(req)
    records = HistoryDAO.get_all_history()
    return page_layout("All History",
                      admin_history_content(records),
                      current_page="/admin/history", auth=auth)

# .env
SECRET_KEY=your-random-64-char-hex-string
GITHUB_CLIENT_ID=your_github_client_id
GITHUB_CLIENT_SECRET=your_github_client_secret
ADMIN_EMAIL=[email protected]
ADMIN_PASSWORD=YourStr0ngPassword
DB_PATH=tools.db
DEBUG=true
SESSION_EXPIRY=604800
DEFAULT_MODEL=openai/gpt-4o-mini

pip install python-fasthtml>=0.14,<0.15 python-dotenv
python main.py

from fasthtml.test import FastHTMLTestClient

client = FastHTMLTestClient(app)

# Test registration
resp = client.post('/auth/email/register', data={
    'username': 'testuser',
    'email': '[email protected]',
    'password': 'TestPass123',
    'confirm_password': 'TestPass123',
    'csrf': ''  # bypass CSRF in tests or set session first
})
assert resp.status_code == 303

# Test login
resp = client.post('/auth/email/login', data={
    'email': '[email protected]',
    'password': 'TestPass123',
    'csrf': ''
})
assert resp.status_code == 303
assert 'session' in resp.cookies

conn.execute("PRAGMA journal_mode=WAL")