Anexo con Docker: el módulo de identidad de tu reto

git clone https://github.com/rpversontec/reto-auth-fastapi-docker.git
cd reto-auth-fastapi-docker

[project]
name = "auth-fastapi"
version = "0.1.0"
description = "Modulo de autenticacion y autorizacion de referencia para el reto de TC2007B: FastAPI + PostgreSQL, mismo contrato que la API de la Practica 6."
requires-python = ">=3.11"
dependencies = [
    "fastapi==0.141.1",
    "uvicorn[standard]==0.53.0",
    "sqlalchemy==2.0.54",
    "psycopg[binary]==3.3.6",
    "pyjwt==2.14.0",
    "pwdlib[argon2]==0.3.1",
    "pydantic-settings==2.15.0",
]

# La imagen del servidor: Python 3.12 y las dependencias exactas de uv.lock.
FROM python:3.12-slim-bookworm
COPY --from=ghcr.io/astral-sh/uv:0.12.9 /uv /bin/uv

WORKDIR /code
ENV UV_LINK_MODE=copy \
    UV_COMPILE_BYTECODE=1 \
    PYTHONUNBUFFERED=1

# Primero las dependencias: si solo cambia tu código, Docker reutiliza esta capa
# y reconstruir tarda segundos, no minutos.
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-install-project --no-dev

COPY app ./app

EXPOSE 8000
CMD ["uv", "run", "--no-sync", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# Todo el backend en la red de Docker:
#   docker compose up -d                   la base y el servidor (http://localhost:8000)
#   docker compose --profile tunel up -d   además, el túnel HTTPS
#   docker compose logs tunel              ahí aparece la dirección https://…trycloudflare.com
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: avisos
    volumes:
      - pgdata:/var/lib/postgresql/data
    # Sin `ports`: la base solo se ve dentro de la red de Docker. Así no choca con
    # un PostgreSQL que ya tengas en el 5432. Para entrar: docker compose exec db psql -U postgres avisos
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d avisos"]
      interval: 2s
      retries: 30

  api:
    build: .
    env_file: .env
    environment:
      # Dentro de la red de Docker, la base se llama `db`, no `localhost`.
      DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/avisos
    # --reload + el volumen: editas app/ en tu máquina y el servidor se reinicia solo.
    command: ["uv", "run", "--no-sync", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
    volumes:
      - ./app:/code/app
    # Solo en 127.0.0.1: tu máquina y el emulador (10.0.2.2) llegan; el resto del wifi, no.
    ports:
      - "127.0.0.1:8000:8000"
    depends_on:
      db:
        condition: service_healthy

  tunel:
    image: cloudflare/cloudflared:2026.9.3
    # Llega a la API por su nombre en la red de Docker. http2 porque muchas redes bloquean QUIC (UDP).
    command: ["tunnel", "--no-autoupdate", "--protocol", "http2", "--url", "http://api:8000"]
    depends_on:
      - api
    profiles: ["tunel"]

volumes:
  pgdata:

cp .env.example .env

# Copia este archivo a .env y cambia los valores. .env NUNCA se sube al repositorio.
# DATABASE_URL no va aquí: la pone docker-compose.yml, porque dentro de Docker la base se llama `db`.
JWT_SECRET=cambia-esto-por-48-bytes-al-azar
CODIGO_PROFESOR=cambia-este-codigo
ACCESS_TTL_S=300
REFRESH_TTL_S=604800

docker compose up -d
docker compose logs -f api

== 5 login mal (401)                           [401]
== 8 avisos con token (200)                    [200]
== 10 alumno publica (403)                     [403]
== 15 reusar refresh viejo (401)               [401]
== 17 token alterado (401)                     [401]

class Settings(BaseSettings):
    """Todo lo que cambia entre tu máquina y el servidor vive en el entorno, no en el código.

    Se lee de variables de entorno o de un archivo `.env` (que nunca se sube al repo).
    Si falta `JWT_SECRET` o `CODIGO_PROFESOR`, la app no arranca — a propósito.
    """

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/avisos"
    jwt_secret: str
    codigo_profesor: str

    # Cinco minutos de acceso y siete días de refresco: los mismos números de la Práctica 6.
    access_ttl_s: int = 5 * 60
    refresh_ttl_s: int = 7 * 24 * 60 * 60

class Usuario(Base):
    """Quién es quién. La contraseña NO está: está su hash Argon2 (que ya trae su sal adentro)."""

    __tablename__ = "usuarios"

    usuario: Mapped[str] = mapped_column(String(32), primary_key=True)
    password_hash: Mapped[str] = mapped_column(Text)
    rol: Mapped[str] = mapped_column(String(16))  # "alumno" | "profesor"
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())


class Sesion(Base):
    """Una fila por login. Guarda el HASH del refresh token, nunca el token."""

    __tablename__ = "sesiones"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    usuario: Mapped[str] = mapped_column(ForeignKey("usuarios.usuario"), index=True)
    refresh_hash: Mapped[str] = mapped_column(String(64), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    expires_at: Mapped[int] = mapped_column(BigInteger)  # epoch en segundos
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

# `recommended()` es Argon2id con parámetros razonables. No los ajustes sin leer por qué.
_hasher = PasswordHash.recommended()

ALGORITMO = "HS256"


def hash_password(password: str) -> str:
    return _hasher.hash(password)


def verify_password(password: str, password_hash: str) -> bool:
    return _hasher.verify(password, password_hash)


def sign_jwt(usuario: str, rol: str, ttl_s: int, secret: str) -> str:
    """Un JWT es JSON firmado, no cifrado: cualquiera lo lee, nadie lo altera."""
    ahora = int(time.time())
    payload = {"sub": usuario, "rol": rol, "iat": ahora, "exp": ahora + ttl_s}
    return jwt.encode(payload, secret, algorithm=ALGORITMO)


def verify_jwt(token: str, secret: str) -> dict[str, Any]:
    """PyJWT verifica la firma Y la fecha. El orden importa: un token alterado ni se lee."""
    try:
        return jwt.decode(token, secret, algorithms=[ALGORITMO])
    except jwt.ExpiredSignatureError as e:
        raise TokenInvalido("expirado") from e
    except jwt.InvalidTokenError as e:
        raise TokenInvalido("invalido") from e


def random_token() -> str:
    """32 bytes al azar en base64url. No dice nada de nadie: es una llave, no una credencial."""
    return secrets.token_urlsafe(32)

class ApiError(Exception):
    def __init__(self, status: int, error: str, code: str | None = None, hint: str | None = None) -> None:
        super().__init__(error)
        self.status = status
        self.error = error
        self.code = code
        self.hint = hint


def instalar(app: FastAPI) -> None:
    @app.exception_handler(ApiError)
    async def api_error(_: Request, e: ApiError) -> JSONResponse:
        cuerpo: dict[str, str] = {"error": e.error}
        if e.code:
            cuerpo["code"] = e.code
        if e.hint:
            cuerpo["hint"] = e.hint
        headers = {}
        if e.status == 401:
            # El estándar: un 401 dice CÓMO autenticarse. Los clientes serios lo leen.
            headers["WWW-Authenticate"] = 'Bearer realm="reto"'
        return JSONResponse(cuerpo, status_code=e.status, headers=headers)

# Declarar el esquema Bearer es lo que hace aparecer «Authorize» y los candados en /docs.
# auto_error=False: si falta el token, el 401 lo da usuario_actual con la forma del
# contrato ({error, code, hint}), no FastAPI con {"detail": ...}, que la app no lee.
_bearer = HTTPBearer(auto_error=False)


def usuario_actual(credenciales: HTTPAuthorizationCredentials | None = Depends(_bearer)) -> Identidad:
    """Lee `Authorization: Bearer <jwt>`, verifica la firma y la fecha."""
    if credenciales is None:
        raise ApiError(401, "Falta el token de acceso", "falta_token", "Manda `Authorization: Bearer <accessToken>`.")
    token = credenciales.credentials
    try:
        payload = verify_jwt(token, settings.jwt_secret)
    except TokenInvalido as e:
        if e.reason == "expirado":
            raise ApiError(401, "El token de acceso expiró", "token_expirado", "Pide otro con POST /api/auth/refresh.")
        raise ApiError(401, "El token no es válido", "token_invalido", "La firma no coincide o el token está mal formado.")
    return Identidad(usuario=payload["sub"], rol=payload["rol"], exp=payload["exp"])


def requiere_profesor(quien: Identidad = Depends(usuario_actual)) -> Identidad:
    """La app puede esconder el botón; el servidor no confía en que lo haya hecho."""
    if quien.rol != "profesor":
        raise ApiError(
            403,
            "Solo un profesor puede publicar o borrar avisos",
            hint=f'Tu sesión es de "{quien.usuario}" con rol "{quien.rol}".',
        )
    return quien

@router.get("", response_model=list[AvisoOut])
def listar(desde: int = 0, _: Identidad = Depends(usuario_actual), db: Session = Depends(get_db)) -> list[Aviso]:
    """Todo el tablón exige sesión, incluso leerlo. `?desde=N`: solo los avisos con id mayor a N."""
    return list(db.scalars(select(Aviso).where(Aviso.id > desde).order_by(Aviso.id.desc()).limit(100)))


@router.post("", response_model=AvisoOut, status_code=201)
def crear(body: NuevoAviso, quien: Identidad = Depends(requiere_profesor), db: Session = Depends(get_db)) -> Aviso:
    aviso = Aviso(titulo=body.titulo, cuerpo=body.cuerpo, autor=quien.usuario)
    db.add(aviso)
    db.commit()
    db.refresh(aviso)
    return aviso

def emitir_tokens(db: Session, usuario: str, rol: str) -> Tokens:
    """Emite el par y registra la sesión. Se usa en register, login y refresh."""
    access = sign_jwt(usuario, rol, settings.access_ttl_s, settings.jwt_secret)
    refresh = random_token()
    db.add(Sesion(usuario=usuario, refresh_hash=sha256(refresh), expires_at=ahora() + settings.refresh_ttl_s))
    db.commit()
    return Tokens(accessToken=access, refreshToken=refresh, expiresIn=settings.access_ttl_s, usuario=usuario, rol=rol)


@router.post("/login", response_model=Tokens)
def login(body: Credenciales, db: Session = Depends(get_db)) -> Tokens:
    fila = db.get(Usuario, body.usuario)
    # Un solo mensaje para "no existe" y "contraseña mal": no se le dice a un
    # atacante cuál de las dos cosas acertó.
    if fila is None or not verify_password(body.password, fila.password_hash):
        raise ApiError(401, "Usuario o contraseña incorrectos", "credenciales", "Revisa los dos; el mensaje es el mismo a propósito.")
    return emitir_tokens(db, fila.usuario, fila.rol)


@router.post("/refresh", response_model=Tokens)
def refresh(body: RefreshBody, db: Session = Depends(get_db)) -> Tokens:
    sesion = db.scalar(select(Sesion).where(Sesion.refresh_hash == sha256(body.refreshToken)))
    if sesion is None or sesion.revoked_at is not None or sesion.expires_at <= ahora():
        raise ApiError(401, "El refresh token no sirve", "refresh_invalido", "Fue revocado, ya expiró, o nunca existió. Hay que volver a entrar.")

    # Rotación: el token usado se retira y sale uno nuevo. Si alguien lo copió,
    # solo uno de los dos —el ladrón o el dueño— logra usarlo.
    sesion.revoked_at = datetime.now(timezone.utc)
    usuario = db.get(Usuario, sesion.usuario)
    assert usuario is not None
    return emitir_tokens(db, usuario.usuario, usuario.rol)

    # El rol lo decide el servidor, nunca el cliente. "profesor" solo con el código
    # que se da en clase; cualquier otra cosa —incluido mandar rol: "profesor"— es alumno.
    rol = "profesor" if body.codigoProfesor and body.codigoProfesor == settings.codigo_profesor else "alumno"

"POST /api/auth/login HTTP/1.1" 200 OK
"GET /api/avisos HTTP/1.1" 200 OK
"GET /api/avisos HTTP/1.1" 401 Unauthorized
"POST /api/auth/refresh HTTP/1.1" 200 OK
"GET /api/avisos HTTP/1.1" 200 OK

    // Tu servidor, visto desde el emulador. En un teléfono físico sería la IP de tu máquina en la red local.
    private const val BASE_URL = "http://10.0.2.2:8000/api/"

<network-security-config>
    <domain-config cleartextTrafficPermitted="true">
        <domain includeSubdomains="false">10.0.2.2</domain>
    </domain-config>
</network-security-config>

<manifest xmlns:android="http://schemas.android.com/apk/res/android">
    <application android:networkSecurityConfig="@xml/network_security_config" />
</manifest>

docker compose --profile tunel up -d
docker compose logs tunel

private const val BASE_URL = "https://palabras-al-azar.trycloudflare.com/api/"

def requiere_rol(*roles: str):
    def dependencia(quien: Identidad = Depends(usuario_actual)) -> Identidad:
        if quien.rol not in roles:
            raise ApiError(403, "No tienes permiso para esto")
        return quien
    return dependencia

@router.delete("/{id}")
def borrar(id: int, quien: Identidad = Depends(requiere_rol("admin", "supervisor"))): ...