modelo-fastapi/main.py

"""Encurtador de links em FastAPI, pronto para publicar na Dado.

Rodar local:  uvicorn main:app --reload   (ou: python main.py)
"""

import html
import json
import os
import platform
import re
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import fastapi
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response

from links import FORMATO_CODIGO, ErroDeValidacao, buscar_link, criar_link, host_do_link, listar_links, registrar_clique

# Variável de ambiente de exemplo: na Dado, ela se configura no painel do app.
MENSAGEM = os.environ.get("MENSAGEM_BOAS_VINDAS") or "Defina a variável MENSAGEM_BOAS_VINDAS para trocar este texto."
INICIO = datetime.now(timezone.utc)
# Brasília não tem horário de verão desde 2019: UTC-3 o ano todo.
_inicio_brasilia = INICIO.astimezone(timezone(timedelta(hours=-3)))
INICIO_FORMATADO = _inicio_brasilia.strftime("%d/%m/%Y às %H:%M") + " (horário de Brasília)"

PASTA_PAGINAS = Path(__file__).parent / "paginas"

# A página é montada uma vez só: nada nela muda enquanto o processo vive.
PAGINA_INICIAL = (
    (PASTA_PAGINAS / "index.html")
    .read_text(encoding="utf-8")
    .replace("{{RUNTIME}}", html.escape(f"FastAPI {fastapi.__version__} · Python {platform.python_version()}"))
    .replace("{{INICIO_ISO}}", INICIO.isoformat(timespec="seconds").replace("+00:00", "Z"))
    .replace("{{INICIO}}", html.escape(INICIO_FORMATADO))
    .replace("{{MENSAGEM}}", html.escape(MENSAGEM))
)
PAGINA_NAO_ENCONTRADA = (PASTA_PAGINAS / "nao-encontrado.html").read_text(encoding="utf-8")
MODELO_AVISO = (PASTA_PAGINAS / "aviso.html").read_text(encoding="utf-8")
# A página de aviso não roda script nenhum: o navegador só aceita o CSS dela.
CSP_AVISO = "default-src 'none'; style-src 'unsafe-inline'; img-src data:; frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
MARCADORES_AVISO = re.compile(r"\{\{(HOST|URL|CODIGO)\}\}")


def pagina_aviso(link: dict) -> str:
    """Página de aviso antes de sair: mostra para onde o link leva. Todo valor é
    escapado (o link foi digitado por alguém) e entra numa passada só, para um
    valor nunca ser lido como outro marcador."""
    valores = {"HOST": host_do_link(link["url"]), "URL": link["url"], "CODIGO": link["codigo"]}
    return MARCADORES_AVISO.sub(lambda m: html.escape(valores[m.group(1)]), MODELO_AVISO)


# Freio simples contra abuso: no máximo 30 links novos a cada 10 minutos por IP.
LIMITE_POR_IP = 30
JANELA_SEGUNDOS = 10 * 60
_tentativas: dict[str, tuple[int, float]] = {}


def passou_do_limite(ip: str) -> bool:
    agora = time.monotonic()
    if len(_tentativas) > 10_000:
        _tentativas.clear()
    total, expira_em = _tentativas.get(ip, (0, 0.0))
    if expira_em <= agora:
        _tentativas[ip] = (1, agora + JANELA_SEGUNDOS)
        return False
    _tentativas[ip] = (total + 1, expira_em)
    return total + 1 > LIMITE_POR_IP


app = FastAPI(title="Encurtador de links", description="Encurta links e conta os cliques.")

CORS = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
}


@app.middleware("http")
async def cors_so_na_api(request: Request, call_next):
    """CORS aberto só na API: qualquer site (ou outro app seu) pode chamar /api/*."""
    if not request.url.path.startswith("/api/"):
        return await call_next(request)
    if request.method == "OPTIONS":
        resposta = Response(status_code=204)
    else:
        resposta = await call_next(request)
    resposta.headers.update(CORS)
    return resposta


def endereco_base(request: Request) -> str:
    # O app roda atrás de um proxy HTTPS, que avisa o protocolo e o host originais.
    protocolo = request.headers.get("x-forwarded-proto", request.url.scheme).split(",")[0].strip()
    host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc
    return f"{protocolo}://{host}"


def para_json(link: dict, request: Request) -> dict:
    return {
        "codigo": link["codigo"],
        "url": link["url"],
        "curto": f"{endereco_base(request)}/{link['codigo']}",
        "cliques": link["cliques"],
        "criadoEm": link["criadoEm"],
    }


CORPO_INVALIDO = 'Corpo inválido. Envie JSON no formato {"url": "https://..."}.'
# Um pedido de encurtar link cabe com folga em 10 KB; mais que isso é recusado
# sem ler o resto (o app é público: ninguém deve conseguir mandar megabytes).
TAMANHO_MAXIMO_CORPO = 10 * 1024


class CorpoInvalido(Exception):
    pass


async def ler_json_limitado(request: Request) -> dict:
    """Lê o corpo em pedaços e para assim que passar do limite (não confia só no Content-Length)."""
    declarado = request.headers.get("content-length")
    if declarado is not None and (not declarado.isdigit() or int(declarado) > TAMANHO_MAXIMO_CORPO):
        raise CorpoInvalido()
    corpo = bytearray()
    async for pedaco in request.stream():
        corpo += pedaco
        if len(corpo) > TAMANHO_MAXIMO_CORPO:
            raise CorpoInvalido()
    try:
        dados = json.loads(corpo)
    except ValueError:
        raise CorpoInvalido() from None
    if not isinstance(dados, dict):
        raise CorpoInvalido()
    return dados


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def pagina_inicial():
    return PAGINA_INICIAL


@app.get("/api/links", summary="Lista os últimos 20 links")
def listar(request: Request):
    return JSONResponse(
        {"links": [para_json(link, request) for link in listar_links()]},
        headers={"Cache-Control": "no-store"},
    )


@app.post(
    "/api/links",
    status_code=201,
    summary="Encurta um link",
    # O corpo é lido à mão (com limite de tamanho); isto só descreve o formato no /docs.
    openapi_extra={
        "requestBody": {
            "required": True,
            "content": {
                "application/json": {
                    "schema": {"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]},
                    "example": {"url": "https://exemplo.com.br/uma-pagina"},
                }
            },
        }
    },
)
async def criar(request: Request):
    ip = request.headers.get("cf-connecting-ip") or (request.client.host if request.client else "?")
    if passou_do_limite(ip):
        return JSONResponse(
            status_code=429,
            content={"erro": "Muitos links em pouco tempo. Espere alguns minutos e tente de novo."},
        )
    try:
        dados = await ler_json_limitado(request)
    except CorpoInvalido:
        return JSONResponse(status_code=400, content={"erro": CORPO_INVALIDO})
    try:
        link = criar_link(dados.get("url"))
    except ErroDeValidacao as erro:
        return JSONResponse(status_code=400, content={"erro": str(erro)})
    return JSONResponse(status_code=201, content=para_json(link, request))


# GET /<código> não redireciona direto: mostra antes uma página de aviso com o
# destino. Sem ela, qualquer pessoa usaria o seu endereço para disfarçar um
# link de golpe. A página não conta clique — quem só olha não foi ao destino.
# Estas rotas ficam por último: as rotas fixas acima têm prioridade.
@app.get("/{codigo}", include_in_schema=False)
def aviso(codigo: str):
    link = buscar_link(codigo) if FORMATO_CODIGO.match(codigo) else None
    if link is None:
        return HTMLResponse(PAGINA_NAO_ENCONTRADA, status_code=404)
    return HTMLResponse(
        pagina_aviso(link),
        headers={
            "Cache-Control": "no-store",
            "Referrer-Policy": "no-referrer",
            "X-Robots-Tag": "noindex, nofollow",
            "Content-Security-Policy": CSP_AVISO,
        },
    )


# O botão "Continuar" leva para GET /<código>/ir: aqui o servidor soma o
# clique e redireciona (302) para o link.
@app.get("/{codigo}/ir", include_in_schema=False)
def redirecionar(codigo: str):
    link = registrar_clique(codigo) if FORMATO_CODIGO.match(codigo) else None
    if link is None:
        return HTMLResponse(PAGINA_NAO_ENCONTRADA, status_code=404)
    return RedirectResponse(
        link["url"], status_code=302, headers={"Cache-Control": "no-store", "Referrer-Policy": "no-referrer"}
    )


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))