245 lines
9.7 KiB
Python
245 lines
9.7 KiB
Python
import hmac
|
|
import ipaddress
|
|
import re
|
|
import secrets
|
|
import socket
|
|
import threading
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from datetime import datetime
|
|
from urllib.parse import urlsplit
|
|
|
|
from flask import request, session
|
|
|
|
|
|
USERNAME_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{2,31}$")
|
|
COLOR_RE = re.compile(r"^#[0-9A-Fa-f]{6}$")
|
|
CATEGORIES = {"geral", "estudos", "trabalho", "pessoal", "provas", "entregas"}
|
|
PRIORITIES = {"baixa", "normal", "alta", "urgente"}
|
|
SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}
|
|
COMMON_PASSWORDS = {
|
|
"123456789012", "password1234", "qwerty123456", "senha12345678",
|
|
"admin12345678", "studycalendar",
|
|
}
|
|
|
|
|
|
class ValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
def reject_unknown_fields(data, allowed):
|
|
unknown = set(data) - set(allowed)
|
|
if unknown:
|
|
raise ValidationError(f"Campos desconhecidos: {', '.join(sorted(unknown))}.")
|
|
|
|
|
|
def require_json():
|
|
if not request.is_json:
|
|
raise ValidationError("Content-Type application/json é obrigatório.")
|
|
data = request.get_json(silent=True)
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("JSON deve ser um objeto.")
|
|
return data
|
|
|
|
|
|
def text(data, field, *, required=False, minimum=0, maximum=255, default=None):
|
|
value = data.get(field, default)
|
|
if value is None:
|
|
if required:
|
|
raise ValidationError(f"{field} é obrigatório.")
|
|
return None
|
|
if not isinstance(value, str):
|
|
raise ValidationError(f"{field} deve ser texto.")
|
|
value = value.strip()
|
|
if required and not value:
|
|
raise ValidationError(f"{field} é obrigatório.")
|
|
if len(value) < minimum or len(value) > maximum:
|
|
raise ValidationError(f"{field} deve ter entre {minimum} e {maximum} caracteres.")
|
|
return value
|
|
|
|
|
|
def integer(data, field, *, minimum, maximum, default=None):
|
|
value = data.get(field, default)
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise ValidationError(f"{field} deve ser um número inteiro.")
|
|
if not minimum <= value <= maximum:
|
|
raise ValidationError(f"{field} deve estar entre {minimum} e {maximum}.")
|
|
return value
|
|
|
|
|
|
def boolean(data, field, default=False):
|
|
value = data.get(field, default)
|
|
if not isinstance(value, bool):
|
|
raise ValidationError(f"{field} deve ser booleano.")
|
|
return value
|
|
|
|
|
|
def iso_datetime(value, field, *, required=False):
|
|
if value in (None, ""):
|
|
if required:
|
|
raise ValidationError(f"{field} é obrigatório.")
|
|
return None
|
|
if not isinstance(value, str) or len(value) > 40:
|
|
raise ValidationError(f"{field} deve ser uma data ISO 8601 válida.")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise ValidationError(f"{field} deve ser uma data ISO 8601 válida.") from exc
|
|
return parsed.isoformat()
|
|
|
|
|
|
def validate_credentials(data, registering=False):
|
|
allowed = {"username", "password", "display_name"} if registering else {"username", "password"}
|
|
reject_unknown_fields(data, allowed)
|
|
username = text(data, "username", required=True, minimum=3, maximum=32).lower()
|
|
if not USERNAME_RE.fullmatch(username):
|
|
raise ValidationError("username aceita apenas letras minúsculas, números, ponto, hífen e sublinhado.")
|
|
password = data.get("password")
|
|
if not isinstance(password, str) or len(password) > 128:
|
|
raise ValidationError("password inválida.")
|
|
if registering and (len(password) < 12 or password.isspace()):
|
|
raise ValidationError("password deve ter entre 12 e 128 caracteres.")
|
|
if registering and password.casefold() in COMMON_PASSWORDS:
|
|
raise ValidationError("password é muito comum.")
|
|
if not registering and not password:
|
|
raise ValidationError("password é obrigatória.")
|
|
display_name = None
|
|
if registering:
|
|
display_name = text(data, "display_name", required=True, minimum=1, maximum=80)
|
|
if any(ord(char) < 32 for char in display_name):
|
|
raise ValidationError("display_name contém caracteres inválidos.")
|
|
return username, password, display_name
|
|
|
|
|
|
def validate_event(data):
|
|
reject_unknown_fields(data, {
|
|
"title", "start_datetime", "end_datetime", "description", "location",
|
|
"all_day", "category", "priority",
|
|
})
|
|
title = text(data, "title", required=True, minimum=1, maximum=200)
|
|
start = iso_datetime(data.get("start_datetime"), "start_datetime", required=True)
|
|
end = iso_datetime(data.get("end_datetime"), "end_datetime")
|
|
start_value = datetime.fromisoformat(start)
|
|
end_value = datetime.fromisoformat(end) if end else None
|
|
if end_value and bool(start_value.tzinfo) != bool(end_value.tzinfo):
|
|
raise ValidationError("start_datetime e end_datetime devem usar o mesmo tipo de fuso horário.")
|
|
if end_value and end_value <= start_value:
|
|
raise ValidationError("end_datetime deve ser posterior a start_datetime.")
|
|
category = data.get("category", "geral")
|
|
priority = data.get("priority", "normal")
|
|
if category not in CATEGORIES:
|
|
raise ValidationError("category inválida.")
|
|
if priority not in PRIORITIES:
|
|
raise ValidationError("priority inválida.")
|
|
return {
|
|
"title": title,
|
|
"start_datetime": start,
|
|
"end_datetime": end,
|
|
"description": text(data, "description", maximum=4000, default=""),
|
|
"location": text(data, "location", maximum=300, default=""),
|
|
"all_day": boolean(data, "all_day"),
|
|
"category": category,
|
|
"priority": priority,
|
|
}
|
|
|
|
|
|
def validate_feed(data, https_only=True):
|
|
reject_unknown_fields(data, {"url", "name", "color", "sync_interval_minutes"})
|
|
url = text(data, "url", required=True, minimum=8, maximum=2048)
|
|
parsed = urlsplit(url)
|
|
allowed = {"https"} if https_only else {"http", "https"}
|
|
if parsed.scheme.lower() not in allowed or not parsed.hostname or parsed.username or parsed.password:
|
|
raise ValidationError("URL de feed inválida ou protocolo não permitido.")
|
|
if parsed.fragment:
|
|
raise ValidationError("URL de feed não pode conter fragmento.")
|
|
color = data.get("color", "#4A90D9")
|
|
if not isinstance(color, str) or not COLOR_RE.fullmatch(color):
|
|
raise ValidationError("color deve usar o formato #RRGGBB.")
|
|
return {
|
|
"url": url,
|
|
"name": text(data, "name", required=True, minimum=1, maximum=100),
|
|
"color": color.upper(),
|
|
"sync_interval_minutes": integer(
|
|
data, "sync_interval_minutes", minimum=5, maximum=10080, default=30
|
|
),
|
|
}
|
|
|
|
|
|
def validate_public_url(url, https_only=True):
|
|
if not isinstance(url, str) or len(url) > 2048 or any(ord(char) < 32 for char in url):
|
|
raise ValidationError("URL de feed inválida.")
|
|
parsed = urlsplit(url)
|
|
allowed = {"https"} if https_only else {"http", "https"}
|
|
if parsed.scheme.lower() not in allowed or not parsed.hostname:
|
|
raise ValidationError("Protocolo de feed não permitido.")
|
|
if parsed.username or parsed.password or parsed.fragment:
|
|
raise ValidationError("URL de feed inválida.")
|
|
try:
|
|
port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
|
|
except ValueError as exc:
|
|
raise ValidationError("Porta inválida.") from exc
|
|
expected_port = 443 if parsed.scheme.lower() == "https" else 80
|
|
if port != expected_port:
|
|
raise ValidationError("Porta de feed não permitida.")
|
|
try:
|
|
records = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM)
|
|
except socket.gaierror as exc:
|
|
raise ValidationError("Host do feed não pôde ser resolvido.") from exc
|
|
addresses = {
|
|
str(ipaddress.ip_address(record[4][0].split("%", 1)[0])) for record in records
|
|
}
|
|
if not addresses:
|
|
raise ValidationError("Host do feed não possui endereço válido.")
|
|
for address in addresses:
|
|
ip = ipaddress.ip_address(address)
|
|
if (not ip.is_global or ip.is_private or ip.is_reserved or ip.is_loopback
|
|
or ip.is_link_local or ip.is_multicast or ip.is_unspecified):
|
|
raise ValidationError("Host do feed aponta para uma rede não pública.")
|
|
return parsed, addresses
|
|
|
|
|
|
def csrf_token():
|
|
token = session.get("csrf_token")
|
|
if not token:
|
|
token = secrets.token_urlsafe(32)
|
|
session["csrf_token"] = token
|
|
return token
|
|
|
|
|
|
def valid_csrf(origin):
|
|
if request.method in SAFE_METHODS or not request.path.startswith("/api/"):
|
|
return True
|
|
supplied = request.headers.get("X-CSRF-Token", "")
|
|
expected = session.get("csrf_token", "")
|
|
if not supplied or not expected or not hmac.compare_digest(supplied, expected):
|
|
return False
|
|
source = request.headers.get("Origin") or request.headers.get("Referer")
|
|
if not source:
|
|
return True
|
|
source_url = urlsplit(source)
|
|
target_url = urlsplit(request.host_url)
|
|
return source_url.scheme == target_url.scheme and source_url.netloc == target_url.netloc
|
|
|
|
|
|
class RateLimiter:
|
|
def __init__(self):
|
|
self._hits = defaultdict(deque)
|
|
self._lock = threading.Lock()
|
|
|
|
def check(self, key, limit, window=60):
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
if key not in self._hits and len(self._hits) >= 10000:
|
|
self._hits = defaultdict(deque, {
|
|
existing_key: hits for existing_key, hits in self._hits.items()
|
|
if hits and hits[-1] > now - window
|
|
})
|
|
if len(self._hits) >= 10000:
|
|
return False, window
|
|
hits = self._hits[key]
|
|
while hits and hits[0] <= now - window:
|
|
hits.popleft()
|
|
if len(hits) >= limit:
|
|
return False, max(1, int(window - (now - hits[0])))
|
|
hits.append(now)
|
|
return True, 0
|