212 lines
8.3 KiB
Python
212 lines
8.3 KiB
Python
"""Sincronização segura de feeds iCalendar."""
|
|
|
|
import logging
|
|
import os
|
|
import ipaddress
|
|
import socket
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
from urllib.parse import urljoin
|
|
|
|
import icalendar
|
|
import pytz
|
|
import requests
|
|
|
|
import database
|
|
from security import ValidationError, validate_public_url
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
sp_tz = pytz.timezone('America/Sao_Paulo')
|
|
|
|
MAX_REDIRECTS = int(os.environ.get('ICS_MAX_REDIRECTS', '3'))
|
|
MAX_BYTES = int(os.environ.get('ICS_MAX_BYTES', str(5 * 1024 * 1024)))
|
|
MAX_EVENTS = int(os.environ.get('ICS_MAX_EVENTS', '5000'))
|
|
CONNECT_TIMEOUT = float(os.environ.get('ICS_CONNECT_TIMEOUT', '5'))
|
|
READ_TIMEOUT = float(os.environ.get('ICS_READ_TIMEOUT', '15'))
|
|
TOTAL_TIMEOUT = float(os.environ.get('ICS_TOTAL_TIMEOUT', '30'))
|
|
HTTPS_ONLY = os.environ.get('ICS_HTTPS_ONLY', 'true').lower() == 'true'
|
|
_DNS_LOCK = threading.Lock()
|
|
|
|
if not 0 <= MAX_REDIRECTS <= 5:
|
|
raise RuntimeError('ICS_MAX_REDIRECTS deve estar entre 0 e 5.')
|
|
if not 1024 <= MAX_BYTES <= 10 * 1024 * 1024:
|
|
raise RuntimeError('ICS_MAX_BYTES deve estar entre 1 KiB e 10 MiB.')
|
|
if not 1 <= MAX_EVENTS <= 10000:
|
|
raise RuntimeError('ICS_MAX_EVENTS deve estar entre 1 e 10000.')
|
|
if not 0 < CONNECT_TIMEOUT <= 30 or not 0 < READ_TIMEOUT <= 30 or not 1 <= TOTAL_TIMEOUT <= 120:
|
|
raise RuntimeError('Timeouts ICS fora dos limites permitidos.')
|
|
|
|
|
|
def parse_ics_datetime(dt_prop):
|
|
"""Converte data iCalendar para ISO no fuso de São Paulo."""
|
|
if not dt_prop:
|
|
return None, False
|
|
value = dt_prop.dt
|
|
if isinstance(value, datetime):
|
|
if value.tzinfo is None:
|
|
value = sp_tz.localize(value)
|
|
return value.astimezone(sp_tz).isoformat(), False
|
|
value = datetime(value.year, value.month, value.day)
|
|
return sp_tz.localize(value).isoformat(), True
|
|
|
|
|
|
def _peer_address(response):
|
|
"""Obtém o peer conectado antes de consumir o corpo, quando exposto pelo urllib3."""
|
|
raw = response.raw
|
|
candidates = [
|
|
getattr(getattr(raw, '_connection', None), 'sock', None),
|
|
getattr(getattr(raw, 'connection', None), 'sock', None),
|
|
getattr(getattr(getattr(getattr(raw, '_fp', None), 'fp', None), 'raw', None), '_sock', None),
|
|
]
|
|
for sock in candidates:
|
|
if sock:
|
|
return str(ipaddress.ip_address(sock.getpeername()[0].split('%', 1)[0]))
|
|
return None
|
|
|
|
|
|
@contextmanager
|
|
def _pinned_dns(hostname, address):
|
|
"""Fixa a resolução durante a conexão para eliminar a janela de DNS rebinding."""
|
|
original = socket.getaddrinfo
|
|
|
|
def canonical_host(value):
|
|
if isinstance(value, bytes):
|
|
value = value.decode('ascii')
|
|
return str(value).rstrip('.').encode('idna').decode('ascii').lower()
|
|
|
|
normalized_host = canonical_host(hostname)
|
|
ip = ipaddress.ip_address(address)
|
|
|
|
def resolve(host, port, family=0, type=0, proto=0, flags=0):
|
|
candidate = canonical_host(host)
|
|
if candidate != normalized_host:
|
|
return original(host, port, family, type, proto, flags)
|
|
selected_family = socket.AF_INET6 if ip.version == 6 else socket.AF_INET
|
|
sockaddr = (address, port, 0, 0) if ip.version == 6 else (address, port)
|
|
return [(selected_family, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', sockaddr)]
|
|
|
|
with _DNS_LOCK:
|
|
socket.getaddrinfo = resolve
|
|
try:
|
|
yield
|
|
finally:
|
|
socket.getaddrinfo = original
|
|
|
|
|
|
def fetch_ics(url, session_factory=requests.Session):
|
|
"""Baixa um ICS validando DNS, redirects, peer, TLS e tamanho."""
|
|
client = session_factory()
|
|
client.trust_env = False
|
|
current = url
|
|
deadline = time.monotonic() + TOTAL_TIMEOUT
|
|
try:
|
|
for redirect_count in range(MAX_REDIRECTS + 1):
|
|
parsed, approved_addresses = validate_public_url(current, HTTPS_ONLY)
|
|
approved_address = sorted(approved_addresses)[0]
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise ValidationError('Tempo total de download excedido.')
|
|
with _pinned_dns(parsed.hostname, approved_address):
|
|
response = client.get(
|
|
current,
|
|
headers={'User-Agent': 'StudyCalendar/2.0', 'Accept': 'text/calendar,*/*;q=0.1'},
|
|
timeout=(min(CONNECT_TIMEOUT, remaining), min(READ_TIMEOUT, remaining)),
|
|
allow_redirects=False,
|
|
stream=True,
|
|
verify=True,
|
|
)
|
|
try:
|
|
peer = _peer_address(response)
|
|
if not peer:
|
|
raise ValidationError('Não foi possível validar o endereço conectado.')
|
|
if peer not in approved_addresses:
|
|
raise ValidationError('Endereço conectado diverge do DNS validado.')
|
|
if response.is_redirect or response.is_permanent_redirect:
|
|
if redirect_count >= MAX_REDIRECTS:
|
|
raise ValidationError('Limite de redirecionamentos excedido.')
|
|
location = response.headers.get('Location')
|
|
if not location:
|
|
raise ValidationError('Redirecionamento sem destino.')
|
|
current = urljoin(current, location)
|
|
validate_public_url(current, HTTPS_ONLY)
|
|
continue
|
|
response.raise_for_status()
|
|
declared = response.headers.get('Content-Length')
|
|
if declared and int(declared) > MAX_BYTES:
|
|
raise ValidationError('Feed excede o tamanho permitido.')
|
|
body = bytearray()
|
|
for chunk in response.iter_content(64 * 1024):
|
|
if time.monotonic() > deadline:
|
|
raise ValidationError('Tempo total de download excedido.')
|
|
body.extend(chunk)
|
|
if len(body) > MAX_BYTES:
|
|
raise ValidationError('Feed excede o tamanho permitido.')
|
|
return bytes(body)
|
|
finally:
|
|
response.close()
|
|
finally:
|
|
client.close()
|
|
raise ValidationError('Não foi possível obter o feed.')
|
|
|
|
|
|
def _field(component, name, maximum, default=''):
|
|
value = component.get(name)
|
|
if value is None:
|
|
return default
|
|
return str(value).replace('\\n', '\n').replace('\\,', ',')[:maximum]
|
|
|
|
|
|
def parse_ics(content):
|
|
calendar = icalendar.Calendar.from_ical(content)
|
|
events = []
|
|
seen_uids = set()
|
|
for component in calendar.walk('VEVENT'):
|
|
if len(events) >= MAX_EVENTS:
|
|
raise ValidationError('Feed excede o número de eventos permitido.')
|
|
uid = _field(component, 'uid', 512).strip()
|
|
if not uid or uid in seen_uids:
|
|
continue
|
|
start_iso, all_day = parse_ics_datetime(component.get('dtstart'))
|
|
if not start_iso:
|
|
continue
|
|
end_iso, _ = parse_ics_datetime(component.get('dtend'))
|
|
if end_iso and datetime.fromisoformat(end_iso) <= datetime.fromisoformat(start_iso):
|
|
end_iso = None
|
|
seen_uids.add(uid)
|
|
events.append({
|
|
'uid': uid,
|
|
'title': _field(component, 'summary', 200, 'Sem Título') or 'Sem Título',
|
|
'description': _field(component, 'description', 4000),
|
|
'location': _field(component, 'location', 300),
|
|
'start_datetime': start_iso,
|
|
'end_datetime': end_iso,
|
|
'all_day': all_day,
|
|
})
|
|
return events
|
|
|
|
|
|
def sync_feed(feed_id):
|
|
"""Sincroniza um feed sem registrar sua URL ou tokens em logs."""
|
|
feed = database.get_feed(feed_id)
|
|
if not feed or not feed.get('user_id'):
|
|
logger.warning('Feed inexistente ou sem proprietário (id=%s).', feed_id)
|
|
return False
|
|
try:
|
|
events = parse_ics(fetch_ics(feed['url']))
|
|
success = database.replace_ics_events(feed_id, feed['user_id'], events)
|
|
if success:
|
|
logger.info('Feed sincronizado (id=%s, eventos=%s).', feed_id, len(events))
|
|
return success
|
|
except Exception as exc:
|
|
database.mark_feed_sync_attempt(feed_id)
|
|
logger.warning('Falha ao sincronizar feed (id=%s, tipo=%s).', feed_id, type(exc).__name__)
|
|
return False
|
|
|
|
|
|
def sync_all_feeds():
|
|
"""Sincroniza todos os feeds; deve ser chamado apenas pelo processo scheduler."""
|
|
for feed in database.get_all_feeds():
|
|
sync_feed(feed['id'])
|